fix(ui): reconcile stale active tools after reload (#3396)

* fix: reconcile stale active tools on materialization * fix: recover stale turns after reload
2026-09-07 20:25:42 +03:00
parent bf4262dbc9
commit a059d54b44
7 changed files with 251 additions and 26 deletions
@@ -246,7 +246,10 @@ Incomplete-session materialization is deduplicated by runtime, directory, and se

 When `session.idle` or `session.error` settles a session but the trailing assistant message still contains a `pending` or `running` tool, sync refreshes that session tail. This narrowly reconciles a missed terminal tool-part event without refetching normally completed turns or stale tools from older turns. A stale refresh or delayed part event cannot regress a locally observed terminal tool to an active status.

+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.

 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.

@@ -1,5 +1,5 @@
 import { describe, expect, test } from "bun:test"
-import type { Message, Part } from "@opencode-ai/sdk/v2/client"
+import type { Message, Part, ToolPart } from "@opencode-ai/sdk/v2/client"
 import {
  getSessionMaterializationRequestKey,
  getSessionMaterializationStatus,
@@ -16,6 +16,23 @@ function userMessage(id: string, sessionID = "ses_1"): Message {
  return { id, sessionID, role: "user", time: { created: 1 } } as Message
 }

+function completedAssistantMessage(id: string, sessionID = "ses_1"): Message {
+  return {
+    id,
+    sessionID,
+    role: "assistant",
+    time: { created: 1, completed: 4000 },
+    parentID: "msg_parent",
+    modelID: "model",
+    providerID: "provider",
+    mode: "mode",
+    agent: "agent",
+    path: { cwd: "/repo", root: "/repo" },
+    cost: 0,
+    tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
+  }
+}
+
 function part(id: string, messageID: string, type = "text", text = id): Part {
  return { id, messageID, sessionID: "ses_1", type, text } as Part
 }
@@ -28,6 +45,69 @@ describe("getSessionMaterializationRequestKey", () => {
 })

 describe("materializeSessionSnapshots", () => {
+  test("finalizes an active tool under a completed assistant message", () => {
+    const completedMessage = completedAssistantMessage("msg_1")
+    const staleRunningTool = {
+      id: "prt_1",
+      messageID: "msg_1",
+      sessionID: "ses_1",
+      type: "tool",
+      tool: "bash",
+      state: { status: "running", input: { command: "ls" }, time: { start: 1000 } },
+      callID: "call-prt_1",
+    } satisfies ToolPart
+
+    const result = materializeSessionSnapshots(
+      { message: {}, part: {} },
+      "ses_1",
+      [{ info: completedMessage, parts: [staleRunningTool] }],
+    )
+
+    const reconciledPart = result.part.msg_1[0]
+    if (!reconciledPart || reconciledPart.type !== "tool") throw new Error("Expected tool part")
+    if (reconciledPart.state.status !== "error") throw new Error("Expected interrupted tool part")
+    expect(reconciledPart.state.error).toBe("Interrupted")
+    expect(reconciledPart.state.time).toEqual({ start: 1000, end: 4000 })
+    expect(getStaleRunningToolMessageID(result, "ses_1")).toBe(undefined)
+  })
+
+  test("preserves a terminal tool already observed when a completed snapshot is stale", () => {
+    const completedMessage = completedAssistantMessage("msg_1")
+    const terminalTool = {
+      id: "prt_1",
+      messageID: "msg_1",
+      sessionID: "ses_1",
+      type: "tool",
+      tool: "bash",
+      state: {
+        status: "completed",
+        input: { command: "ls" },
+        output: "done",
+        title: "bash",
+        metadata: {},
+        time: { start: 1000, end: 2000 },
+      },
+      callID: "call-prt_1",
+    } satisfies ToolPart
+    const staleRunningTool = {
+      ...terminalTool,
+      state: { status: "running", input: {}, time: { start: 1000 } },
+    } satisfies ToolPart
+    const state = {
+      message: { ses_1: [completedMessage] },
+      part: { msg_1: [terminalTool] },
+    }
+
+    const result = materializeSessionSnapshots(
+      state,
+      "ses_1",
+      [{ info: completedMessage, parts: [staleRunningTool] }],
+    )
+
+    expect(result.part).toBe(state.part)
+    expect(result.part.msg_1[0]).toBe(terminalTool)
+  })
+
  test("marks an empty successful page as materialized", () => {
    const result = materializeSessionSnapshots(
      { message: {}, part: {} },
@@ -7,7 +7,7 @@
 */
 import { beforeEach, describe, expect, mock, test } from "bun:test"
 import { create, type StoreApi } from "zustand"
-import type { SessionStatus } from "@opencode-ai/sdk/v2/client"
+import type { Message, Part, SessionStatus } from "@opencode-ai/sdk/v2/client"
 import { INITIAL_STATE } from "../types"
 import type { DirectoryStore } from "../child-store"

@@ -29,17 +29,43 @@ mock.module("@/lib/runtime-switch", () => ({
  getRuntimeKey: () => "test-runtime",
 }))

-import { maybePollStatusAfterMessageCompletion, MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS } from "../sync-context"
+import {
+  maybePollStatusAfterMessageCompletion,
+  MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS,
+  recoverInterruptedTurnAfterMessageLoad,
+} from "../sync-context"

-const createStore = (status: SessionStatus): StoreApi<DirectoryStore> => {
+const createStore = (status?: SessionStatus): StoreApi<DirectoryStore> => {
+  const session_status: DirectoryStore["session_status"] = {}
+  if (status) session_status.ses_1 = status
  return create<DirectoryStore>()((set) => ({
    ...INITIAL_STATE,
-    session_status: { ses_1: status },
+    session_status,
    patch: (partial) => set(partial),
    replace: (next) => set(next),
  }))
 }

+// SAFETY: The recovery path reads only the identity, role, and completion time
+// fields from this synthetic assistant message.
+const unfinishedAssistant = {
+  id: "msg_1",
+  sessionID: "ses_1",
+  role: "assistant",
+  time: { created: 1 },
+} as Message
+
+// SAFETY: The recovery path reads only the tool discriminator and state fields
+// from this synthetic part.
+const runningTool = {
+  id: "part_1",
+  messageID: "msg_1",
+  sessionID: "ses_1",
+  type: "tool",
+  tool: "bash",
+  state: { status: "running", time: { start: 1 }, input: {} },
+} as Part
+
 const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms))

 /** Past the deferral, plus room for the background-network task chain. */
@@ -138,4 +164,23 @@ describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => {
    expect(statusSnapshotCalls).toEqual(["/test/project", "/test/project"])
    expect(store.getState().session_status?.ses_1?.type).toBe("idle")
  })
+
+  test("recovers an unfinished turn after reload when status was initially unknown", async () => {
+    const store = createStore()
+    store.getState().patch({
+      message: { ses_1: [unfinishedAssistant] },
+      part: { msg_1: [runningTool] },
+    })
+
+    await recoverInterruptedTurnAfterMessageLoad("/test/project", store, "ses_1")
+
+    expect(statusSnapshotCalls).toEqual(["/test/project"])
+    expect(store.getState().session_status?.ses_1?.type).toBe("idle")
+    const message = store.getState().message.ses_1[0]
+    expect(message?.role).toBe("assistant")
+    if (message?.role === "assistant") expect(message.time.completed).toBeDefined()
+    const part = store.getState().part.msg_1[0]
+    expect(part?.type).toBe("tool")
+    if (part?.type === "tool") expect(part.state.status).toBe("error")
+  })
 })
@@ -86,6 +86,13 @@ describe("applySessionStatusSnapshot", () => {
      expect(changed).toBe(true)
      expect(store.getState().session_status.ses_a).toEqual({ type: "idle" })
    })
+
+    test("seeds idle for a candidate with no previous status entry", () => {
+      const store = createDirectoryStore({ session_status: {} })
+      const changed = applySessionStatusSnapshot(store, {} as StatusSnapshot, ["ses_a"], "authoritative")
+      expect(changed).toBe(true)
+      expect(store.getState().session_status.ses_a).toEqual({ type: "idle" })
+    })
  })
 })

@@ -98,6 +98,31 @@ function filterMaterializedParts(parts: Part[], skipPartTypes: ReadonlySet<strin
    .filter((part) => !!part?.id && !skipPartTypes.has(part.type))
 }

+function finalizeActiveToolsInCompletedMessage(message: Message, parts: Part[]): Part[] {
+  if (message.role !== "assistant" || message.time.completed === undefined) return parts
+
+  const completedAt = message.time.completed
+  let reconciledParts = parts
+  for (let index = 0; index < parts.length; index += 1) {
+    const part = parts[index]
+    if (part.type !== "tool" || !ACTIVE_TOOL_STATUSES.has(part.state.status)) continue
+
+    const start = getPartStateTime(part)?.start ?? completedAt
+    if (reconciledParts === parts) reconciledParts = [...parts]
+    reconciledParts[index] = {
+      ...part,
+      state: {
+        ...part.state,
+        status: "error" as const,
+        error: "Interrupted",
+        time: { start, end: completedAt },
+      },
+    }
+  }
+
+  return reconciledParts
+}
+
 function haveEquivalentPartSnapshots(left: Part[] | undefined, right: Part[]): boolean {
  // `undefined` means "parts never fetched", which is NOT equivalent to a
  // fetched-empty snapshot — the empty array must be committed so
@@ -297,12 +322,13 @@ export function materializeSessionSnapshots(

    const isAssistant = record.info.role === "assistant"
    const existing = nextPartState[messageID]
-    const nextParts = mergeMaterializedParts(
+    const mergedParts = mergeMaterializedParts(
      existing,
      filterMaterializedParts(record.parts ?? [], skipPartTypes),
      skipPartTypes,
      isAssistant,
    )
+    const nextParts = finalizeActiveToolsInCompletedMessage(record.info, mergedParts)
    // For non-assistant messages an empty snapshot keeps the old "absent"
    // representation; only assistant messages need the explicit [] marker
    // (getSessionMaterializationStatus checks only assistant messages).
@@ -476,6 +476,9 @@ async function materializeSessionFromServer(
  if (statusBeforeMaterialization && statusBeforeMaterialization.type !== "idle" && !options?.isStale?.()) {
    await resyncDirectorySessionStatuses(directory, store, [sessionID], "authoritative")
  }
+  if (!options?.isStale?.()) {
+    await recoverInterruptedTurnAfterMessageLoad(directory, store, sessionID)
+  }
 }

 // Module-level refs for notification viewed check.
@@ -718,7 +721,10 @@ export function applySessionStatusSnapshot(
      if (mode === "monotonic") continue

      const existing = current[sessionId]
-      if (existing && existing.type !== "idle") {
+      // Keep the successful snapshot distinguishable from "status has never
+      // been observed". Interrupted-turn recovery requires this explicit
+      // settle marker after a cold reload.
+      if (!existing || existing.type !== "idle") {
        draft()[sessionId] = { type: "idle" }
        changed = true
      }
@@ -751,20 +757,7 @@ async function resyncDirectorySessionStatuses(
    // which is the gate the helper requires — a session the snapshot reports
    // busy stays untouched.
    for (const sessionId of candidateSessionIds) {
-      const interrupted = interruptedTurnToolParts(store.getState(), sessionId)
-      if (interrupted) {
-        if (!interrupted.parts) {
-          store.setState((state) => ({
-            message: { ...state.message, [sessionId]: interrupted.messages },
-          }))
-          continue
-        }
-        const interruptedParts = interrupted.parts
-        store.setState((state) => ({
-          message: { ...state.message, [sessionId]: interrupted.messages },
-          part: { ...state.part, [interrupted.messageID]: interruptedParts },
-        }))
-      }
+      applyInterruptedTurnReconciliation(store, sessionId)
    }
  }
  return nextStatuses
@@ -1546,6 +1539,7 @@ async function resyncDirectoryAfterReconnect(
      }).catch(() => null),
      loader?.refreshTail({ directory, sessionID: sessionId }, RECONNECT_MESSAGE_LIMIT) ?? Promise.resolve(),
    ])
+    await recoverInterruptedTurnAfterMessageLoad(directory, store, sessionId)
    const session = sessionResponse?.data
    if (!session) return

@@ -2144,6 +2138,67 @@ export function interruptedTurnToolParts(
  }
 }

+function hasUnfinishedAssistantTurn(state: DirectoryStore, sessionID: string): boolean {
+  const messages = state.message[sessionID] ?? []
+  for (let index = messages.length - 1; index >= 0; index -= 1) {
+    const message = messages[index]
+    if (message.role === "user") return false
+    if (message.role !== "assistant") continue
+    return message.time.completed === undefined
+  }
+  return false
+}
+
+function applyInterruptedTurnReconciliation(store: StoreApi<DirectoryStore>, sessionID: string): void {
+  const interrupted = interruptedTurnToolParts(store.getState(), sessionID)
+  if (!interrupted) return
+
+  const interruptedParts = interrupted.parts
+  if (!interruptedParts) {
+    store.setState((state) => ({
+      message: { ...state.message, [sessionID]: interrupted.messages },
+    }))
+    return
+  }
+
+  store.setState((state) => ({
+    message: { ...state.message, [sessionID]: interrupted.messages },
+    part: { ...state.part, [interrupted.messageID]: interruptedParts },
+  }))
+}
+
+/**
+ * Re-checks a hydrated session whose trailing assistant turn is unfinished.
+ * A cold reload can hydrate messages after the initial status snapshot, so the
+ * settle decision must be repeated after the message records are available.
+ * If no per-session status exists yet, fetch one authoritative snapshot first;
+ * a successful snapshot that omits the session establishes it as idle.
+ */
+export async function recoverInterruptedTurnAfterMessageLoad(
+  directory: string,
+  store: StoreApi<DirectoryStore>,
+  sessionID: string,
+): Promise<void> {
+  const initial = store.getState()
+  if (!hasUnfinishedAssistantTurn(initial, sessionID)) return
+  if ((initial.question?.[sessionID] ?? []).length > 0) return
+  if ((initial.permission?.[sessionID] ?? []).length > 0) return
+
+  if (!initial.session_status?.[sessionID]) {
+    const snapshot = await opencodeClient.getSessionStatusForDirectory(directory)
+    if (snapshot === null) return
+
+    // Do not overwrite a live status event that arrived while the snapshot was
+    // in flight. The snapshot only fills the previously unknown state.
+    if (!store.getState().session_status?.[sessionID]) {
+      applySessionStatusSnapshot(store, snapshot, [sessionID], "authoritative")
+      applyGlobalSessionStatusSnapshot(directory, snapshot, [sessionID])
+    }
+  }
+
+  applyInterruptedTurnReconciliation(store, sessionID)
+}
+
 // ---------------------------------------------------------------------------
 // Provider
 // ---------------------------------------------------------------------------
@@ -15,6 +15,7 @@ import {
  useSyncRuntime,
  resyncBlockingRequestsForDirectory,
  buildSessionMessageRecordsSnapshot,
+  recoverInterruptedTurnAfterMessageLoad,
 } from "./sync-context"
 import { stripSessionDiffSnapshots } from "./sanitize"
 import { isVSCodeRuntime } from "@/lib/desktop"
@@ -244,7 +245,10 @@ export function useSync() {
      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) return
+      if (cachedReady && hasSession && !force) {
+        await recoverInterruptedTurnAfterMessageLoad(targetDirectory, targetStore, sessionID)
+        return
+      }
      const shouldLoadMessages = Boolean(!cachedReady || force)
      const shouldFetchSession = shouldFetchSessionForRenderableSync({ hasSession, shouldLoadMessages, force: Boolean(force) })
      const promise = (async () => {
@@ -271,10 +275,15 @@ export function useSync() {
              })()
            : Promise.resolve(),
          shouldLoadMessages
-            ? messageLoader.ensure(
-                { directory: targetDirectory, sessionID },
-                { force, reason: "reactive" },
-              )
+            ? (async () => {
+                await messageLoader.ensure(
+                  { directory: targetDirectory, sessionID },
+                  { force, reason: "reactive" },
+                )
+                if (!isStale()) {
+                  await recoverInterruptedTurnAfterMessageLoad(targetDirectory, targetStore, sessionID)
+                }
+              })()
            : Promise.resolve(),
        ])
      })()