返回 DeepSeek-Reasonix
remote_tab.go
根目录 / desktop / remote_tab.go
1 package main
2
3 import (
4 "bufio"
5 "bytes"
6 "context"
7 "encoding/json"
8 "fmt"
9 "io"
10 "log"
11 "net/http"
12 "strconv"
13 "strings"
14 "time"
15
16 "reasonix/internal/agent"
17 "reasonix/internal/control"
18 "reasonix/internal/event"
19 )
20
21 // The remote-tab bridge exchanges its pre-shared token for an HttpOnly
22 // session cookie over the loopback tunnel. Subsequent API and SSE requests
23 // use that cookie, keeping the token out of request lines and access logs.
24
25 const remoteTabStreamOpenStability = 50 * time.Millisecond
26
27 func remoteSessionTransitionBusy(err error) bool {
28 if err == nil {
29 return false
30 }
31 message := err.Error()
32 return strings.Contains(message, "while a turn is running") ||
33 strings.Contains(message, "while another session change is in progress") ||
34 strings.Contains(message, "session is finishing background teardown")
35 }
36
37 // attachRemoteTabServe starts the event pump before entering the session so
38 // /new or /resume frames are not missed. The caller's context owns the pump;
39 // handshake and session entry use a bounded child context.
40 func (a *App) attachRemoteTabServe(ctx context.Context, tabID, base, token, instanceID string, opts RemoteTabOpenOptions) (bool, error) {
41 callCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
42 defer cancel()
43
44 client, err := newServeHTTPClient(base)
45 if err != nil {
46 return false, err
47 }
48 capabilities, err := serveHandshakeCapabilities(callCtx, client, base, token)
49 if err != nil {
50 log.Printf("[remote] attachRemoteTabServe: handshake FAILED tab=%s base=%q err=%v", tabID, base, err)
51 return false, err
52 }
53 a.remoteTabMu.Lock()
54 tab := a.remoteTabs[tabID]
55 a.remoteTabMu.Unlock()
56 if tab == nil {
57 return false, fmt.Errorf("remote tab %q closed during bootstrap", tabID)
58 }
59 a.remoteTabMu.Lock()
60 if current := a.remoteTabs[tabID]; current == tab {
61 tab.capabilities = make(map[string]bool, len(capabilities))
62 for _, capability := range capabilities {
63 tab.capabilities[capability] = true
64 }
65 }
66 a.remoteTabMu.Unlock()
67 tab.sessionMu.Lock()
68 defer tab.sessionMu.Unlock()
69
70 // Resolve every non-new target before opening the all-session pump. A
71 // detached controller may replay pending prompts as soon as /resume starts;
72 // publishing its route first keeps those frames on the foreground surface.
73 focusOnly := !opts.NewSession && strings.TrimSpace(opts.SessionName) == "" && strings.TrimSpace(opts.SessionPath) == "" && strings.TrimSpace(opts.SessionID) == ""
74 var target serveSessionEntry
75 if !opts.NewSession {
76 target, err = preflightRemoteSessionTarget(callCtx, client, base, opts)
77 if err != nil {
78 return false, err
79 }
80 }
81
82 pumpCtx, gen, attachPathRevision, err := a.installRemoteTabAttachPump(ctx, tabID, tab, client, base, token, remoteSessionRoute(target), !opts.NewSession)
83 if err != nil {
84 return false, err
85 }
86
87 opened := make(chan error, 1)
88 a.goRemoteTabSafe("remoteTabPump", func() { a.remoteTabPump(pumpCtx, tabID, gen, opened) })
89 select {
90 case err = <-opened:
91 if err != nil {
92 a.retireRemoteTabGeneration(tabID, gen)
93 return false, err
94 }
95 case <-callCtx.Done():
96 a.retireRemoteTabGeneration(tabID, gen)
97 return false, callCtx.Err()
98 }
99 entered := true
100 if !focusOnly {
101 enterOpts := opts
102 if !opts.NewSession {
103 enterOpts.SessionName, enterOpts.SessionPath, enterOpts.SessionID, enterOpts.SessionTitle = target.Name, target.Path, target.SessionID, target.Title
104 }
105 target, err = enterRemoteSessionTarget(callCtx, client, base, enterOpts)
106 entered = err == nil && !target.TakenOver
107 }
108 if err == nil && target.TakenOver {
109 // The serve mounted this caller as a read-only spectator (another
110 // runtime owns the session writer). The tab stays attached to render
111 // the file/mirrored view and the take-back banner drives /reclaim.
112 log.Printf("[remote] attachRemoteTabServe: enterRemoteSession SPECTATOR (writer owned elsewhere) tab=%s session=%q", tabID, remoteSessionRoute(target))
113 }
114 if err != nil {
115 // A busy serve refuses session transitions with 409 but retains its
116 // usable current session. Keep the attach so pending work remains visible.
117 if remoteSessionTransitionBusy(err) {
118 log.Printf("[remote] attachRemoteTabServe: enterRemoteSession BUSY (attached to current session) tab=%s err=%v", tabID, err)
119 entered = false
120 target, _ = serveCurrentSession(callCtx, client, base)
121 } else if remoteSessionTakenOver(err) {
122 // A local runtime on the serve host owns the session. Pin the tab to
123 // the requested session as a read-only spectator: the mirror's frames
124 // route here, /history and /status serve the file-backed view, and
125 // the take-back banner drives /reclaim. No serve-frontend transition
126 // ran, but the tab must stay attached to render the mirror.
127 log.Printf("[remote] attachRemoteTabServe: enterRemoteSession TAKEN OVER (read-only spectator) tab=%s session=%q err=%v", tabID, target.Path, err)
128 entered = false
129 if remoteSessionRoute(target) == "" {
130 current, _ := serveCurrentSession(callCtx, client, base)
131 target = current
132 }
133 target.TakenOver = true
134 } else {
135 log.Printf("[remote] attachRemoteTabServe: enterRemoteSession FAILED tab=%s err=%v", tabID, err)
136 a.retireRemoteTabGeneration(tabID, gen)
137 return false, err
138 }
139 }
140 if !a.commitRemoteTabAttachResponse(tabID, tab, gen, attachPathRevision, target, opts.NewSession) {
141 entered = false
142 }
143 if !a.waitRemoteTabStreamStable(callCtx, tabID, gen) {
144 return false, fmt.Errorf("remote tab %q event stream closed during session attach", tabID)
145 }
146 a.remoteTabMu.Lock()
147 if current := a.remoteTabs[tabID]; current == tab && current.gen == gen {
148 current.session.instanceID = instanceID
149 }
150 a.remoteTabMu.Unlock()
151 // A 200 response is only the stream-open barrier. The stream can still die
152 // while /new or /resume is in flight; publish readiness only if its pump has
153 // not already moved this same generation into reconnecting/error.
154 if !a.markRemoteTabAttached(tabID, gen) {
155 return false, fmt.Errorf("remote tab %q event stream closed during session attach", tabID)
156 }
157 return entered, nil
158 }
159
160 // commitRemoteTabAttachResponse applies an attach response only while it still
161 // owns the foreground route. A session_changed frame for a newer adoption is
162 // authoritative even when the older /new or /resume response arrives later.
163 func (a *App) commitRemoteTabAttachResponse(tabID string, tab *remoteTab, gen, requestPathRevision uint64, target serveSessionEntry, reset bool) bool {
164 tab.routeEventMu.Lock()
165 defer tab.routeEventMu.Unlock()
166 a.remoteTabMu.Lock()
167 defer a.remoteTabMu.Unlock()
168 current := a.remoteTabs[tabID]
169 if current != tab || current.gen != gen {
170 return false
171 }
172 target.Path = strings.TrimSpace(target.Path)
173 route := remoteSessionRoute(target)
174 if current.routing.pathRevision != requestPathRevision && current.routing.currentPath != route {
175 return false
176 }
177 alreadyAdopted := current.routing.pathRevision != requestPathRevision
178 if !alreadyAdopted {
179 commitRemoteTabAttachRoute(current, route, reset)
180 }
181 current.session.takenOver = target.TakenOver
182 current.session.path = target.Path
183 current.session.sessionID = target.SessionID
184 if name := strings.TrimSpace(target.Name); name != "" {
185 current.session.name = name
186 }
187 if title := strings.TrimSpace(target.Title); title != "" {
188 current.topicTitle = title
189 }
190 if current.routing.running == nil {
191 current.routing.running = map[string]bool{}
192 }
193 return true
194 }
195
196 func (a *App) waitRemoteTabStreamStable(ctx context.Context, tabID string, gen uint64) bool {
197 timer := time.NewTimer(remoteTabStreamOpenStability)
198 defer timer.Stop()
199 select {
200 case <-ctx.Done():
201 return false
202 case <-timer.C:
203 return a.remoteTabGenerationCurrent(tabID, gen)
204 }
205 }
206
207 func (a *App) markRemoteTabAttached(tabID string, gen uint64) bool {
208 a.remoteTabMu.Lock()
209 defer a.remoteTabMu.Unlock()
210 tab := a.remoteTabs[tabID]
211 if tab == nil || tab.gen != gen || tab.state != "connecting" {
212 return false
213 }
214 tab.attachedGen = gen
215 return true
216 }
217
218 func (a *App) publishRemoteTabAttachedReady(tabID string, gen uint64) bool {
219 tab := a.lockRemoteTabPublication(tabID)
220 if tab == nil {
221 return false
222 }
223 a.remoteTabMu.Lock()
224 if a.remoteTabs[tabID] != tab || tab.gen != gen || tab.attachedGen != gen || tab.state != "connecting" {
225 a.remoteTabMu.Unlock()
226 tab.routeEventMu.Unlock()
227 return false
228 }
229 tab.attachedGen = 0
230 tab.state = "ready"
231 tab.err = ""
232 a.remoteTabMu.Unlock()
233 a.emitRemoteEvent(fmt.Sprintf("remote-tab:%s:state", tabID), RemoteTabStateView{State: "ready"})
234 tab.routeEventMu.Unlock()
235 a.applyPendingRemoteTabOpenSelection(tabID)
236 return true
237 }
238
239 func (a *App) remoteTabGenerationCurrent(tabID string, gen uint64) bool {
240 a.remoteTabMu.Lock()
241 defer a.remoteTabMu.Unlock()
242 tab := a.remoteTabs[tabID]
243 return tab != nil && tab.gen == gen
244 }
245
246 // remoteTabPump forwards Serve events for one tab generation. Cancellation,
247 // stream death, or a generation mismatch retires the pump.
248 func (a *App) remoteTabPump(ctx context.Context, tabID string, gen uint64, opened chan<- error) {
249 signalOpened := func(err error) {
250 if opened == nil {
251 return
252 }
253 select {
254 case opened <- err:
255 default:
256 }
257 opened = nil
258 }
259 a.remoteTabMu.Lock()
260 tab := a.remoteTabs[tabID]
261 var client *http.Client
262 var base string
263 if tab != nil && tab.gen == gen {
264 client, base = tab.client, tab.base
265 }
266 a.remoteTabMu.Unlock()
267 if client == nil || base == "" {
268 if opened != nil {
269 opened <- fmt.Errorf("remote tab %q event stream was retired before opening", tabID)
270 }
271 return
272 }
273
274 req, err := http.NewRequestWithContext(ctx, http.MethodGet, serveURL(base, "/events?all=1"), nil)
275 if err != nil {
276 signalOpened(err)
277 a.emitRemoteTabStateForGeneration(tabID, gen, "error", err.Error())
278 return
279 }
280 req.Header.Set("Accept", "text/event-stream")
281 resp, err := client.Do(req)
282 if err != nil {
283 // Schedule recovery before signalling the opener: the reattach
284 // retirement bumps the generation first, so the opener's own retire
285 // for this error becomes a no-op instead of racing the recovery.
286 if ctx.Err() == nil {
287 log.Printf("[remote] remoteTabPump: /events DO-FAILED tab=%s err=%v", tabID, err)
288 // A tunnel that just dropped the old stream often refuses the
289 // replacement too; parking in error would strand a healthy tab.
290 // Route through the reattach loop, which re-ensures the server
291 // and retries while the transport heals.
292 a.startRemoteTabReattach(tabID, gen)
293 }
294 signalOpened(err)
295 return
296 }
297 defer resp.Body.Close()
298 if resp.StatusCode != http.StatusOK {
299 err = fmt.Errorf("serve /events: status %d", resp.StatusCode)
300 if ctx.Err() == nil {
301 log.Printf("[remote] remoteTabPump: /events BAD-STATUS tab=%s status=%d", tabID, resp.StatusCode)
302 a.startRemoteTabReattach(tabID, gen)
303 }
304 signalOpened(err)
305 return
306 }
307 signalOpened(nil)
308 scanner := bufio.NewScanner(resp.Body)
309 scanner.Buffer(make([]byte, 64<<10), serveEventMaxBytes)
310 for scanner.Scan() {
311 line := scanner.Text()
312 line = strings.TrimRight(line, "\r\n")
313 if !strings.HasPrefix(line, "data:") {
314 continue // ": ping" keepalives and other SSE fields
315 }
316 frame := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
317 if frame == "" {
318 continue
319 }
320 if !a.remoteTabGenerationCurrent(tabID, gen) {
321 return
322 }
323 kind, framePath, current, reset := probeRemoteTabFrame(frame)
324 if kind == "runtime_state" {
325 a.acceptRemoteRuntimeFrame(tabID, gen, framePath, json.RawMessage(frame))
326 continue
327 }
328 // A takeover notice for the session this tab is viewing flips the
329 // spectator pin live: the entry-time probe only runs when the tab
330 // enters a session, so a mid-view takeover (or its reversal) would
331 // otherwise leave the banner and the composer locked to stale state.
332 if kind == "notice" && framePath != "" &&
333 (strings.Contains(frame, event.NoticeCodeSessionTakenOver) ||
334 strings.Contains(frame, event.NoticeCodeSessionReclaimed)) {
335 a.goRemoteTabSafe("remoteTabTakeoverNoticeProbe", func() {
336 a.probeSpectatorAfterNotice(tabID, gen, client, base, framePath)
337 })
338 }
339 if !a.routeRemoteTabWireFrame(tabID, gen, framePath, kind, current, reset) {
340 continue
341 }
342 if a.bufferRemoteTabResumeFrame(tabID, gen, framePath, kind, json.RawMessage(frame)) {
343 continue
344 }
345 a.publishRemoteTabFrame(tabID, gen, framePath, kind, json.RawMessage(frame))
346 }
347 if err := scanner.Err(); err != nil {
348 log.Printf("[remote] remoteTabPump: READ-EXIT tab=%s gen=%d err=%v ctxErr=%v", tabID, gen, err, ctx.Err())
349 }
350 // Only the current generation reacts to an unexpected stream death.
351 // Reattach now; the host status hook also retries on connection recovery.
352 if ctx.Err() == nil {
353 a.startRemoteTabReattach(tabID, gen)
354 }
355 }
356
357 func (a *App) completeRemoteTabTurn(tabID string, gen uint64) {
358 a.remoteTabMu.Lock()
359 tab := a.remoteTabs[tabID]
360 if tab == nil || tab.gen != gen {
361 a.remoteTabMu.Unlock()
362 return
363 }
364 tab.runtime.revision++
365 tab.pendingEvents = nil
366 tab.runtime.running = false
367 tab.runtime.turnStartedAt = 0
368 tab.runtime.pendingPrompt = false
369 tab.runtime.cancelRequested = false
370 tab.runtime.cancellable = tab.runtime.backgroundJobs > 0
371 // A completed turn makes the fresh session non-blank even when the
372 // best-effort /sessions title lookup fails. New Topic must never reuse
373 // a conversation that already has a completed turn.
374 tab.session.reset = false
375 meta := remoteTabMetaLocked(tab)
376 a.remoteTabMu.Unlock()
377 a.emitRemoteEvent("remote-tab:updated", meta)
378 }
379
380 func (a *App) recordRemoteTabTurnStarted(tabID string, gen uint64, frame json.RawMessage) {
381 var payload struct {
382 TurnStartedAt int64 `json:"turnStartedAt"`
383 }
384 _ = json.Unmarshal(frame, &payload)
385 if payload.TurnStartedAt <= 0 {
386 payload.TurnStartedAt = time.Now().UnixMilli()
387 }
388 a.remoteTabMu.Lock()
389 tab := a.remoteTabs[tabID]
390 if tab == nil || tab.gen != gen {
391 a.remoteTabMu.Unlock()
392 return
393 }
394 tab.runtime.revision++
395 tab.runtime.running = true
396 tab.runtime.turnStartedAt = payload.TurnStartedAt
397 tab.runtime.pendingPrompt = false
398 tab.runtime.cancelRequested = false
399 tab.runtime.cancellable = true
400 meta := remoteTabMetaLocked(tab)
401 a.remoteTabMu.Unlock()
402 a.emitRemoteEvent("remote-tab:updated", meta)
403 }
404
405 func (a *App) cacheRemotePendingEvent(tabID string, gen uint64, kind string, frame json.RawMessage) {
406 key := remotePendingEventKey(kind, frame)
407 a.remoteTabMu.Lock()
408 tab := a.remoteTabs[tabID]
409 if tab == nil || tab.gen != gen {
410 a.remoteTabMu.Unlock()
411 return
412 }
413 tab.runtime.revision++
414 if tab.pendingEvents == nil {
415 tab.pendingEvents = make(map[string]json.RawMessage)
416 }
417 tab.pendingEvents[key] = append(json.RawMessage(nil), frame...)
418 tab.runtime.pendingPrompt = true
419 tab.runtime.cancellable = true
420 meta := remoteTabMetaLocked(tab)
421 a.remoteTabMu.Unlock()
422 a.emitRemoteEvent("remote-tab:updated", meta)
423 }
424
425 func (a *App) clearRemotePendingEvent(tabID, kind, callID string) {
426 a.remoteTabMu.Lock()
427 var meta TabMeta
428 changed := false
429 if tab := a.remoteTabs[tabID]; tab != nil {
430 tab.runtime.revision++
431 delete(tab.pendingEvents, kind+":"+strings.TrimSpace(callID))
432 pending := len(tab.pendingEvents) > 0
433 changed = tab.runtime.pendingPrompt != pending
434 tab.runtime.pendingPrompt = pending
435 meta = remoteTabMetaLocked(tab)
436 }
437 a.remoteTabMu.Unlock()
438 if changed {
439 a.emitRemoteEvent("remote-tab:updated", meta)
440 }
441 }
442
443 // serveGet fetches a JSON member of the tab snapshot, returning the raw
444 // payload for verbatim passthrough.
445 func serveGet(ctx context.Context, client *http.Client, url string) (json.RawMessage, error) {
446 req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
447 if err != nil {
448 return nil, err
449 }
450 resp, err := client.Do(req)
451 if err != nil {
452 return nil, err
453 }
454 defer resp.Body.Close()
455 data, err := io.ReadAll(io.LimitReader(resp.Body, serveSnapshotMaxBytes+1))
456 if err != nil {
457 return nil, err
458 }
459 if len(data) > serveSnapshotMaxBytes {
460 return nil, fmt.Errorf("%s: response exceeds %d bytes", url, serveSnapshotMaxBytes)
461 }
462 if resp.StatusCode != http.StatusOK {
463 return nil, fmt.Errorf("%s: status %d", url, resp.StatusCode)
464 }
465 return json.RawMessage(data), nil
466 }
467
468 // commandContext bounds one proxied command. Boot context when available;
469 // the timeout keeps a wedged tunnel from hanging the binding call.
470 func commandContext(a *App) (context.Context, context.CancelFunc) {
471 ctx := a.bootContext()
472 if ctx == nil {
473 ctx = context.Background()
474 }
475 return context.WithTimeout(ctx, 15*time.Second)
476 }
477
478 // remoteTabCommandClient resolves a tabID to its live serve client. A tab
479 // that has not finished bootstrap, is reconnecting, or has failed is an
480 // error, not a silent no-op.
481 func (a *App) remoteTabCommandClient(tabID string) (*http.Client, string, error) {
482 client, base, _, err := a.remoteTabCommandTarget(tabID)
483 return client, base, err
484 }
485
486 func (a *App) remoteTabCommandTarget(tabID string) (*http.Client, string, string, error) {
487 a.remoteTabMu.Lock()
488 tab := a.remoteTabs[tabID]
489 var client *http.Client
490 var base, expectedPath string
491 switching := tab != nil && tab.routing.rehydratingPath != ""
492 usable := tab != nil && tab.client != nil && tab.state == "ready" && !switching
493 if usable {
494 client, base = tab.client, tab.base
495 expectedPath = tab.routing.currentPath
496 }
497 a.remoteTabMu.Unlock()
498 if !usable {
499 if switching {
500 return nil, "", "", fmt.Errorf("remote tab %q is switching sessions; wait for it to become ready", tabID)
501 }
502 return nil, "", "", fmt.Errorf("remote tab %q is not connected", tabID)
503 }
504 return client, base, expectedPath, nil
505 }
506
507 // remoteTabAdmissionCurrent reports whether the tab still runs the generation
508 // a run-admission decision (model-settings revision or legacy skip) was made
509 // for. Generation 0 marks an ungated decision; any other replaced generation
510 // must be re-admitted so a reconnect's newer Serve never receives an unfenced
511 // request that was approved against the retired connection.
512 func (a *App) remoteTabAdmissionCurrent(tabID string, generation uint64) bool {
513 if generation == 0 {
514 return true
515 }
516 a.remoteTabMu.Lock()
517 defer a.remoteTabMu.Unlock()
518 tab := a.remoteTabs[tabID]
519 return tab != nil && tab.gen == generation
520 }
521
522 func (a *App) isRemoteTab(tabID string) bool {
523 if strings.TrimSpace(tabID) == "" {
524 return false
525 }
526 a.remoteTabMu.Lock()
527 _, ok := a.remoteTabs[tabID]
528 a.remoteTabMu.Unlock()
529 return ok
530 }
531
532 // remoteTabRefFor returns the host+workspace ref when tabID belongs to a
533 // remote tab; view builders use it to mark remote-shaped metas.
534 func (a *App) remoteTabRefFor(tabID string) (RemoteTabRef, bool) {
535 a.remoteTabMu.Lock()
536 defer a.remoteTabMu.Unlock()
537 if tab := a.remoteTabs[tabID]; tab != nil {
538 return tab.ref, true
539 }
540 return RemoteTabRef{}, false
541 }
542
543 func (a *App) remoteTabCurrentModel(tabID string) (string, bool) {
544 if !a.isRemoteTab(tabID) {
545 return "", false
546 }
547 a.remoteTabMu.Lock()
548 tab := a.remoteTabs[tabID]
549 cur := ""
550 if tab != nil {
551 cur = tab.model
552 }
553 a.remoteTabMu.Unlock()
554 return cur, true
555 }
556
557 // ReclaimRemoteTabSession takes a mirrored session back from the local
558 // runtime that took it over. Serve long-polls until the local writer yields,
559 // so this call can outlast a normal command timeout.
560 func (a *App) ReclaimRemoteTabSession(tabID string) error {
561 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
562 return err
563 }
564 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
565 if err != nil {
566 return err
567 }
568 if strings.TrimSpace(expectedPath) == "" {
569 return fmt.Errorf("remote tab %q has no active session", tabID)
570 }
571 observed, err := a.observeRemoteTabForReclaim(tabID, client)
572 if err != nil {
573 return err
574 }
575 observedTab, observedGen := observed.tab, observed.gen
576 stillCurrent := func(tab *remoteTab) bool {
577 return tab != nil && tab == observedTab && tab.client == client && tab.gen == observed.gen &&
578 tab.runtime.revision == observed.runtimeRevision && tab.selectionRevision == observed.selectionRevision &&
579 agent.CanonicalSessionPath(tab.routing.currentPath) == agent.CanonicalSessionPath(expectedPath)
580 }
581 reconcileOwnership := func() { a.reconcileRemoteTabReclaimOwnership(tabID, client, base, expectedPath, stillCurrent) }
582 // Short timeout: the serve caps un-mirrored reclaims at 10s and mirrored
583 // ones use the writer's cooperative heartbeat (seconds, not minutes). A
584 // long client-side timeout only hangs the UI button.
585 ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
586 defer cancel()
587 body, _ := json.Marshal(map[string]any{
588 "sessionPath": expectedPath,
589 "mode": "wait",
590 "timeoutMs": 15000,
591 })
592 resp, err := serveDo(ctx, client, http.MethodPost, serveURL(base, "/reclaim"), body)
593 if err != nil {
594 reconcileOwnership()
595 return fmt.Errorf("reclaim session: %w", err)
596 }
597 defer resp.Body.Close()
598 respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
599 if resp.StatusCode != http.StatusNoContent {
600 errMsg := strings.TrimSpace(string(respBody))
601 // A failed reclaim is not proof that ownership changed — generation
602 // conflicts and transient 5xx included. Keep the spectator pin until a
603 // fenced probe proves this exact binding is no longer locally owned.
604 reconcileOwnership()
605 return fmt.Errorf("reclaim session: %s", errMsg)
606 }
607 // Reclaim succeeded: Serve now owns the session again. Clear the spectator
608 // pin immediately so the composer un-locks without waiting for the next
609 // status poll to observe takenOver=false.
610 observedTab.routeEventMu.Lock()
611 defer observedTab.routeEventMu.Unlock()
612 a.remoteTabMu.Lock()
613 if tab := a.remoteTabs[tabID]; stillCurrent(tab) {
614 tab.session.takenOver = false
615 // Fence status payloads reserved before this reclaim: they may still
616 // be in flight and carry the pre-reclaim takenOver=true, which would
617 // re-pin the spectator banner the moment ownership returned.
618 tab.ownership.reclaimRevision = tab.runtime.revision + 1
619 deferBarrier := tab.runtime.running || tab.runtime.pendingPrompt
620 tab.ownership.readyBarrierPending = deferBarrier
621 meta := remoteTabMetaLocked(tab)
622 a.remoteTabMu.Unlock()
623 a.emitRemoteEvent("remote-tab:updated", meta)
624 // The spectator era froze the projection, so publish the ready barrier
625 // to re-hydrate the view and accept the re-owned writer's frames. Defer
626 // it mid-turn: the barrier bumps the frontend connection generation.
627 if !deferBarrier {
628 a.transitionRemoteTabStateLocked(tab, observedGen, "ready", "ready", "")
629 }
630 } else {
631 a.remoteTabMu.Unlock()
632 }
633 a.goRemoteTabSafe("reclaimStatusRefresh", func() { _, _ = a.RemoteTabStatus(tabID) })
634 return nil
635 }
636
637 func (a *App) SubmitRemoteTab(tabID, text string) error {
638 return a.SubmitRemoteTabWithSubmission(tabID, text, "")
639 }
640
641 func (a *App) SubmitRemoteTabWithSubmission(tabID, text, submissionID string) error {
642 // Report the connection state before capability negotiation so a tab that
643 // has not finished bootstrap is never misdiagnosed as a legacy Serve.
644 if _, _, _, err := a.remoteTabCommandTarget(tabID); err != nil {
645 return err
646 }
647 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
648 return err
649 }
650 if err := a.requireRemotePermissionPresets(tabID); err != nil {
651 return err
652 }
653 for {
654 revision, admittedGen, err := a.ensureRemoteModelSettings(tabID)
655 if err != nil {
656 return err
657 }
658 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
659 if err != nil {
660 return err
661 }
662 if !a.remoteTabAdmissionCurrent(tabID, admittedGen) {
663 continue
664 }
665 ctx, cancel := commandContext(a)
666 input := map[string]string{"input": text}
667 if submissionID != "" {
668 input["submissionId"] = submissionID
669 }
670 body, _ := json.Marshal(input)
671 err = servePostForSession(ctx, client, serveURL(base, "/submit"), body, expectedPath, revision)
672 cancel()
673 return err
674 }
675 }
676
677 func (a *App) CancelRemoteTab(tabID string) error {
678 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
679 return err
680 }
681 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
682 if err != nil {
683 return err
684 }
685 ctx, cancel := commandContext(a)
686 defer cancel()
687 return servePostForSession(ctx, client, serveURL(base, "/cancel"), nil, expectedPath)
688 }
689
690 // ApproveRemoteTab answers a tool-approval request. Only one-shot and scoped
691 // session grants are supported; durable approval rules were intentionally
692 // removed from the permission model.
693 func (a *App) ApproveRemoteTab(tabID, callID, decision string) error {
694 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
695 return err
696 }
697 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
698 if err != nil {
699 return err
700 }
701 ctx, cancel := commandContext(a)
702 defer cancel()
703 decision = strings.ToLower(strings.TrimSpace(decision))
704 allow, session := false, false
705 switch decision {
706 case "allow", "once":
707 allow = true
708 case "session":
709 allow, session = true, true
710 case "persist", "persistent", "project":
711 return fmt.Errorf("permanent remote approval is no longer supported")
712 case "deny":
713 default:
714 return fmt.Errorf("invalid remote approval decision %q", decision)
715 }
716 body, _ := json.Marshal(map[string]any{"id": callID, "allow": allow, "session": session, "persist": false})
717 if err := servePostForSession(ctx, client, serveURL(base, "/approve"), body, expectedPath); err != nil {
718 return err
719 }
720 a.clearRemotePendingEvent(tabID, "approval_request", callID)
721 return nil
722 }
723
724 // ResolveRemoteTabPlanDecision preserves the three distinct exit_plan_mode
725 // outcomes that the generic approval boolean cannot represent. Revision text
726 // travels in the same Serve request so the controller can durably stage it
727 // before resolving the approval; a tunnel failure can no longer split the
728 // decision from the requested revision.
729 func (a *App) ResolveRemoteTabPlanDecision(tabID, callID, action, feedback string) error {
730 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
731 return err
732 }
733 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
734 if err != nil {
735 return err
736 }
737 action = strings.ToLower(strings.TrimSpace(action))
738 switch action {
739 case "start_execution", "revise_plan", "exit_plan":
740 default:
741 return fmt.Errorf("invalid remote plan decision %q", action)
742 }
743 ctx, cancel := commandContext(a)
744 defer cancel()
745 body, _ := json.Marshal(map[string]string{"id": callID, "action": action, "feedback": strings.TrimSpace(feedback)})
746 if err := servePostForSession(ctx, client, serveURL(base, "/plan-decision"), body, expectedPath); err != nil {
747 return err
748 }
749 a.clearRemotePendingEvent(tabID, "approval_request", callID)
750 return nil
751 }
752
753 type RemoteAskAnswer struct {
754 QuestionID string `json:"QuestionID"`
755 Selected []string `json:"Selected"`
756 }
757
758 // AnswerRemoteTab preserves the batch ask id at the top level and sends every
759 // question's own id/selections in the Serve AskAnswer wire shape.
760 func (a *App) AnswerRemoteTab(tabID, callID string, answers []RemoteAskAnswer) error {
761 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
762 return err
763 }
764 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
765 if err != nil {
766 return err
767 }
768 ctx, cancel := commandContext(a)
769 defer cancel()
770 body, _ := json.Marshal(map[string]any{
771 "id": callID,
772 "answers": answers,
773 })
774 if err := servePostForSession(ctx, client, serveURL(base, "/answer"), body, expectedPath); err != nil {
775 return err
776 }
777 a.clearRemotePendingEvent(tabID, "ask_request", callID)
778 return nil
779 }
780
781 func (a *App) SubmitRemoteTabExtensionForm(tabID, pluginID, surfaceID string, values map[string]any) error {
782 if err := a.remoteTabPost(tabID, "/extension-form", map[string]any{
783 "pluginId": pluginID, "surfaceId": surfaceID, "values": values,
784 }); err != nil {
785 return err
786 }
787 a.clearRemotePendingExtensionForm(tabID, pluginID, surfaceID)
788 return nil
789 }
790
791 // RewindRemoteTab rewinds to a checkpoint. Serve identifies checkpoints by
792 // TURN index and takes {turn, scope}; the checkpointID string is that turn.
793 func (a *App) RewindRemoteTab(tabID, checkpointID, scope string) error {
794 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
795 return err
796 }
797 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
798 if err != nil {
799 return err
800 }
801 ctx, cancel := commandContext(a)
802 defer cancel()
803 turn, convErr := strconv.Atoi(strings.TrimSpace(checkpointID))
804 if convErr != nil {
805 return fmt.Errorf("invalid checkpoint id %q: want the turn index", checkpointID)
806 }
807 scope = strings.TrimSpace(scope)
808 switch scope {
809 case "code", "conversation", "both":
810 default:
811 return fmt.Errorf("invalid rewind scope %q", scope)
812 }
813 body, _ := json.Marshal(map[string]any{"turn": turn, "scope": scope})
814 return servePostForSession(ctx, client, serveURL(base, "/rewind"), body, expectedPath)
815 }
816
817 func (a *App) SetRemoteTabToolApprovalMode(tabID, mode string) error {
818 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
819 return err
820 }
821 if err := a.requireRemotePermissionPresets(tabID); err != nil {
822 return err
823 }
824 ctx, cancel := commandContext(a)
825 defer cancel()
826 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
827 if err != nil {
828 return err
829 }
830 snapshot, err := remotePermissionSnapshot(ctx, client, base, expectedPath)
831 if err != nil {
832 return err
833 }
834 _, err = setRemotePermissionPresetAt(ctx, client, base, expectedPath, mode, snapshot.Revision)
835 return err
836 }
837
838 func setRemotePermissionPresetAt(ctx context.Context, client *http.Client, base, expectedPath, mode string, revision uint64) (control.PermissionSnapshot, error) {
839 var snapshot control.PermissionSnapshot
840 body, _ := json.Marshal(map[string]any{"preset": mode, "expectedRevision": revision})
841 resp, err := serveDoForSession(ctx, client, http.MethodPost, serveURL(base, "/permission/preset"), body, expectedPath)
842 if err != nil {
843 return snapshot, err
844 }
845 defer resp.Body.Close()
846 if resp.StatusCode < 200 || resp.StatusCode >= 300 {
847 data, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<10))
848 return snapshot, fmt.Errorf("set remote permission preset: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data)))
849 }
850 var envelope struct {
851 Snapshot control.PermissionSnapshot `json:"snapshot"`
852 }
853 if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&envelope); err != nil {
854 return snapshot, fmt.Errorf("decode remote permission update: %w", err)
855 }
856 return envelope.Snapshot, nil
857 }
858
859 func (a *App) SetRemoteTabComposerProfile(tabID, collaborationMode, toolApprovalMode, goal string) ([]string, error) {
860 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
861 return nil, err
862 }
863 if err := a.requireRemotePermissionPresets(tabID); err != nil {
864 return nil, err
865 }
866 if strings.EqualFold(strings.TrimSpace(collaborationMode), "goal") || strings.TrimSpace(goal) != "" {
867 if err := a.requireRemoteGoalLifecycle(tabID); err != nil {
868 return nil, err
869 }
870 }
871 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
872 if err != nil {
873 return nil, err
874 }
875 ctx, cancel := commandContext(a)
876 defer cancel()
877 snapshot, err := remotePermissionSnapshot(ctx, client, base, expectedPath)
878 if err != nil {
879 return nil, err
880 }
881 body, _ := json.Marshal(map[string]any{
882 "collaborationMode": collaborationMode,
883 "toolApprovalMode": toolApprovalMode,
884 "goal": goal,
885 "expectedPermissionRevision": snapshot.Revision,
886 })
887 resp, err := serveDoForSession(ctx, client, http.MethodPost, serveURL(base, "/composer-profile"), body, expectedPath)
888 if err != nil {
889 return nil, err
890 }
891 defer resp.Body.Close()
892 data, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<10))
893 if resp.StatusCode < 200 || resp.StatusCode >= 300 {
894 if message := strings.TrimSpace(string(data)); message != "" {
895 return nil, fmt.Errorf("%s: status %d: %s", serveURL(base, "/composer-profile"), resp.StatusCode, message)
896 }
897 return nil, fmt.Errorf("%s: status %d", serveURL(base, "/composer-profile"), resp.StatusCode)
898 }
899 if len(bytes.TrimSpace(data)) == 0 {
900 return []string{}, nil
901 }
902 var result struct {
903 DrainedApprovalIDs []string `json:"drainedApprovalIDs"`
904 }
905 if err := json.Unmarshal(data, &result); err != nil {
906 return nil, fmt.Errorf("decode remote composer profile response: %w", err)
907 }
908 return result.DrainedApprovalIDs, nil
909 }
910
911 func remotePermissionSnapshot(ctx context.Context, client *http.Client, base, expectedPath string) (control.PermissionSnapshot, error) {
912 var snapshot control.PermissionSnapshot
913 resp, err := serveDoForSession(ctx, client, http.MethodGet, serveURL(base, "/permission"), nil, expectedPath)
914 if err != nil {
915 return snapshot, err
916 }
917 defer resp.Body.Close()
918 if resp.StatusCode < 200 || resp.StatusCode >= 300 {
919 data, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<10))
920 return snapshot, fmt.Errorf("query remote permission snapshot: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data)))
921 }
922 if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&snapshot); err != nil {
923 return snapshot, fmt.Errorf("decode remote permission snapshot: %w", err)
924 }
925 return snapshot, nil
926 }
927
928 func revokeRemotePermissionGrantAt(ctx context.Context, client *http.Client, base, expectedPath, scope, target string, revision uint64) (control.PermissionSnapshot, error) {
929 var snapshot control.PermissionSnapshot
930 body, _ := json.Marshal(map[string]any{"scope": scope, "target": target, "expectedRevision": revision})
931 resp, err := serveDoForSession(ctx, client, http.MethodPost, serveURL(base, "/permission/grants/revoke"), body, expectedPath)
932 if err != nil {
933 return snapshot, err
934 }
935 defer resp.Body.Close()
936 if resp.StatusCode < 200 || resp.StatusCode >= 300 {
937 data, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<10))
938 return snapshot, fmt.Errorf("revoke remote permission grant: status %d: %s", resp.StatusCode, strings.TrimSpace(string(data)))
939 }
940 if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&snapshot); err != nil {
941 return snapshot, fmt.Errorf("decode remote permission revocation: %w", err)
942 }
943 return snapshot, nil
944 }
945
946 func (a *App) requireRemotePermissionPresets(tabID string) error {
947 a.remoteTabMu.Lock()
948 tab := a.remoteTabs[tabID]
949 supported := tab != nil && tab.capabilities["permission-presets-v1"]
950 a.remoteTabMu.Unlock()
951 if tab == nil {
952 return fmt.Errorf("remote tab %q is not open", tabID)
953 }
954 if !supported {
955 return fmt.Errorf("this remote Reasonix Serve is read-only because it does not support permission-presets-v1; upgrade the remote service to run tools or change permissions")
956 }
957 return nil
958 }
959
960 // requireRemoteExecutionProtocol fences every state-changing command at the
961 // authenticated Serve capability boundary. A legacy Serve remains usable for
962 // history reads, but Desktop never emulates the v3 runtime over older RPCs.
963 func (a *App) requireRemoteExecutionProtocol(tabID string) error {
964 a.remoteTabMu.Lock()
965 tab := a.remoteTabs[tabID]
966 supported := tab != nil && tab.capabilities[serveCapabilityExecutionV2] && tab.capabilities[serveCapabilitySessions] && tab.capabilities[serveCapabilitySessionIdentityV1] && tab.capabilities[serveCapabilitySessionOwnershipV1]
967 a.remoteTabMu.Unlock()
968 if tab == nil {
969 return fmt.Errorf("remote tab %q is not open", tabID)
970 }
971 if !supported {
972 return fmt.Errorf("this remote Reasonix Serve is read-only because it does not support %s, %s, %s, and %s; upgrade the remote service to execute or control a session", serveCapabilityExecutionV2, serveCapabilitySessions, serveCapabilitySessionIdentityV1, serveCapabilitySessionOwnershipV1)
973 }
974 return nil
975 }
976
977 func (a *App) SetRemoteTabGoal(tabID, goal string) error {
978 if err := a.requireRemoteExecutionProtocol(tabID); err != nil {
979 return err
980 }
981 if err := a.requireRemoteGoalLifecycle(tabID); err != nil {
982 return err
983 }
984 client, base, expectedPath, err := a.remoteTabCommandTarget(tabID)
985 if err != nil {
986 return err
987 }
988 ctx, cancel := commandContext(a)
989 defer cancel()
990 body, _ := json.Marshal(map[string]string{"goal": goal})
991 return servePostForSession(ctx, client, serveURL(base, "/goal"), body, expectedPath)
992 }
993
994 func (a *App) requireRemoteGoalLifecycle(tabID string) error {
995 a.remoteTabMu.Lock()
996 tab := a.remoteTabs[tabID]
997 supported := tab != nil && tab.capabilities[serveCapabilityGoalLifecycleV2]
998 a.remoteTabMu.Unlock()
999 if tab == nil {
1000 return fmt.Errorf("remote tab %q is not open", tabID)
1001 }
1002 if !supported {
1003 return fmt.Errorf("this remote Reasonix Serve does not support %s; upgrade it before creating or controlling goals", serveCapabilityGoalLifecycleV2)
1004 }
1005 return nil
1006 }
1007
1008 func (a *App) SetRemoteTabQualityFloor(tabID, floor string) error {
1009 return a.validateRemoteQualityFloor(tabID, floor)
1010 }
1011
1011 lines GO