fix(sync): reconcile tools left running after settlement

packages/ui/src/sync/DOCUMENTATION.md

@@ -185,7 +185,9 @@ The event pipeline delivers each ordered per-directory flush as one reducer batc
 
 Streaming lifecycle derivation has two paths. Directory attach, switch, bootstrap, and reconnect may perform a full reconciliation. Normal store publications reconcile only sessions whose `session_status` or `message` bucket changed; part-only events update the affected streaming message heartbeat directly and must not rescan all busy sessions.
 
-Incomplete-session materialization is deduplicated by runtime, directory, and session for the full cooldown window, including after a fast success or failure. Deferred recovery is dropped if its captured runtime is no longer active. If recovery requests a tail refresh while an older load is in flight, one refresh runs after that load instead of losing the newer authority demand. Completion retains the cooldown marker until expiry, and an older completion cannot clear a newer request marker. Recovery starts after the current ordered event batch and rechecks whether local state already contains the requested entity before starting HTTP. An explicit empty part bucket is authoritative fetched-empty state, not a missing snapshot. This prevents repeated orphan/missing-part events from creating message-tail and status request storms while preserving later recovery.
+Incomplete-session materialization is deduplicated by runtime, directory, and session for the full cooldown window, including after a fast success or failure. A settled-running-tool recovery may supersede a different request in that window so an earlier pre-settlement refresh cannot consume the only terminal recovery signal. Deferred recovery is dropped if its captured runtime is no longer active. If recovery requests a tail refresh while an older load is in flight, one refresh runs after that load instead of losing the newer authority demand. Completion retains the cooldown marker until expiry, and an older completion cannot clear a newer request marker. Recovery starts after the current ordered event batch and rechecks whether local state already contains the requested entity before starting HTTP. An explicit empty part bucket is authoritative fetched-empty state, not a missing snapshot. This prevents repeated orphan/missing-part events from creating message-tail and status request storms while preserving later recovery.
+
+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.
 
 Directory stores also own session-keyed sidecar notification channels for permissions and message materialization. High-frequency realtime part events annotate the exact session/message before committing, so visible records, user history, renderability, and sidebar permission 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__/materialization.test.ts

@@ -3,6 +3,7 @@ import type { Message, Part } from "@opencode-ai/sdk/v2/client"
 import {
   getSessionMaterializationRequestKey,
   getSessionMaterializationStatus,
+  getStaleRunningToolMessageID,
   isSessionMaterializationStillNeeded,
   materializeSessionSnapshots,
 } from "../materialization"
@@ -166,6 +167,37 @@ describe("materializeSessionSnapshots", () => {
     expect(mergedPart.state?.time?.end).toBe(2000)
   })
 
+  test("does not regress a completed tool when a stale running snapshot arrives", () => {
+    const completedTool = {
+      id: "prt_1",
+      messageID: "msg_1",
+      sessionID: "ses_1",
+      type: "tool",
+      state: { status: "completed", output: "done", time: { start: 1000, end: 2000 } },
+    } as unknown as Part
+    const staleRunningTool = {
+      id: "prt_1",
+      messageID: "msg_1",
+      sessionID: "ses_1",
+      type: "tool",
+      state: { status: "running", time: { start: 1000 } },
+    } as unknown as Part
+    const state = {
+      message: { ses_1: [message("msg_1")] },
+      part: { msg_1: [completedTool] },
+    }
+
+    const result = materializeSessionSnapshots(
+      state,
+      "ses_1",
+      [{ info: message("msg_1"), parts: [staleRunningTool] }],
+    )
+
+    expect(result.part).toBe(state.part)
+    expect(result.part.msg_1[0]).toBe(completedTool)
+    expect(getStaleRunningToolMessageID(result, "ses_1")).toBe(undefined)
+  })
+
   test("preserves state.attachments from existing part when completed snapshot lacks them", () => {
     const livePart = {
       id: "prt_1",
@@ -459,4 +491,50 @@ describe("isSessionMaterializationStillNeeded", () => {
       partID: "prt_1",
     })).toBe(true)
   })
+
+  test("recovers a settled session whose trailing assistant still has a running tool", () => {
+    const runningTool = {
+      id: "prt_1",
+      messageID: "msg_1",
+      sessionID: "ses_1",
+      type: "tool",
+      tool: "read",
+      state: { status: "running" },
+    } as Part
+    const state = { message: { ses_1: [message("msg_1")] }, part: { msg_1: [runningTool] } }
+
+    expect(getStaleRunningToolMessageID(state, "ses_1")).toBe("msg_1")
+    expect(isSessionMaterializationStillNeeded(state, "ses_1", {
+      reason: "settled-running-tool",
+      messageID: "msg_1",
+    })).toBe(true)
+
+    const completedState = {
+      ...state,
+      part: { msg_1: [{ ...runningTool, state: { status: "completed" } } as Part] },
+    }
+    expect(getStaleRunningToolMessageID(completedState, "ses_1")).toBe(undefined)
+    expect(isSessionMaterializationStillNeeded(completedState, "ses_1", {
+      reason: "settled-running-tool",
+      messageID: "msg_1",
+    })).toBe(false)
+  })
+
+  test("does not recover an older running tool after a newer user turn", () => {
+    const state = {
+      message: { ses_1: [message("msg_1"), userMessage("msg_2")] },
+      part: {
+        msg_1: [{
+          id: "prt_1",
+          messageID: "msg_1",
+          sessionID: "ses_1",
+          type: "tool",
+          tool: "read",
+          state: { status: "running" },
+        } as Part],
+      },
+    }
+
+    expect(getStaleRunningToolMessageID(state, "ses_1")).toBe(undefined)
+  })
 })

packages/ui/src/sync/event-reducer.ts

@@ -163,6 +163,7 @@ export type SessionMaterializationReason =
   | "stream-reconnect"
   | "transport-switch"
   | "stale-status-resync"
+  | "settled-running-tool"
 
 export type DirectoryEventResult = boolean | {
   changed: boolean

packages/ui/src/sync/materialization.ts

@@ -4,6 +4,8 @@ import type { SessionMaterializationReason } from "./event-reducer"
 
 const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0)
 const STREAMING_PART_FIELDS = ["text", "output"] as const
+const ACTIVE_TOOL_STATUSES = new Set(["pending", "running"])
+const FINAL_TOOL_STATUSES = new Set(["completed", "error", "aborted", "failed", "timeout", "cancelled"])
 
 export type MaterializedMessageRecord = {
   info: Message
@@ -65,9 +67,32 @@ export function isSessionMaterializationStillNeeded(
     return !(state.part[request.messageID] ?? []).some((part) => part.id === request.partID)
   }
 
+  if (request.reason === "settled-running-tool") {
+    return getStaleRunningToolMessageID(state, sessionID) === request.messageID
+  }
+
   return true
 }
 
+export function getStaleRunningToolMessageID(
+  state: MaterializedState,
+  sessionID: string,
+): string | undefined {
+  const messages = state.message[sessionID] ?? []
+  for (let index = messages.length - 1; index >= 0; index -= 1) {
+    const message = messages[index]
+    if (message.role === "user") return undefined
+    if (message.role !== "assistant") continue
+    const hasActiveTool = (state.part[message.id] ?? []).some((part) => {
+      if (part.type !== "tool") return false
+      const status = (part as { state?: { status?: unknown } }).state?.status
+      return typeof status === "string" && ACTIVE_TOOL_STATUSES.has(status)
+    })
+    return hasActiveTool ? message.id : undefined
+  }
+  return undefined
+}
+
 function sortParts(parts: Part[], skipPartTypes: ReadonlySet<string>) {
   return parts
     .filter((part) => !!part?.id && !skipPartTypes.has(part.type))
@@ -135,6 +160,19 @@ function getPartStateTime(part: Part): { start?: number; end?: number } | undefi
 function mergeMaterializedPart(existing: Part | undefined, next: Part): Part {
   if (!existing) return next
 
+  if (existing.type === "tool" && next.type === "tool") {
+    const existingStatus = (existing as { state?: { status?: unknown } }).state?.status
+    const nextStatus = (next as { state?: { status?: unknown } }).state?.status
+    if (
+      typeof existingStatus === "string"
+      && FINAL_TOOL_STATUSES.has(existingStatus)
+      && typeof nextStatus === "string"
+      && ACTIVE_TOOL_STATUSES.has(nextStatus)
+    ) {
+      return existing
+    }
+  }
+
   if (getPartEndTime(next) !== undefined) {
     const existingAttachments = getPartStateAttachments(existing)
     if (existingAttachments?.length && getPartStateAttachments(next) === undefined) {

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

@@ -53,6 +53,7 @@ import type { QuestionRequest } from "@/types/question"
 import {
   getSessionMaterializationRequestKey,
   getSessionMaterializationStatus,
+  getStaleRunningToolMessageID,
   isSessionMaterializationStillNeeded,
   type SessionMaterializationRequest,
 } from "./materialization"
@@ -284,7 +285,11 @@ function enqueueSessionMaterialization(
   const runtimeKey = getRuntimeKey()
   const k = getSessionMaterializationRequestKey(runtimeKey, directory, sessionID)
   const existing = pendingSessionMaterializations.get(k)
-  if (existing && Date.now() - existing.enqueuedAt < SESSION_MATERIALIZATION_COOLDOWN_MS) return
+  if (existing && Date.now() - existing.enqueuedAt < SESSION_MATERIALIZATION_COOLDOWN_MS) {
+    const settlementMustFollowEarlierRecovery = request.reason === "settled-running-tool"
+      && existing.request.reason !== "settled-running-tool"
+    if (!settlementMustFollowEarlierRecovery) return
+  }
 
   const pending = { runtimeKey, sessionID, directory, enqueuedAt: Date.now(), request }
   pendingSessionMaterializations.set(k, pending)
@@ -1728,6 +1733,18 @@ function handleEvent(
     }
   }
 
+  if (payload.type === "session.idle" || payload.type === "session.error") {
+    const sessionID = getSessionIdFromPayload(payload) ?? undefined
+    const state = getDirectoryEventState(store, batch)
+    const messageID = sessionID ? getStaleRunningToolMessageID(state, sessionID) : undefined
+    if (sessionID && messageID) {
+      enqueueSessionMaterialization(resolvedDirectory, sessionID, childStores, {
+        reason: "settled-running-tool",
+        messageID,
+      })
+    }
+  }
+
   updateRoutingIndexFromEvent(routingIndex, resolvedDirectory, payload)
 }