fix(sync): settle completed turns and finished messages promptly

Two remaining stuck/incorrect busy-state edge cases from the post-#483
spinner audit (OPE-193):

- B1: when a turn ended but the session.idle SSE event was delayed or
  lost, the busy spinner kept showing until the next watchdog poll tick
  (~5s) and its escalation (~10s). An assistant message.updated that
  carries time.completed now triggers one immediate directory status
  poll (monotonic confirm, authoritative settle when the snapshot
  reports the session idle) — recovery drops to a single round-trip,
  with one in-flight fetch per directory and the watchdog poll as the
  backstop.

- C1: the streaming derivation marked the trailing assistant message as
  streaming while the session stayed busy even after the server stamped
  time.completed (whole response incl. tools finished) — the typing
  indicator and streaming part-update suspension lingered on finished
  content until the session settled or the next message started. A
  completed trailing message is now never marked streaming; both the
  full and incremental derivations complete the previous streaming
  message instead.

Refs OPE-193

packages/ui/src/sync/DOCUMENTATION.md

@@ -200,6 +200,10 @@ 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.
 
+A trailing assistant message that the server stamped `time.completed` is never marked as streaming: the stamp means the whole response (text plus every tool call) finished, so even while the session stays busy for the next step of the turn, the typing indicator and the streaming part-update suspension must not linger on finished content. The message-level streaming state (`streamingMessageIds` / `messageStreamStates`) is therefore a *message* lifecycle, not a turn lifecycle — it is completed by an explicit `time.completed`, by a newer trailing message, or by the session leaving `busy`.
+
+When an assistant `message.updated` event carries `time.completed` and the store still believes the session busy, sync fires one immediate directory status poll (`maybePollStatusAfterMessageCompletion`): the monotonic pass confirms/raises active status but never lowers it, and when the snapshot reports the session idle while the store believes it busy — a delayed or lost `session.idle` — an authoritative resync settles the status at once. This narrows the stuck-spinner window after turn completion from a full watchdog poll interval to a single round-trip; one in-flight fetch per directory bounds the fan-out and the 5s watchdog poll remains the backstop.
+
 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.

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

@@ -0,0 +1,149 @@
+/**
+ * Tests for the immediate status poll fired when an assistant message
+ * completes (issue OPE-193, B1): the busy spinner must not linger for up to a
+ * full watchdog poll interval after a turn completed when the session.idle
+ * event was delayed or lost.
+ */
+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 { INITIAL_STATE } from "../types"
+import type { DirectoryStore } from "../child-store"
+
+type StatusSnapshot = Record<string, SessionStatus | undefined>
+
+let statusSnapshotResult: StatusSnapshot | null = { ses_1: { type: "idle" } }
+let statusSnapshotErrors = 0
+const statusSnapshotCalls: string[] = []
+
+mock.module("@/lib/opencode/client", () => ({
+  opencodeClient: {
+    getSessionStatusForDirectory: mock((directory: string) => {
+      statusSnapshotCalls.push(directory)
+      if (statusSnapshotErrors > 0) {
+        statusSnapshotErrors -= 1
+        return Promise.resolve(null)
+      }
+      return Promise.resolve(statusSnapshotResult)
+    }),
+  },
+}))
+
+mock.module("@/lib/runtime-switch", () => ({
+  getRuntimeKey: () => "test-runtime",
+}))
+
+import { maybePollStatusAfterMessageCompletion } from "../sync-context"
+
+const createStore = (status: SessionStatus | undefined): StoreApi<DirectoryStore> => {
+  return create<DirectoryStore>()((set) => ({
+    ...INITIAL_STATE,
+    ...(status ? { session_status: { ses_1: status } } : {}),
+    patch: (partial) => set(partial),
+    replace: (next) => set(next),
+  }))
+}
+
+const waitForPollSettled = async (): Promise<void> => {
+  // The helper runs under the background-network concurrency gate; give the
+  // task chain (and any promise-based snapshot) real time to finish.
+  await new Promise((resolve) => setTimeout(resolve, 25))
+  await new Promise((resolve) => setTimeout(resolve, 25))
+}
+
+describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => {
+  beforeEach(() => {
+    statusSnapshotResult = { ses_1: { type: "idle" } }
+    statusSnapshotErrors = 0
+    statusSnapshotCalls.length = 0
+  })
+
+  test("does not poll when the store believes the session is already idle", async () => {
+    const store = createStore({ type: "idle" })
+
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    await waitForPollSettled()
+
+    expect(statusSnapshotCalls).toEqual([])
+    expect(store.getState().session_status?.ses_1?.type).toBe("idle")
+  })
+
+  test("does not poll without a directory or session id", async () => {
+    const store = createStore({ type: "busy" })
+
+    maybePollStatusAfterMessageCompletion("", store, "ses_1")
+    maybePollStatusAfterMessageCompletion("global", store, "ses_1")
+    maybePollStatusAfterMessageCompletion("/test/project", store, "")
+    await waitForPollSettled()
+
+    expect(statusSnapshotCalls).toEqual([])
+  })
+
+  test("settles a busy session to idle immediately when the snapshot omits it", async () => {
+    const store = createStore({ type: "busy" })
+
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    await waitForPollSettled()
+
+    expect(statusSnapshotCalls).toEqual(["/test/project", "/test/project"])
+    expect(store.getState().session_status?.ses_1?.type).toBe("idle")
+  })
+
+  test("keeps the session busy when the snapshot confirms it is still active", async () => {
+    const store = createStore({ type: "busy" })
+    statusSnapshotResult = { ses_1: { type: "busy" } }
+
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    await waitForPollSettled()
+
+    // Monotonic poll confirms busy; the snapshot is not idle, so no
+    // authoritative escalation runs.
+    expect(statusSnapshotCalls).toEqual(["/test/project"])
+    expect(store.getState().session_status?.ses_1?.type).toBe("busy")
+  })
+
+  test("preserves the busy status when the status fetch fails", async () => {
+    const store = createStore({ type: "busy" })
+    statusSnapshotErrors = 1
+
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    await waitForPollSettled()
+
+    expect(statusSnapshotCalls).toEqual(["/test/project"])
+    // Failure is not treated as authoritative empty: the busy status stays
+    // until the watchdog poll (or a live event) corrects it.
+    expect(store.getState().session_status?.ses_1?.type).toBe("busy")
+  })
+
+  test("deduplicates concurrent polls for the same directory", async () => {
+    const store = createStore({ type: "busy" })
+    statusSnapshotResult = new Promise((resolve) => {
+      setTimeout(() => resolve({ ses_1: { type: "idle" } }), 10)
+    }) as unknown as StatusSnapshot
+
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    await waitForPollSettled()
+
+    expect(statusSnapshotCalls).toEqual(["/test/project", "/test/project"])
+  })
+
+  test("does not poll again while a previous poll for the directory is in flight", async () => {
+    const store = createStore({ type: "busy" })
+    let release: () => void = () => {}
+    statusSnapshotResult = new Promise((resolve) => {
+      release = () => resolve({ ses_1: { type: "idle" } })
+    }) as unknown as StatusSnapshot
+
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1")
+    release()
+    await waitForPollSettled()
+
+    // First call ran the poll; the second call was deduped by the in-flight
+    // guard. The escalation (second fetch) is the authoritative resync.
+    expect(statusSnapshotCalls).toEqual(["/test/project", "/test/project"])
+    expect(store.getState().session_status?.ses_1?.type).toBe("idle")
+  })
+})

packages/ui/src/sync/streaming.test.ts

@@ -18,6 +18,12 @@ const message = (id: string, role: "user" | "assistant"): Message => ({
   role,
 } as unknown as Message)
 
+const completedAssistantMessage = (id: string): Message => ({
+  id,
+  role: "assistant",
+  time: { created: 1, completed: 100 },
+} as unknown as Message)
+
 const stateWithMessages = (messages: Message[], status: SessionStatus = { type: "busy" } as SessionStatus): State => ({
   ...INITIAL_STATE,
   session_status: {
@@ -163,4 +169,72 @@ describe("updateStreamingState", () => {
     expect(streaming.messageStreamStates.get("msg_assistant_1")?.phase).toBe("completed")
     expect(streaming.messageStreamStates.get("msg_assistant_2")?.phase).toBe("streaming")
   })
+
+  test("completes a streaming message when the trailing assistant message finishes while the session stays busy", () => {
+    updateStreamingState(stateWithMessages([
+      message("msg_user_1", "user"),
+      message("msg_assistant_1", "assistant"),
+    ]))
+    expect(useStreamingStore.getState().streamingMessageIds.get("ses_1")).toBe("msg_assistant_1")
+
+    // The message completed (time.completed) but the turn keeps running
+    // (next step / tool phase) — the finished message must not stay marked
+    // as streaming with the typing indicator and part-update suspension on it.
+    updateStreamingState(stateWithMessages([
+      message("msg_user_1", "user"),
+      completedAssistantMessage("msg_assistant_1"),
+    ]))
+
+    const streaming = useStreamingStore.getState()
+    expect(streaming.streamingMessageIds.get("ses_1")).toBeNull()
+    expect(streaming.messageStreamStates.get("msg_assistant_1")?.phase).toBe("completed")
+  })
+
+  test("does not mark an already-completed trailing assistant message as streaming", () => {
+    updateStreamingState(stateWithMessages([
+      message("msg_user_1", "user"),
+      completedAssistantMessage("msg_assistant_1"),
+    ]))
+
+    const streaming = useStreamingStore.getState()
+    expect(streaming.streamingMessageIds.get("ses_1") ?? null).toBeNull()
+    expect(streaming.messageStreamStates.has("msg_assistant_1")).toBe(false)
+  })
+
+  test("incrementally clears the streaming marker when the trailing message completes while busy", () => {
+    const previous = stateWithMessages([
+      message("msg_user_1", "user"),
+      message("msg_assistant_1", "assistant"),
+    ])
+    updateStreamingState(previous, 10)
+    expect(useStreamingStore.getState().streamingMessageIds.get("ses_1")).toBe("msg_assistant_1")
+
+    const next = stateWithMessages([
+      message("msg_user_1", "user"),
+      completedAssistantMessage("msg_assistant_1"),
+    ])
+    updateChangedStreamingSessions(next, previous, 20)
+
+    const streaming = useStreamingStore.getState()
+    expect(streaming.streamingMessageIds.get("ses_1")).toBeNull()
+    expect(streaming.messageStreamStates.get("msg_assistant_1")?.phase).toBe("completed")
+  })
+
+  test("keeps the next assistant message streaming after an intermediate message completed while busy", () => {
+    const previous = stateWithMessages([
+      message("msg_user_1", "user"),
+      completedAssistantMessage("msg_assistant_1"),
+    ])
+    updateStreamingState(previous, 10)
+    expect(useStreamingStore.getState().streamingMessageIds.get("ses_1") ?? null).toBeNull()
+
+    const next = stateWithMessages([
+      message("msg_user_1", "user"),
+      completedAssistantMessage("msg_assistant_1"),
+      message("msg_assistant_2", "assistant"),
+    ])
+    updateChangedStreamingSessions(next, previous, 20)
+
+    expect(useStreamingStore.getState().streamingMessageIds.get("ses_1")).toBe("msg_assistant_2")
+  })
 })

packages/ui/src/sync/streaming.ts

@@ -58,6 +58,18 @@ const findTrailingAssistantMessage = (messages: Message[] | undefined): Message
   return null
 }
 
+/**
+ * The server stamps `time.completed` on an assistant message only after its
+ * whole response (text + every tool call) finished. A completed trailing
+ * message therefore means the message itself is done even when the turn keeps
+ * running (next step, follow-up tool phase) — it must not stay marked as
+ * streaming, or the typing indicator and the part-update suspension linger on
+ * finished content until the session settles.
+ */
+const isTrailingMessageComplete = (message: Message): boolean => {
+  return typeof (message as { time?: { completed?: unknown } }).time?.completed === "number"
+}
+
 export function updateStreamingState(state: State, now = Date.now()) {
   countSyncPerformance("streamingFullReconciliations")
   const currentStore = useStreamingStore.getState()
@@ -108,6 +120,18 @@ export function updateStreamingState(state: State, now = Date.now()) {
       continue
     }
 
+    // The trailing assistant message already finished (time.completed), so
+    // nothing is streaming right now even though the session stays busy for
+    // the rest of the turn. Complete any previously streaming message instead
+    // of re-marking the finished one as streaming.
+    if (isTrailingMessageComplete(streamingMsg)) {
+      const prevId = currentStreamingIds.get(sessionID)
+      if (prevId) {
+        completeStreamingMessage(sessionID, prevId)
+      }
+      continue
+    }
+
     const prevId = currentStreamingIds.get(sessionID)
     if (prevId !== streamingMsg.id) changed = true
     nextStreamingIds.set(sessionID, streamingMsg.id)
@@ -222,6 +246,14 @@ export function updateChangedStreamingSessions(state: State, previous: State, no
       continue
     }
 
+    // Completed trailing message while the turn keeps running: nothing is
+    // streaming — clear the marker and any previous streaming message instead
+    // of keeping the finished message flagged as streaming.
+    if (isTrailingMessageComplete(streamingMessage)) {
+      if (previousMessageID) complete(sessionID, previousMessageID)
+      continue
+    }
+
     if (previousMessageID && previousMessageID !== streamingMessage.id) {
       complete(sessionID, previousMessageID)
     }

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

@@ -290,6 +290,11 @@ type PendingSessionMaterialization = {
 const SESSION_MATERIALIZATION_COOLDOWN_MS = 5_000
 const pendingSessionMaterializations = new Map<string, PendingSessionMaterialization>()
 
+// In-flight guard for the immediate status poll fired when an assistant
+// message completes (see maybePollStatusAfterMessageCompletion). Keyed by
+// directory so a burst of completing messages shares one status fetch.
+const messageCompletionStatusPolls = new Set<string>()
+
 function enqueueSessionMaterialization(
   directory: string,
   sessionID: string,
@@ -632,6 +637,49 @@ async function resyncDirectorySessionStatuses(
   return nextStatuses
 }
 
+/**
+ * Immediately re-check the session status after an assistant message
+ * completes. The turn-ending `session.idle` event can be delayed or lost; left
+ * alone, the busy spinner keeps showing until the next watchdog poll tick
+ * (up to ~5s) and its escalation (up to ~10s). One cheap status fetch right
+ * after the completion confirms the turn really ended, mirroring the watchdog
+ * escalation: the monotonic pass confirms/raises busy but never lowers it, and
+ * when the snapshot reports the session idle while the store still believes it
+ * busy, an authoritative resync settles the status immediately.
+ *
+ * Bounded: one in-flight fetch per directory, only for sessions the store
+ * currently believes active, best-effort (the watchdog poll remains the
+ * backstop). This narrows recovery latency without restructuring the polling
+ * design.
+ */
+export function maybePollStatusAfterMessageCompletion(
+  directory: string,
+  store: StoreApi<DirectoryStore>,
+  sessionID: string,
+): void {
+  if (!directory || directory === "global" || !sessionID) return
+  const current = store.getState().session_status?.[sessionID]
+  if (!current || current.type === "idle") return
+  if (messageCompletionStatusPolls.has(directory)) return
+
+  messageCompletionStatusPolls.add(directory)
+  void (async () => {
+    try {
+      const statuses = await runBackgroundNetworkTask(() =>
+        resyncDirectorySessionStatuses(directory, store, [sessionID], "monotonic"))
+      if (!statuses) return
+      if (needsSnapshotAfterStatusPoll(store.getState(), sessionID, statuses[sessionID])) {
+        await runBackgroundNetworkTask(() =>
+          resyncDirectorySessionStatuses(directory, store, [sessionID], "authoritative"))
+      }
+    } catch {
+      // Best-effort — the watchdog poll retries on its own cadence.
+    } finally {
+      messageCompletionStatusPolls.delete(directory)
+    }
+  })()
+}
+
 // After a monotonic poll, decide whether to escalate to a full authoritative
 // resync: the store believes the session is active but the snapshot reports it
 // idle/absent — a suspected missed idle that the monotonic poll deliberately
@@ -1722,6 +1770,12 @@ function handleEvent(
           messageID,
         })
       }
+      // An assistant message that finished is strong evidence the turn may
+      // have ended; if the session.idle event was delayed or lost, settle the
+      // busy status immediately instead of waiting for the next watchdog poll.
+      if (info.role === "assistant" && typeof info.time?.completed === "number") {
+        maybePollStatusAfterMessageCompletion(resolvedDirectory, store, sessionID)
+      }
     }
   } else {
     const sessionID = getSessionIdFromPayload(payload) ?? undefined