fix(sync): reject obsolete interrupted-turn recovery responses

Follow up #3396 with runtime, SDK, and request ownership guards across hydration and reconnect.

packages/ui/src/sync/DOCUMENTATION.md

@@ -248,8 +248,7 @@ When `session.idle` or `session.error` settles a session but the trailing assist
 
 A completed assistant message is authoritative for its own tool parts. During materialization, a `pending` or `running` tool under `time.completed` becomes `error`/`Interrupted` with an end time. This handles stale persisted tool state during reload. The merge preserves a terminal part already observed live, and a later terminal server snapshot can replace the local interrupted marker.
 
-When a session is authoritatively settled — `session.idle`/`session.error` event, or an authoritative status snapshot that lowers a previously busy session — and the trailing assistant message is still *unfinished* (`time.completed` missing) with no pending question/permission, the turn is treated as interrupted (managed OpenCode process died mid-turn; the server never finalizes the message or parts, see openchamber#2577 / anomalyco/opencode#19023). The unfinished assistant message is completed locally with `MessageAbortedError`, including text-only turns and turns whose tools had already finished, so the chat shows a visible interrupted state. Any active parts are also finalized as `error`/`Interrupted` with an end time, so tool timers stop and cards render the error state. The mark is gated on an explicit idle status (absent status is "unknown", never judged), never applies while the session is busy (including question/permission waits), and a later terminal event can supersede it while a stale unfinished refresh cannot regress the locally finalized message or parts.
-When a session is authoritatively settled — `session.idle`/`session.error` event, or an authoritative status snapshot that lowers a previously busy session — and the trailing assistant message is still *unfinished* (`time.completed` missing) with no pending question/permission, the turn is treated as interrupted (managed OpenCode process died mid-turn; the server never finalizes the message or parts, see openchamber#2577 / anomalyco/opencode#19023). The unfinished assistant message is completed locally with `MessageAbortedError`, including text-only turns and turns whose tools had already finished, so the chat shows a visible interrupted state. Any active parts are also finalized as `error`/`Interrupted` with an end time, so tool timers stop and cards render the error state. The mark is gated on an explicit idle status (absent status is "unknown", never judged), never applies while the session is busy (including question/permission waits), and a later terminal event can supersede it while a stale unfinished refresh cannot regress the locally finalized message or parts. A successful authoritative snapshot records explicit idle for previously unknown candidates, and message hydration retries this reconciliation after the transcript arrives; a failed status fetch leaves the session unknown.
+When a session is authoritatively settled — `session.idle`/`session.error` event, or an authoritative status snapshot that lowers a previously busy session — and the trailing assistant message is still *unfinished* (`time.completed` missing) with no pending question/permission, the turn is treated as interrupted (managed OpenCode process died mid-turn; the server never finalizes the message or parts, see openchamber#2577 / anomalyco/opencode#19023). The unfinished assistant message is completed locally with `MessageAbortedError`, including text-only turns and turns whose tools had already finished, so the chat shows a visible interrupted state. Any active parts are also finalized as `error`/`Interrupted` with an end time, so tool timers stop and cards render the error state. The mark is gated on an explicit idle status (absent status is "unknown", never judged), never applies while the session is busy (including question/permission waits), and a later terminal event can supersede it while a stale unfinished refresh cannot regress the locally finalized message or parts. A successful authoritative snapshot records explicit idle for previously unknown candidates, and message hydration retries this reconciliation after the transcript arrives; a failed status fetch leaves the session unknown. Recovery rejects responses after a runtime or SDK switch, request invalidation, or directory-store disposal before publishing local or global state.
 
 Directory stores also own session-keyed sidecar notification channels for permissions, questions, and message materialization. High-frequency realtime part events annotate the exact session/message before committing, so visible records, user history, renderability, and sidebar permission and question rows are not notified by unrelated sessions. Structural message replacements notify only changed subscribed session buckets; unannotated bulk part replacement conservatively resets active message subscribers so bootstrap, pagination, rollback, and legacy writers cannot leave stale projections.
 

packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts

@@ -15,9 +15,12 @@ type StatusSnapshot = Record<string, SessionStatus | undefined>
 
 let respondWithSnapshot: () => Promise<StatusSnapshot | null> = () => Promise.resolve({ ses_1: { type: "idle" } })
 const statusSnapshotCalls: string[] = []
+let runtimeKey = "test-runtime"
+let sdkIdentity = {}
 
 mock.module("@/lib/opencode/client", () => ({
   opencodeClient: {
+    getSdkClient: () => sdkIdentity,
     getSessionStatusForDirectory: mock((directory: string) => {
       statusSnapshotCalls.push(directory)
       return respondWithSnapshot()
@@ -26,9 +29,13 @@ mock.module("@/lib/opencode/client", () => ({
 }))
 
 mock.module("@/lib/runtime-switch", () => ({
-  getRuntimeKey: () => "test-runtime",
+  getRuntimeKey: () => runtimeKey,
 }))
 
+import { applyGlobalSessionStatusSnapshot, useGlobalSessionStatusStore } from "../global-session-status"
+import { useSessionOrderingStore } from "../session-ordering"
+import { useSessionActivityTimingStore } from "../session-activity-timing"
+
 import {
   maybePollStatusAfterMessageCompletion,
   MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS,
@@ -78,6 +85,8 @@ describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => {
   beforeEach(() => {
     respondWithSnapshot = () => Promise.resolve({ ses_1: { type: "idle" } })
     statusSnapshotCalls.length = 0
+    runtimeKey = "test-runtime"
+    sdkIdentity = {}
   })
 
   test("does not poll when the store believes the session is already idle", async () => {
@@ -183,4 +192,36 @@ describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => {
     expect(part?.type).toBe("tool")
     if (part?.type === "tool") expect(part.state.status).toBe("error")
   })
+
+  for (const change of ["runtime", "sdk", "request"] as const) {
+    test(`discards delayed recovery after ${change} ownership changes`, async () => {
+      const store = createStore()
+      store.getState().patch({
+        message: { ses_1: [unfinishedAssistant] },
+        part: { msg_1: [runningTool] },
+      })
+      const before = store.getState()
+      let resolveSnapshot: (snapshot: StatusSnapshot) => void = () => { throw new Error("Request not started") }
+      respondWithSnapshot = () => new Promise((resolve) => { resolveSnapshot = resolve })
+      let stale = false
+      const recovery = recoverInterruptedTurnAfterMessageLoad("/test/project", store, "ses_1", () => stale)
+      expect(statusSnapshotCalls).toEqual(["/test/project"])
+
+      if (change === "runtime") runtimeKey = "runtime-b"
+      if (change === "sdk") sdkIdentity = {}
+      if (change === "request") stale = true
+      applyGlobalSessionStatusSnapshot("/test/project", { ses_new: { type: "busy" } })
+      const statuses = useGlobalSessionStatusStore.getState()
+      const ordering = useSessionOrderingStore.getState()
+      const timing = useSessionActivityTimingStore.getState()
+      resolveSnapshot({ ses_old: { type: "busy" } })
+      await recovery
+
+      expect(store.getState()).toBe(before)
+      expect(useGlobalSessionStatusStore.getState()).toBe(statuses)
+      expect(useSessionOrderingStore.getState()).toBe(ordering)
+      expect(useSessionActivityTimingStore.getState()).toBe(timing)
+    })
+  }
+
 })

packages/ui/src/sync/sync-context.tsx

@@ -428,7 +428,11 @@ function enqueueSessionMaterialization(
         return
       }
       countSyncPerformance("materializationRequests")
-      await materializeSessionFromServer(directory, sessionID, store, request)
+      await materializeSessionFromServer(directory, sessionID, store, {
+        ...request,
+        isStale: () => childStores.children.get(directory) !== store
+          || pendingSessionMaterializations.get(k) !== pending,
+      })
     } catch {
       // Transient failure — next SSE event or reconnect will catch up.
     } finally {
@@ -458,6 +462,10 @@ async function materializeSessionFromServer(
   store: StoreApi<DirectoryStore>,
   options?: SessionMaterializationRequest & { isStale?: () => boolean },
 ) {
+  const runtimeKey = getRuntimeKey()
+  const sdk = opencodeClient.getSdkClient()
+  const isStale = () => options?.isStale?.() || getRuntimeKey() !== runtimeKey
+    || opencodeClient.getSdkClient() !== sdk
   const statusBeforeMaterialization = store.getState().session_status?.[sessionID]
   syncDebug.recovery.materializing({
     reason: options?.reason ?? "ensure-session-messages",
@@ -467,17 +475,18 @@ async function materializeSessionFromServer(
     partID: options?.partID,
   })
   const loader = getImperativeSessionMessageLoader()
-  if (!loader || options?.isStale?.()) return
+  if (!loader || isStale()) return
   await loader.refreshTail({ directory, sessionID }, SESSION_MATERIALIZATION_MESSAGE_LIMIT)
+  if (isStale()) return
   if (loader.getSnapshot({ directory, sessionID }).status === "error") {
     throw loader.getSnapshot({ directory, sessionID }).error ?? new Error("Session materialization failed")
   }
 
-  if (statusBeforeMaterialization && statusBeforeMaterialization.type !== "idle" && !options?.isStale?.()) {
-    await resyncDirectorySessionStatuses(directory, store, [sessionID], "authoritative")
+  if (statusBeforeMaterialization && statusBeforeMaterialization.type !== "idle" && !isStale()) {
+    await resyncDirectorySessionStatuses(directory, store, [sessionID], "authoritative", isStale)
   }
-  if (!options?.isStale?.()) {
-    await recoverInterruptedTurnAfterMessageLoad(directory, store, sessionID)
+  if (!isStale()) {
+    await recoverInterruptedTurnAfterMessageLoad(directory, store, sessionID, isStale)
   }
 }
 
@@ -741,11 +750,12 @@ async function resyncDirectorySessionStatuses(
   store: StoreApi<DirectoryStore>,
   candidateSessionIds: string[],
   mode: StatusSnapshotMode,
+  isStale?: () => boolean,
 ): Promise<DirectorySessionStatusSnapshot | null> {
   const nextStatuses = await opencodeClient.getSessionStatusForDirectory(directory)
   // null = fetch failed; preserve existing state. {} or populated = a snapshot
   // of active sessions — reconciled per `mode` (absence ≠ idle under monotonic).
-  if (nextStatuses === null) return null
+  if (nextStatuses === null || isStale?.()) return null
   applySessionStatusSnapshot(store, nextStatuses, candidateSessionIds, mode)
   if (mode === "authoritative") {
     store.setState({ sessionStatusReady: true })
@@ -1520,12 +1530,15 @@ async function resyncDirectoryAfterReconnect(
   store: StoreApi<DirectoryStore>,
   routingIndex: EventRoutingIndex,
   reason: SessionMaterializationReason,
+  isStale: () => boolean,
 ) {
+  if (isStale()) return
   const current = store.getState()
   const candidateSessionIds = getActiveSessionCandidateIds(directory, current)
   if (candidateSessionIds.length === 0) return
 
-  await resyncDirectorySessionStatuses(directory, store, candidateSessionIds, "authoritative")
+  await resyncDirectorySessionStatuses(directory, store, candidateSessionIds, "authoritative", isStale)
+  if (isStale()) return
 
   const scopedClient = opencodeClient.getScopedSdkClient(directory)
   await Promise.all(candidateSessionIds.map(async (sessionId) => {
@@ -1539,7 +1552,9 @@ async function resyncDirectoryAfterReconnect(
       }).catch(() => null),
       loader?.refreshTail({ directory, sessionID: sessionId }, RECONNECT_MESSAGE_LIMIT) ?? Promise.resolve(),
     ])
-    await recoverInterruptedTurnAfterMessageLoad(directory, store, sessionId)
+    if (isStale()) return
+    await recoverInterruptedTurnAfterMessageLoad(directory, store, sessionId, isStale)
+    if (isStale()) return
     const session = sessionResponse?.data
     if (!session) return
 
@@ -1563,8 +1578,10 @@ async function resyncDirectoryAfterReconnect(
     setIndexedSessionMessages(routingIndex, sessionId, directory, store.getState().message[sessionId] ?? [])
   }))
 
+  if (isStale()) return
   await resyncBlockingRequestsForDirectory(directory, store, candidateSessionIds)
 
+  if (isStale()) return
   ingestDirectoryStateIntoRoutingIndex(routingIndex, directory, store.getState())
 }
 
@@ -2178,7 +2195,11 @@ export async function recoverInterruptedTurnAfterMessageLoad(
   directory: string,
   store: StoreApi<DirectoryStore>,
   sessionID: string,
+  isStale?: () => boolean,
 ): Promise<void> {
+  if (isStale?.()) return
+  const runtimeKey = getRuntimeKey()
+  const sdk = opencodeClient.getSdkClient()
   const initial = store.getState()
   if (!hasUnfinishedAssistantTurn(initial, sessionID)) return
   if ((initial.question?.[sessionID] ?? []).length > 0) return
@@ -2186,7 +2207,8 @@ export async function recoverInterruptedTurnAfterMessageLoad(
 
   if (!initial.session_status?.[sessionID]) {
     const snapshot = await opencodeClient.getSessionStatusForDirectory(directory)
-    if (snapshot === null) return
+    if (snapshot === null || isStale?.()
+      || getRuntimeKey() !== runtimeKey || opencodeClient.getSdkClient() !== sdk) return
 
     // Do not overwrite a live status event that arrived while the snapshot was
     // in flight. The snapshot only fills the previously unknown state.
@@ -2279,7 +2301,11 @@ export function SyncProvider(props: {
 
     lastFullResyncAtByDirectoryRef.current.set(directory, Date.now())
     resyncing.add(directory)
-    void resyncDirectoryAfterReconnect(directory, store, routingIndex, reason)
+    const sdk = opencodeClient.getSdkClient()
+    const expectedRuntimeKey = getRuntimeKey()
+    const isStale = () => getRuntimeKey() !== expectedRuntimeKey
+      || opencodeClient.getSdkClient() !== sdk || childStores.children.get(directory) !== store
+    void resyncDirectoryAfterReconnect(directory, store, routingIndex, reason, isStale)
       .catch(() => {
         // Transient failure — the watchdog, next SSE event, or reconnect will catch up.
       })

packages/ui/src/sync/use-sync.ts

@@ -236,17 +236,19 @@ export function useSync() {
       // knows it is stale and should not write to the store.
       const generation = (syncSessionGenerationByKey.get(key) ?? 0) + 1
       syncSessionGenerationByKey.set(key, generation)
-      const isStale = () => syncSessionGenerationByKey.get(key) !== generation
 
       const targetStore = targetDirectory === directory
         ? store
         : childStores.ensureChild(targetDirectory, { bootstrap: false })
+      const isStale = () => getRuntimeKey() !== runtimeKey
+        || syncSessionGenerationByKey.get(key) !== generation
+        || childStores.children.get(targetDirectory) !== targetStore
       const current = targetStore.getState()
       const materialization = getSessionMaterializationStatus(current, sessionID)
       const cachedReady = materialization.hasMessages && materialization.renderable
       const hasSession = Binary.search(current.session, sessionID, (s) => s.id).found
       if (cachedReady && hasSession && !force) {
-        await recoverInterruptedTurnAfterMessageLoad(targetDirectory, targetStore, sessionID)
+        await recoverInterruptedTurnAfterMessageLoad(targetDirectory, targetStore, sessionID, isStale)
         return
       }
       const shouldLoadMessages = Boolean(!cachedReady || force)
@@ -281,7 +283,7 @@ export function useSync() {
                   { force, reason: "reactive" },
                 )
                 if (!isStale()) {
-                  await recoverInterruptedTurnAfterMessageLoad(targetDirectory, targetStore, sessionID)
+                  await recoverInterruptedTurnAfterMessageLoad(targetDirectory, targetStore, sessionID, isStale)
                 }
               })()
             : Promise.resolve(),