From 9977b25540e828b7a6ded38699d0a9e454ae6e39 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Thu, 30 Jul 2026 00:28:46 +0300 Subject: [PATCH] fix(sync): reconcile tools left running after settlement --- packages/ui/src/sync/DOCUMENTATION.md | 4 +- .../sync/__tests__/materialization.test.ts | 78 +++++++++++++++++++ packages/ui/src/sync/event-reducer.ts | 1 + packages/ui/src/sync/materialization.ts | 38 +++++++++ packages/ui/src/sync/sync-context.tsx | 19 ++++- 5 files changed, 138 insertions(+), 2 deletions(-) diff --git a/packages/ui/src/sync/DOCUMENTATION.md b/packages/ui/src/sync/DOCUMENTATION.md index c72ebdfb..710127c5 100644 --- a/packages/ui/src/sync/DOCUMENTATION.md +++ b/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. diff --git a/packages/ui/src/sync/__tests__/materialization.test.ts b/packages/ui/src/sync/__tests__/materialization.test.ts index d830c791..cb7ce546 100644 --- a/packages/ui/src/sync/__tests__/materialization.test.ts +++ b/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) + }) }) diff --git a/packages/ui/src/sync/event-reducer.ts b/packages/ui/src/sync/event-reducer.ts index ffcaa0d0..3d7833c4 100644 --- a/packages/ui/src/sync/event-reducer.ts +++ b/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 diff --git a/packages/ui/src/sync/materialization.ts b/packages/ui/src/sync/materialization.ts index 122418b9..6ec52f26 100644 --- a/packages/ui/src/sync/materialization.ts +++ b/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) { 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) { diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index 69152a46..5a5bb100 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/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) }