From 10238cabb5f644883feaa6d461259b2858db559c Mon Sep 17 00:00:00 2001 From: heunghingwan Date: Wed, 4 Nov 2026 09:15:49 +0800 Subject: [PATCH 1/6] perf(transcript): coalesce streaming projections to one pass per frame Every native transcript event reduced synchronously and then projected into the message store in the same task, so a reply delivered as one burst blocked the renderer for the whole projection series: a 400-delta reply measured a single ~1.1s long task at ~2.8-4ms per delta, freezing input on local and remote transcripts alike (#1017 explicitly left frame-level batching as follow-up work). Events still reduce immediately and in order through the SDK reducer; only the projection into the store and the re-render it causes moves to the frame level. stores/opencode-projection-frame.ts runs at most one projection per animation frame per session, always from the reducer's newest state: same-frame schedules merge, the last event decides pending-send confirmation, and usage updates accumulate. Hidden windows fall back to a bounded 250ms timer (they pause animation frames), and environments without a frame scheduler keep the previous synchronous projection. Store readers lag at most one frame. Ordering and lifetime edges: a session selection write and a visibility transition flush pending pages first; transcript eviction and instance teardown cancel them so no later frame resurrects a destroyed store; destructive events (session.deleted, inbox cancellation, revert commit) and forced authoritative publications still project synchronously and subsume the pending frame; an event whose session stopped being active at frame time invalidates the loaded page so the next activation reads authoritatively. The reference-anchored projection-skip prohibition and hydrateMessages record reuse are untouched. Covered by opencode-projection-frame.test.ts (manual frame clock) and tests/browser/streaming-projection.test.ts (real SessionView with a 400-delta burst: projections 400 -> 1, burst dispatch ~1094ms -> ~25ms, no long task); compaction-responsiveness settles the compaction row's creation frame before resetting its counters. --- dev-docs/CACHE_REFRESH_CONVENTIONS.md | 2 +- dev-docs/TRANSCRIPT.md | 1 + packages/ui/src/stores/instances.ts | 63 +++++-- .../stores/opencode-projection-frame.test.ts | 178 ++++++++++++++++++ .../src/stores/opencode-projection-frame.ts | 160 ++++++++++++++++ packages/ui/src/stores/session-state.ts | 4 + .../src/stores/session-transcript-memory.ts | 2 + .../browser/compaction-responsiveness.test.ts | 10 +- .../browser/fixtures/streaming-projection.tsx | 122 ++++++++++++ .../browser/streaming-projection.test.ts | 88 +++++++++ 10 files changed, 615 insertions(+), 15 deletions(-) create mode 100644 packages/ui/src/stores/opencode-projection-frame.test.ts create mode 100644 packages/ui/src/stores/opencode-projection-frame.ts create mode 100644 packages/ui/tests/browser/fixtures/streaming-projection.tsx create mode 100644 packages/ui/tests/browser/streaming-projection.test.ts diff --git a/dev-docs/CACHE_REFRESH_CONVENTIONS.md b/dev-docs/CACHE_REFRESH_CONVENTIONS.md index 370997f9b9..ced7237810 100644 --- a/dev-docs/CACHE_REFRESH_CONVENTIONS.md +++ b/dev-docs/CACHE_REFRESH_CONVENTIONS.md @@ -48,7 +48,7 @@ content before promoting it, and retain the last good display on expiry. | Worktree display | Last successful server snapshot; demand-driven refresh after 10 s or invalidation; one scan per workspace; obsolete scans discarded and followed by validation; UI display requests coalesce with one trailing demand. Pending recovery can explicitly request `?fresh=true` once after a missing/not-owned inventory-only candidate. Concurrent fresh callers share their fresh read, never a cached completion. Fresh and display outcomes are independent on one serialized read/publication lane: a failed fresh read rejects its callers without dropping queued display demand or leaking freshness into later requests. | server `workspaces/worktree-inventory.ts`, UI `stores/worktrees.ts` | | Worktree authority | Validated reads await stale-inventory revalidation; family transactions force scans. Ownership misses can bypass a warm inventory once per directory-cache lifetime. Create/remove requires a validated next display read. | server `workspaces/worktree-directory.ts`, `manager.ts` | | Git ownership preflight | Common-directory reads resolve the enclosing checkout's administrative files (`.git`, `commondir`, `HEAD`/`objects`) directly on every call; concurrent reads share one pending read per directory. Environment overrides, paths inside `.git` and invalid layouts defer to Git, whose commands run with two process slots in a lazy worker, keeping Windows process creation off the HTTP/SSE thread. Completed/failed reads are removed immediately; a matching identity still requires native inventory validation. | server `workspaces/git-common-directory.ts`, `git-process.ts`, `git-worktrees.ts` | -| Transcript pages | Session reducer reads and first latest-window loads merge their page with the rows streamed during the fetch; only destructive events (deletion, revert commit, inbox cancellation, compaction boundaries) restart a reducer read. Reloads and history windows keep revision checks. Rules in [TRANSCRIPT.md](TRANSCRIPT.md). | UI `stores/opencode-data.ts`, `opencode-data-read-merge.ts`, `session-api.ts` | +| Transcript pages | Session reducer reads and first latest-window loads merge their page with the rows streamed during the fetch; only destructive events (deletion, revert commit, inbox cancellation, compaction boundaries) restart a reducer read. Reloads and history windows keep revision checks. Event-driven projections coalesce to one pass per animation frame. Rules in [TRANSCRIPT.md](TRANSCRIPT.md). | UI `stores/opencode-data.ts`, `opencode-projection-frame.ts`, `opencode-data-read-merge.ts`, `session-api.ts` | | Mission bodies | Display-only content-route reads keyed by workspace, project, mission and selector; continuations pinned to their first slice's digest, read only as far as the shown page needs. One read at a time per body with coalesced demand. The panel's maps are compact: a descriptor in place of each body. A shown body is read again when its map's descriptor names another digest (a map naming no descriptor expects nothing); a plugin update, reconnect or new stream advances the entry's invalidation epoch, so a request sent before it neither settles nor parks the body and one fresh read follows. The last good body stays shown until the newer one covers it; failures (a plugin without content reads included) show beside it and park until demand, an invalidation or Retry. Intents copying inputs (edit prefill, Save as brief) accept only bodies whose digests are their snapshot's descriptors. Unshown bodies stay within 4 MiB / 128 bodies, LRU. Never authority. Rules in [MISSIONS.md](MISSIONS.md) (Body display, Compact maps). | UI `stores/mission-content-store.ts`, `stores/mission-content.ts`, `components/mission-target-bodies.ts` | | Mission task pages | A compact map holds a page of its tasks; reads further (`loadTasks`: a window of 50 or tasks by name) name the map's display order and run in turn per mission, an identical waiting read coalesced and a running one followed by one trailing read. Each read is recorded before it is sent; rows, reports and reads each keep the revision of the page that answered them, so a page vouches only for what it answered. An answer in the shown order merges into the list read's own map (a row it holds stays unless a page answered it later), a refusal or another order reads the list again. List publications keep the rows of the same order (last good display), drop caches of maps no longer listed, and read again each read last answered before the map's revision, from scratch in another order. A failure keeps the rows with Try again and parks its read until a later revision or an explicit retry; it stays shown until a read answers it, whatever else succeeds. Attention windows count as covered only for the attention sequence they answered. A body a row read further describes is expected at that page's revision; an older plugin's undescribed history is read again when its report metadata changes. `clear` fences reads in flight (epoch and cache identity). Reads share the secondary background budget. Never authority. Rules in [MISSIONS.md](MISSIONS.md) (Task pages). | UI `stores/mission-store.ts`, `stores/mission-task-pages.ts` | | Mission panel inventory | Live picker/cleanup discovery uses explicit 50-row logical windows through `panelPage`. Content digests verify that window; a separate order digest covers the collection's row values and fences further selections. Only selected/reader-wanted missions load bounded headers and task windows; root collection/extension descriptors permit exact field reads. Every window retains its own revision and failed selector. A failed middle window stays failed through unrelated successes; changed cursors park for explicit restart/Retry. Passive refresh reads first windows and wanted headers, preserving expanded displays without replaying every read after unrelated revisions. Initial discovery point-reads the durable finish counter before hints; completion seeding happens once, independently of partial summary maxima. Exact cleanup reconciliation keeps its existing single-mission path. Every transfer retains the shared secondary slot through bounded stream cleanup, with one 30-second logical deadline and source-specific preallocation limits. Hints serialization/order work remains in the append lane and linear in the whole chosen collection per transfer. | UI `stores/mission-panel-inventory.ts`, `lib/mission-task-page-read.ts`; server `missions/mission-panel-page.ts`, `mission-panel-header.ts`, `MissionControl.panelPage` | diff --git a/dev-docs/TRANSCRIPT.md b/dev-docs/TRANSCRIPT.md index aafa136c87..128ab622a0 100644 --- a/dev-docs/TRANSCRIPT.md +++ b/dev-docs/TRANSCRIPT.md @@ -15,5 +15,6 @@ Message content visibility and presets, token lines, tool renderers, native sess - Native websearch text is parsed conservatively by `tool-call/renderers/websearch-data.ts`; unknown/oversized responses use the shared raw renderer. Recognized provider-consent Forms retain the generic schema/answer path, with layout in `styles/messaging/tool-call/websearch.css`. - Session lifecycle features use OpenCode V2's published signals: one `session.execution.{succeeded,failed,interrupted}` per busy period (`lib/native-execution-events.ts`) and `session.retry.scheduled` for provider backoff. The schema-declared `session.idle`/`session.status` are never published; never wire a feature only to them. A terminal event is a hint until `session.active` confirms settlement, which alone announces "A session finished", only for root sessions ending in success, failure or an inactivity stop. Transcript reducers retire at settlement, deferred while prompts stay parked in their inbox or a user shell or unsettled tool call still needs reconciling there. A tool part still `streaming` (rendered "pending") once a settlement drained has lost its `called` transition: `session.tool.success` reduces solely from the running status (a later `failed` can still settle it). Symmetrically, admission drops a `session.tool.called` whose newest matching part already holds a terminal status: the reducer would overwrite the settled state wholesale (flipping completed tool images back to "running" until a settlement read healed them), while a legitimate re-call is always preceded by a fresh `session.tool.input.started` part that becomes the reducer's match. The SDK reducer already answers a terminal event whose transcript holds a streaming or running tool by invalidating and re-reading messages, but that refresh is dropped while the stream is not connected and never publishes its result; the terminal event therefore finishes the read itself (`finishOpenCodeDataEvent`: one `message.sync` that shares the SDK's in-flight read when connected and starts it when not) and projects the reconciled page, which keeps the display window's membership. Only a newer admitted execution suppresses that publication — same-round activity such as a late terminal tool event must not leave the repair unpublished. Non-event settlement evidence finishes the same read through the same fence: the liveness activity read observing that a run the UI still believed working has settled (`observeNonEventSettlement`, reported by `session-api`), and an authoritative inbox sync that drops an item the reducer admitted (a delivery or cancellation without events, whose user message may never be admitted either). Token/cost lines (`components/message-usage.tsx`) appear once per agent message: the row of an assistant message's final text (the one carrying "delete tools and reasoning") renders, once that message has native usage, the sum (`sumMessageUsage`) of its own usage and of every message holding a tool or reasoning part that cleanup selects (`technicalCleanupParts`). Reuse that selection rather than a separate projection. The line renders inside the content row, so it never shifts `` items or splits a technical group (whose Delete needs two tools); tool-only steps get no line. A message whose text is followed by tools also counts in the next message's line, as its tools belong to that selection. Sums cover the loaded transcript window only, like cleanup. Capture real streams with `scripts/test-session-lifecycle-native.mjs`. - Native transcript events reduce only through `stores/opencode-data.ts` (never add another partial-part reducer), which publishes bounded snapshots through `hydrateMessages` (message-v2/instance-store.ts); these per-event re-projections must not invalidate unchanged messages: `hydrateMessages` reuses the resident `MessageRecord` object whenever content, explicit timestamps and prompt-display metadata are unchanged, and an identical whole snapshot returns before any store write or session-revision bump. Part/info comparisons use allocation-free structural equality (`partDataEqual`), never `JSON.stringify`. The SDK reduce mutates rows in place, so reference identity can never prove a session unchanged — never add a reference-anchored projection skip. A session's reducer is replaced (authoritative resync after an overflow, rotation failure or reconnect; pruning invalidation) only through `replaceTranscriptEntry`, which bumps the transcript's `entryRevision`; reactive readers of the reducer, such as the inbox behind queued badges (`getOpenCodeSessionInbox`), track it so they rebind instead of keeping the disposed reducer's last state (`opencode-inbox-rebinding.test.ts`, `tests/browser/session-rendering.test.ts`). +- Event-driven projections coalesce to the frame level through `stores/opencode-projection-frame.ts`: events still reduce synchronously and in order, but the projection into the message store runs at most once per animation frame per session, always from the reducer's newest state (a bounded timer while the window is hidden, since hidden windows pause animation frames; environments without a frame scheduler keep the previous synchronous projection). Store readers lag at most one frame. Session selection writes (`writeSessionSelection`) and visibility transitions flush pending pages; transcript eviction and instance teardown cancel them; destructive events (`session.deleted`, `session.inbox.cancelled`, `session.revert.committed`) and forced authoritative publications still project synchronously and subsume the pending frame. An event admitted while its session was active but no longer active at frame time invalidates the loaded page so the next activation reads authoritatively. Covered by `opencode-projection-frame.test.ts` and `tests/browser/streaming-projection.test.ts`. - Reducer page reads (the SDK's `message.sync` through the `message.list` proxy in `createDataEntry`) flush buffered compaction text, snapshot the session's live rows, fetch, flush again and answer with the page merged against that snapshot (`stores/opencode-data-read-merge.ts`), so buffered text is reduced exactly once: the page is the base and keeps its cursor, a live row new or changed since the snapshot keeps its live version (a row changed only in its root `time.completed`, which SDK step boundaries stamp, takes the page's version) except for its identified content parts, such as tool calls, left unchanged since the snapshot, which take the page's version so progress on one tool never hides authority for another, and rows created during the fetch and absent from the page ride along newest-first. Only events that remove or replace content (`invalidatesFetchedPage`: deletion, revert commit, inbox cancellation, compaction start/end/failure) restart the read; streamed text, reasoning, tool progress, compaction text and new rows merge, so a continuously streaming session completes its read in one fetch. Live versions are reconciled by the native full-text endings and the SDK's end-of-round refresh. A first latest-window load (`loadMessages` with no resident rows and no stored window) merges its 200-row page the same way and commits it while live revisions advance, then, unless the session is known idle, seeds a session reducer that holds no rows with the page's newest 20 rows, the SDK's own page size, without another read (`seedOpenCodeSessionMessages`), because the reducer applies streamed events only to messages it holds. Reloads and history windows keep the revision checks. Covered by `opencode-data-trailing-read.test.ts`, `opencode-data-read-merge.test.ts` and `session-api-running-open.test.ts`. - Shared speech byte/MediaSource transport lives in `lib/audio-utils.ts`; playback ownership and cancellation stay with its callers. diff --git a/packages/ui/src/stores/instances.ts b/packages/ui/src/stores/instances.ts index 983da5f13d..4b0ed2de35 100644 --- a/packages/ui/src/stores/instances.ts +++ b/packages/ui/src/stores/instances.ts @@ -54,6 +54,11 @@ import { import { setHasInstances } from "./ui" import { messageStoreBus } from "./message-v2/bus" import { applyOpenCodeDataEvent, destroyOpenCodeData, finishOpenCodeDataEvent, projectOpenCodeMessages, syncOpenCodeSessionInbox, type OpenCodeDataEventAdmission } from "./opencode-data" +import { + cancelTranscriptProjections, + scheduleTranscriptProjection, + setTranscriptProjectionGates, +} from "./opencode-projection-frame" import { updateSessionInfo } from "./message-v2/session-info" import { isLatestWindow } from "./message-v2/message-window" import { upsertPermissionV2, removePermissionV2, removeMessageV2 } from "./message-v2/bridge" @@ -661,6 +666,7 @@ function attachClient(descriptor: WorkspaceDescriptor) { if (instance.client) { sdkManager.destroyClientsForInstance(descriptor.id) destroyOpenCodeData(descriptor.id) + cancelTranscriptProjections(descriptor.id) } const client = sdkManager.createClient(descriptor.id, nextProxyPath, () => activeInstanceId() === descriptor.id) @@ -739,6 +745,7 @@ function releaseInstanceResources(instanceId: string) { sdkManager.destroyClientsForInstance(instanceId) } destroyOpenCodeData(instanceId) + cancelTranscriptProjections(instanceId) sseManager.seedStatus(instanceId, "disconnected") } @@ -1487,6 +1494,7 @@ function updateInstance(id: string, updates: Partial) { clearCommands(id) clearInstanceMetadata(id) destroyOpenCodeData(id) + cancelTranscriptProjections(id) volatileInstanceRefreshes.delete(id) for (const sessionId of sessions().get(id)?.keys() ?? []) invalidateSessionMessageLoad(id, sessionId) } @@ -1555,6 +1563,7 @@ function removeInstance(id: string, options: { authoritative?: boolean } = {}) { clearPendingFormQueue(id) clearInstanceMetadata(id) destroyOpenCodeData(id) + cancelTranscriptProjections(id) clearPermissionAutoAcceptForInstance(id) clearSyncedYoloSessionsForInstance(id) initialHydrations.delete(id) @@ -2245,6 +2254,15 @@ const USAGE_EVENT_TYPES = new Set([ "session.step.failed", ]) +// These events publish their page in the same task: teardown must not leave a +// coalesced frame behind, and each pairs with direct store surgery whose +// result a delayed projection would only repeat. +const SYNCHRONOUS_PROJECTION_EVENTS = new Set([ + "session.deleted", + "session.inbox.cancelled", + "session.revert.committed", +]) + // One stable, explicitly chunk-safe publication identity for ordinary replay. // Destructive event-specific effects below retain per-event callback barriers. const publishTranscriptChunk: NonNullable = ({ instanceId, sessionId, data, events }) => { @@ -2255,10 +2273,10 @@ const publishTranscriptChunk: NonNullable USAGE_EVENT_TYPES.has(event.type) || event.type === "session.compaction.delta")) { - updateSessionInfo(instanceId, sessionId) - } + scheduleTranscriptProjection(instanceId, sessionId, data, { + confirmPending: sessionEvents.at(-1)?.type !== "session.inbox.enqueued", + updateUsage: sessionEvents.some((event) => USAGE_EVENT_TYPES.has(event.type) || event.type === "session.compaction.delta"), + }) } function handleInstanceInvalidation(instanceId: string, event: Parameters>[1]): void { @@ -2287,14 +2305,23 @@ function handleInstanceInvalidation(instanceId: string, event: Parameters, preserveOmitted = true) => { @@ -2359,6 +2386,18 @@ function handleInstanceInvalidation(instanceId: string, event: Parameters activeSessionId().get(instanceId) === sessionId, + isLatestMessageWindow: (instanceId, sessionId) => + isLatestWindow(messageStoreBus.getOrCreate(instanceId).getMessageWindow(sessionId)), + invalidateInactiveLoadedSession: (instanceId, sessionId) => { + if (messagesLoaded().get(instanceId)?.has(sessionId)) invalidateSessionMessageLoad(instanceId, sessionId) + }, + onProjectedUsage: updateSessionInfo, +}) + sseManager.onConnectionLost = (instanceId, reason) => { const instance = instances().get(instanceId) if (!instance) { diff --git a/packages/ui/src/stores/opencode-projection-frame.test.ts b/packages/ui/src/stores/opencode-projection-frame.test.ts new file mode 100644 index 0000000000..13c5f851d4 --- /dev/null +++ b/packages/ui/src/stores/opencode-projection-frame.test.ts @@ -0,0 +1,178 @@ +import assert from "node:assert/strict" +import { test } from "node:test" +import { Schema } from "effect" +import { EventManifest } from "@opencode/schema/event-manifest" +import type { OpenCodeEvent } from "@opencode/client" +import { sdkManager } from "../lib/sdk-manager.ts" +import { addInstance, handleInstanceInvalidation, removeInstance } from "./instances.ts" +import { messageStoreBus } from "./message-v2/bus.ts" +import { applyOpenCodeDataEvent, destroyOpenCodeData, projectOpenCodeMessages } from "./opencode-data.ts" +import { setActiveSession, setSessions } from "./session-state.ts" + +const sessionId = "ses_frame" +const model = { providerID: "fixture", id: "fixture" } +const tokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } } +type Data = ReturnType + +// Validate the actual installed native wire schema, including durable envelopes. +function eventFactory() { + let sequence = 0 + return (type: string, payload: Record = {}): OpenCodeEvent => { + const definition = EventManifest.Latest.get(type as any) as any + assert.ok(definition, type) + const seq = ++sequence + return Schema.decodeUnknownSync(definition)({ + id: `evt_${seq}`, type, created: seq, location: { directory: "/fixture" }, + ...(definition.durable ? { durable: { aggregateID: sessionId, seq, version: definition.durable.version } } : {}), + data: { sessionID: sessionId, ...payload }, + }) as OpenCodeEvent + } +} + +/** A manual animation-frame clock: the renderer coalesces projections onto it. */ +function frameClock() { + const queue: (() => void)[] = [] + ;(globalThis as any).requestAnimationFrame = (callback: () => void) => { + queue.push(callback) + return queue.length + } + ;(globalThis as any).cancelAnimationFrame = () => {} + return { + fire() { + const pending = queue.splice(0) + for (const callback of pending) callback() + }, + restore() { + delete (globalThis as any).requestAnimationFrame + delete (globalThis as any).cancelAnimationFrame + }, + } +} + +function fixture(instanceId: string) { + const make = eventFactory() + const nativeSession = { + id: sessionId, title: "Frame", projectID: "prj_fixture", location: { directory: "/fixture" }, + time: { created: 1, updated: 1 }, cost: 0, tokens, + } + const client: any = { + session: { active: async () => ({}), get: async () => nativeSession, + inbox: { list: async () => [] }, form: { list: async () => [] } }, + permission: { list: async () => [] }, + message: { list: async () => ({ data: [], cursor: {} }) }, + } + ;(sdkManager as any).clients.set(`${instanceId}:/workspaces/${instanceId}/instance`, client) + const store = messageStoreBus.getOrCreate(instanceId) + let data!: Data + const apply = (type: string, payload: Record = {}) => { + data = applyOpenCodeDataEvent(instanceId, "/fixture", make(type, payload)) + } + // A short resident history establishes the session's latest display window. + for (let index = 0; index < 2; index++) { + const assistantMessageID = `msg_${String(index).padStart(3, "0")}` + apply("session.step.started", { assistantMessageID, agent: "build", model, started: index }) + apply("session.step.ended", { assistantMessageID, finish: "stop", cost: 0, tokens }) + } + projectOpenCodeMessages(instanceId, sessionId, data) + addInstance({ id: instanceId, folder: "/fixture", port: 0, pid: 0, proxyPath: "", status: "ready", client }) + setSessions(previous => new Map(previous).set(instanceId, new Map([[sessionId, { + ...nativeSession, instanceId, parentId: null, status: "working", agent: "build", + model: { providerId: "fixture", modelId: "fixture" }, + } as any]]))) + setActiveSession(instanceId, sessionId) + return { + instanceId, make, store, + dispatch: (type: string, payload: Record = {}) => handleInstanceInvalidation(instanceId, make(type, payload)), + stream: (count: number) => { + const streamId = "msg_live" + handleInstanceInvalidation(instanceId, make("session.step.started", { assistantMessageID: streamId, agent: "build", model, started: 500 })) + handleInstanceInvalidation(instanceId, make("session.text.started", { assistantMessageID: streamId, ordinal: 0 })) + for (let index = 0; index < count; index++) { + handleInstanceInvalidation(instanceId, make("session.text.delta", { assistantMessageID: streamId, ordinal: 0, delta: `[${index}]` })) + } + return { streamId, joined: Array.from({ length: count }, (_, index) => `[${index}]`).join("") } + }, + storeText: (messageId: string) => Object.values(store.getMessage(messageId)?.parts ?? {}) + .filter(part => part.data.type === "text").map(part => (part.data as any).text).join(""), + cleanup() { + destroyOpenCodeData(instanceId) + removeInstance(instanceId, { authoritative: false }) + sdkManager.destroyClientsForInstance(instanceId) + if (messageStoreBus.getInstance(instanceId)) messageStoreBus.unregisterInstance(instanceId) + }, + } +} + +test("a same-task delta burst projects once at the next frame", ctx => { + const clock = frameClock() + const f = fixture("frame-burst") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + const { streamId, joined } = f.stream(40) + assert.equal(hydration.mock.callCount(), 0, + "deltas admitted in one task must not project before the frame; store readers lag at most one frame") + assert.equal(f.store.getMessage(streamId), undefined, "the streaming row is not yet resident") + clock.fire() + assert.equal(hydration.mock.callCount(), 1, "one frame projects the whole burst once") + assert.equal(f.storeText(streamId), joined, "the projected page carries every admitted delta") + } finally { + clock.restore() + f.cleanup() + } +}) + +test("switching the active session flushes unprojected deltas", ctx => { + const clock = frameClock() + const f = fixture("frame-switch-flush") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + const { streamId, joined } = f.stream(10) + assert.equal(hydration.mock.callCount(), 0) + setActiveSession(f.instanceId, "ses_other") + assert.equal(hydration.mock.callCount(), 1, "a selection change flushes the pending projection") + assert.equal(f.storeText(streamId), joined) + } finally { + clock.restore() + f.cleanup() + } +}) + +test("instance teardown drops pending projections instead of resurrecting stores", ctx => { + const clock = frameClock() + const f = fixture("frame-teardown") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + f.stream(5) + assert.equal(hydration.mock.callCount(), 0) + removeInstance(f.instanceId, { authoritative: false }) + clock.fire() + assert.equal(hydration.mock.callCount(), 0, "a destroyed instance projects nothing") + assert.equal(messageStoreBus.getInstance(f.instanceId), undefined, + "no later frame may resurrect the destroyed instance's store") + } finally { + clock.restore() + f.cleanup() + } +}) + +test("a destructive event projects synchronously and subsumes the pending frame", ctx => { + const clock = frameClock() + const f = fixture("frame-destructive") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + f.stream(5) + assert.equal(hydration.mock.callCount(), 0) + f.dispatch("session.inbox.cancelled", { inboxID: "msg_inbox_1" }) + assert.equal(hydration.mock.callCount(), 1, "the destructive event projects its page synchronously") + clock.fire() + assert.equal(hydration.mock.callCount(), 1, "the subsumed frame projects nothing more") + // Teardown of a deleted session drops its pending frame the same way. + f.stream(5) + f.dispatch("session.deleted") + clock.fire() + assert.equal(hydration.mock.callCount(), 1, "a deleted session never projects a later frame") + } finally { + clock.restore() + f.cleanup() + } +}) diff --git a/packages/ui/src/stores/opencode-projection-frame.ts b/packages/ui/src/stores/opencode-projection-frame.ts new file mode 100644 index 0000000000..1c84b72cf5 --- /dev/null +++ b/packages/ui/src/stores/opencode-projection-frame.ts @@ -0,0 +1,160 @@ +import type { Data } from "@opencode/client/solid" +import { projectOpenCodeMessages } from "./opencode-data" +import { getLogger } from "../lib/logger" + +const log = getLogger("session") + +/** + * Frame-level coalescing for event-driven transcript projection. + * + * Native transcript events reduce synchronously and in order through the SDK + * reducer (`stores/opencode-data.ts`); only their *projection* — copying the + * reducer window into the message store, which re-renders the transcript — is + * coalesced here to at most one pass per animation frame per session, always + * reading the reducer's newest state. Projections never skip on reference + * identity (the reducer mutates rows in place); `hydrateMessages` record reuse + * remains what keeps an unchanged window cheap. + */ +export type TranscriptProjectionGates = { + isActiveSession(instanceId: string, sessionId: string): boolean + isLatestMessageWindow(instanceId: string, sessionId: string): boolean + /** Inactive-session recovery: the next activation loads the page authoritatively. */ + invalidateInactiveLoadedSession(instanceId: string, sessionId: string): void + /** Cost/token totals published after usage-affecting events. */ + onProjectedUsage(instanceId: string, sessionId: string): void +} + +type PendingProjection = { + instanceId: string + sessionId: string + data: Data + confirmPending: boolean + updateUsage: boolean +} + +/** Hidden windows pause animation frames; this timer keeps projections bounded-stale instead. */ +const HIDDEN_PROJECTION_INTERVAL_MS = 250 + +let gates: TranscriptProjectionGates | undefined +const pending = new Map() +let cancelScheduledRun: (() => void) | undefined + +export function setTranscriptProjectionGates(next: TranscriptProjectionGates): void { + gates = next +} + +function projectionKey(instanceId: string, sessionId: string): string { + return `${instanceId}\u0000${sessionId}` +} + +/** Defers one projection for a session to the current frame's end. Same-data + * schedules merge (every admitted event then lands in one pass); a replaced + * `data` reference (authoritative resync) simply wins with its own flags. */ +export function scheduleTranscriptProjection( + instanceId: string, + sessionId: string, + data: Data, + options: { confirmPending?: boolean; updateUsage?: boolean } = {}, +): void { + if (!gates) { + // No instance layer registered (unit harnesses): project synchronously. + projectOpenCodeMessages(instanceId, sessionId, data, true, options.confirmPending ?? true) + return + } + const key = projectionKey(instanceId, sessionId) + const previous = pending.get(key) + const merged = previous !== undefined && previous.data === data + pending.set(key, { + instanceId, + sessionId, + data, + // The merged pass replaces a sequence of per-event projections, so the last + // event decides confirmation (an `inbox.enqueued` tail must keep a parked + // prompt's optimistic row pending), while usage updates accumulate. + confirmPending: options.confirmPending ?? true, + updateUsage: (merged ? previous.updateUsage : false) || Boolean(options.updateUsage), + }) + requestScheduledRun() +} + +/** Publishes every pending projection now. Selection changes call this so a + * page admitted under the previous selection still lands before its gates move. */ +export function flushTranscriptProjections(): void { + drain(true) +} + +/** Drops pending projections for a session (or whole instance): destroyed + * reducers must not be projected by a later frame into a resurrected store. */ +export function cancelTranscriptProjections(instanceId: string, sessionId?: string): void { + if (sessionId === undefined) { + const prefix = `${instanceId}\u0000` + for (const key of pending.keys()) { + if (key.startsWith(prefix)) pending.delete(key) + } + } else { + pending.delete(projectionKey(instanceId, sessionId)) + } + if (!pending.size && cancelScheduledRun) { + cancelScheduledRun() + cancelScheduledRun = undefined + } +} + +function drain(bypassActiveGate: boolean): void { + cancelScheduledRun?.() + cancelScheduledRun = undefined + if (!pending.size || !gates) return + const entries = [...pending.values()] + pending.clear() + for (const entry of entries) { + try { + applyProjection(entry, bypassActiveGate) + } catch (error) { + log.error("Failed to project coalesced transcript page", { + instanceId: entry.instanceId, sessionId: entry.sessionId, error, + }) + } + } +} + +function applyProjection(entry: PendingProjection, bypassActiveGate: boolean): void { + const gate = gates + if (!gate) return + if (!bypassActiveGate && !gate.isActiveSession(entry.instanceId, entry.sessionId)) { + // Admitted under an older selection: the next activation reads the page + // authoritatively instead of trusting a stale display window. + gate.invalidateInactiveLoadedSession(entry.instanceId, entry.sessionId) + return + } + if (!gate.isLatestMessageWindow(entry.instanceId, entry.sessionId)) return + projectOpenCodeMessages(entry.instanceId, entry.sessionId, entry.data, true, entry.confirmPending) + if (entry.updateUsage) gate.onProjectedUsage(entry.instanceId, entry.sessionId) +} + +function requestScheduledRun(): void { + if (cancelScheduledRun) return + if (typeof document !== "undefined" && document.hidden) { + const timer = setTimeout(() => drain(false), HIDDEN_PROJECTION_INTERVAL_MS) + cancelScheduledRun = () => clearTimeout(timer) + return + } + if (typeof requestAnimationFrame === "function") { + const cancelFrame = typeof cancelAnimationFrame === "function" ? cancelAnimationFrame : () => {} + const frame = requestAnimationFrame(() => { + cancelScheduledRun = undefined + drain(false) + }) + cancelScheduledRun = () => cancelFrame(frame) + return + } + // No frame scheduler exists (server-side/unit contexts): there is no frame to + // coalesce onto, so projections keep their synchronous behavior. + drain(false) +} + +if (typeof document !== "undefined") { + // Flushing on visibility transitions keeps a hidden window from holding an + // unprojected page (animation frames pause there) and a repainted window + // from starting stale. + document.addEventListener("visibilitychange", () => flushTranscriptProjections()) +} diff --git a/packages/ui/src/stores/session-state.ts b/packages/ui/src/stores/session-state.ts index f1c422d465..3915e2c90e 100644 --- a/packages/ui/src/stores/session-state.ts +++ b/packages/ui/src/stores/session-state.ts @@ -15,6 +15,7 @@ import { applySessionPendingState } from "./session-pending-state" import { markSessionInterruptionSeen } from "./session-interruption-seen" import { cleanupBlankSession } from "./blank-session-cleanup" import { getOpenCodeInstanceGeneration } from "./opencode-data" +import { flushTranscriptProjections } from "./opencode-projection-frame" import { resolveAuthoritativeGenerationRecovery, resolveHydratedGenerationRecovery, @@ -703,6 +704,9 @@ function writeSessionSelection( instanceId: string, sessionId: string | null, ): void { + // A coalesced transcript page admitted under the previous selection lands + // before the gates move; nothing pending may cross a selection change. + flushTranscriptProjections() setter((prev) => { const next = new Map(prev) if (sessionId) next.set(instanceId, sessionId) diff --git a/packages/ui/src/stores/session-transcript-memory.ts b/packages/ui/src/stores/session-transcript-memory.ts index 4876ce40b0..9f448893ac 100644 --- a/packages/ui/src/stores/session-transcript-memory.ts +++ b/packages/ui/src/stores/session-transcript-memory.ts @@ -6,6 +6,7 @@ import { isSessionTranscriptProtected, SessionTranscriptLru } from "../lib/sessi import { messageStoreBus } from "./message-v2/bus" import { loading, sessions } from "./session-state" import { destroyOpenCodeData } from "./opencode-data" +import { cancelTranscriptProjections } from "./opencode-projection-frame" // Inactive transcript budget. Protected active/live 200-message windows retain authoritative content and may exceed it. export const SESSION_TRANSCRIPT_BYTE_BUDGET = 64 * 1024 * 1024 @@ -30,6 +31,7 @@ const coordinator = new SessionTranscriptLru({ }, evict: (instanceId, sessionId) => { log.info("Evicting inactive session transcript", { instanceId, sessionId }) + cancelTranscriptProjections(instanceId, sessionId) destroyOpenCodeData(instanceId, sessionId) messageStoreBus.getInstance(instanceId)?.evictSessionTranscript(sessionId) }, diff --git a/packages/ui/tests/browser/compaction-responsiveness.test.ts b/packages/ui/tests/browser/compaction-responsiveness.test.ts index e88b5c9928..3b88cd97f1 100644 --- a/packages/ui/tests/browser/compaction-responsiveness.test.ts +++ b/packages/ui/tests/browser/compaction-responsiveness.test.ts @@ -73,7 +73,10 @@ const snapshot = (page: Page) => page.evaluate(() => (window as any).compactionF const joined = (count: number) => Array.from({ length: count }, (_, index) => `[${index}]`).join("") test("128 active deltas reduce and project once, preserve immediate revision, and end authoritatively", () => fixture(async page => { - await page.evaluate(() => { const f = (window as any).compactionFixture; f.start(); f.resetCounts() }) + await page.evaluate(() => { const f = (window as any).compactionFixture; f.start() }) + // The compaction row's own creation projection lands on its frame first. + await page.evaluate(() => new Promise(resolve => requestAnimationFrame(() => requestAnimationFrame(resolve)))) + await page.evaluate(() => { const f = (window as any).compactionFixture; f.resetCounts() }) const before = await snapshot(page) const dispatchMs = await page.evaluate(() => (window as any).compactionFixture.burst("active")) const pending = await snapshot(page) @@ -96,7 +99,10 @@ test("128 active deltas reduce and project once, preserve immediate revision, an })) test("spaced native events coalesce per interval without losing chunks or affecting the other session", () => fixture(async page => { - await page.evaluate(() => { const f = (window as any).compactionFixture; f.start(); f.resetCounts() }) + await page.evaluate(() => { const f = (window as any).compactionFixture; f.start() }) + // The compaction row's own creation projection lands on its frame first. + await page.evaluate(() => new Promise(resolve => requestAnimationFrame(() => requestAnimationFrame(resolve)))) + await page.evaluate(() => { const f = (window as any).compactionFixture; f.resetCounts() }) for (let phase = 1; phase <= 2; phase++) { await page.evaluate(() => (window as any).compactionFixture.phase("active")) await page.waitForFunction(phase => (window as any).compactionFixture.snapshot().active === diff --git a/packages/ui/tests/browser/fixtures/streaming-projection.tsx b/packages/ui/tests/browser/fixtures/streaming-projection.tsx new file mode 100644 index 0000000000..bc97ad55d7 --- /dev/null +++ b/packages/ui/tests/browser/fixtures/streaming-projection.tsx @@ -0,0 +1,122 @@ +import { createSignal, Show } from "solid-js" +import { render } from "solid-js/web" +import SessionView from "../../../src/components/session/session-view" +import { ConfigProvider } from "../../../src/stores/preferences" +import { I18nProvider } from "../../../src/lib/i18n" +import { ThemeProvider } from "../../../src/lib/theme" +import { sdkManager } from "../../../src/lib/sdk-manager" +import { addInstance } from "../../../src/stores/instances" +import { sessions, setSessions, setActiveSession, setProviders } from "../../../src/stores/session-state" +import { messageStoreBus } from "../../../src/stores/message-v2/bus" +import { sseManager } from "../../../src/lib/sse-manager" +import { getOpenCodeMessageRevision } from "../../../src/stores/opencode-data" +import "../../../src/index.css" + +// Real SessionView, real native SSE dispatcher, real reducer and projection: +// only HTTP is fake. Streams an assistant text reply one delta per event. +const instanceId = "streaming-fixture", sessionId = "active" +const streamId = "msg_live" +const model = { providerID: "fixture", id: "fixture" } +let sequence = 1000, messageLists = 0, projections = 0 +// The authoritative page the HTTP read serves; a reply's final row lands here. +;(window as any).__nativePage = [] as any[] +const [selected, select] = createSignal(sessionId) +const [visible, show] = createSignal(true) +const client: any = { + session: { + active: async () => ({}), inbox: { list: async () => [] }, + form: { list: async () => [] }, + get: async ({ sessionID }: any) => ({ id: sessionID, title: sessionID, + location: { directory: "/fixture" }, time: { created: 1, updated: 1 } }), + instructions: { entry: { remove: async () => {}, put: async () => {} } }, + }, + model: { default: async () => ({ data: model }) }, + permission: { list: async () => [] }, + message: { list: async ({ limit = 200 }: any) => { + messageLists++ + const page: any[] = (window as any).__nativePage ?? [] + return { data: structuredClone(page.slice(-limit)).reverse(), cursor: {} } + } }, +} +;(sdkManager as any).clients.set(`${instanceId}:/workspaces/${instanceId}/instance`, client) +addInstance({ id: instanceId, folder: "/fixture", port: 0, pid: 0, proxyPath: "", status: "ready", client }) +setSessions(previous => new Map(previous).set(instanceId, new Map([[sessionId, { + id: sessionId, instanceId, parentId: null, title: sessionId, agent: "build", + model: { providerId: "fixture", modelId: "fixture" }, status: "idle", retry: null, idleSince: null, + generationRecovery: null, runtimeStatusKnown: true, version: "1", projectID: "fixture", + location: { directory: "/fixture" }, cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, time: { created: 1, updated: 1 }, +} as any]]))) +setProviders(previous => new Map(previous).set(instanceId, [{ id: "fixture", name: "Fixture", models: [ + { id: "fixture", name: "Fixture", providerId: "fixture", limit: { context: 10000, output: 1000 }, cost: { input: 0, output: 0 } }, +] }])) +setActiveSession(instanceId, sessionId) +const store = messageStoreBus.getOrCreate(instanceId) +// Projections are the store writes a frame coalesces; revisions are bumped per +// event at admission, so they must keep streaming while projections wait. +const hydrate = store.hydrateMessages.bind(store) +store.hydrateMessages = (...args: Parameters) => { + projections++ + return hydrate(...args) +} +const longtasks: number[] = [] +new PerformanceObserver(list => { + for (const entry of list.getEntries()) longtasks.push(entry.duration) +}).observe({ entryTypes: ["longtask"] }) +render(() => + +, document.getElementById("root")!) + +const emit = (type: string, data: any = {}) => { + const started = performance.now() + ;(sseManager as any).handleEvent(instanceId, { + id: `event_${++sequence}`, type, created: sequence, location: { directory: "/fixture" }, data: { sessionID: sessionId, ...data }, + }) + return performance.now() - started +} +let currentStream = streamId +const storeText = () => Object.values(store.getMessage(currentStream)?.parts ?? {}) + .filter(part => part.data.type === "text").map(part => (part.data as any).text).join("") +;(window as any).streamingFixture = { + resetCounts: () => { projections = 0; messageLists = 0; longtasks.length = 0 }, + snapshot: () => ({ projections, messageLists, revision: getOpenCodeMessageRevision(instanceId, sessionId), + longtasks: [...longtasks], text: storeText(), ids: store.getSessionMessageIds(sessionId) }), + startReply: (messageId = streamId) => { + currentStream = messageId + emit("session.step.started", { assistantMessageID: messageId, agent: "build", model, started: 500 }) + emit("session.text.started", { assistantMessageID: messageId, ordinal: 0 }) + }, + /** Dispatches `count` deltas in one JavaScript task, measuring each dispatch. */ + burst: (count = 400) => { + const perDelta: number[] = [] + const started = performance.now() + for (let index = 0; index < count; index++) perDelta.push(emit("session.text.delta", { assistantMessageID: currentStream, ordinal: 0, delta: `[${index}]` })) + perDelta.sort((a, b) => a - b) + return { dispatchMs: performance.now() - started, perDeltaMax: perDelta.at(-1), + perDeltaP95: perDelta[Math.floor(perDelta.length * 0.95)] } + }, + /** Spreads one delta across `count` spaced tasks, as a live tunnel delivers them. */ + spaced: async (count = 24) => { + const perDelta: number[] = [] + for (let index = 0; index < count; index++) { + perDelta.push(emit("session.text.delta", { assistantMessageID: currentStream, ordinal: 0, delta: `<${index}>` })) + await new Promise(resolve => setTimeout(resolve, 4)) + } + return { perDeltaMax: Math.max(...perDelta) } + }, + endReply: (text: string) => { + ;(window as any).__nativePage = [{ id: currentStream, type: "assistant", agent: "build", model, + time: { created: sequence, completed: sequence }, content: [{ type: "text", text }] }] + emit("session.text.ended", { assistantMessageID: currentStream, ordinal: 0, text }) + emit("session.step.ended", { assistantMessageID: currentStream, finish: "stop", cost: 0, + tokens: { input: 1, output: 1, reasoning: 0, cache: { read: 0, write: 0 } } }) + }, + activate: (sessionID: string) => { + show(false) + select(sessionID) + setActiveSession(instanceId, sessionID) + show(true) + }, + visibility: () => document.dispatchEvent(new Event("visibilitychange")), +} diff --git a/packages/ui/tests/browser/streaming-projection.test.ts b/packages/ui/tests/browser/streaming-projection.test.ts new file mode 100644 index 0000000000..16963bb55c --- /dev/null +++ b/packages/ui/tests/browser/streaming-projection.test.ts @@ -0,0 +1,88 @@ +import assert from "node:assert/strict" +import { after, before, test } from "node:test" +import { borrowFixture } from "./fixture-borrow" +import { chromium, type Browser, type Page } from "playwright" +import { armFixtureNavigationBudget } from "./fixture-navigation-budget" + +let browser: Browser, url: string +before(async () => { + const fixture = await borrowFixture({ + name: "streaming-projection", + routes: [ + { path: "/streaming", html: '
' }, + ], + }) + url = fixture.url + browser = await chromium.launch({ executablePath: process.env.CODENOMAD_BROWSER_PATH || undefined }) + armFixtureNavigationBudget(browser) +}) +after(async () => { await browser?.close() }) + +async function open(run: (page: Page) => Promise) { + const page = await browser.newPage({ viewport: { width: 1200, height: 800 } }) + const errors: string[] = [] + page.on("pageerror", error => errors.push(error.message)) + await page.route("**/*", route => { + const request = new URL(route.request().url()) + if (request.hostname !== "127.0.0.1") return route.abort() + if (request.pathname.startsWith("/api/")) return route.fulfill({ json: {} }) + return route.continue() + }) + try { + await page.goto(url, { waitUntil: "domcontentloaded" }) + await page.waitForFunction(() => Boolean((window as any).streamingFixture), undefined, { timeout: 60_000 }) + await page.waitForTimeout(300) + await run(page) + assert.deepEqual(errors, []) + } finally { await page.close() } +} + +const snapshot = (page: Page) => page.evaluate(() => (window as any).streamingFixture.snapshot()) +const frames = (page: Page) => page.evaluate(() => new Promise(resolve => + requestAnimationFrame(() => requestAnimationFrame(() => resolve())))) +const joined = (count: number) => Array.from({ length: count }, (_, index) => `[${index}]`).join("") +/** Starts a reply and lets its own frame land before counts reset. */ +const startSettled = async (page: Page, messageId?: string) => { + await page.evaluate((id) => (window as any).streamingFixture.startReply(id ?? undefined), messageId) + await frames(page) + await page.evaluate(() => (window as any).streamingFixture.resetCounts()) +} + +test("a same-task delta burst projects once at the next frame without an authoritative read", () => open(async page => { + await startSettled(page) + const ingress = await page.evaluate(() => (window as any).streamingFixture.burst(400)) + const during = await snapshot(page) + console.log("streaming-burst", JSON.stringify({ variant: process.env.CODENOMAD_STREAMING_VARIANT ?? "default", ingress, during: { projections: during.projections, longtasks: during.longtasks } })) + assert.ok(during.projections <= 1, `a burst dispatched in one task projects at most once before the snapshot (got ${during.projections})`) + assert.equal(during.messageLists, 0, "streamed deltas must not trigger HTTP message reads") + assert.ok(during.revision >= 400, "reducer revisions keep streaming per admitted event") + await page.waitForFunction((text) => (window as any).streamingFixture.snapshot().text === text, joined(400), { timeout: 15_000 }) + const flushed = await snapshot(page) + assert.equal(flushed.projections, 1, "one frame projects the whole burst once") + await page.waitForFunction(() => document.querySelector(".message-stream")?.textContent?.includes("[399]"), undefined, { timeout: 15_000 }) + await page.evaluate(() => { const f = (window as any).streamingFixture; f.resetCounts(); f.endReply("authoritative final") }) + await page.waitForFunction(() => (window as any).streamingFixture.snapshot().text === "authoritative final", undefined, { timeout: 15_000 }) + const final = await snapshot(page) + assert.equal(final.messageLists, 0, "the terminal events reconcile through the reducer, not a refetch") +})) + +test("spaced deltas across tasks stay frame-coalesced and complete", () => open(async page => { + await startSettled(page) + const spaced = await page.evaluate(() => (window as any).streamingFixture.spaced(24)) + await frames(page) + const state = await snapshot(page) + console.log("streaming-spaced", JSON.stringify({ variant: process.env.CODENOMAD_STREAMING_VARIANT ?? "default", spaced, projections: state.projections })) + assert.ok(state.text.endsWith("<23>"), "the projected page carries every spaced delta") +})) + +test("visibility changes and session switches flush pending pages", () => open(async page => { + await startSettled(page) + await page.evaluate(() => { const f = (window as any).streamingFixture; f.burst(30); f.visibility() }) + const visible = await snapshot(page) + assert.equal(visible.projections, 1, "a visibility change flushes the pending projection") + assert.equal(visible.text, joined(30)) + await page.evaluate(() => { const f = (window as any).streamingFixture; f.startReply("msg_second"); f.burst(60); f.activate("other") }) + const switched = await snapshot(page) + assert.equal(switched.projections, 2, "a selection change flushes the pending projection") + assert.equal(switched.text, joined(60), "the second reply's page flushed with its own stream") +})) From 2c4da0162fc6964cae69dd3e763f4c22cc9ba286 Mon Sep 17 00:00:00 2001 From: heunghingwan Date: Wed, 4 Nov 2026 13:14:42 +0800 Subject: [PATCH 2/6] fix(transcript): keep coalesced projections off a live frame scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps the native pruning UI check exposed on a quiet page: An idle compositor stops producing frames (a hidden or headless page with nothing to draw), so a schedule armed only on requestAnimationFrame could hang a pending projection forever — no projection means no DOM change, which means no next frame. Every visible-window schedule now also arms a 32ms fallback; whichever wake wins cancels the other, so a live page still projects exactly once per frame and an idle one projects within the fallback window. The hidden-window timer stays at its calmer 250ms. An execution settlement (session.execution.{succeeded,failed,interrupted}) now projects synchronously like the destructive events: it is the stable point interactive reads synchronize on — part deletion rejects a row whose status still reads "streaming" — and it fires once per turn, so the synchronous cost is one projection at settlement. Verified by reproducing the failing native pruning UI run locally: the traced page showed the final event batch scheduled but never projected on an idle compositor, and the deletion click reading a settled row a frame behind. Both fixed runs pass all nine UI checks twice; unit tests cover the idle-compositor fallback and the same-task settlement status. --- packages/ui/src/stores/instances.ts | 10 +++-- .../stores/opencode-projection-frame.test.ts | 40 +++++++++++++++++++ .../src/stores/opencode-projection-frame.ts | 31 ++++++++++++-- 3 files changed, 75 insertions(+), 6 deletions(-) diff --git a/packages/ui/src/stores/instances.ts b/packages/ui/src/stores/instances.ts index 4b0ed2de35..fdc284a0be 100644 --- a/packages/ui/src/stores/instances.ts +++ b/packages/ui/src/stores/instances.ts @@ -2254,13 +2254,17 @@ const USAGE_EVENT_TYPES = new Set([ "session.step.failed", ]) -// These events publish their page in the same task: teardown must not leave a -// coalesced frame behind, and each pairs with direct store surgery whose -// result a delayed projection would only repeat. +// These events publish their page in the same task. Destructive events pair +// with direct store surgery a delayed projection would only repeat, and an +// execution settlement is the stable point interactive reads (part deletion, +// copy) synchronize on: its final status must never lag a frame behind. const SYNCHRONOUS_PROJECTION_EVENTS = new Set([ "session.deleted", "session.inbox.cancelled", "session.revert.committed", + "session.execution.succeeded", + "session.execution.failed", + "session.execution.interrupted", ]) // One stable, explicitly chunk-safe publication identity for ordinary replay. diff --git a/packages/ui/src/stores/opencode-projection-frame.test.ts b/packages/ui/src/stores/opencode-projection-frame.test.ts index 13c5f851d4..29120b9e27 100644 --- a/packages/ui/src/stores/opencode-projection-frame.test.ts +++ b/packages/ui/src/stores/opencode-projection-frame.test.ts @@ -155,6 +155,46 @@ test("instance teardown drops pending projections instead of resurrecting stores } }) +test("an idle compositor that never frames still projects through the fallback", async ctx => { + // Headless/hidden pages can stop producing frames entirely; every visible + // schedule arms a short fallback so a pending page can never hang. + ;(globalThis as any).requestAnimationFrame = () => 1 + ;(globalThis as any).cancelAnimationFrame = () => {} + const f = fixture("frame-fallback") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + const { streamId, joined } = f.stream(20) + await new Promise(resolve => setTimeout(resolve, 120)) + assert.equal(hydration.mock.callCount(), 1, "the frame fallback projects without a compositor frame") + assert.equal(f.storeText(streamId), joined) + } finally { + delete (globalThis as any).requestAnimationFrame + delete (globalThis as any).cancelAnimationFrame + f.cleanup() + } +}) + +test("an execution settlement projects its final status in the same task", ctx => { + const clock = frameClock() + const f = fixture("frame-settlement") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + const { streamId } = f.stream(5) + assert.equal(hydration.mock.callCount(), 0) + assert.equal(f.store.getMessage(streamId), undefined, "the streaming row is not resident before a frame") + f.dispatch("session.step.ended", { assistantMessageID: "msg_live", finish: "stop", cost: 0, tokens: { input: 1, output: 1, reasoning: 0, cache: { read: 0, write: 0 } } }) + f.dispatch("session.execution.succeeded", {}) + const settled = f.store.getMessage("msg_live") + assert.ok(settled, "the settlement projected synchronously") + assert.equal(settled.status, "complete", "interactive reads never see a settled row lag a frame behind") + clock.fire() + assert.equal(hydration.mock.callCount(), 1, "the settlement subsumed the pending frame") + } finally { + clock.restore() + f.cleanup() + } +}) + test("a destructive event projects synchronously and subsumes the pending frame", ctx => { const clock = frameClock() const f = fixture("frame-destructive") diff --git a/packages/ui/src/stores/opencode-projection-frame.ts b/packages/ui/src/stores/opencode-projection-frame.ts index 1c84b72cf5..6f29f46afd 100644 --- a/packages/ui/src/stores/opencode-projection-frame.ts +++ b/packages/ui/src/stores/opencode-projection-frame.ts @@ -35,6 +35,15 @@ type PendingProjection = { /** Hidden windows pause animation frames; this timer keeps projections bounded-stale instead. */ const HIDDEN_PROJECTION_INTERVAL_MS = 250 +/** + * An idle compositor can stop producing frames (a hidden or headless page with + * nothing to draw), and a frame-aligned callback would then never run — a + * scheduled projection could hang until the next paint. Every visible-window + * schedule therefore also arms this short fallback; whichever wake wins, the + * other is cancelled, so a live page still projects exactly once per frame. + */ +const FRAME_FALLBACK_MS = 32 + let gates: TranscriptProjectionGates | undefined const pending = new Map() let cancelScheduledRun: (() => void) | undefined @@ -140,11 +149,27 @@ function requestScheduledRun(): void { } if (typeof requestAnimationFrame === "function") { const cancelFrame = typeof cancelAnimationFrame === "function" ? cancelAnimationFrame : () => {} - const frame = requestAnimationFrame(() => { + let frame: number | undefined + let fallback: ReturnType | undefined + let settled = false + const settle = () => { + if (settled) return + settled = true + if (frame !== undefined) cancelFrame(frame) + if (fallback !== undefined) clearTimeout(fallback) cancelScheduledRun = undefined drain(false) - }) - cancelScheduledRun = () => cancelFrame(frame) + } + const cancel = () => { + if (settled) return + settled = true + if (frame !== undefined) cancelFrame(frame) + if (fallback !== undefined) clearTimeout(fallback) + cancelScheduledRun = undefined + } + frame = requestAnimationFrame(settle) + fallback = setTimeout(settle, FRAME_FALLBACK_MS) + cancelScheduledRun = cancel return } // No frame scheduler exists (server-side/unit contexts): there is no frame to From 534bdfb7689ad8abc91ba1dada35420980017554 Mon Sep 17 00:00:00 2001 From: heunghingwan Date: Wed, 4 Nov 2026 14:09:42 +0800 Subject: [PATCH 3/6] test(ui): settle a frame before counting Code Mode progress rows session.tool.progress renders through the coalesced frame projection, so the call list assertion must wait for the frame it lands on instead of reading the DOM in the same task as the event. --- packages/ui/tests/browser/tool-images.test.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/ui/tests/browser/tool-images.test.ts b/packages/ui/tests/browser/tool-images.test.ts index 375065b526..d9776b3afb 100644 --- a/packages/ui/tests/browser/tool-images.test.ts +++ b/packages/ui/tests/browser/tool-images.test.ts @@ -36,8 +36,11 @@ async function withPage(run: (page: Page) => Promise) { } finally { await page.close() } } -async function decodedImages(page: Page, count: number) { - await page.waitForFunction(count => { +/** Event-driven projections coalesce onto the next animation frame. */ +const frames = (page: Page) => page.evaluate(() => + new Promise(resolve => requestAnimationFrame(() => requestAnimationFrame(() => resolve())))) + +async function decodedImages(page: Page, count: number) { await page.waitForFunction(count => { const images = Array.from(document.querySelectorAll(".tool-call-output-images img")) return images.length === count && images.every(img => img.complete && img.naturalWidth === 800) }, count) @@ -147,6 +150,7 @@ test("Code Mode renders native progress, nested errors, script, images and histo await firstCall.click() await firstCall.focus() await page.evaluate(() => (window as any).fixture.execute("progress")) + await frames(page) assert.equal(await page.locator(".execute-calls li").count(), 3) assert.equal(await page.locator(".execute-calls details").first().getAttribute("open"), "") assert.equal(await firstCall.evaluate(el => el === document.activeElement), true) From d90d7506854a1c3a33a08169a16f4ebbd6a7f5d5 Mon Sep 17 00:00:00 2001 From: heunghingwan Date: Thu, 5 Nov 2026 02:20:04 +0800 Subject: [PATCH 4/6] fix(transcript): close the review's three projection-lifetime gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adopting the review's three reproductions, each now covered by a unit case in opencode-projection-frame.test.ts: A synchronous settlement or destructive event now adopts the pending frame's usage obligation instead of dropping it (a step end with usage followed by execution.succeeded in one task used to leave session info at zero while the message store carried the tokens). destroyOpenCodeData cancels coalesced frames at the shared lifetime boundary — a hard reconnect's reset previously deleted the transcript entry while a pending frame kept the old Data, and the missing entry slipped past the projection guard, resurrecting the pre-reset rows over an authoritative empty load. The frame module now reaches the store only through the registered gates, so the reducer module can cancel without an import cycle. A replayed chunk carrying an execution settlement publishes synchronously through the same barrier as the immediate path, so a drained-and-retired rotation can no longer lose the completed reply it held to a later frame or a newer execution's Data. Also per review: the frame test joins the workflow's browser-conditioned unit list; streaming-projection borrows the quiet fixture event stream and keeps strict console/network failure checks (which immediately caught the unmocked Yolo probe); its spaced-stream case now asserts the complete coalesced text and fewer passes than deltas; and history-navigation awaits the scheduled row's actual publication before rechecking the distant marker's identity. Verified: 252/252 across the transcript/instance unit families, typecheck clean, streaming-projection 3/3, compaction-responsiveness 5/5, tool-images 6/6, and the isolated native pruning UI run passes all nine checks on this head. --- .github/workflows/pr-build.yml | 1 + packages/ui/src/stores/instances.ts | 52 +++++-- packages/ui/src/stores/opencode-data.ts | 5 + .../stores/opencode-projection-frame.test.ts | 140 +++++++++++++++++- .../src/stores/opencode-projection-frame.ts | 38 +++-- .../tests/browser/history-navigation.test.ts | 7 +- .../browser/streaming-projection.test.ts | 18 ++- 7 files changed, 230 insertions(+), 31 deletions(-) diff --git a/.github/workflows/pr-build.yml b/.github/workflows/pr-build.yml index 912b965d58..c6233af77d 100644 --- a/.github/workflows/pr-build.yml +++ b/.github/workflows/pr-build.yml @@ -492,6 +492,7 @@ jobs: packages/ui/src/stores/message-v2/transcript-eviction.test.ts packages/ui/src/stores/opencode-data.test.ts packages/ui/src/stores/opencode-data-idle.test.ts + packages/ui/src/stores/opencode-projection-frame.test.ts packages/ui/src/stores/opencode-data-settlement.test.ts packages/ui/src/stores/opencode-data-reconnect.test.ts packages/ui/src/stores/opencode-data-trailing-read.test.ts diff --git a/packages/ui/src/stores/instances.ts b/packages/ui/src/stores/instances.ts index fdc284a0be..e54d8ce9b1 100644 --- a/packages/ui/src/stores/instances.ts +++ b/packages/ui/src/stores/instances.ts @@ -53,11 +53,14 @@ import { } from "./session-state" import { setHasInstances } from "./ui" import { messageStoreBus } from "./message-v2/bus" -import { applyOpenCodeDataEvent, destroyOpenCodeData, finishOpenCodeDataEvent, projectOpenCodeMessages, syncOpenCodeSessionInbox, type OpenCodeDataEventAdmission } from "./opencode-data" +import { + applyOpenCodeDataEvent, destroyOpenCodeData, finishOpenCodeDataEvent, projectOpenCodeMessages, syncOpenCodeSessionInbox, type OpenCodeDataEventAdmission, +} from "./opencode-data" import { cancelTranscriptProjections, scheduleTranscriptProjection, setTranscriptProjectionGates, + takeTranscriptProjection, } from "./opencode-projection-frame" import { updateSessionInfo } from "./message-v2/session-info" import { isLatestWindow } from "./message-v2/message-window" @@ -2257,7 +2260,9 @@ const USAGE_EVENT_TYPES = new Set([ // These events publish their page in the same task. Destructive events pair // with direct store surgery a delayed projection would only repeat, and an // execution settlement is the stable point interactive reads (part deletion, -// copy) synchronize on: its final status must never lag a frame behind. +// copy) synchronize on: its final status must never lag a frame behind. A +// replayed chunk carrying one publishes the same way, before the drained +// reducer retires and the page it holds is lost to a later frame. const SYNCHRONOUS_PROJECTION_EVENTS = new Set([ "session.deleted", "session.inbox.cancelled", @@ -2267,6 +2272,21 @@ const SYNCHRONOUS_PROJECTION_EVENTS = new Set([ "session.execution.interrupted", ]) +const isSettlementEvent = (type: string) => SYNCHRONOUS_PROJECTION_EVENTS.has(type) + +/** Publishes a page in this task, subsuming the pending frame: the dropped + * entry's obligations (usage updates) transfer to this publication. */ +const publishSynchronousPage = ( + instanceId: string, + sessionId: string, + data: ReturnType, + options: { preserveOmitted?: boolean; confirmPending: boolean; updateUsage: boolean }, +) => { + const dropped = takeTranscriptProjection(instanceId, sessionId) + projectOpenCodeMessages(instanceId, sessionId, data, options.preserveOmitted ?? true, options.confirmPending) + if (options.updateUsage || Boolean(dropped?.updateUsage)) updateSessionInfo(instanceId, sessionId) +} + // One stable, explicitly chunk-safe publication identity for ordinary replay. // Destructive event-specific effects below retain per-event callback barriers. const publishTranscriptChunk: NonNullable = ({ instanceId, sessionId, data, events }) => { @@ -2277,10 +2297,15 @@ const publishTranscriptChunk: NonNullable USAGE_EVENT_TYPES.has(event.type) || event.type === "session.compaction.delta"), - }) + } + if (sessionEvents.some((event) => isSettlementEvent(event.type))) { + publishSynchronousPage(instanceId, sessionId, data, options) + return + } + scheduleTranscriptProjection(instanceId, sessionId, data, options) } function handleInstanceInvalidation(instanceId: string, event: Parameters>[1]): void { @@ -2309,17 +2334,12 @@ function handleInstanceInvalidation(instanceId: string, event: Parameters + projectOpenCodeMessages(instanceId, sessionId, data, true, confirmPending), }) sseManager.onConnectionLost = (instanceId, reason) => { diff --git a/packages/ui/src/stores/opencode-data.ts b/packages/ui/src/stores/opencode-data.ts index b80e64cefe..19b3b7cb54 100644 --- a/packages/ui/src/stores/opencode-data.ts +++ b/packages/ui/src/stores/opencode-data.ts @@ -16,6 +16,7 @@ import { } from "./opencode-data-events" import { mergePageWithSnapshot, snapshotRows, type OpenCodeRowSnapshot } from "./opencode-data-read-merge" import { createReconnect, type ReconnectCapture } from "./opencode-reconnect" +import { cancelTranscriptProjections } from "./opencode-projection-frame" const log = getLogger("session") @@ -1029,6 +1030,10 @@ function isStaleToolCalled(data: Data, event: OpenCodeEvent): boolean { } export function destroyOpenCodeData(instanceId: string, sessionId?: string): void { + // The shared lifetime boundary for reducers, reconnect resets and instance + // teardown: a coalesced frame holding this reducer's page must never publish + // after the entry is gone (its guard cannot see a missing entry). + cancelTranscriptProjections(instanceId, sessionId) if (sessionId !== undefined) { compactionDeltas.cancel(instanceId, sessionId) unobservedCompactions.delete(messageRevisionKey(instanceId, sessionId)) diff --git a/packages/ui/src/stores/opencode-projection-frame.test.ts b/packages/ui/src/stores/opencode-projection-frame.test.ts index 29120b9e27..d55828ac37 100644 --- a/packages/ui/src/stores/opencode-projection-frame.test.ts +++ b/packages/ui/src/stores/opencode-projection-frame.test.ts @@ -7,7 +7,7 @@ import { sdkManager } from "../lib/sdk-manager.ts" import { addInstance, handleInstanceInvalidation, removeInstance } from "./instances.ts" import { messageStoreBus } from "./message-v2/bus.ts" import { applyOpenCodeDataEvent, destroyOpenCodeData, projectOpenCodeMessages } from "./opencode-data.ts" -import { setActiveSession, setSessions } from "./session-state.ts" +import { setActiveSession, setSessions, getSessionInfo } from "./session-state.ts" const sessionId = "ses_frame" const model = { providerID: "fixture", id: "fixture" } @@ -49,7 +49,7 @@ function frameClock() { } } -function fixture(instanceId: string) { +function fixture(instanceId: string, options: { seed?: number } = {}) { const make = eventFactory() const nativeSession = { id: sessionId, title: "Frame", projectID: "prj_fixture", location: { directory: "/fixture" }, @@ -67,8 +67,9 @@ function fixture(instanceId: string) { const apply = (type: string, payload: Record = {}) => { data = applyOpenCodeDataEvent(instanceId, "/fixture", make(type, payload)) } - // A short resident history establishes the session's latest display window. - for (let index = 0; index < 2; index++) { + // A resident history establishes the session's latest display window; a full + // 200-message window additionally forces later replies through rotation replay. + for (let index = 0; index < (options.seed ?? 2); index++) { const assistantMessageID = `msg_${String(index).padStart(3, "0")}` apply("session.step.started", { assistantMessageID, agent: "build", model, started: index }) apply("session.step.ended", { assistantMessageID, finish: "stop", cost: 0, tokens }) @@ -195,6 +196,137 @@ test("an execution settlement projects its final status in the same task", ctx = } }) +test("a synchronous settlement adopts the pending frame's usage obligation", ctx => { + const clock = frameClock() + const f = fixture("frame-usage-adoption") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + f.dispatch("session.step.started", { assistantMessageID: "msg_live", agent: "build", model, started: 500 }) + f.dispatch("session.text.started", { assistantMessageID: "msg_live", ordinal: 0 }) + f.dispatch("session.text.delta", { assistantMessageID: "msg_live", ordinal: 0, delta: "done" }) + // A usage-carrying step end schedules the totals update; the settlement + // that follows in the same task publishes synchronously and must adopt it. + f.dispatch("session.step.ended", { assistantMessageID: "msg_live", finish: "stop", cost: 1, tokens: { input: 60, output: 58, reasoning: 0, cache: { read: 0, write: 0 } } }) + assert.equal(hydration.mock.callCount(), 0, "the delta stream only scheduled") + assert.equal(getSessionInfo(f.instanceId, sessionId), undefined, "no totals published before the barrier") + f.dispatch("session.execution.succeeded", {}) + assert.equal(hydration.mock.callCount(), 1, "the settlement published its page synchronously") + assert.notEqual(getSessionInfo(f.instanceId, sessionId), undefined, + "the settlement adopted the pending frame's usage update") + clock.fire() + assert.equal(hydration.mock.callCount(), 1) + } finally { + clock.restore() + f.cleanup() + } +}) + +test("destroying the reducer cancels its pending frame at the shared boundary", ctx => { + const clock = frameClock() + const f = fixture("frame-destroy-boundary") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + f.stream(5) + assert.equal(hydration.mock.callCount(), 0) + destroyOpenCodeData(f.instanceId, sessionId) + clock.fire() + assert.equal(hydration.mock.callCount(), 0, "a destroyed reducer never publishes a later frame") + assert.equal(f.store.getMessage("msg_live"), undefined, "no frame resurrects the destroyed reducer's rows") + } finally { + clock.restore() + f.cleanup() + } +}) + +test("a replayed settlement chunk publishes before the drained reducer retires", async ctx => { + const clock = frameClock() + const f = fixture("frame-replay-settlement", { seed: 200 }) + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + const overflow = (type: string, payload: Record = {}) => f.dispatch(type, { + assistantMessageID: "msg_over", ...payload, + }) + // The full window queues the reply; the replayed chunk carrying the + // settlement must publish in the replay task, not on a later frame. + overflow("session.step.started", { agent: "build", model, started: 900 }) + overflow("session.text.started", { ordinal: 0 }) + overflow("session.text.delta", { ordinal: 0, delta: "completed reply" }) + overflow("session.text.ended", { ordinal: 0, text: "completed reply" }) + overflow("session.step.ended", { finish: "stop", cost: 0, tokens }) + overflow("session.execution.succeeded", {}) + const deadline = Date.now() + 5_000 + while (f.store.getMessage("msg_over")?.status !== "complete" && Date.now() < deadline) { + await new Promise(resolve => setTimeout(resolve, 25)) + } + assert.equal(f.store.getMessage("msg_over")?.status, "complete", + "the queued reply's settlement page reached the store without any frame") + assert.ok(f.storeText("msg_over").includes("completed reply")) + } finally { + clock.restore() + f.cleanup() + } +}) + +test("a pending frame respects the latest-window gate", ctx => { + const clock = frameClock() + const f = fixture("frame-window-gate") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + f.stream(3) + f.store.setMessageWindow(sessionId, { kind: "history", resumeCursor: "cursor", newerCursors: [] }) + clock.fire() + assert.equal(hydration.mock.callCount(), 0, "a history window must not take streaming projections") + assert.equal(f.store.getMessage("msg_live"), undefined) + } finally { + clock.restore() + f.cleanup() + } +}) + +test("a hidden window projects on its bounded timer instead of animation frames", async ctx => { + ;(globalThis as any).document = { hidden: true } + const f = fixture("frame-hidden-timer") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + const { streamId, joined } = f.stream(5) + await new Promise(resolve => setTimeout(resolve, 60)) + assert.equal(hydration.mock.callCount(), 0, "the hidden interval has not elapsed yet") + await new Promise(resolve => setTimeout(resolve, 320)) + assert.equal(hydration.mock.callCount(), 1, "the hidden timer projects the coalesced page") + assert.equal(f.storeText(streamId), joined) + } finally { + delete (globalThis as any).document + f.cleanup() + } +}) + +test("a merged frame keeps the last event's confirmation and accumulates usage", ctx => { + const clock = frameClock() + try { + const confirming = fixture("frame-confirm-true") + const confirmHydration = ctx.mock.method(confirming.store, "hydrateMessages") + confirming.dispatch("session.text.delta", { assistantMessageID: "msg_live", ordinal: 0, delta: "a" }) + confirming.dispatch("session.inbox.enqueued", { inboxID: "msg_queued", item: { type: "user", payload: { text: "next" }, delivery: "queue" } }) + clock.fire() + assert.equal(confirmHydration.mock.callCount(), 1) + assert.equal(confirmHydration.mock.calls[0].arguments[3]?.confirmPending, false, + "an inbox.enqueued tail keeps the parked prompt's row pending") + + const settling = fixture("frame-confirm-false") + const settleHydration = ctx.mock.method(settling.store, "hydrateMessages") + settling.dispatch("session.inbox.enqueued", { inboxID: "msg_queued2", item: { type: "user", payload: { text: "next" }, delivery: "queue" } }) + settling.dispatch("session.text.delta", { assistantMessageID: "msg_live", ordinal: 0, delta: "b" }) + clock.fire() + assert.equal(settleHydration.mock.callCount(), 1) + assert.equal(settleHydration.mock.calls[0].arguments[3]?.confirmPending, true, + "an ordinary tail confirms pending sends again") + settling.cleanup() + confirming.cleanup() + } finally { + clock.restore() + } +}) + test("a destructive event projects synchronously and subsumes the pending frame", ctx => { const clock = frameClock() const f = fixture("frame-destructive") diff --git a/packages/ui/src/stores/opencode-projection-frame.ts b/packages/ui/src/stores/opencode-projection-frame.ts index 6f29f46afd..8a9486729b 100644 --- a/packages/ui/src/stores/opencode-projection-frame.ts +++ b/packages/ui/src/stores/opencode-projection-frame.ts @@ -1,5 +1,4 @@ import type { Data } from "@opencode/client/solid" -import { projectOpenCodeMessages } from "./opencode-data" import { getLogger } from "../lib/logger" const log = getLogger("session") @@ -14,6 +13,10 @@ const log = getLogger("session") * reading the reducer's newest state. Projections never skip on reference * identity (the reducer mutates rows in place); `hydrateMessages` record reuse * remains what keeps an unchanged window cheap. + * + * The module stays free of store imports so the reducer's own lifetime + * boundaries (`destroyOpenCodeData`) can cancel pending frames; every store + * touch arrives through the gates the instance layer registers. */ export type TranscriptProjectionGates = { isActiveSession(instanceId: string, sessionId: string): boolean @@ -22,6 +25,8 @@ export type TranscriptProjectionGates = { invalidateInactiveLoadedSession(instanceId: string, sessionId: string): void /** Cost/token totals published after usage-affecting events. */ onProjectedUsage(instanceId: string, sessionId: string): void + /** Publishes a reduced page into the message store. */ + projectPage(instanceId: string, sessionId: string, data: Data, confirmPending: boolean): void } type PendingProjection = { @@ -65,11 +70,7 @@ export function scheduleTranscriptProjection( data: Data, options: { confirmPending?: boolean; updateUsage?: boolean } = {}, ): void { - if (!gates) { - // No instance layer registered (unit harnesses): project synchronously. - projectOpenCodeMessages(instanceId, sessionId, data, true, options.confirmPending ?? true) - return - } + if (!gates) return const key = projectionKey(instanceId, sessionId) const previous = pending.get(key) const merged = previous !== undefined && previous.data === data @@ -92,6 +93,15 @@ export function flushTranscriptProjections(): void { drain(true) } +/** Removes and returns a session's pending projection. A synchronous + * publication takes its obligations (usage updates) over before dropping it. */ +export function takeTranscriptProjection(instanceId: string, sessionId: string): PendingProjection | undefined { + const key = projectionKey(instanceId, sessionId) + const entry = pending.get(key) + if (entry !== undefined) disarmIfEmpty(key) + return entry +} + /** Drops pending projections for a session (or whole instance): destroyed * reducers must not be projected by a later frame into a resurrected store. */ export function cancelTranscriptProjections(instanceId: string, sessionId?: string): void { @@ -103,15 +113,23 @@ export function cancelTranscriptProjections(instanceId: string, sessionId?: stri } else { pending.delete(projectionKey(instanceId, sessionId)) } - if (!pending.size && cancelScheduledRun) { + if (!pending.size) disarm() +} + +function disarmIfEmpty(removedKey: string): void { + pending.delete(removedKey) + if (!pending.size) disarm() +} + +function disarm(): void { + if (cancelScheduledRun) { cancelScheduledRun() cancelScheduledRun = undefined } } function drain(bypassActiveGate: boolean): void { - cancelScheduledRun?.() - cancelScheduledRun = undefined + disarm() if (!pending.size || !gates) return const entries = [...pending.values()] pending.clear() @@ -136,7 +154,7 @@ function applyProjection(entry: PendingProjection, bypassActiveGate: boolean): v return } if (!gate.isLatestMessageWindow(entry.instanceId, entry.sessionId)) return - projectOpenCodeMessages(entry.instanceId, entry.sessionId, entry.data, true, entry.confirmPending) + gate.projectPage(entry.instanceId, entry.sessionId, entry.data, entry.confirmPending) if (entry.updateUsage) gate.onProjectedUsage(entry.instanceId, entry.sessionId) } diff --git a/packages/ui/tests/browser/history-navigation.test.ts b/packages/ui/tests/browser/history-navigation.test.ts index b0ea9d6e5f..47c8268f3c 100644 --- a/packages/ui/tests/browser/history-navigation.test.ts +++ b/packages/ui/tests/browser/history-navigation.test.ts @@ -982,7 +982,12 @@ test("resident streaming keeps a distant timeline marker mounted", async () => { return element.isConnected }) assert(retained, "a resident content update must not remount an unchanged distant click target") - assert((await snapshot(f.page)).ids.includes("msg_streaming"), "the resident transcript did receive the streamed message") + // The streamed row publishes on its coalesced frame; wait for it before + // rechecking the marker, so the identity assertion protects the DOM + // invariant instead of racing the publication. + await f.page.waitForFunction(() => (window as any).fixture.snapshot().ids.includes("msg_streaming"), undefined, { timeout: 15_000 }) + assert.equal(await marker.evaluate(element => element.isConnected), true, + "the distant marker stays mounted after the streamed row publishes") assert.deepEqual(f.errors, []) } finally { await f.close() } }) diff --git a/packages/ui/tests/browser/streaming-projection.test.ts b/packages/ui/tests/browser/streaming-projection.test.ts index 16963bb55c..0d0c42449b 100644 --- a/packages/ui/tests/browser/streaming-projection.test.ts +++ b/packages/ui/tests/browser/streaming-projection.test.ts @@ -3,8 +3,10 @@ import { after, before, test } from "node:test" import { borrowFixture } from "./fixture-borrow" import { chromium, type Browser, type Page } from "playwright" import { armFixtureNavigationBudget } from "./fixture-navigation-budget" +import { fixtureEventStream } from "./fixture-event-stream" let browser: Browser, url: string +let stream: Awaited> before(async () => { const fixture = await borrowFixture({ name: "streaming-projection", @@ -13,23 +15,33 @@ before(async () => { ], }) url = fixture.url + stream = await fixtureEventStream(url) browser = await chromium.launch({ executablePath: process.env.CODENOMAD_BROWSER_PATH || undefined }) armFixtureNavigationBudget(browser) }) -after(async () => { await browser?.close() }) +after(async () => { try { await browser?.close() } finally { await stream?.close() } }) async function open(run: (page: Page) => Promise) { const page = await browser.newPage({ viewport: { width: 1200, height: 800 } }) const errors: string[] = [] page.on("pageerror", error => errors.push(error.message)) + page.on("console", message => { if (message.type() === "error") errors.push(message.text()) }) + page.on("response", response => { if (response.status() >= 400) errors.push(`${response.status()} ${response.url()}`) }) + page.on("requestfailed", request => errors.push(`${request.method()} ${request.url()}: ${request.failure()?.errorText}`)) await page.route("**/*", route => { const request = new URL(route.request().url()) if (request.hostname !== "127.0.0.1") return route.abort() + if (request.pathname === "/workspaces/streaming-fixture/yolo/sessions/active" && route.request().method() === "GET") { + return route.fulfill({ json: { enabled: false } }) + } if (request.pathname.startsWith("/api/")) return route.fulfill({ json: {} }) return route.continue() }) try { + await stream.route(page) + const opened = page.waitForResponse(response => new URL(response.url()).pathname === "/api/events") await page.goto(url, { waitUntil: "domcontentloaded" }) + assert.match((await opened).headers()["content-type"], /text\/event-stream/) await page.waitForFunction(() => Boolean((window as any).streamingFixture), undefined, { timeout: 60_000 }) await page.waitForTimeout(300) await run(page) @@ -41,6 +53,7 @@ const snapshot = (page: Page) => page.evaluate(() => (window as any).streamingFi const frames = (page: Page) => page.evaluate(() => new Promise(resolve => requestAnimationFrame(() => requestAnimationFrame(() => resolve())))) const joined = (count: number) => Array.from({ length: count }, (_, index) => `[${index}]`).join("") +const spacedText = (count: number) => Array.from({ length: count }, (_, index) => `<${index}>`).join("") /** Starts a reply and lets its own frame land before counts reset. */ const startSettled = async (page: Page, messageId?: string) => { await page.evaluate((id) => (window as any).streamingFixture.startReply(id ?? undefined), messageId) @@ -72,7 +85,8 @@ test("spaced deltas across tasks stay frame-coalesced and complete", () => open( await frames(page) const state = await snapshot(page) console.log("streaming-spaced", JSON.stringify({ variant: process.env.CODENOMAD_STREAMING_VARIANT ?? "default", spaced, projections: state.projections })) - assert.ok(state.text.endsWith("<23>"), "the projected page carries every spaced delta") + assert.equal(state.text, spacedText(24), "the projected page carries every spaced delta, exactly once, in order") + assert.ok(state.projections < 24, `frame coalescing projects fewer passes than deltas (got ${state.projections})`) })) test("visibility changes and session switches flush pending pages", () => open(async page => { From 4df39cb51a5561b5bec04658d960001b37484f17 Mon Sep 17 00:00:00 2001 From: heunghingwan Date: Thu, 5 Nov 2026 16:22:51 +0800 Subject: [PATCH 5/6] fix(transcript): publish a retiring reducer's final coalesced page Round-2 review: the shared destroy boundary cancelled the page a late tool settlement had just scheduled, so retiring the drained reducer right after session.tool.success dropped the completed output and the display kept showing the running row. Normal retirement now publishes the session's pending page (and its usage obligations) through a scoped flush before the entry goes away, while resets, eviction and destructive teardown stay cancel-only. The scoped drain leaves other sessions on their armed frame. The history-navigation marker check pins the original node with an ElementHandle, so a clone replacing it during the published frame can no longer satisfy the identity assertion the way a re-resolving Locator did. Covered by "a retirement-scoped flush publishes the pending final page and its usage". Verified: 259/259 across the transcript/instance unit families, typecheck clean, streaming-projection 3/3, compaction -responsiveness 5/5, the history case green, and the isolated native pruning UI run passes all nine checks. --- packages/ui/src/stores/opencode-data.ts | 6 +++- .../stores/opencode-projection-frame.test.ts | 23 ++++++++++++ .../src/stores/opencode-projection-frame.ts | 35 ++++++++++++++----- .../tests/browser/history-navigation.test.ts | 9 +++-- 4 files changed, 61 insertions(+), 12 deletions(-) diff --git a/packages/ui/src/stores/opencode-data.ts b/packages/ui/src/stores/opencode-data.ts index 19b3b7cb54..a9e594063a 100644 --- a/packages/ui/src/stores/opencode-data.ts +++ b/packages/ui/src/stores/opencode-data.ts @@ -16,7 +16,7 @@ import { } from "./opencode-data-events" import { mergePageWithSnapshot, snapshotRows, type OpenCodeRowSnapshot } from "./opencode-data-read-merge" import { createReconnect, type ReconnectCapture } from "./opencode-reconnect" -import { cancelTranscriptProjections } from "./opencode-projection-frame" +import { cancelTranscriptProjections, flushTranscriptProjections } from "./opencode-projection-frame" const log = getLogger("session") @@ -921,6 +921,10 @@ function retireDrainedTranscript(instanceId: string, sessionId: string, transcri // settle only in it. const data = transcript.entry.data if (data.session.pending.list(sessionId).length || data.session.message.list(sessionId).some(isUnsettledAtSettlement)) return + // The reducer's final coalesced page — e.g. a tool that settled after the + // execution did — publishes here, before normal retirement destroys the + // entry; resets, eviction and destructive teardown cancel instead. + flushTranscriptProjections(instanceId, sessionId) destroyOpenCodeData(instanceId, sessionId) } diff --git a/packages/ui/src/stores/opencode-projection-frame.test.ts b/packages/ui/src/stores/opencode-projection-frame.test.ts index d55828ac37..8f79b5be74 100644 --- a/packages/ui/src/stores/opencode-projection-frame.test.ts +++ b/packages/ui/src/stores/opencode-projection-frame.test.ts @@ -7,6 +7,7 @@ import { sdkManager } from "../lib/sdk-manager.ts" import { addInstance, handleInstanceInvalidation, removeInstance } from "./instances.ts" import { messageStoreBus } from "./message-v2/bus.ts" import { applyOpenCodeDataEvent, destroyOpenCodeData, projectOpenCodeMessages } from "./opencode-data.ts" +import { flushTranscriptProjections } from "./opencode-projection-frame.ts" import { setActiveSession, setSessions, getSessionInfo } from "./session-state.ts" const sessionId = "ses_frame" @@ -327,6 +328,28 @@ test("a merged frame keeps the last event's confirmation and accumulates usage", } }) +test("a retirement-scoped flush publishes the pending final page and its usage", ctx => { + const clock = frameClock() + const f = fixture("frame-retire-flush") + try { + const hydration = ctx.mock.method(f.store, "hydrateMessages") + const { streamId, joined } = f.stream(8) + f.dispatch("session.step.ended", { assistantMessageID: "msg_live", finish: "stop", cost: 2, tokens: { input: 30, output: 88, reasoning: 0, cache: { read: 0, write: 0 } } }) + assert.equal(hydration.mock.callCount(), 0, "the final tool page only scheduled") + // Normal retirement publishes the session's page before destroying the + // reducer; the display must not stay on the pre-settlement running row. + flushTranscriptProjections(f.instanceId, sessionId) + assert.equal(hydration.mock.callCount(), 1, "the scoped flush publishes without a frame") + assert.equal(f.storeText(streamId), joined) + assert.notEqual(getSessionInfo(f.instanceId, sessionId), undefined, "usage obligations publish with the page") + clock.fire() + assert.equal(hydration.mock.callCount(), 1, "nothing further remains scheduled") + } finally { + clock.restore() + f.cleanup() + } +}) + test("a destructive event projects synchronously and subsumes the pending frame", ctx => { const clock = frameClock() const f = fixture("frame-destructive") diff --git a/packages/ui/src/stores/opencode-projection-frame.ts b/packages/ui/src/stores/opencode-projection-frame.ts index 8a9486729b..414e19f805 100644 --- a/packages/ui/src/stores/opencode-projection-frame.ts +++ b/packages/ui/src/stores/opencode-projection-frame.ts @@ -87,10 +87,22 @@ export function scheduleTranscriptProjection( requestScheduledRun() } -/** Publishes every pending projection now. Selection changes call this so a - * page admitted under the previous selection still lands before its gates move. */ -export function flushTranscriptProjections(): void { - drain(true) +/** Publishes pending projections now. Selection changes call this so a + * page admitted under the previous selection still lands before its gates move; + * a normal reducer retirement publishes its session's final page (including + * usage obligations) before the reducer goes away, while resets, eviction and + * destructive teardown stay cancel-only. An optional scope publishes only the + * matching sessions and leaves the rest on their frame. */ +export function flushTranscriptProjections(instanceId?: string, sessionId?: string): void { + if (instanceId === undefined) { + drain(true) + return + } + const prefix = `${instanceId}\u0000` + const matches = sessionId === undefined + ? (key: string) => key.startsWith(prefix) + : (key: string) => key === projectionKey(instanceId, sessionId) + drain(true, matches) } /** Removes and returns a session's pending projection. A synchronous @@ -128,11 +140,16 @@ function disarm(): void { } } -function drain(bypassActiveGate: boolean): void { +function drain(bypassActiveGate: boolean, matches?: (key: string) => boolean): void { + if (!pending.size || !gates) { + disarm() + return + } + const keys = matches === undefined ? [...pending.keys()] : [...pending.keys()].filter(matches) + if (!keys.length) return disarm() - if (!pending.size || !gates) return - const entries = [...pending.values()] - pending.clear() + const entries = keys.map(key => pending.get(key)!) + for (const key of keys) pending.delete(key) for (const entry of entries) { try { applyProjection(entry, bypassActiveGate) @@ -142,6 +159,8 @@ function drain(bypassActiveGate: boolean): void { }) } } + // A scoped drain leaves other sessions pending; keep their frame armed. + if (pending.size) requestScheduledRun() } function applyProjection(entry: PendingProjection, bypassActiveGate: boolean): void { diff --git a/packages/ui/tests/browser/history-navigation.test.ts b/packages/ui/tests/browser/history-navigation.test.ts index 47c8268f3c..f1129ff3bd 100644 --- a/packages/ui/tests/browser/history-navigation.test.ts +++ b/packages/ui/tests/browser/history-navigation.test.ts @@ -977,7 +977,10 @@ test("resident streaming keeps a distant timeline marker mounted", async () => { await f.page.locator(".message-timeline").evaluate(element => { element.scrollTop = element.scrollHeight * 250 / 1500 }) const marker = f.page.locator('.message-timeline-segment[data-message-id="msg_00250"]') await marker.waitFor() - const retained = await marker.evaluate(element => { + // A Locator re-resolves on every evaluate; an ElementHandle pins the + // original node so the identity check catches a clone replacing it. + const markerNode = await marker.elementHandle() + const retained = await markerNode!.evaluate(element => { ;(window as any).fixture.stream("A new resident response while browsing old timeline markers") return element.isConnected }) @@ -986,8 +989,8 @@ test("resident streaming keeps a distant timeline marker mounted", async () => { // rechecking the marker, so the identity assertion protects the DOM // invariant instead of racing the publication. await f.page.waitForFunction(() => (window as any).fixture.snapshot().ids.includes("msg_streaming"), undefined, { timeout: 15_000 }) - assert.equal(await marker.evaluate(element => element.isConnected), true, - "the distant marker stays mounted after the streamed row publishes") + assert.equal(await markerNode!.evaluate(element => element.isConnected), true, + "the original distant marker node stays mounted after the streamed row publishes") assert.deepEqual(f.errors, []) } finally { await f.close() } }) From 6f7f3f129cff7695abee2a732e75da1a99551988 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pascal=20Andr=C3=A9?= Date: Thu, 5 Nov 2026 10:15:31 +0100 Subject: [PATCH 6/6] fix(transcript): keep other sessions' wake through a scoped projection flush A normal reducer retirement publishes its own session's pending page through a scoped flush. That flush cancelled the shared wake and armed a fresh one for the sessions it left pending, so in a hidden window every retirement elsewhere pushed an unrelated session's 250ms publication back again; a steady trickle of late tool results could postpone it indefinitely. A scoped drain now leaves the armed wake alone and only disarms once nothing remains pending; projections scheduled during the drain still request a wake, which is a no-op while one is armed. Tests: the retirement case now goes through real events (a tool still running at settlement, then its late success), so dropping the flush before retirement fails it; a hidden-window case retires late tools in four other instances 100ms apart and requires the waiting session to publish on its original 250ms wake, which fails with the re-arming drain. --- .../stores/opencode-projection-frame.test.ts | 90 +++++++++++++++---- .../src/stores/opencode-projection-frame.ts | 6 +- 2 files changed, 79 insertions(+), 17 deletions(-) diff --git a/packages/ui/src/stores/opencode-projection-frame.test.ts b/packages/ui/src/stores/opencode-projection-frame.test.ts index 8f79b5be74..a07b0c9261 100644 --- a/packages/ui/src/stores/opencode-projection-frame.test.ts +++ b/packages/ui/src/stores/opencode-projection-frame.test.ts @@ -6,8 +6,7 @@ import type { OpenCodeEvent } from "@opencode/client" import { sdkManager } from "../lib/sdk-manager.ts" import { addInstance, handleInstanceInvalidation, removeInstance } from "./instances.ts" import { messageStoreBus } from "./message-v2/bus.ts" -import { applyOpenCodeDataEvent, destroyOpenCodeData, projectOpenCodeMessages } from "./opencode-data.ts" -import { flushTranscriptProjections } from "./opencode-projection-frame.ts" +import { applyOpenCodeDataEvent, captureOpenCodeRowSnapshot, destroyOpenCodeData, projectOpenCodeMessages } from "./opencode-data.ts" import { setActiveSession, setSessions, getSessionInfo } from "./session-state.ts" const sessionId = "ses_frame" @@ -328,28 +327,89 @@ test("a merged frame keeps the last event's confirmation and accumulates usage", } }) -test("a retirement-scoped flush publishes the pending final page and its usage", ctx => { +/** Settles an execution while its tool still runs: the settlement read returns + * the running tool, so the reducer stays resident until the tool's terminal event. */ +async function settleWithRunningTool(f: ReturnType) { + const client = (sdkManager as any).clients.get(`${f.instanceId}:/workspaces/${f.instanceId}/instance`) + client.message.list = async () => ({ data: [{ + id: "msg_tool", type: "assistant", agent: "build", model, time: { created: 500, completed: 600 }, + content: [{ type: "tool", id: "tool", name: "read", time: { created: 501, ran: 502 }, state: { status: "running", input: {}, metadata: {} } }], + }], cursor: {} }) + f.dispatch("session.step.started", { assistantMessageID: "msg_tool", agent: "build", model, started: 500 }) + f.dispatch("session.tool.input.started", { assistantMessageID: "msg_tool", id: "tool", name: "read" }) + f.dispatch("session.tool.called", { assistantMessageID: "msg_tool", id: "tool", input: {}, executed: true }) + f.dispatch("session.step.ended", { assistantMessageID: "msg_tool", finish: "stop", cost: 3, tokens: { input: 100, output: 12, reasoning: 1, cache: { read: 2, write: 3 } } }) + f.dispatch("session.execution.succeeded") + await new Promise(resolve => setImmediate(resolve)) +} + +function finishTool(f: ReturnType) { + f.dispatch("session.tool.success", { assistantMessageID: "msg_tool", id: "tool", content: [{ type: "text", text: "finished" }], metadata: {}, executed: true }) +} + +test("a late tool result publishes before its drained reducer retires", async () => { const clock = frameClock() - const f = fixture("frame-retire-flush") + const f = fixture("frame-retire-late-tool") + const tool = () => Object.values(f.store.getMessage("msg_tool")?.parts ?? {}) + .find(part => part.data.type === "tool")?.data as any try { - const hydration = ctx.mock.method(f.store, "hydrateMessages") - const { streamId, joined } = f.stream(8) - f.dispatch("session.step.ended", { assistantMessageID: "msg_live", finish: "stop", cost: 2, tokens: { input: 30, output: 88, reasoning: 0, cache: { read: 0, write: 0 } } }) - assert.equal(hydration.mock.callCount(), 0, "the final tool page only scheduled") - // Normal retirement publishes the session's page before destroying the - // reducer; the display must not stay on the pre-settlement running row. - flushTranscriptProjections(f.instanceId, sessionId) - assert.equal(hydration.mock.callCount(), 1, "the scoped flush publishes without a frame") - assert.equal(f.storeText(streamId), joined) - assert.notEqual(getSessionInfo(f.instanceId, sessionId), undefined, "usage obligations publish with the page") + await settleWithRunningTool(f) + assert.equal(tool()?.state?.status, "running") + assert.ok(captureOpenCodeRowSnapshot(f.instanceId, sessionId).size > 0, "the running tool keeps its reducer resident") + finishTool(f) + assert.equal(captureOpenCodeRowSnapshot(f.instanceId, sessionId).size, 0, "the terminal tool event retired the reducer") + // Retirement publishes the scheduled final page instead of cancelling it. + assert.equal(tool()?.state?.status, "completed") + assert.equal(tool()?.state?.output, "finished") + assert.equal(getSessionInfo(f.instanceId, sessionId)?.actualUsageTokens, 118) clock.fire() - assert.equal(hydration.mock.callCount(), 1, "nothing further remains scheduled") + assert.equal(tool()?.state?.status, "completed") } finally { clock.restore() f.cleanup() } }) +test("retirements in other instances keep a hidden session's original wake", async () => { + const retiring = Array.from({ length: 4 }, (_, index) => fixture(`frame-wake-retire-${index}`)) + const waiting = fixture("frame-wake-waiting") + const realSetTimeout = globalThis.setTimeout + const realClearTimeout = globalThis.clearTimeout + // A virtual clock for the hidden-window interval only. + let now = 0 + let nextTimer = 0 + const timers = new Map void }>() + try { + for (const f of retiring) await settleWithRunningTool(f) + ;(globalThis as any).document = { hidden: true } + ;(globalThis as any).setTimeout = (callback: () => void, ms: number, ...args: unknown[]) => { + if (ms !== 250) return (realSetTimeout as any)(callback, ms, ...args) + timers.set(++nextTimer, { at: now + ms, callback }) + return nextTimer + } + ;(globalThis as any).clearTimeout = (timer: any) => { + if (!timers.delete(timer)) realClearTimeout(timer) + } + const { streamId, joined } = waiting.stream(3) + for (const f of retiring) { + now += 100 + for (const [timer, entry] of [...timers]) { + if (entry.at > now) continue + timers.delete(timer) + entry.callback() + } + finishTool(f) + } + assert.equal(waiting.storeText(streamId), joined, "unrelated retirements must not postpone the 250ms publication") + } finally { + globalThis.setTimeout = realSetTimeout + globalThis.clearTimeout = realClearTimeout + delete (globalThis as any).document + for (const f of retiring) f.cleanup() + waiting.cleanup() + } +}) + test("a destructive event projects synchronously and subsumes the pending frame", ctx => { const clock = frameClock() const f = fixture("frame-destructive") diff --git a/packages/ui/src/stores/opencode-projection-frame.ts b/packages/ui/src/stores/opencode-projection-frame.ts index 414e19f805..11a8535079 100644 --- a/packages/ui/src/stores/opencode-projection-frame.ts +++ b/packages/ui/src/stores/opencode-projection-frame.ts @@ -147,9 +147,11 @@ function drain(bypassActiveGate: boolean, matches?: (key: string) => boolean): v } const keys = matches === undefined ? [...pending.keys()] : [...pending.keys()].filter(matches) if (!keys.length) return - disarm() const entries = keys.map(key => pending.get(key)!) for (const key of keys) pending.delete(key) + // A scoped drain leaves the other sessions on their already armed wake: + // re-arming it would let repeated retirements elsewhere postpone them. + if (!pending.size) disarm() for (const entry of entries) { try { applyProjection(entry, bypassActiveGate) @@ -159,7 +161,7 @@ function drain(bypassActiveGate: boolean, matches?: (key: string) => boolean): v }) } } - // A scoped drain leaves other sessions pending; keep their frame armed. + // Projections scheduled while draining need a wake (a no-op when one is armed). if (pending.size) requestScheduledRun() }