| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "net/http" |
| 6 | "strings" |
| 7 | ) |
| 8 | |
| 9 | type remoteTabProvisionalResume struct { |
| 10 | targetPath string |
| 11 | previousPath string |
| 12 | pathRevision uint64 |
| 13 | previousPending map[string]json.RawMessage |
| 14 | previousRuntime remoteTabRuntimeState |
| 15 | active bool |
| 16 | selectionRevision uint64 |
| 17 | previousSelection *remoteTabOpenSelection |
| 18 | } |
| 19 | |
| 20 | func probeRemoteTabFrame(frame string) (kind, path string, current, reset bool) { |
| 21 | var probe struct { |
| 22 | Kind string `json:"kind"` |
| 23 | SessionID string `json:"sessionId"` |
| 24 | SessionPath string `json:"sessionPath"` |
| 25 | SessionCurrent bool `json:"sessionCurrent"` |
| 26 | SessionReset bool `json:"sessionReset"` |
| 27 | } |
| 28 | kind = "?" |
| 29 | if json.Unmarshal([]byte(frame), &probe) == nil && probe.Kind != "" { |
| 30 | kind = probe.Kind |
| 31 | } |
| 32 | return kind, remoteSessionIdentityRoute(probe.SessionPath, probe.SessionID), probe.SessionCurrent, probe.SessionReset |
| 33 | } |
| 34 | |
| 35 | func (a *App) beginRemoteTabProvisionalResume(tabID string, tab *remoteTab, client *http.Client, gen uint64, targetPath string) remoteTabProvisionalResume { |
| 36 | route := remoteTabProvisionalResume{targetPath: strings.TrimSpace(targetPath)} |
| 37 | tab.routeEventMu.Lock() |
| 38 | defer tab.routeEventMu.Unlock() |
| 39 | a.remoteTabMu.Lock() |
| 40 | defer a.remoteTabMu.Unlock() |
| 41 | current := a.remoteTabs[tabID] |
| 42 | if current != tab || current.client != client || current.gen != gen || current.state != "ready" { |
| 43 | return route |
| 44 | } |
| 45 | route.selectionRevision = current.selectionRevision |
| 46 | route.previousPath = current.routing.currentPath |
| 47 | route.pathRevision = current.routing.pathRevision |
| 48 | if route.targetPath == route.previousPath { |
| 49 | return route |
| 50 | } |
| 51 | route.previousPending = cloneRemotePendingEvents(current.pendingEvents) |
| 52 | route.previousRuntime = current.runtime |
| 53 | route.active = true |
| 54 | current.routing.currentPath = route.targetPath |
| 55 | current.routing.pathRevision++ |
| 56 | current.routing.rehydratingPath = route.targetPath |
| 57 | current.routing.rehydratingFrames = nil |
| 58 | current.routing.revision++ |
| 59 | resetRemoteTabForegroundRuntimeLocked(current) |
| 60 | current.runtime.running = current.routing.running[route.targetPath] |
| 61 | current.runtime.cancellable = current.runtime.running |
| 62 | return route |
| 63 | } |
| 64 | |
| 65 | func (a *App) rollbackRemoteTabProvisionalResume(tabID string, tab *remoteTab, client *http.Client, gen uint64, route remoteTabProvisionalResume) bool { |
| 66 | tab.routeEventMu.Lock() |
| 67 | defer tab.routeEventMu.Unlock() |
| 68 | a.remoteTabMu.Lock() |
| 69 | defer a.remoteTabMu.Unlock() |
| 70 | current := a.remoteTabs[tabID] |
| 71 | if current != tab || current.client != client || current.gen != gen || current.state != "ready" || |
| 72 | current.selectionRevision != route.selectionRevision || |
| 73 | current.routing.currentPath != route.targetPath { |
| 74 | return false |
| 75 | } |
| 76 | if !route.active { |
| 77 | // Re-selecting the already current session creates no rehydration epoch. |
| 78 | // The path revision still proves whether this failed request owns the |
| 79 | // visible route or a newer adoption has already superseded it. |
| 80 | return current.routing.pathRevision == route.pathRevision |
| 81 | } |
| 82 | if current.routing.rehydratingPath != route.targetPath { |
| 83 | return false |
| 84 | } |
| 85 | restoreRemoteTabProvisionalRouteLocked(current, route) |
| 86 | return true |
| 87 | } |
| 88 | |
| 89 | // closeRemoteTabProvisionalRouteLocked ends the provisional route epoch when |
| 90 | // its pump generation is retired. Every commit and rollback path fences on |
| 91 | // that generation, so the buffered frames can never be drained and the gate |
| 92 | // would otherwise refuse commands until the next identity change. The route |
| 93 | // itself stays: the reattach reconciles it against Serve's foreground. Caller |
| 94 | // holds remoteTabMu. |
| 95 | func closeRemoteTabProvisionalRouteLocked(tab *remoteTab) { |
| 96 | tab.routing.rehydratingPath = "" |
| 97 | tab.routing.rehydratingFrames = nil |
| 98 | } |
| 99 | |
| 100 | func restoreRemoteTabProvisionalRouteLocked(current *remoteTab, route remoteTabProvisionalResume) { |
| 101 | current.routing.currentPath = route.previousPath |
| 102 | current.routing.pathRevision++ |
| 103 | current.routing.rehydratingPath = "" |
| 104 | current.routing.rehydratingFrames = nil |
| 105 | current.routing.revision++ |
| 106 | current.pendingEvents = route.previousPending |
| 107 | restoredRuntime := route.previousRuntime |
| 108 | restoredRuntime.revision = max(current.runtime.revision, route.previousRuntime.revision) + 1 |
| 109 | current.runtime = restoredRuntime |
| 110 | } |
| 111 | |
| 112 | // reconcileRemoteTabRejectedResume installs the route Serve reports after an |
| 113 | // ambiguous transport failure. The common unchanged case restores the exact |
| 114 | // preflight snapshot; an externally changed route drops controller-local state |
| 115 | // and publishes the authoritative identity behind a new ready barrier. It |
| 116 | // commits rejection and any pre-open restoration before publishing its error. |
| 117 | func (a *App) reconcileRemoteTabRejectedResume(tabID string, tab *remoteTab, client *http.Client, gen uint64, route remoteTabProvisionalResume, authoritative serveSessionEntry, resumeErr error) bool { |
| 118 | authoritative.Path = strings.TrimSpace(authoritative.Path) |
| 119 | authoritativeRoute := remoteSessionRoute(authoritative) |
| 120 | if authoritativeRoute == route.previousPath || route.previousSelection != nil && authoritativeRoute == route.previousSelection.currentPath { |
| 121 | return a.completeRemoteTabResumeFailure(tabID, tab, client, gen, route, resumeErr.Error()) |
| 122 | } |
| 123 | tab.routeEventMu.Lock() |
| 124 | defer tab.routeEventMu.Unlock() |
| 125 | a.remoteTabMu.Lock() |
| 126 | current := a.remoteTabs[tabID] |
| 127 | if current != tab || current.client != client || current.gen != gen || current.state != "ready" || |
| 128 | current.selectionRevision != route.selectionRevision || |
| 129 | current.routing.currentPath != route.targetPath || |
| 130 | route.active && current.routing.rehydratingPath != route.targetPath || |
| 131 | !route.active && current.routing.pathRevision != route.pathRevision { |
| 132 | a.remoteTabMu.Unlock() |
| 133 | return true |
| 134 | } |
| 135 | if !adoptRemoteTabSessionPathLocked(current, authoritativeRoute) { |
| 136 | current.routing.rehydratingPath = "" |
| 137 | current.routing.rehydratingFrames = nil |
| 138 | } |
| 139 | current.session.name = strings.TrimSpace(authoritative.Name) |
| 140 | current.session.path = authoritative.Path |
| 141 | current.session.sessionID = authoritative.SessionID |
| 142 | current.session.takenOver = authoritative.TakenOver |
| 143 | current.session.newSession = false |
| 144 | current.session.reset = false |
| 145 | current.runtime.running = authoritative.Running || current.routing.running[authoritativeRoute] |
| 146 | current.runtime.cancellable = current.runtime.running |
| 147 | title := strings.TrimSpace(authoritative.Title) |
| 148 | if title == "" { |
| 149 | title = strings.TrimSpace(authoritative.Name) |
| 150 | } |
| 151 | if title == "" { |
| 152 | title = remoteWorkspaceName(current.ref.Workspace) |
| 153 | } |
| 154 | current.topicTitle = title |
| 155 | meta := remoteTabMetaLocked(current) |
| 156 | a.remoteTabMu.Unlock() |
| 157 | a.emitRemoteEvent("remote-tab:updated", meta) |
| 158 | a.saveTabsFromRemote() |
| 159 | a.transitionRemoteTabStateLocked(tab, gen, "ready", "ready", resumeErr.Error()) |
| 160 | // The probed third path is Serve-authoritative. The generic open-selection |
| 161 | // rollback must not replace it with the preflight route. |
| 162 | return true |
| 163 | } |
| 164 | |
| 165 | func (a *App) commitRemoteTabResume(tabID string, tab *remoteTab, client *http.Client, gen uint64, route remoteTabProvisionalResume, target serveSessionEntry, title string) (TabMeta, bool) { |
| 166 | a.remoteTabMu.Lock() |
| 167 | defer a.remoteTabMu.Unlock() |
| 168 | current := a.remoteTabs[tabID] |
| 169 | if current != tab || current.client != client || current.gen != gen || current.state != "ready" || |
| 170 | current.routing.currentPath != route.targetPath || |
| 171 | route.active && current.routing.rehydratingPath != route.targetPath || |
| 172 | !route.active && current.routing.pathRevision != route.pathRevision { |
| 173 | return TabMeta{}, false |
| 174 | } |
| 175 | current.topicTitle = title |
| 176 | current.session.reset = false |
| 177 | current.session.newSession = false |
| 178 | current.session.name = strings.TrimSpace(target.Name) |
| 179 | current.session.path = target.Path |
| 180 | current.session.sessionID = target.SessionID |
| 181 | current.session.takenOver = target.TakenOver |
| 182 | targetRoute := remoteSessionRoute(target) |
| 183 | current.routing.currentPath = targetRoute |
| 184 | // Close the provisional routing epoch so a listing that began while |
| 185 | // /resume was in flight cannot publish its pre-switch snapshot afterward. |
| 186 | current.routing.revision++ |
| 187 | current.runtime.revision++ |
| 188 | current.runtime.running = current.runtime.running || target.Running || current.routing.running[targetRoute] |
| 189 | current.runtime.cancellable = current.runtime.cancellable || current.runtime.running |
| 190 | return remoteTabMetaLocked(current), true |
| 191 | } |
| 192 | |
| 193 | // commitAndPublishRemoteTabResume keeps the successful HTTP commit, metadata |
| 194 | // publication, and ready/replay handoff in the same route epoch. Without this |
| 195 | // fence a newer session_changed adoption could publish between those steps and |
| 196 | // then be overwritten by the older resume metadata. |
| 197 | func (a *App) commitAndPublishRemoteTabResume(tabID string, tab *remoteTab, client *http.Client, gen uint64, route remoteTabProvisionalResume, target serveSessionEntry, title string) bool { |
| 198 | tab.routeEventMu.Lock() |
| 199 | defer tab.routeEventMu.Unlock() |
| 200 | meta, committed := a.commitRemoteTabResume(tabID, tab, client, gen, route, target, title) |
| 201 | if !committed { |
| 202 | return false |
| 203 | } |
| 204 | a.emitRemoteEvent("remote-tab:updated", meta) |
| 205 | a.saveTabsFromRemote() |
| 206 | // Frames received while /resume was in flight were held behind the |
| 207 | // provisional route. Rehydrate the committed session before replaying its |
| 208 | // retained prompts or later live output. |
| 209 | a.publishRemoteTabResumeReadyLocked(tabID, tab, client, gen, route) |
| 210 | return true |
| 211 | } |
| 212 | |
| 213 | func (a *App) publishRemoteTabResumeReady(tabID string, tab *remoteTab, client *http.Client, gen uint64, route remoteTabProvisionalResume) { |
| 214 | // Keep the ready barrier and the complete buffered drain in one ordered |
| 215 | // route-publication epoch. A later session adoption waits until every frame |
| 216 | // from this snapshot is visible, then publishes its own ready barrier. |
| 217 | tab.routeEventMu.Lock() |
| 218 | defer tab.routeEventMu.Unlock() |
| 219 | a.publishRemoteTabResumeReadyLocked(tabID, tab, client, gen, route) |
| 220 | } |
| 221 | |
| 222 | // publishRemoteTabResumeReadyLocked publishes while tab.routeEventMu is held. |
| 223 | func (a *App) publishRemoteTabResumeReadyLocked(tabID string, tab *remoteTab, client *http.Client, gen uint64, route remoteTabProvisionalResume) { |
| 224 | if !a.transitionRemoteTabStateLocked(tab, gen, "ready", "ready", "") { |
| 225 | return |
| 226 | } |
| 227 | for { |
| 228 | a.remoteTabMu.Lock() |
| 229 | current := a.remoteTabs[tabID] |
| 230 | if current != tab || current.client != client || current.gen != gen || current.routing.rehydratingPath != route.targetPath { |
| 231 | a.remoteTabMu.Unlock() |
| 232 | return |
| 233 | } |
| 234 | frames := current.routing.rehydratingFrames |
| 235 | current.routing.rehydratingFrames = nil |
| 236 | if len(frames) == 0 { |
| 237 | // Clearing the path under the same lock that producers use closes the |
| 238 | // replay/live race: a later frame either joined this drain or observes |
| 239 | // the committed live route after every older frame was published. |
| 240 | current.routing.rehydratingPath = "" |
| 241 | a.remoteTabMu.Unlock() |
| 242 | return |
| 243 | } |
| 244 | a.remoteTabMu.Unlock() |
| 245 | for _, frame := range frames { |
| 246 | kind, path, _, _ := probeRemoteTabFrame(string(frame)) |
| 247 | if path != "" && path != route.targetPath { |
| 248 | // The buffer fence only admits the target route; a foreign frame |
| 249 | // here is defensive debris. Dropping it keeps the drain alive — |
| 250 | // aborting would strand the epoch and buffer live frames forever. |
| 251 | continue |
| 252 | } |
| 253 | if !a.publishRemoteTabFrameForRouteLocked(tabID, tab, tab, client, gen, route.targetPath, true, kind, frame) { |
| 254 | return |
| 255 | } |
| 256 | } |
| 257 | } |
| 258 | } |
| 259 | |
| 260 | func cloneRemotePendingEvents(src map[string]json.RawMessage) map[string]json.RawMessage { |
| 261 | if src == nil { |
| 262 | return nil |
| 263 | } |
| 264 | dst := make(map[string]json.RawMessage, len(src)) |
| 265 | for key, frame := range src { |
| 266 | dst[key] = append(json.RawMessage(nil), frame...) |
| 267 | } |
| 268 | return dst |
| 269 | } |
| 270 | |
| 271 | func remotePendingEventKey(kind string, frame json.RawMessage) string { |
| 272 | var probe struct { |
| 273 | Approval *struct { |
| 274 | ID string `json:"id"` |
| 275 | } `json:"approval"` |
| 276 | Ask *struct { |
| 277 | ID string `json:"id"` |
| 278 | } `json:"ask"` |
| 279 | MCPInteraction *struct { |
| 280 | ID string `json:"id"` |
| 281 | } `json:"mcpInteraction"` |
| 282 | } |
| 283 | _ = json.Unmarshal(frame, &probe) |
| 284 | id := "" |
| 285 | if probe.Approval != nil { |
| 286 | id = probe.Approval.ID |
| 287 | } else if probe.Ask != nil { |
| 288 | id = probe.Ask.ID |
| 289 | } else if probe.MCPInteraction != nil { |
| 290 | id = probe.MCPInteraction.ID |
| 291 | } |
| 292 | return kind + ":" + strings.TrimSpace(id) |
| 293 | } |
| 294 | |
| 295 | func (a *App) bufferRemoteTabResumeFrame(tabID string, gen uint64, sessionPath, kind string, frame json.RawMessage) bool { |
| 296 | sessionPath = strings.TrimSpace(sessionPath) |
| 297 | if sessionPath == "" { |
| 298 | return false |
| 299 | } |
| 300 | key := "" |
| 301 | switch kind { |
| 302 | case "approval_request", "ask_request", "mcp_interaction": |
| 303 | key = remotePendingEventKey(kind, frame) |
| 304 | case "extension_surface": |
| 305 | if remotePendingExtensionForm(frame) { |
| 306 | key = remotePendingExtensionFormKey |
| 307 | } |
| 308 | } |
| 309 | a.remoteTabMu.Lock() |
| 310 | defer a.remoteTabMu.Unlock() |
| 311 | tab := a.remoteTabs[tabID] |
| 312 | if tab == nil || tab.gen != gen || tab.routing.rehydratingPath != sessionPath { |
| 313 | return false |
| 314 | } |
| 315 | if key != "" { |
| 316 | tab.runtime.revision++ |
| 317 | if tab.pendingEvents == nil { |
| 318 | tab.pendingEvents = make(map[string]json.RawMessage) |
| 319 | } |
| 320 | tab.pendingEvents[key] = append(json.RawMessage(nil), frame...) |
| 321 | tab.runtime.pendingPrompt = true |
| 322 | tab.runtime.cancellable = true |
| 323 | } |
| 324 | // Actionable frames must also cross the same fenced handoff as ordinary |
| 325 | // output. Snapshot hydration deduplicates prompt IDs against live-buffered |
| 326 | // events, while this replay closes the window after snapshot capture. |
| 327 | tab.routing.rehydratingFrames = append(tab.routing.rehydratingFrames, append(json.RawMessage(nil), frame...)) |
| 328 | return true |
| 329 | } |
| 330 |