| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "log/slog" |
| 6 | "strings" |
| 7 | |
| 8 | "reasonix/internal/agent" |
| 9 | "reasonix/internal/control" |
| 10 | "reasonix/internal/event" |
| 11 | "reasonix/internal/turnevent" |
| 12 | ) |
| 13 | |
| 14 | // TurnStartView is the synchronous admission receipt for the new Wails turn |
| 15 | // API. Events remain the streaming authority after admission. |
| 16 | type TurnStartView struct { |
| 17 | TurnID string `json:"turnId"` |
| 18 | Status event.TurnStatus `json:"status"` |
| 19 | Disposition control.SubmitDisposition `json:"disposition"` |
| 20 | RuntimeEpoch string `json:"runtimeEpoch,omitempty"` |
| 21 | SubmissionID string `json:"submissionId,omitempty"` |
| 22 | } |
| 23 | |
| 24 | // validatePromptIdentity fences a decision to the runtime and turn that |
| 25 | // rendered its card. It is intentionally shared by every decision surface; |
| 26 | // callers must still resolve the prompt on the same controller instance. |
| 27 | func (a *App) validatePromptIdentity(tabID, turnID, runtimeEpoch string) (control.SessionAPI, error) { |
| 28 | tab, ctrl := a.tabAndCtrlByID(tabID) |
| 29 | if ctrl == nil { |
| 30 | return nil, a.workspaceNotReadyErr(tab) |
| 31 | } |
| 32 | status := ctrl.RuntimeStatus() |
| 33 | if strings.TrimSpace(turnID) == "" || status.TurnID != strings.TrimSpace(turnID) { |
| 34 | return nil, fmt.Errorf("turn %q is not the active turn for tab %q", turnID, tabID) |
| 35 | } |
| 36 | if epoch := strings.TrimSpace(runtimeEpoch); epoch != "" && tab != nil && tab.sink != nil && tab.sink.runtimeEpochSnapshot() != epoch { |
| 37 | return nil, fmt.Errorf("runtime changed while resolving prompt for tab %q", tabID) |
| 38 | } |
| 39 | return ctrl, nil |
| 40 | } |
| 41 | |
| 42 | // stoppableCtrl resolves and captures the controller a Stop request targets. |
| 43 | // The request is session-scoped and idle cancellation is idempotent. |
| 44 | func (a *App) stoppableCtrl(tabID, turnID string) (control.SessionAPI, error) { |
| 45 | tab, ctrl := a.tabAndCtrlByID(tabID) |
| 46 | if ctrl == nil { |
| 47 | return nil, a.workspaceNotReadyErr(tab) |
| 48 | } |
| 49 | status := ctrl.RuntimeStatus() |
| 50 | if turnID = strings.TrimSpace(turnID); turnID != status.TurnID { |
| 51 | slog.Info("desktop: stop targeted a stale turn id; interrupting the active turn", "tab", tabID, "requested", turnID, "active", status.TurnID) |
| 52 | } |
| 53 | return ctrl, nil |
| 54 | } |
| 55 | |
| 56 | // CancelSessionForTab is the protocol-v2 Stop operation. It captures the tab's |
| 57 | // current controller exactly once, so later tab switches cannot retarget it. |
| 58 | func (a *App) CancelSessionForTab(tabID string) (control.CancelReceipt, error) { |
| 59 | ctrl, err := a.stoppableCtrl(tabID, "") |
| 60 | if err != nil { |
| 61 | return control.CancelReceipt{}, err |
| 62 | } |
| 63 | if concrete, ok := ctrl.(*control.Controller); ok { |
| 64 | return concrete.CancelSessionFrom("user_stop"), nil |
| 65 | } |
| 66 | if session, ok := ctrl.(interface{ CancelSession() control.CancelReceipt }); ok { |
| 67 | return session.CancelSession(), nil |
| 68 | } |
| 69 | status := ctrl.RuntimeStatus() |
| 70 | ctrl.Cancel() |
| 71 | return control.CancelReceipt{SessionRef: ctrl.SessionPath(), HeadID: agent.BranchID(ctrl.SessionPath()), Accepted: true, AlreadyIdle: !status.Running && !status.PendingPrompt}, nil |
| 72 | } |
| 73 | |
| 74 | // StartTurnForTab is the turn-id-aware replacement for SubmitToTab. Existing |
| 75 | // Submit entry points remain compatibility wrappers during the protocol cutover. |
| 76 | func (a *App) StartTurnForTab(tabID, input, submissionID string) (TurnStartView, error) { |
| 77 | if strings.TrimSpace(submissionID) == "" { |
| 78 | return TurnStartView{}, fmt.Errorf("submissionId is required") |
| 79 | } |
| 80 | if _, ctrl := a.tabAndCtrlByID(tabID); ctrl != nil { |
| 81 | if identified, ok := ctrl.(*control.Controller); ok { |
| 82 | receipt, found, err := identified.LookupSubmission(control.SubmissionRequest{ID: submissionID, Input: input, Display: input}) |
| 83 | if err != nil { |
| 84 | return TurnStartView{}, err |
| 85 | } |
| 86 | if found { |
| 87 | return TurnStartView{TurnID: receipt.TurnID, Status: event.TurnQueued, Disposition: control.SubmitTurnStarted, SubmissionID: submissionID}, nil |
| 88 | } |
| 89 | } |
| 90 | } |
| 91 | result, err := a.submitToTabResult(tabID, input, false, true, submissionID) |
| 92 | if err != nil { |
| 93 | return TurnStartView{}, err |
| 94 | } |
| 95 | if result.Disposition == control.SubmitManagementHandled { |
| 96 | return TurnStartView{Disposition: result.Disposition, SubmissionID: submissionID}, nil |
| 97 | } |
| 98 | tab, ctrl := a.tabAndCtrlByID(tabID) |
| 99 | if ctrl == nil { |
| 100 | return TurnStartView{}, a.workspaceNotReadyErr(tab) |
| 101 | } |
| 102 | turnID := "" |
| 103 | if admitted, ok := ctrl.(interface{ TurnIDForSubmission(string) string }); ok { |
| 104 | turnID = admitted.TurnIDForSubmission(submissionID) |
| 105 | } |
| 106 | if strings.TrimSpace(turnID) == "" { |
| 107 | return TurnStartView{}, fmt.Errorf("turn admission did not produce a durable turn id") |
| 108 | } |
| 109 | epoch := "" |
| 110 | if tab != nil && tab.sink != nil { |
| 111 | epoch = tab.sink.runtimeEpochSnapshot() |
| 112 | } |
| 113 | // This is an admission receipt, not a potentially raced runtime snapshot. |
| 114 | // Ordered events carry every later transition, including a provider that |
| 115 | // completed before the Wails Promise was delivered. |
| 116 | return TurnStartView{TurnID: turnID, Status: event.TurnQueued, Disposition: control.SubmitTurnStarted, RuntimeEpoch: epoch, SubmissionID: submissionID}, nil |
| 117 | } |
| 118 | |
| 119 | func (a *App) StartTurnForTabWithDrafts(tabID, input, submissionID string, draftIDs []string) (TurnStartView, error) { |
| 120 | if strings.TrimSpace(submissionID) == "" { |
| 121 | return TurnStartView{}, fmt.Errorf("submissionId is required") |
| 122 | } |
| 123 | req := control.SubmissionRequest{ID: submissionID, Input: input, Display: input, DraftIDs: append([]string(nil), draftIDs...)} |
| 124 | if _, ctrl := a.tabAndCtrlByID(tabID); ctrl != nil { |
| 125 | if identified, ok := ctrl.(*control.Controller); ok { |
| 126 | receipt, found, err := identified.LookupSubmission(req) |
| 127 | if err != nil { |
| 128 | return TurnStartView{}, err |
| 129 | } |
| 130 | if found { |
| 131 | return TurnStartView{TurnID: receipt.TurnID, Status: event.TurnQueued, Disposition: control.SubmitTurnStarted, SubmissionID: submissionID}, nil |
| 132 | } |
| 133 | } |
| 134 | } |
| 135 | admission, ctrl, err := a.beginTabTurn(tabID, true, submissionID) |
| 136 | if err != nil { |
| 137 | return TurnStartView{}, a.submissionAdmissionError(tabID, req, err) |
| 138 | } |
| 139 | defer admission.abort() |
| 140 | if err := a.ensureTabTopicIndexedForUserTurn(admission.tab); err != nil { |
| 141 | return TurnStartView{}, err |
| 142 | } |
| 143 | identified, ok := ctrl.(*control.Controller) |
| 144 | if !ok { |
| 145 | return TurnStartView{}, fmt.Errorf("unsupported: attachments-v1") |
| 146 | } |
| 147 | if _, err := identified.SubmitIdentified(req); err != nil { |
| 148 | return TurnStartView{}, inboxBridgeError(err) |
| 149 | } |
| 150 | admission.finish(ctrl) |
| 151 | turnID := identified.TurnIDForSubmission(submissionID) |
| 152 | if strings.TrimSpace(turnID) == "" { |
| 153 | return TurnStartView{}, fmt.Errorf("turn admission did not produce a durable turn id") |
| 154 | } |
| 155 | epoch := "" |
| 156 | if admission.tab != nil && admission.tab.sink != nil { |
| 157 | epoch = admission.tab.sink.runtimeEpochSnapshot() |
| 158 | } |
| 159 | return TurnStartView{TurnID: turnID, Status: event.TurnQueued, Disposition: control.SubmitTurnStarted, RuntimeEpoch: epoch, SubmissionID: submissionID}, nil |
| 160 | } |
| 161 | |
| 162 | // InterruptTurnForTab stops the tab's active work. Stop is a session-level |
| 163 | // request: a turn id from a stale button still interrupts whatever is running |
| 164 | // now, because an unstoppable turn is worse than stopping its replacement. |
| 165 | func (a *App) InterruptTurnForTab(tabID, turnID string) error { |
| 166 | ctrl, err := a.stoppableCtrl(tabID, turnID) |
| 167 | if err != nil { |
| 168 | return err |
| 169 | } |
| 170 | if session, ok := ctrl.(interface{ CancelSession() control.CancelReceipt }); ok { |
| 171 | session.CancelSession() |
| 172 | } else { |
| 173 | ctrl.Cancel() |
| 174 | } |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | // InterruptTurnWithInboxItemsForTab is the receipt-capable Stop used by the |
| 179 | // Composer when it also discards queued follow-ups. |
| 180 | func (a *App) InterruptTurnWithInboxItemsForTab(tabID, turnID string, itemIDs []string) (InboxCancelResultView, error) { |
| 181 | view := InboxCancelResultView{DiscardedItemIDs: []string{}} |
| 182 | ctrl, err := a.stoppableCtrl(tabID, turnID) |
| 183 | if err != nil { |
| 184 | return view, err |
| 185 | } |
| 186 | result, err := ctrl.CancelWithInboxItemsResult(itemIDs, "desktop") |
| 187 | if err != nil { |
| 188 | return view, inboxBridgeError(err) |
| 189 | } |
| 190 | view.DiscardedItemIDs = append(view.DiscardedItemIDs, result.DiscardedItemIDs...) |
| 191 | view.Warning = result.Warning |
| 192 | a.emitInboxChanged(tabID) |
| 193 | return view, nil |
| 194 | } |
| 195 | |
| 196 | // AnswerPromptForTab resolves an Ask only when it belongs to the exact active |
| 197 | // turn. Controller-side prompt ids remain independently idempotent. |
| 198 | func (a *App) AnswerPromptForTab(tabID, turnID, promptID string, answers []QuestionAnswer) error { |
| 199 | tab, ctrl := a.tabAndCtrlByID(tabID) |
| 200 | if ctrl == nil { |
| 201 | return a.workspaceNotReadyErr(tab) |
| 202 | } |
| 203 | status := ctrl.RuntimeStatus() |
| 204 | if strings.TrimSpace(turnID) == "" || status.TurnID != strings.TrimSpace(turnID) { |
| 205 | return fmt.Errorf("turn %q is not the active turn for tab %q", turnID, tabID) |
| 206 | } |
| 207 | // Resolve on the controller instance that passed the turn-id fence. Calling |
| 208 | // the legacy app wrapper here would re-resolve the tab and could deliver a |
| 209 | // late answer to a replacement controller after a runtime rebuild. |
| 210 | out := make([]event.AskAnswer, len(answers)) |
| 211 | for i, answer := range answers { |
| 212 | out[i] = event.AskAnswer{QuestionID: answer.QuestionID, Selected: answer.Selected} |
| 213 | } |
| 214 | if checked, ok := ctrl.(interface { |
| 215 | AnswerQuestionChecked(string, []event.AskAnswer) error |
| 216 | }); ok { |
| 217 | return checked.AnswerQuestionChecked(promptID, out) |
| 218 | } |
| 219 | ctrl.AnswerQuestion(promptID, out) |
| 220 | return nil |
| 221 | } |
| 222 | |
| 223 | type turnEventReader interface { |
| 224 | TurnEventReplay(after uint64) (turnevent.ReplayView, error) |
| 225 | } |
| 226 | |
| 227 | type TurnEventReplayView struct { |
| 228 | Events []turnevent.Envelope `json:"events"` |
| 229 | FloorSequence uint64 `json:"floorSeq"` |
| 230 | LatestSequence uint64 `json:"latestSeq"` |
| 231 | NextAfterSequence uint64 `json:"nextAfterSeq"` |
| 232 | HasMore bool `json:"hasMore"` |
| 233 | ResetRequired bool `json:"resetRequired"` |
| 234 | TranscriptRevision int64 `json:"transcriptRevision,omitempty"` |
| 235 | TranscriptDigest string `json:"transcriptDigest,omitempty"` |
| 236 | HeadID string `json:"headId,omitempty"` |
| 237 | LeafMessageID string `json:"leafMessageId,omitempty"` |
| 238 | RuntimeEpoch string `json:"runtimeEpoch,omitempty"` |
| 239 | } |
| 240 | |
| 241 | // TurnEventsForTab supplies the durable suffix used to repair sequence gaps or |
| 242 | // rebuild after a runtime epoch change. |
| 243 | func (a *App) TurnEventsForTab(tabID string, afterSeq uint64) (TurnEventReplayView, error) { |
| 244 | empty := TurnEventReplayView{Events: []turnevent.Envelope{}} |
| 245 | tab, ctrl := a.tabAndCtrlByID(tabID) |
| 246 | if ctrl == nil { |
| 247 | return empty, a.workspaceNotReadyErr(tab) |
| 248 | } |
| 249 | reader, ok := ctrl.(turnEventReader) |
| 250 | if !ok { |
| 251 | return empty, fmt.Errorf("turn event replay is unavailable") |
| 252 | } |
| 253 | // Re-check the controller under the app lock before sampling the epoch. |
| 254 | // This prevents pairing an old controller with a replacement runtime after |
| 255 | // a session rebind races tabAndCtrlByID. |
| 256 | epoch := "" |
| 257 | a.mu.RLock() |
| 258 | bound := tab != nil && a.tabs[tabID] == tab && tab.Ctrl == ctrl |
| 259 | if bound && tab.sink != nil { |
| 260 | epoch = tab.sink.runtimeEpochSnapshot() |
| 261 | } |
| 262 | a.mu.RUnlock() |
| 263 | if !bound { |
| 264 | return empty, fmt.Errorf("runtime changed while binding turn event replay") |
| 265 | } |
| 266 | replay, err := reader.TurnEventReplay(afterSeq) |
| 267 | if replay.Events == nil { |
| 268 | replay.Events = []turnevent.Envelope{} |
| 269 | } |
| 270 | return TurnEventReplayView{ |
| 271 | Events: replay.Events, FloorSequence: replay.FloorSequence, |
| 272 | LatestSequence: replay.LatestSequence, NextAfterSequence: replay.NextAfterSequence, |
| 273 | HasMore: replay.HasMore, ResetRequired: replay.ResetRequired, |
| 274 | TranscriptRevision: replay.TranscriptRevision, TranscriptDigest: replay.TranscriptDigest, |
| 275 | HeadID: replay.HeadID, LeafMessageID: replay.LeafMessageID, |
| 276 | RuntimeEpoch: epoch, |
| 277 | }, err |
| 278 | } |
| 279 | |
| 280 | var _ control.SessionAPI = (*control.Controller)(nil) |
| 281 |