diff --git a/.github/workflows/pr-build.yml b/.github/workflows/pr-build.yml index 912b965d5..c6233af77 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/dev-docs/CACHE_REFRESH_CONVENTIONS.md b/dev-docs/CACHE_REFRESH_CONVENTIONS.md index 370997f9b..ced723781 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 aafa136c8..128ab622a 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 983da5f13..e54d8ce9b 100644 --- a/packages/ui/src/stores/instances.ts +++ b/packages/ui/src/stores/instances.ts @@ -53,7 +53,15 @@ 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" import { upsertPermissionV2, removePermissionV2, removeMessageV2 } from "./message-v2/bridge" @@ -661,6 +669,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 +748,7 @@ function releaseInstanceResources(instanceId: string) { sdkManager.destroyClientsForInstance(instanceId) } destroyOpenCodeData(instanceId) + cancelTranscriptProjections(instanceId) sseManager.seedStatus(instanceId, "disconnected") } @@ -1487,6 +1497,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 +1566,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 +2257,36 @@ const USAGE_EVENT_TYPES = new Set([ "session.step.failed", ]) +// 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. 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", + "session.revert.committed", + "session.execution.succeeded", + "session.execution.failed", + "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 }) => { @@ -2255,10 +2297,15 @@ const publishTranscriptChunk: NonNullable USAGE_EVENT_TYPES.has(event.type) || event.type === "session.compaction.delta")) { - updateSessionInfo(instanceId, sessionId) + const options = { + confirmPending: sessionEvents.at(-1)?.type !== "session.inbox.enqueued", + updateUsage: sessionEvents.some((event) => 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 { @@ -2287,14 +2334,18 @@ function handleInstanceInvalidation(instanceId: string, event: Parameters, preserveOmitted = true) => { @@ -2359,6 +2410,22 @@ 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, + // Frame publications always keep omitted history; only their confirmation + // flag varies (projectOpenCodeMessages takes preserveOmitted first). + projectPage: (instanceId, sessionId, data, confirmPending) => + projectOpenCodeMessages(instanceId, sessionId, data, true, confirmPending), +}) + sseManager.onConnectionLost = (instanceId, reason) => { const instance = instances().get(instanceId) if (!instance) { diff --git a/packages/ui/src/stores/opencode-data.ts b/packages/ui/src/stores/opencode-data.ts index b80e64cef..a9e594063 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, flushTranscriptProjections } from "./opencode-projection-frame" const log = getLogger("session") @@ -920,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) } @@ -1029,6 +1034,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 new file mode 100644 index 000000000..a07b0c926 --- /dev/null +++ b/packages/ui/src/stores/opencode-projection-frame.test.ts @@ -0,0 +1,433 @@ +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, captureOpenCodeRowSnapshot, destroyOpenCodeData, projectOpenCodeMessages } from "./opencode-data.ts" +import { setActiveSession, setSessions, getSessionInfo } 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, options: { seed?: number } = {}) { + 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 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 }) + } + 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("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 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() + } +}) + +/** 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-late-tool") + const tool = () => Object.values(f.store.getMessage("msg_tool")?.parts ?? {}) + .find(part => part.data.type === "tool")?.data as any + try { + 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(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") + 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 000000000..11a853507 --- /dev/null +++ b/packages/ui/src/stores/opencode-projection-frame.ts @@ -0,0 +1,224 @@ +import type { Data } from "@opencode/client/solid" +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. + * + * 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 + 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 + /** Publishes a reduced page into the message store. */ + projectPage(instanceId: string, sessionId: string, data: Data, confirmPending: boolean): 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 + +/** + * 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 + +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) 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 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 + * 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 { + 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) 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, 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 + 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) + } catch (error) { + log.error("Failed to project coalesced transcript page", { + instanceId: entry.instanceId, sessionId: entry.sessionId, error, + }) + } + } + // Projections scheduled while draining need a wake (a no-op when one is armed). + if (pending.size) requestScheduledRun() +} + +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 + gate.projectPage(entry.instanceId, entry.sessionId, entry.data, 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 : () => {} + 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) + } + 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 + // 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 f1c422d46..3915e2c90 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 4876ce40b..9f448893a 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 e88b5c992..3b88cd97f 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 000000000..bc97ad55d --- /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/history-navigation.test.ts b/packages/ui/tests/browser/history-navigation.test.ts index b0ea9d6e5..f1129ff3b 100644 --- a/packages/ui/tests/browser/history-navigation.test.ts +++ b/packages/ui/tests/browser/history-navigation.test.ts @@ -977,12 +977,20 @@ 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 }) 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 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() } }) 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 000000000..0d0c42449 --- /dev/null +++ b/packages/ui/tests/browser/streaming-projection.test.ts @@ -0,0 +1,102 @@ +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" +import { fixtureEventStream } from "./fixture-event-stream" + +let browser: Browser, url: string +let stream: Awaited> +before(async () => { + const fixture = await borrowFixture({ + name: "streaming-projection", + routes: [ + { path: "/streaming", html: '
' }, + ], + }) + url = fixture.url + stream = await fixtureEventStream(url) + browser = await chromium.launch({ executablePath: process.env.CODENOMAD_BROWSER_PATH || undefined }) + armFixtureNavigationBudget(browser) +}) +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) + 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("") +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) + 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.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 => { + 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") +})) diff --git a/packages/ui/tests/browser/tool-images.test.ts b/packages/ui/tests/browser/tool-images.test.ts index 375065b52..d9776b3af 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)