From 498a029e51c7d396e5d4fa987efd3f28ca641d34 Mon Sep 17 00:00:00 2001 From: Serhii Dziupin Date: Wed, 5 Aug 2026 11:57:12 +0300 Subject: [PATCH 1/2] fix(sync): settle completed turns and finished messages promptly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 4 + .../message-completion-status-poll.test.ts | 149 ++++++++++++++++++ packages/ui/src/sync/streaming.test.ts | 74 +++++++++ packages/ui/src/sync/streaming.ts | 32 ++++ packages/ui/src/sync/sync-context.tsx | 54 +++++++ 5 files changed, 313 insertions(+) create mode 100644 packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts diff --git a/packages/ui/src/sync/DOCUMENTATION.md b/packages/ui/src/sync/DOCUMENTATION.md index 5699fc93..f05573d6 100644 --- a/packages/ui/src/sync/DOCUMENTATION.md +++ b/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. diff --git a/packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts b/packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts new file mode 100644 index 00000000..e40d3065 --- /dev/null +++ b/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 + +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 => { + return create()((set) => ({ + ...INITIAL_STATE, + ...(status ? { session_status: { ses_1: status } } : {}), + patch: (partial) => set(partial), + replace: (next) => set(next), + })) +} + +const waitForPollSettled = async (): Promise => { + // 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") + }) +}) diff --git a/packages/ui/src/sync/streaming.test.ts b/packages/ui/src/sync/streaming.test.ts index 326f07d6..5db05915 100644 --- a/packages/ui/src/sync/streaming.test.ts +++ b/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") + }) }) diff --git a/packages/ui/src/sync/streaming.ts b/packages/ui/src/sync/streaming.ts index ab62ac99..3c396292 100644 --- a/packages/ui/src/sync/streaming.ts +++ b/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) } diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index b4ff6c19..909c798a 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -290,6 +290,11 @@ type PendingSessionMaterialization = { const SESSION_MATERIALIZATION_COOLDOWN_MS = 5_000 const pendingSessionMaterializations = new Map() +// 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() + 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, + 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 From 10634c451263b450219ead3003debdf491bd36ed Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Sat, 29 Aug 2026 00:29:42 +0300 Subject: [PATCH 2/2] fix(sync): settle completed turns and finished messages promptly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A lost or delayed turn-ending `session.idle` left the busy spinner up until the watchdog poll caught it (5-10s). An assistant `message.updated` carrying `time.completed` now schedules one status check for that session, and `streaming.ts` stops treating a completed trailing message as streaming. The check is deferred by 750ms and re-reads the session status when the timer fires, so the overwhelmingly common case — the turn's own `session.idle` arriving right behind the completed message — settles on its own and costs zero extra requests; only a session the store still believes busy spends a fetch. The poll shares the watchdog's in-flight directory guard, so the deferred check and the periodic poll cannot overlap on one directory. Status authority is unchanged: the monotonic pass never lowers status, and an authoritative resync runs only when the snapshot disagrees. --- packages/ui/src/sync/DOCUMENTATION.md | 2 +- .../message-completion-status-poll.test.ts | 80 ++++++++--------- packages/ui/src/sync/streaming.test.ts | 12 +-- packages/ui/src/sync/streaming.ts | 2 +- packages/ui/src/sync/sync-context.tsx | 90 +++++++++++-------- 5 files changed, 100 insertions(+), 86 deletions(-) diff --git a/packages/ui/src/sync/DOCUMENTATION.md b/packages/ui/src/sync/DOCUMENTATION.md index b2bee68f..59308c5f 100644 --- a/packages/ui/src/sync/DOCUMENTATION.md +++ b/packages/ui/src/sync/DOCUMENTATION.md @@ -222,7 +222,7 @@ Streaming lifecycle derivation has two paths. Directory attach, switch, bootstra 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. +When an assistant `message.updated` event carries `time.completed` and the store still believes the session busy, sync schedules one deferred status check (`maybePollStatusAfterMessageCompletion`, ~750ms). The status is re-read when the timer fires, so a normal turn whose `session.idle` lands inside that window issues no request at all; only a still-busy session spends a directory status poll, sharing the watchdog's one-in-flight-per-directory guard. The invariant is unchanged from the watchdog escalation: the monotonic pass confirms or raises active status and never lowers it, and an authoritative resync runs only when the snapshot disagrees with a store that still believes the session busy. This narrows the stuck-spinner window after a lost `session.idle` from a watchdog interval to one round-trip; 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. diff --git a/packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts b/packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts index e40d3065..09aa3d58 100644 --- a/packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts +++ b/packages/ui/src/sync/__tests__/message-completion-status-poll.test.ts @@ -1,8 +1,9 @@ /** - * 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. + * Tests for the deferred status poll fired when an assistant message completes + * (issue OPE-193): 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 — and a normal turn, whose session.idle arrives promptly, must not + * cost a single extra request. */ import { beforeEach, describe, expect, mock, test } from "bun:test" import { create, type StoreApi } from "zustand" @@ -12,19 +13,14 @@ import type { DirectoryStore } from "../child-store" type StatusSnapshot = Record -let statusSnapshotResult: StatusSnapshot | null = { ses_1: { type: "idle" } } -let statusSnapshotErrors = 0 +let respondWithSnapshot: () => Promise = () => Promise.resolve({ ses_1: { type: "idle" } }) 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) + return respondWithSnapshot() }), }, })) @@ -33,28 +29,28 @@ mock.module("@/lib/runtime-switch", () => ({ getRuntimeKey: () => "test-runtime", })) -import { maybePollStatusAfterMessageCompletion } from "../sync-context" +import { maybePollStatusAfterMessageCompletion, MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS } from "../sync-context" -const createStore = (status: SessionStatus | undefined): StoreApi => { +const createStore = (status: SessionStatus): StoreApi => { return create()((set) => ({ ...INITIAL_STATE, - ...(status ? { session_status: { ses_1: status } } : {}), + session_status: { ses_1: status }, patch: (partial) => set(partial), replace: (next) => set(next), })) } +const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)) + +/** Past the deferral, plus room for the background-network task chain. */ const waitForPollSettled = async (): Promise => { - // 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)) + await sleep(MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS + 50) + await sleep(50) } describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => { beforeEach(() => { - statusSnapshotResult = { ses_1: { type: "idle" } } - statusSnapshotErrors = 0 + respondWithSnapshot = () => Promise.resolve({ ses_1: { type: "idle" } }) statusSnapshotCalls.length = 0 }) @@ -79,10 +75,25 @@ describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => { expect(statusSnapshotCalls).toEqual([]) }) - test("settles a busy session to idle immediately when the snapshot omits it", async () => { + test("issues no request when session.idle arrives inside the deferral window", async () => { const store = createStore({ type: "busy" }) maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1") + // The turn's own session.idle event lands well before the timer fires. + await sleep(50) + store.getState().patch({ session_status: { ses_1: { type: "idle" } } }) + await waitForPollSettled() + + expect(statusSnapshotCalls).toEqual([]) + }) + + test("settles a busy session to idle when the idle event never arrives", async () => { + const store = createStore({ type: "busy" }) + + maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1") + // Nothing settles the session inside the window; the poll must run. + expect(statusSnapshotCalls).toEqual([]) + await waitForPollSettled() expect(statusSnapshotCalls).toEqual(["/test/project", "/test/project"]) @@ -91,7 +102,7 @@ describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => { test("keeps the session busy when the snapshot confirms it is still active", async () => { const store = createStore({ type: "busy" }) - statusSnapshotResult = { ses_1: { type: "busy" } } + respondWithSnapshot = () => Promise.resolve({ ses_1: { type: "busy" } }) maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1") await waitForPollSettled() @@ -104,7 +115,7 @@ describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => { test("preserves the busy status when the status fetch fails", async () => { const store = createStore({ type: "busy" }) - statusSnapshotErrors = 1 + respondWithSnapshot = () => Promise.resolve(null) maybePollStatusAfterMessageCompletion("/test/project", store, "ses_1") await waitForPollSettled() @@ -115,34 +126,15 @@ describe("maybePollStatusAfterMessageCompletion (issue OPE-193)", () => { expect(store.getState().session_status?.ses_1?.type).toBe("busy") }) - test("deduplicates concurrent polls for the same directory", async () => { + test("schedules one check for a burst of completions on the same session", 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. + // One monotonic poll plus its authoritative escalation, not three. expect(statusSnapshotCalls).toEqual(["/test/project", "/test/project"]) expect(store.getState().session_status?.ses_1?.type).toBe("idle") }) diff --git a/packages/ui/src/sync/streaming.test.ts b/packages/ui/src/sync/streaming.test.ts index 5db05915..0062b65c 100644 --- a/packages/ui/src/sync/streaming.test.ts +++ b/packages/ui/src/sync/streaming.test.ts @@ -16,13 +16,15 @@ import { const message = (id: string, role: "user" | "assistant"): Message => ({ id, role, + time: { created: 1 }, } as unknown as Message) -const completedAssistantMessage = (id: string): Message => ({ - id, - role: "assistant", - time: { created: 1, completed: 100 }, -} as unknown as Message) +const completedAssistantMessage = (id: string): Message => { + const base = message(id, "assistant") + // SAFETY: test fixture — the streaming reducers read only `id`, `role`, and + // `time.completed`, which this literal provides. + return { ...base, time: { created: 1, completed: 100 } } as Message +} const stateWithMessages = (messages: Message[], status: SessionStatus = { type: "busy" } as SessionStatus): State => ({ ...INITIAL_STATE, diff --git a/packages/ui/src/sync/streaming.ts b/packages/ui/src/sync/streaming.ts index 3c396292..5ca44ca3 100644 --- a/packages/ui/src/sync/streaming.ts +++ b/packages/ui/src/sync/streaming.ts @@ -67,7 +67,7 @@ const findTrailingAssistantMessage = (messages: Message[] | undefined): Message * finished content until the session settles. */ const isTrailingMessageComplete = (message: Message): boolean => { - return typeof (message as { time?: { completed?: unknown } }).time?.completed === "number" + return message.role === "assistant" && message.time.completed !== undefined } export function updateStreamingState(state: State, now = Date.now()) { diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index b316d1ac..2b6215d2 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -341,10 +341,17 @@ type PendingSessionMaterialization = { const SESSION_MATERIALIZATION_COOLDOWN_MS = 5_000 const pendingSessionMaterializations = new Map() -// 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() +// One in-flight directory status fetch at a time, shared by the active-session +// watchdog poll and the deferred completion poll so the two cannot overlap on +// the same directory. +const statusPollingDirectories = new Set() + +// Deferred completion polls awaiting their delay, keyed by directory+session so +// a burst of completing messages schedules one check. +const pendingMessageCompletionPolls = new Map>() + +// How long to wait for the turn's own `session.idle` before spending a request. +export const MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS = 750 function enqueueSessionMaterialization( directory: string, @@ -737,19 +744,22 @@ async function resyncDirectorySessionStatuses( } /** - * 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. + * Re-check the session status shortly 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). * - * 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. + * The check is deferred by `MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS`, and the + * status is read again when the timer fires: a normal turn whose `session.idle` + * arrives inside that window settles on its own and issues no request at all. + * Only a session the store still believes busy costs one status fetch, which + * mirrors 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. + * + * Bounded: one scheduled check per session, one in-flight status fetch per + * directory (shared with the watchdog poll), best-effort — the watchdog poll + * remains the backstop. */ export function maybePollStatusAfterMessageCompletion( directory: string, @@ -759,24 +769,35 @@ export function maybePollStatusAfterMessageCompletion( 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")) + const pendingKey = `${directory}\u0000${sessionID}` + if (pendingMessageCompletionPolls.has(pendingKey)) return + + const timer = setTimeout(() => { + pendingMessageCompletionPolls.delete(pendingKey) + const latest = store.getState().session_status?.[sessionID] + if (!latest || latest.type === "idle") return + if (statusPollingDirectories.has(directory)) return + + statusPollingDirectories.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 { + statusPollingDirectories.delete(directory) } - } catch { - // Best-effort — the watchdog poll retries on its own cadence. - } finally { - messageCompletionStatusPolls.delete(directory) - } - })() + })() + }, MESSAGE_COMPLETION_STATUS_POLL_DELAY_MS) + + pendingMessageCompletionPolls.set(pendingKey, timer) } // After a monotonic poll, decide whether to escalate to a full authoritative @@ -2123,7 +2144,6 @@ export function SyncProvider(props: { const lastChildDiscoveryAtByDirectoryRef = useRef(new Map()) const resyncingDirectoriesRef = useRef(new Set()) const blockingRequestResyncingDirectoriesRef = useRef(new Set()) - const statusPollingDirectoriesRef = useRef(new Set()) const pipelineReconnectRef = useRef<((reason?: string) => void) | null>(null) const pipelineHasConnectedRef = useRef(false) const pipelineDisconnectedBeforeFirstConnectRef = useRef(false) @@ -2486,7 +2506,7 @@ export function SyncProvider(props: { store: StoreApi, candidateSessionIds: string[], ) => { - const polling = statusPollingDirectoriesRef.current + const polling = statusPollingDirectories if (polling.has(directory)) return polling.add(directory) try { @@ -2544,7 +2564,7 @@ export function SyncProvider(props: { .finally(() => { running = false if (stopped) { - statusPollingDirectoriesRef.current.clear() + statusPollingDirectories.clear() } }) }