From 23aebaf4c2b0ca7d43867b96532dd417a39063d4 Mon Sep 17 00:00:00 2001 From: Gh0st352 Date: Sat, 19 Sep 2026 16:27:50 -0600 Subject: [PATCH] refactor: extract verified transcript transport foundation --- .../transcript-transport-foundation.md | 37 + package.json | 3 +- packages/types/src/vscode-extension-host.ts | 38 + scripts/check-transcript-transport.ts | 17 + .../__tests__/transcriptTransport.model.ts | 1042 +++++++++++++++++ .../__tests__/transcriptTransport.spec.ts | 945 +++++++++++++++ src/core/webview/transcriptTransport.ts | 412 +++++++ 7 files changed, 2493 insertions(+), 1 deletion(-) create mode 100644 docs/architecture/transcript-transport-foundation.md create mode 100644 scripts/check-transcript-transport.ts create mode 100644 src/core/webview/__tests__/transcriptTransport.model.ts create mode 100644 src/core/webview/__tests__/transcriptTransport.spec.ts create mode 100644 src/core/webview/transcriptTransport.ts diff --git a/docs/architecture/transcript-transport-foundation.md b/docs/architecture/transcript-transport-foundation.md new file mode 100644 index 0000000000..23e01d8d21 --- /dev/null +++ b/docs/architecture/transcript-transport-foundation.md @@ -0,0 +1,37 @@ +# Transcript transport foundation + +## Staged delivery + +This is the independently testable foundation extracted from [PR #1360](https://github.com/Zoo-Code-Org/Zoo-Code/pull/1360), related to [issue #630](https://github.com/Zoo-Code-Org/Zoo-Code/issues/630). It adds the transport implementation, additive wire declarations, driver tests, and a bounded model checker. **It does not activate the transport.** Task producers, provider ownership, renderer disposal, focus publication, resync handling, and the React receiver remain integration work in #1360. Existing state and legacy message delivery are unchanged. + +The split preserves the reviewed transport implementation rather than removing safety checks or weakening mutation limits. Against the extraction base, the foundation contains 262 changed executable extension lines; after it lands unchanged, #1360's extension contribution falls from 567 to 305. A separate commit or an unmerged prerequisite does not reduce #1360's scope against main. Both PRs require validation against their actual integration bases, and line scope passing does not establish mutation execution or review acceptance. + +## Ownership and protocol + +[`TranscriptTransport`](../../src/core/webview/transcriptTransport.ts:274) owns captured payloads and caller resolvers. Its pure [`reduceTranscriptTransport()`](../../src/core/webview/transcriptTransport.ts:92) owns generation, task/instance-scoped job descriptors, task-keyed sequence allocation, snapshot progress, and a physical-send barrier. The driver and bounded explorer share that reducer and [`transcriptFrameMessage()`](../../src/core/webview/transcriptTransport.ts:227); the checker is not a second queue implementation. + +- Requests must match the current generation, task, and originating instance before deep cloning. Reducer admission checks ownership again after capture. An absent instance cannot adopt a live instance. +- Append/update requests increment the task's sequence and carry complete captured message values. Empty deltas are rejected without capture or allocation. Snapshots optionally bump the sequence and send start, bounded chunks of 200 messages, and end; empty snapshots send start/end only. +- Invalidation releases queued payloads and the active snapshot's unsent suffix, settling discarded waiting callers. It does not reset the physical-send barrier: an already-started send settles before the next generation can send. +- Renderer shutdown closes admission, releases all owned work, settles pending callers, clears callbacks, and detaches the physical completion slot. Late completion cannot pump a replacement renderer. Reopen installs fresh callbacks and advances generation without resetting IDs or sequences. +- Task-sequence pruning is explicit. Sequence monotonicity is not global across removed/recreated task lifetimes. Snapshot identity is opaque and distinct from the transcript revision. + +The additive fields in [`ExtensionMessage`](../../packages/types/src/vscode-extension-host.ts:30) describe dedicated frames only. They do not remove legacy transcript state or implement consumer validation. The future adapter must supply exact task/instance focus and bind sends to the originating renderer. The future receiver must publish focus before accepting frames, reject stale scopes, validate revisions and contiguous snapshot ranges, and apply snapshots atomically. + +## Verification + +Run the focused checker with **pnpm transcript-transport:model-check**. It is also included in **pnpm lifecycle:model-check** through [`package.json`](../../package.json:16), without changing persisted lifecycle transitions. Run the focused Vitest suite from the extension package: **pnpm --dir src exec vitest run core/webview/**tests**/transcriptTransport.spec.ts**. + +[`transcriptTransport.spec.ts`](../../src/core/webview/__tests__/transcriptTransport.spec.ts:1) verifies admission, exact chunk boundaries, deep capture, failure recovery, held physical sends, repeated invalidation, same-task instance replacement, shutdown/reopen, callback detachment, and successful or rejected late completions. It also runs scenario coverage and one exhaustive check per injected fault, with unchanged test timeouts and exploration bounds. + +The deterministic breadth-first [`explorer`](../../src/core/webview/__tests__/transcriptTransport.model.ts:1) checks ten scenarios with two task IDs plus no task, at most two instances of the first task, at most five admitted jobs, two invalidations, two shutdown calls and one reopen, snapshots up to four messages with chunk size two, and at most one failed physical send per trace. Each scenario has a 30,000-state budget and depth limit 40. Exceeding either budget fails closed. The extraction's diagnostics are 40,099 states and 61,303 transitions, with all 27 actions and 34 named landmarks reached. Twenty-four test-only reducer, wire, and receiver-policy faults must yield their expected shortest counterexamples. + +Invariants cover scope ownership, a single physical send per renderer, immediate release of obsolete work, exactly-once caller settlement, monotonic allocation within retained task scope, exact captured ranges, atomic snapshot application in the independent receiver oracle, and isolation of retired renderer completions. The [`CLI checker`](../../scripts/check-transcript-transport.ts:1) reports scenario counts, action/landmark coverage, bounds, and fault witnesses. + +## Limits and integration obligations + +An already-initiated physical send cannot be revoked. A same-instance end marker started before invalidation may still complete; the guarantee is that no later stale post is initiated. Shutdown detaches ownership but does not cancel an editor-owned operation or prove immediate garbage collection. + +The receiver is an independent oracle, not the React implementation. The model assumes ordered successful delivery at settlement and no delivery for rejection. It does not prove browser timers, real delivery acknowledgements, provider focus/metadata ordering, persistence, rendering, restart behavior, or adapter callback binding. Driver tests verify concrete callback detachment; the full provider and UI integration tests remain in #1360. + +There is no fairness or unbounded liveness claim. A held physical send blocks that renderer until settlement or shutdown. The bounded model does not prove arbitrary queue lengths, sequence overflow, unlimited instance replacement or renderer cycles, instance-ID collision resistance, or arbitrary payload values. It neither imports nor modifies persisted task lifecycle reducers or scheduler transitions. diff --git a/package.json b/package.json index e4d3867e0b..c75151114c 100644 --- a/package.json +++ b/package.json @@ -13,8 +13,9 @@ "check-types": "turbo check-types --log-order grouped --output-logs new-only", "test": "turbo test --log-order grouped --output-logs new-only", "test:mutation-ci": "node --test scripts/stryker-diff.test.mjs", - "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts && tsx scripts/check-provider-handoff-scheduler.ts && pnpm cleanup-protocol:model-check && pnpm parser-scope:model-check && tsx scripts/check-completion-persistence.ts && tsx scripts/check-delegated-mode-readers.ts", + "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts && tsx scripts/check-provider-handoff-scheduler.ts && pnpm cleanup-protocol:model-check && pnpm parser-scope:model-check && tsx scripts/check-completion-persistence.ts && tsx scripts/check-delegated-mode-readers.ts && pnpm transcript-transport:model-check", "fanout-protocol:model-check": "tsx scripts/check-task-fanout-protocol.ts", + "transcript-transport:model-check": "tsx scripts/check-transcript-transport.ts", "cleanup-protocol:model-check": "tsx scripts/check-task-cleanup-protocol.ts", "parser-scope:model-check": "node scripts/run-native-tool-call-parser-scoping.mjs", "mcp:integration-check": "tsx scripts/check-mcp-oauth-integration.ts", diff --git a/packages/types/src/vscode-extension-host.ts b/packages/types/src/vscode-extension-host.ts index 5f6b579779..1112866673 100644 --- a/packages/types/src/vscode-extension-host.ts +++ b/packages/types/src/vscode-extension-host.ts @@ -38,6 +38,11 @@ export interface ExtensionMessage { | "workspaceUpdated" | "invoke" | "messageUpdated" + | "clineMessageAppended" + | "clineMessageUpdated" + | "clineMessagesSnapshotStart" + | "clineMessagesSnapshotChunk" + | "clineMessagesSnapshotEnd" | "mcpServers" | "enhancedPrompt" | "commitSearchResults" @@ -138,7 +143,40 @@ export interface ExtensionMessage { isActive: boolean path?: string }> + /** + * Task scope for dedicated transcript deltas and snapshot frames. Omitted for + * the no-task scope, whose snapshot is empty with sequence 0. + */ + taskId?: string + /** + * Originating task instance for dedicated transcript frames. Both taskId and + * taskInstanceId must match receiver focus; never retag an old instance's frame. + * Omitted for no-task frames and legacy consumers without instance metadata. + */ + taskInstanceId?: string clineMessage?: ClineMessage + /** + * Nonempty, ordered slice for a snapshot chunk, beginning at snapshotStartIndex. + * Buffer until the matching end frame; start/end frames carry no messages. + * Legacy full transcripts remain in state.clineMessages. + */ + clineMessages?: ClineMessage[] + /** + * Nonnegative safe-integer revision scoped to a task in retained transport state. + * Accepted deltas and bumping snapshots increment it; resync snapshots reuse it. + * Every frame of one snapshot shares a revision. No-task snapshots use 0. + * This field does not change legacy state/messageUpdated delivery semantics. + */ + clineMessagesSeq?: number + /** Opaque, nonempty correlation ID shared by a snapshot's start, chunks, and end. */ + snapshotId?: string + /** Contiguous zero-based chunk offset; omitted on start/end frames. */ + snapshotStartIndex?: number + /** + * Total message count declared by start and repeated by end. Apply only when + * the complete buffered count matches. Zero means start/end without chunks. + */ + snapshotTotal?: number routerModels?: RouterModels openAiModels?: string[] ollamaModels?: ModelRecord diff --git a/scripts/check-transcript-transport.ts b/scripts/check-transcript-transport.ts new file mode 100644 index 0000000000..af824d1d29 --- /dev/null +++ b/scripts/check-transcript-transport.ts @@ -0,0 +1,17 @@ +import { + checkTranscriptTransportModel, + TRANSPORT_MODEL_BOUNDS, +} from "../src/core/webview/__tests__/transcriptTransport.model" + +const result = checkTranscriptTransportModel() +console.log(`Transcript transport model passed; bounds=${JSON.stringify(TRANSPORT_MODEL_BOUNDS)}`) +for (const scenario of result.results) { + console.log( + `${scenario.name}: ${scenario.states} states, ${scenario.transitions} transitions, maximum depth ${scenario.maximumDepth}`, + ) +} +console.log(`Actions (${result.actions.length}): ${result.actions.join(", ")}`) +console.log(`Landmarks (${result.landmarks.length}): ${result.landmarks.join(", ")}`) +for (const counterexample of result.counterexamples) { + console.log(`Mutant ${counterexample.name}: ${counterexample.violation}\n ${counterexample.trace.join(" -> ")}`) +} diff --git a/src/core/webview/__tests__/transcriptTransport.model.ts b/src/core/webview/__tests__/transcriptTransport.model.ts new file mode 100644 index 0000000000..06bb7c6ed9 --- /dev/null +++ b/src/core/webview/__tests__/transcriptTransport.model.ts @@ -0,0 +1,1042 @@ +import type { ExtensionMessage } from "@roo-code/types" +import { + createTranscriptTransportState, + reduceTranscriptTransport, + transcriptFrameMessage, + type TranscriptAction, + type TranscriptFrame, + type TranscriptJob, + type TranscriptTransportState, +} from "../transcriptTransport" + +type TaskId = "a" | "b" +type Intent = + | "snapshot" + | "append" + | "update" + | "resync" + | "invalidate" + | "switch" + | "clear" + | "focus" + | "stale-snapshot" + | "replace-instance" + | "sync-instance" + | "stale-instance-append" + | "stale-instance-snapshot" + | "empty-append" + | "empty-update" + | "empty-snapshot" + | "shutdown" + | "reopen" +type Capture = { + job: TranscriptJob + renderer: number + taskInstanceId: string | undefined + scope: string + values: number[] + failed: boolean +} +type ModelState = { + transport: TranscriptTransportState + renderer: number + shutdowns: number + shutdownWithQueued: boolean + closedAdmissionRejected: boolean + retired: Array<{ frame: TranscriptFrame; message: ExtensionMessage }> + lateOutcomes: string[] + lateWhileSending: string[] + focus: TaskId | undefined + focusInstance: string | undefined + instanceSyncPending: boolean + staleInstanceRejected: boolean + discardedInstanceJobs: string[] + rejectedInstancePhases: string[] + appliedInstanceDeltas: string[] + producer: number + controller: number + failures: number + data: Record + epochs: Record + captures: Capture[] + payloads: number[] + callers: number[] + physical?: TranscriptFrame + physicalMessage?: ExtensionMessage + allocated: Record + sent: Record + visible: number[] + appliedSeq: number + staging?: { id: number; values: number[] } + committed: number[] + staleCompletions: number + staleCommitCompletions: number +} +type Scenario = { name: string; producer: Intent[]; controller: Intent[] } +type Event = { name: string; actor?: "producer" | "controller"; intent?: Intent; action?: TranscriptAction } +type Node = { state: ModelState; parent: number; event: string; depth: number } +type Reducer = typeof reduceTranscriptTransport +type ReceiverScope = Pick +type Faults = { + wire?: typeof transcriptFrameMessage + accepts?: (message: ExtensionMessage, scope: ReceiverScope) => boolean +} +type Mutation = Faults & { name: string; expected: string; reduce?: Reducer } + +export const TRANSPORT_MODEL_BOUNDS = { depth: 40, states: 30_000, chunkSize: 2, failures: 1 } as const +export const TRANSPORT_SCENARIOS: Scenario[] = [ + { + name: "queued-deltas-repeated-resync", + producer: ["snapshot", "append", "update"], + controller: ["resync", "resync"], + }, + { name: "task-switch-and-clear", producer: ["snapshot", "append", "snapshot"], controller: ["switch", "clear"] }, + { + name: "invalidation-and-recovery", + producer: ["snapshot", "update", "snapshot"], + controller: ["invalidate", "resync"], + }, + { + name: "focus-before-sync-and-stale-request", + producer: ["snapshot", "append", "update"], + controller: ["focus", "resync", "stale-snapshot"], + }, + { + name: "same-task-instance-snapshot-before-sync", + producer: ["snapshot", "stale-instance-append"], + controller: ["replace-instance", "sync-instance", "append", "update"], + }, + { + name: "same-task-instance-deltas-before-sync", + producer: ["append", "update", "stale-instance-snapshot"], + controller: ["replace-instance", "sync-instance"], + }, + { + name: "empty-deltas-before-valid-work", + producer: ["empty-append", "empty-update", "append", "empty-snapshot"], + controller: [], + }, + { + name: "renderer-disposal-and-reopen", + producer: ["snapshot", "append"], + controller: ["shutdown", "shutdown", "reopen", "snapshot"], + }, + { + name: "renderer-reopen-without-task", + producer: ["snapshot"], + controller: ["clear", "shutdown", "reopen", "empty-snapshot"], + }, + { + name: "renderer-reopen-with-new-instance", + producer: ["snapshot", "stale-instance-append"], + controller: ["shutdown", "replace-instance", "reopen", "sync-instance"], + }, +] +export const TRANSPORT_ACTIONS = [ + "snapshot", + "append", + "update", + "resync", + "invalidate", + "switch", + "clear", + "focus", + "stale-snapshot", + "replace-instance", + "sync-instance", + "stale-instance-append", + "stale-instance-snapshot", + "empty-append", + "empty-update", + "empty-snapshot", + "pump", + "start", + "chunk", + "end", + "settle", + "fail", + "discard", + "shutdown", + "reopen", + "late-resolve", + "late-reject", +] +export const TRANSPORT_LANDMARKS = { + "held-post-with-queued-delta": (s: ModelState) => + !!s.physical && s.transport.queue.some((job) => job.kind !== "snapshot"), + "repeated-invalidation-while-held": (s: ModelState) => + !!s.physical && s.transport.generation - s.physical.job.generation >= 2, + "cancelled-active-suffix-released": (s: ModelState) => + !!s.physical && s.physical.job.generation < s.transport.generation && !s.payloads.includes(s.physical.job.id), + "new-generation-waits-for-old-send": (s: ModelState) => + !!s.physical && s.physical.job.generation < s.transport.generation && s.transport.queue.length > 0, + "stale-physical-completion": (s: ModelState) => s.staleCompletions > 0, + "already-initiated-stale-end-can-complete": (s: ModelState) => s.staleCommitCompletions > 0, + "task-switch-with-held-send": (s: ModelState) => s.focus === "b" && s.physical?.job.taskId === "a", + "focus-changed-before-invalidation": (s: ModelState) => + s.focus === "b" && s.transport.generation === 0 && !!s.transport.active, + "clear-prunes-task-sequences": (s: ModelState) => !s.focus && s.transport.sequences.size === 0, + "empty-snapshot-committed": (s: ModelState) => s.committed.some((id) => s.captures[id - 1].job.total === 0), + "multi-chunk-snapshot-committed": (s: ModelState) => + s.committed.some((id) => s.captures[id - 1].job.total > TRANSPORT_MODEL_BOUNDS.chunkSize), + "failed-post-with-queued-recovery": (s: ModelState) => + s.failures > 0 && s.transport.queue.some((job) => job.kind === "snapshot"), + "snapshot-recovery-after-failure": (s: ModelState) => + s.committed.some((id) => s.captures.some((c) => c.failed && c.job.id < id)), + "delta-applied-after-snapshot": (s: ModelState) => + s.committed.length > 0 && s.appliedSeq > s.captures[s.committed.at(-1)! - 1].job.seq, + "same-task-instance-published-before-sync": (s: ModelState) => + s.focus === "a" && s.focusInstance === "a:1" && s.instanceSyncPending && s.transport.generation === 0, + "instance-replacement-with-held-send": (s: ModelState) => + s.focusInstance === "a:1" && s.physical?.job.taskInstanceId === "a:0", + "stale-instance-current-generation-rejected": (s: ModelState) => s.staleInstanceRejected, + "stale-instance-queued-job-discarded": (s: ModelState) => s.discardedInstanceJobs.includes("queued"), + "stale-instance-active-suffix-discarded": (s: ModelState) => s.discardedInstanceJobs.includes("active"), + "old-instance-end-ignored": (s: ModelState) => s.rejectedInstancePhases.includes("end"), + "old-instance-append-ignored": (s: ModelState) => s.rejectedInstancePhases.includes("append"), + "old-instance-update-ignored": (s: ModelState) => s.rejectedInstancePhases.includes("update"), + "new-instance-snapshot-committed": (s: ModelState) => + s.committed.some((id) => s.captures[id - 1].job.taskInstanceId === "a:1"), + "new-instance-append-applied": (s: ModelState) => s.appliedInstanceDeltas.includes("append"), + "new-instance-update-applied": (s: ModelState) => s.appliedInstanceDeltas.includes("update"), + "new-instance-recovers-after-old-end-rejected": (s: ModelState) => + s.rejectedInstancePhases.includes("end") && + s.committed.some((id) => s.captures[id - 1].job.taskInstanceId === "a:1") && + s.appliedInstanceDeltas.includes("append") && + s.appliedInstanceDeltas.includes("update"), + "shutdown-releases-held-and-queued-callers": (s: ModelState) => + s.shutdownWithQueued && s.transport.closed && s.callers.length === 0 && s.payloads.length === 0, + "repeated-shutdown-with-held-send": (s: ModelState) => s.shutdowns === 2 && s.retired.length > 0, + "closed-admission-rejected": (s: ModelState) => s.closedAdmissionRejected, + "reopened-renderer-sends-before-dead-renderer-settles": (s: ModelState) => + s.renderer > 0 && !!s.physical && s.retired.length > 0, + "late-resolve-does-not-settle-new-send": (s: ModelState) => s.lateWhileSending.includes("resolve"), + "late-reject-does-not-settle-new-send": (s: ModelState) => s.lateWhileSending.includes("reject"), + "reopened-snapshot-committed": (s: ModelState) => s.committed.some((id) => s.captures[id - 1].renderer > 0), + "reopened-no-task-snapshot-committed": (s: ModelState) => + s.committed.some((id) => { + const capture = s.captures[id - 1] + return capture.renderer > 0 && capture.job.taskId === undefined + }), +} satisfies Record boolean> + +function initialState(): ModelState { + return { + transport: createTranscriptTransportState(TRANSPORT_MODEL_BOUNDS.chunkSize), + renderer: 0, + shutdowns: 0, + shutdownWithQueued: false, + closedAdmissionRejected: false, + retired: [], + lateOutcomes: [], + lateWhileSending: [], + focus: "a", + focusInstance: "a:0", + instanceSyncPending: false, + staleInstanceRejected: false, + discardedInstanceJobs: [], + rejectedInstancePhases: [], + appliedInstanceDeltas: [], + producer: 0, + controller: 0, + failures: 0, + data: { a: [1, 2, 3], b: [7] }, + epochs: { a: 0, b: 0 }, + captures: [], + payloads: [], + callers: [], + allocated: {}, + sent: {}, + visible: [], + appliedSeq: 0, + committed: [], + staleCompletions: 0, + staleCommitCompletions: 0, + } +} + +function enabled(s: ModelState, scenario: Scenario): Event[] { + const events: Event[] = [] + for (const actor of ["producer", "controller"] as const) { + const intent = scenario[actor][s[actor]] + // A delayed old producer resumes only after the replacement has been published. + if (intent?.startsWith("stale-instance-") && s.focusInstance !== "a:1") continue + if (intent) events.push({ name: `${actor}:${intent}`, actor, intent }) + } + if (!s.transport.inFlight && (s.transport.active || s.transport.queue.length)) { + events.push({ + name: "pump", + action: { type: "pump", focusedTaskId: s.focus, focusedTaskInstanceId: s.focusInstance }, + }) + } + if (s.transport.inFlight) { + events.push({ name: "settle", action: { type: "settle", frame: s.transport.inFlight, success: true } }) + if (s.failures < TRANSPORT_MODEL_BOUNDS.failures) + events.push({ name: "fail", action: { type: "settle", frame: s.transport.inFlight, success: false } }) + } + for (const { frame } of s.retired) { + events.push({ name: "late-resolve", action: { type: "settle", frame, success: true } }) + if (s.failures < TRANSPORT_MODEL_BOUNDS.failures) + events.push({ name: "late-reject", action: { type: "settle", frame, success: false } }) + } + return events +} + +function requireInvariant(condition: unknown, message: string): asserts condition { + if (!condition) throw new Error(message) +} + +function acceptsFocusedMessage(message: ExtensionMessage, scope: ReceiverScope): boolean { + return message.taskId === scope.focus && message.taskInstanceId === scope.focusInstance +} + +/** Independent receiver oracle. It sees captured wire identity, not private generation tokens. */ +function deliver(s: ModelState, frame: TranscriptFrame, message: ExtensionMessage, faults: Faults): void { + const { job, phase } = frame + const accepted = (faults.accepts ?? acceptsFocusedMessage)(message, s) + const current = message.taskId === s.focus && message.taskInstanceId === s.focusInstance + requireInvariant(!accepted || current, "receiver accepted a stale-instance frame") + if (!accepted) { + if (message.taskId === s.focus && message.taskInstanceId !== s.focusInstance) { + s.rejectedInstancePhases = [...new Set([...s.rejectedInstancePhases, phase])].sort() + } + return + } + const capture = s.captures[job.id - 1] + const oldVisible = [...s.visible] + const oldSeq = s.appliedSeq + const seq = message.clineMessagesSeq ?? 0 + if (phase === "start") { + if (seq >= s.appliedSeq) s.staging = { id: job.id, values: [] } + } else if (phase === "chunk") { + if (s.staging?.id === job.id) { + requireInvariant(message.snapshotStartIndex === s.staging.values.length, "non-contiguous snapshot chunk") + s.staging.values.push(...(message.clineMessages ?? []).map((m) => m.ts)) + } + } else if (phase === "end") { + requireInvariant(s.staging?.id === job.id, "snapshot commit without matching start") + requireInvariant( + JSON.stringify(s.staging.values) === JSON.stringify(capture.values), + "snapshot commit before complete chunks", + ) + if (seq >= s.appliedSeq) { + s.visible = s.staging.values + s.appliedSeq = seq + s.committed.push(job.id) + } + s.staging = undefined + } else if (!s.staging && seq === s.appliedSeq + 1) { + requireInvariant(message.clineMessage, "delta lacks a wire payload") + if (phase === "append") s.visible.push(message.clineMessage.ts) + else if (s.visible.length) s.visible[0] = message.clineMessage.ts + s.appliedSeq = seq + if (message.taskInstanceId === "a:1") { + s.appliedInstanceDeltas = [...new Set([...s.appliedInstanceDeltas, phase])].sort() + } + } + if (phase === "start" || phase === "chunk") { + requireInvariant( + JSON.stringify(s.visible) === JSON.stringify(oldVisible), + "snapshot exposed a partial transcript", + ) + requireInvariant(s.appliedSeq === oldSeq, "snapshot applied sequence before commit") + } + requireInvariant(s.appliedSeq >= oldSeq, "applied sequence regressed within focus scope") +} + +class ModelViolation extends Error { + constructor( + message: string, + readonly state: ModelState, + ) { + super(message) + } +} + +function step(source: ModelState, event: Event, reducer: Reducer, coverage: Set, faults: Faults): ModelState { + const s = structuredClone(source) + try { + return executeStep(s, event, reducer, coverage, faults) + } catch (error) { + throw new ModelViolation(error instanceof Error ? error.message : String(error), s) + } +} + +function executeStep(s: ModelState, event: Event, reducer: Reducer, coverage: Set, faults: Faults): ModelState { + const apply = (action: TranscriptAction, values: number[] = []) => { + const before = s.transport + const transition = reducer(before, action) + s.transport = transition.state + const generationChange = + action.type === "invalidate" || + (action.type === "shutdown" && !before.closed) || + (action.type === "reopen" && before.closed) + requireInvariant( + s.transport.generation === before.generation + (generationChange ? 1 : 0), + "generation is not monotonic", + ) + if (action.type === "enqueue") { + const unchanged = + !transition.accepted && + s.transport === before && + !transition.post && + transition.release.length === 0 && + transition.settle.length === 0 + if (before.closed) { + requireInvariant(unchanged, "closed admission allocated work") + s.closedAdmissionRejected = true + } + if (action.request.kind !== "snapshot" && action.total === 0) { + requireInvariant(unchanged, "empty delta allocated work") + } + if (action.request.taskInstanceId !== action.focusedTaskInstanceId) { + requireInvariant(unchanged, "stale-instance request allocated work") + if (action.request.generation === before.generation) s.staleInstanceRejected = true + } + } + if ( + action.type === "enqueue" && + action.request.generation !== undefined && + action.request.generation !== before.generation + ) { + requireInvariant( + !transition.accepted && + s.transport.nextJobId === before.nextJobId && + s.transport.nextSnapshotId === before.nextSnapshotId, + "stale-generation request allocated work", + ) + } + if (transition.accepted) { + const job = transition.accepted + requireInvariant( + action.type === "enqueue" && job.taskInstanceId === action.request.taskInstanceId, + "descriptor lost originating instance identity", + ) + const scope = job.taskId ? `${job.taskId}:${s.epochs[job.taskId as TaskId]}` : "none" + const previousSeq = s.allocated[scope] ?? 0 + const expectedSeq = + previousSeq + + (action.type === "enqueue" && + (action.request.kind !== "snapshot" || action.request.bumpSeq) && + job.taskId + ? 1 + : 0) + requireInvariant(job.seq === expectedSeq, "allocated sequence diverged from capture order") + requireInvariant(job.total === values.length, "job total differs from captured payload") + if (job.kind === "snapshot") { + requireInvariant( + typeof job.snapshotId === "string" && + job.snapshotId.length > 0 && + !s.captures.some((capture) => capture.job.snapshotId === job.snapshotId), + "snapshot lacks a unique identity", + ) + } else { + requireInvariant(!("snapshotId" in job), "delta carries snapshot metadata") + } + s.allocated[scope] = job.seq + s.captures.push({ + job, + renderer: s.renderer, + taskInstanceId: action.request.taskInstanceId, + scope, + values: [...values], + failed: false, + }) + s.payloads.push(job.id) + s.callers.push(job.id) + } + for (const id of transition.release) { + if (action.type === "pump") { + const job = s.captures[id - 1].job + if (job.taskId === s.focus && job.taskInstanceId !== s.focusInstance) { + const location = before.active?.job.id === id ? "active" : "queued" + s.discardedInstanceJobs = [...new Set([...s.discardedInstanceJobs, location])].sort() + } + } + s.payloads = s.payloads.filter((value) => value !== id) + } + for (const { id } of transition.settle) { + requireInvariant(s.callers.includes(id), "settlement lacks a registered caller") + s.callers = s.callers.filter((value) => value !== id) + } + if (action.type === "invalidate") { + requireInvariant( + s.transport.queue.length === 0 && !s.transport.active && s.payloads.length === 0, + "invalidation retained obsolete queue or payload", + ) + requireInvariant( + s.callers.every((id) => id === s.physical?.job.id), + "discarded caller did not settle immediately", + ) + } + if (action.type === "shutdown" || action.type === "reopen") { + requireInvariant( + s.transport.nextJobId === before.nextJobId && + s.transport.nextSnapshotId === before.nextSnapshotId && + JSON.stringify([...s.transport.sequences]) === JSON.stringify([...before.sequences]), + "renderer boundary reset sequences or identities", + ) + if (action.type === "shutdown") { + s.shutdowns++ + s.shutdownWithQueued ||= !!before.inFlight && before.queue.length > 0 + requireInvariant( + s.transport.closed && + !s.transport.inFlight && + !s.transport.active && + s.transport.queue.length === 0 && + s.payloads.length === 0 && + s.callers.length === 0, + "shutdown retained transport ownership", + ) + if (s.physical) { + requireInvariant(s.physicalMessage, "physical send lost its wire message") + s.retired.push({ frame: s.physical, message: s.physicalMessage }) + } + s.physical = undefined + s.physicalMessage = undefined + } else if (before.closed) { + requireInvariant(!s.transport.closed, "reopen did not enable renderer") + s.renderer++ + s.visible = [] + s.appliedSeq = 0 + s.staging = undefined + } + } + if (action.type === "settle") { + const retired = s.retired.find(({ frame }) => frame.job.id === action.frame.job.id) + if (retired) { + requireInvariant( + transition.state === before && + !transition.post && + transition.release.length === 0 && + transition.settle.length === 0, + "late completion changed live transport", + ) + const outcome = action.success ? "resolve" : "reject" + s.lateOutcomes = [...new Set([...s.lateOutcomes, outcome])].sort() + if (s.physical) s.lateWhileSending = [...new Set([...s.lateWhileSending, outcome])].sort() + if (!action.success) s.failures++ + // This wire belongs to the disposed renderer, never its replacement. + s.retired = s.retired.filter((entry) => entry !== retired) + } else { + const physical = s.physical + requireInvariant(physical, "settled without physical send") + if (physical.job.generation < s.transport.generation) s.staleCompletions++ + if ( + action.success && + physical.phase === "end" && + physical.job.generation < s.transport.generation && + physical.job.taskId === s.focus && + physical.job.taskInstanceId === s.focusInstance + ) + s.staleCommitCompletions++ + if (action.success) { + requireInvariant(s.physicalMessage, "physical send lost its captured wire message") + requireInvariant( + s.physicalMessage.taskInstanceId === s.captures[physical.job.id - 1].taskInstanceId, + "wire lost originating instance identity", + ) + deliver(s, physical, s.physicalMessage, faults) + } else { + s.captures[physical.job.id - 1].failed = true + s.failures++ + } + s.physical = undefined + s.physicalMessage = undefined + } + } + if (transition.post) { + const frame = transition.post + const capture = s.captures[frame.job.id - 1] + requireInvariant( + !s.transport.closed && capture.renderer === s.renderer, + "post initiated for a disposed renderer", + ) + requireInvariant(!s.physical, "overlapping physical sends") + requireInvariant( + capture.job.generation === s.transport.generation && frame.job.taskId === s.focus, + "post or commit initiated after invalidation", + ) + requireInvariant(capture.taskInstanceId === s.focusInstance, "post initiated for a stale instance") + requireInvariant( + frame.job.taskInstanceId === capture.taskInstanceId, + "frame lost originating instance identity", + ) + const message = (faults.wire ?? transcriptFrameMessage)( + frame, + capture.values.map((value) => ({ ts: value, type: "say", text: String(value) })), + ) + requireInvariant( + message.taskId === capture.job.taskId && message.taskInstanceId === capture.taskInstanceId, + "wire lost originating instance identity", + ) + requireInvariant(!capture.failed, "failed snapshot continued posting") + if (frame.phase === "chunk") { + // Check the descriptor before wire slicing can clamp an overlarge count. + requireInvariant( + Number.isSafeInteger(frame.start) && + frame.start >= 0 && + frame.start === s.staging?.values.length && + frame.count > 0 && + frame.count === capture.values.slice(frame.start, frame.start + before.chunkSize).length && + frame.start + frame.count <= capture.values.length, + "chunk descriptor differs from captured payload range", + ) + } else { + requireInvariant(frame.start === 0 && frame.count === 0, "non-chunk frame carries a payload range") + } + requireInvariant( + frame.job.seq >= (s.sent[capture.scope] ?? 0), + "sent sequence regressed within task lifetime", + ) + requireInvariant(frame.job.seq <= s.allocated[capture.scope], "sent sequence exceeds allocation") + requireInvariant(s.payloads.includes(frame.job.id), "post without payload ownership") + s.sent[capture.scope] = frame.job.seq + s.physical = frame + s.physicalMessage = message + coverage.add(frame.phase) + } + if ((action.type === "pump" || action.type === "invalidate") && transition.release.length) + coverage.add("discard") + const owned = [ + ...s.transport.queue.map((job) => job.id), + ...(s.transport.active ? [s.transport.active.job.id] : []), + ].sort((a, b) => a - b) + requireInvariant( + JSON.stringify(s.payloads) === JSON.stringify(owned), + "payload ownership differs from queue and active job", + ) + const callers = [...new Set([...owned, ...(s.physical ? [s.physical.job.id] : [])])].sort((a, b) => a - b) + requireInvariant( + JSON.stringify(s.callers) === JSON.stringify(callers), + "caller ownership differs from queued and physical work", + ) + } + + if (event.action) { + coverage.add(event.name) + apply(event.action) + } else if (event.intent && event.actor) { + s[event.actor]++ + coverage.add(event.intent) + const intent = event.intent + if (intent === "shutdown" || intent === "reopen") { + apply({ type: intent }) + return s + } + if (intent === "switch" || intent === "clear" || intent === "focus" || intent === "replace-instance") { + const previous = s.focus + s.focus = intent === "replace-instance" ? "a" : intent === "clear" ? undefined : "b" + s.focusInstance = intent === "replace-instance" ? "a:1" : s.focus ? `${s.focus}:0` : undefined + if (intent === "replace-instance") { + // Publication is synchronous; later sync/invalidation may not have resumed yet. + s.instanceSyncPending = true + s.data.a = [5] + } + s.visible = [] + s.appliedSeq = 0 + s.staging = undefined + if (previous && (intent === "switch" || intent === "clear")) { + apply({ type: "forget-task", taskId: previous }) + s.epochs[previous]++ + } + } + if (["switch", "clear", "invalidate", "resync", "sync-instance"].includes(intent)) apply({ type: "invalidate" }) + if (intent === "sync-instance") s.instanceSyncPending = false + if (intent !== "invalidate" && intent !== "focus" && intent !== "replace-instance") { + const staleInstance = intent.startsWith("stale-instance-") + const empty = intent.startsWith("empty-") + const kind = intent.endsWith("append") ? "append" : intent.endsWith("update") ? "update" : "snapshot" + if (s.focus && !staleInstance && !empty) { + if (kind === "append") s.data[s.focus].push(4) + if (kind === "update") s.data[s.focus][0] = 9 + } + const values = + empty || !s.focus + ? [] + : staleInstance + ? [8] + : kind === "snapshot" + ? s.data[s.focus] + : kind === "append" + ? [4] + : [9] + apply( + { + type: "enqueue", + request: { + kind, + taskId: s.focus, + taskInstanceId: staleInstance ? "a:0" : s.focusInstance, + bumpSeq: intent === "snapshot", + generation: s.transport.generation - (intent === "stale-snapshot" ? 1 : 0), + }, + total: values.length, + focusedTaskId: s.focus, + focusedTaskInstanceId: s.focusInstance, + }, + values, + ) + } + } + return s +} + +function canonical(s: ModelState): string { + return JSON.stringify({ ...s, transport: { ...s.transport, sequences: [...s.transport.sequences].sort() } }) +} + +export function exploreTranscriptTransport( + scenario: Scenario, + reducer: Reducer = reduceTranscriptTransport, + bounds: { depth: number; states: number } = TRANSPORT_MODEL_BOUNDS, + faults: Faults = {}, +) { + const nodes: Node[] = [{ state: initialState(), parent: -1, event: "initial", depth: 0 }] + const visited = new Set([canonical(nodes[0].state)]) + const actions = new Set() + const landmarks = new Set() + let transitions = 0 + let maximumDepth = 0 + const trace = (index: number, lastEvent: string, failureState: ModelState) => { + const path: Array<{ event: string; state: ModelState }> = [] + for (let i = index; i >= 0; i = nodes[i].parent) path.push({ event: nodes[i].event, state: nodes[i].state }) + return [...path.reverse(), { event: lastEvent, state: failureState }] + } + for (let index = 0; index < nodes.length; index++) { + const node = nodes[index] + maximumDepth = Math.max(maximumDepth, node.depth) + for (const [name, predicate] of Object.entries(TRANSPORT_LANDMARKS)) + if (predicate(node.state)) landmarks.add(name) + for (const event of enabled(node.state, scenario)) { + let next: ModelState + try { + next = step(node.state, event, reducer, actions, faults) + } catch (error) { + const witness = trace(index, event.name, error instanceof ModelViolation ? error.state : node.state) + return { + states: visited.size, + transitions, + maximumDepth, + actions, + landmarks, + violation: error instanceof Error ? error.message : String(error), + witness, + } + } + transitions++ + const key = canonical(next) + if (visited.has(key)) continue + if (node.depth >= bounds.depth) + throw new Error(`${scenario.name}: depth ${bounds.depth} truncation at ${event.name}`) + if (visited.size >= bounds.states) + throw new Error(`${scenario.name}: state budget ${bounds.states} exceeded`) + visited.add(key) + nodes.push({ state: next, parent: index, event: event.name, depth: node.depth + 1 }) + } + } + return { + states: visited.size, + transitions, + maximumDepth, + actions, + landmarks, + violation: undefined, + witness: undefined, + } +} + +export const TRANSPORT_MUTATIONS: Mutation[] = [ + { + name: "shutdown-retains-physical-caller", + expected: "shutdown retained transport ownership", + reduce: (state, action) => { + if (action.type !== "shutdown" || state.closed) return reduceTranscriptTransport(state, action) + const result = reduceTranscriptTransport(state, { type: "invalidate" }) + result.state = { ...result.state, closed: true } + return result + }, + }, + { + name: "admit-after-shutdown", + expected: "closed admission allocated work", + reduce: (state, action) => + reduceTranscriptTransport( + action.type === "enqueue" && state.closed ? { ...state, closed: false } : state, + action, + ), + }, + { + name: "reopen-resets-identities", + expected: "renderer boundary reset sequences or identities", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (action.type === "reopen") + result.state = { ...result.state, nextJobId: 0, nextSnapshotId: 0, sequences: new Map() } + return result + }, + }, + { + name: "late-completion-settles-live-send", + expected: "late completion changed live transport", + reduce: (state, action) => + reduceTranscriptTransport( + state, + action.type === "settle" && state.inFlight ? { ...action, frame: state.inFlight } : action, + ), + }, + { + name: "stale-completion-starts-end", + expected: "post or commit initiated after invalidation", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if ( + action.type === "settle" && + state.inFlight && + state.inFlight.job.generation < state.generation && + state.inFlight.phase !== "end" + ) { + result.post = { ...state.inFlight, phase: "end" } + result.state = { ...result.state, inFlight: result.post } + } + return result + }, + }, + { + name: "admit-stale-generation", + expected: "stale-generation request allocated work", + reduce: (state, action) => + reduceTranscriptTransport( + state, + action.type === "enqueue" + ? { ...action, request: { ...action.request, generation: state.generation } } + : action, + ), + }, + { + name: "ignore-focus-at-post", + expected: "post or commit initiated after invalidation", + reduce: (state, action) => + reduceTranscriptTransport( + state, + action.type === "pump" + ? { + ...action, + focusedTaskId: state.active?.job.taskId ?? state.queue[0]?.taskId, + focusedTaskInstanceId: state.active?.job.taskInstanceId ?? state.queue[0]?.taskInstanceId, + } + : action, + ), + }, + { + name: "legacy-generation-only-invalidation", + expected: "invalidation retained obsolete queue or payload", + reduce: (state, action) => + action.type === "invalidate" + ? { state: { ...state, generation: state.generation + 1 }, release: [], settle: [] } + : reduceTranscriptTransport(state, action), + }, + { + name: "reset-promise-barrier", + expected: "overlapping physical sends", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (action.type === "invalidate") result.state = { ...result.state, inFlight: undefined } + return result + }, + }, + { + name: "commit-before-chunks", + expected: "snapshot commit before complete chunks", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (result.post?.phase === "chunk") { + result.post = { ...result.post, phase: "end", start: 0, count: 0 } + result.state = { ...result.state, inFlight: result.post } + } + return result + }, + }, + { + name: "reuse-delta-sequence", + expected: "allocated sequence diverged from capture order", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (result.accepted && result.accepted.kind !== "snapshot") { + const job = { ...result.accepted, seq: result.accepted.seq - 1 } + result.accepted = job + result.state = { ...result.state, queue: [...result.state.queue.slice(0, -1), job] } + } + return result + }, + }, + { + name: "continue-after-rejection", + expected: "failed snapshot continued posting", + reduce: (state, action) => + reduceTranscriptTransport(state, action.type === "settle" ? { ...action, success: true } : action), + }, + { + name: "delta-snapshot-metadata", + expected: "delta carries snapshot metadata", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (result.accepted && result.accepted.kind !== "snapshot") { + const job = { ...result.accepted, snapshotId: "unused" } + result.accepted = job + result.state = { ...result.state, queue: [...result.state.queue.slice(0, -1), job] } + } + return result + }, + }, + { + name: "non-chunk-payload-range", + expected: "non-chunk frame carries a payload range", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (result.post && result.post.phase !== "chunk") { + result.post = { ...result.post, start: 1, count: 1 } + result.state = { ...result.state, inFlight: result.post } + } + return result + }, + }, + { + name: "overrun-final-chunk", + expected: "chunk descriptor differs from captured payload range", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (result.post?.phase === "chunk") { + result.post = { ...result.post, count: state.chunkSize } + result.state = { ...result.state, inFlight: result.post } + } + return result + }, + }, + { + name: "settle-caller-twice", + expected: "settlement lacks a registered caller", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + result.settle.push(...result.settle) + return result + }, + }, + { + name: "admit-empty-delta", + expected: "empty delta allocated work", + reduce: (state, action) => + reduceTranscriptTransport( + state, + action.type === "enqueue" && action.request.kind !== "snapshot" && action.total === 0 + ? { ...action, total: 1 } + : action, + ), + }, + { + name: "admit-stale-instance", + expected: "stale-instance request allocated work", + reduce: (state, action) => + reduceTranscriptTransport( + state, + action.type === "enqueue" + ? { ...action, focusedTaskInstanceId: action.request.taskInstanceId } + : action, + ), + }, + { + name: "ignore-instance-at-post", + expected: "post initiated for a stale instance", + reduce: (state, action) => + reduceTranscriptTransport( + state, + action.type === "pump" + ? { ...action, focusedTaskInstanceId: (state.active?.job ?? state.queue[0])?.taskInstanceId } + : action, + ), + }, + { + name: "drop-descriptor-instance", + expected: "descriptor lost originating instance identity", + reduce: (state, action) => { + const result = reduceTranscriptTransport(state, action) + if (result.accepted) { + const job = { ...result.accepted, taskInstanceId: undefined } + result.accepted = job + result.state = { ...result.state, queue: [...result.state.queue.slice(0, -1), job] } + } + return result + }, + }, + { + name: "drop-wire-instance", + expected: "wire lost originating instance identity", + wire: (frame, messages) => ({ ...transcriptFrameMessage(frame, messages), taskInstanceId: undefined }), + }, + { + name: "receiver-ignores-instance", + expected: "receiver accepted a stale-instance frame", + accepts: (message, scope) => message.taskId === scope.focus, + }, + { + name: "receiver-accepts-stale-end", + expected: "receiver accepted a stale-instance frame", + accepts: (message, scope) => + message.taskId === scope.focus && + (message.type === "clineMessagesSnapshotEnd" || message.taskInstanceId === scope.focusInstance), + }, + { + name: "receiver-accepts-stale-delta", + expected: "receiver accepted a stale-instance frame", + accepts: (message, scope) => + message.taskId === scope.focus && + (message.type === "clineMessageAppended" || + message.type === "clineMessageUpdated" || + message.taskInstanceId === scope.focusInstance), + }, +] + +export function checkTranscriptTransportScenarios() { + const results = TRANSPORT_SCENARIOS.map((scenario) => ({ + name: scenario.name, + ...exploreTranscriptTransport(scenario), + })) + for (const result of results) { + if (result.violation) + throw new Error( + `${result.name}: ${result.violation}\nBounds: ${JSON.stringify(TRANSPORT_MODEL_BOUNDS)}\n${JSON.stringify(result.witness, (_key, value: unknown) => (value instanceof Map ? [...value] : value), 2)}`, + ) + } + const actions = new Set(results.flatMap((result) => [...result.actions])) + const landmarks = new Set(results.flatMap((result) => [...result.landmarks])) + for (const action of TRANSPORT_ACTIONS) requireInvariant(actions.has(action), `unreachable action: ${action}`) + for (const landmark of Object.keys(TRANSPORT_LANDMARKS)) + requireInvariant(landmarks.has(landmark), `unreachable landmark: ${landmark}`) + return { results, actions: [...actions].sort(), landmarks: [...landmarks].sort() } +} + +export function checkTranscriptTransportMutation(mutation: Mutation) { + const failures = TRANSPORT_SCENARIOS.map((scenario) => ({ + scenario: scenario.name, + ...exploreTranscriptTransport(scenario, mutation.reduce, TRANSPORT_MODEL_BOUNDS, mutation), + })).filter((result) => result.violation) + const result = failures.sort((a, b) => a.witness!.length - b.witness!.length)[0] + requireInvariant(result, `${mutation.name}: expected a counterexample`) + requireInvariant( + result.violation === mutation.expected, + `${mutation.name}: expected ${mutation.expected}; got ${result.violation}`, + ) + return { + name: mutation.name, + scenario: result.scenario, + violation: result.violation, + trace: result.witness!.map((entry) => entry.event), + } +} + +export function checkTranscriptTransportModel() { + return { + ...checkTranscriptTransportScenarios(), + counterexamples: TRANSPORT_MUTATIONS.map(checkTranscriptTransportMutation), + } +} diff --git a/src/core/webview/__tests__/transcriptTransport.spec.ts b/src/core/webview/__tests__/transcriptTransport.spec.ts new file mode 100644 index 0000000000..da75c488e1 --- /dev/null +++ b/src/core/webview/__tests__/transcriptTransport.spec.ts @@ -0,0 +1,945 @@ +import type { ClineMessage, ExtensionMessage } from "@roo-code/types" +import { + createTranscriptTransportState, + reduceTranscriptTransport, + transcriptFrameMessage, + TranscriptTransport, + type TranscriptFrame, +} from "../transcriptTransport" +import { + checkTranscriptTransportMutation, + checkTranscriptTransportScenarios, + exploreTranscriptTransport, + TRANSPORT_ACTIONS, + TRANSPORT_LANDMARKS, + TRANSPORT_MUTATIONS, + TRANSPORT_SCENARIOS, +} from "./transcriptTransport.model" + +describe("transcript transport bounded model", () => { + test("exhausts all scenarios, actions and landmarks", () => { + const result = checkTranscriptTransportScenarios() + expect(result.results).toHaveLength(TRANSPORT_SCENARIOS.length) + expect(result.actions).toEqual([...TRANSPORT_ACTIONS].sort()) + expect(result.landmarks).toEqual(Object.keys(TRANSPORT_LANDMARKS).sort()) + }) + + // Keep every scenario and fault, but give each exhaustive fault search its own test timeout. + test.each(TRANSPORT_MUTATIONS)("rejects $name with its shortest counterexample", (mutation) => { + const result = checkTranscriptTransportMutation(mutation) + expect(result.name).toBe(mutation.name) + expect(result.violation).toBe(mutation.expected) + expect(result.trace[0]).toBe("initial") + expect(result.trace.length).toBeGreaterThan(1) + }) + + test("fails closed on depth and state truncation", () => { + expect(() => + exploreTranscriptTransport(TRANSPORT_SCENARIOS[0], undefined, { depth: 0, states: 30_000 }), + ).toThrow("depth 0 truncation") + expect(() => exploreTranscriptTransport(TRANSPORT_SCENARIOS[0], undefined, { depth: 40, states: 1 })).toThrow( + "state budget 1 exceeded", + ) + }) + + test("produces deterministic shortest counterexamples", () => { + const mutation = TRANSPORT_MUTATIONS.find(({ name }) => name === "reset-promise-barrier")! + const first = exploreTranscriptTransport(TRANSPORT_SCENARIOS[0], mutation.reduce) + const second = exploreTranscriptTransport(TRANSPORT_SCENARIOS[0], mutation.reduce) + expect(first.witness).toEqual(second.witness) + expect(first.witness?.map(({ event }) => event)).toEqual([ + "initial", + "producer:snapshot", + "pump", + "controller:resync", + "pump", + ]) + }) +}) + +describe("transcript transport reducer", () => { + test.each(["job", "generation", "phase", "start"] as const)("ignores a mismatched %s completion token", (field) => { + const admitted = reduceTranscriptTransport(createTranscriptTransportState(), { + type: "enqueue", + request: { kind: "snapshot", taskId: "a" }, + total: 1, + focusedTaskId: "a", + }) + const sent = reduceTranscriptTransport(admitted.state, { type: "pump", focusedTaskId: "a" }) + const frame = structuredClone(sent.post!) + if (field === "job") frame.job.id++ + if (field === "generation") frame.job.generation++ + if (field === "phase") frame.phase = "end" + if (field === "start") frame.start++ + for (const success of [true, false]) { + const ignored = reduceTranscriptTransport(sent.state, { type: "settle", frame, success }) + expect(ignored).toEqual({ state: sent.state, release: [], settle: [] }) + expect(ignored.state).toBe(sent.state) + } + }) + + test("closed pump and repeated boundaries are no-ops", () => { + const open = createTranscriptTransportState() + expect(reduceTranscriptTransport(open, { type: "reopen" }).state).toBe(open) + const closed = reduceTranscriptTransport(open, { type: "shutdown" }).state + expect(reduceTranscriptTransport(closed, { type: "shutdown" }).state).toBe(closed) + const pump = reduceTranscriptTransport(closed, { type: "pump", focusedTaskId: undefined }) + expect(pump).toEqual({ state: closed, release: [], settle: [] }) + expect(pump.state).toBe(closed) + }) + + test.each(["append", "update"] as const)("rejects an empty %s without allocating protocol state", (kind) => { + const state = createTranscriptTransportState() + const focus = { focusedTaskId: "a", focusedTaskInstanceId: "instance-1" } + const request = { kind, taskId: "a", taskInstanceId: "instance-1" } + const rejected = reduceTranscriptTransport(state, { type: "enqueue", request, total: 0, ...focus }) + + expect(rejected).toEqual({ state, release: [], settle: [] }) + expect(rejected.state).toBe(state) + const valid = reduceTranscriptTransport(rejected.state, { type: "enqueue", request, total: 1, ...focus }) + expect(valid.accepted).toMatchObject({ id: 1, seq: 1, total: 1, taskInstanceId: "instance-1" }) + const snapshot = reduceTranscriptTransport(valid.state, { + type: "enqueue", + request: { ...request, kind: "snapshot" }, + total: 0, + ...focus, + }) + expect(snapshot.accepted).toMatchObject({ id: 2, seq: 1, total: 0, snapshotId: "a:1" }) + }) + + test.each(["append", "update", "snapshot"] as const)( + "rejects stale or missing instance ownership for %s admission", + (kind) => { + for (const [taskInstanceId, focusedTaskInstanceId] of [ + ["instance-1", "instance-2"], + [undefined, "instance-2"], + ["instance-1", undefined], + ]) { + const state = createTranscriptTransportState() + const transition = reduceTranscriptTransport(state, { + type: "enqueue", + request: { kind, taskId: "a", taskInstanceId, generation: state.generation }, + total: 1, + focusedTaskId: "a", + focusedTaskInstanceId, + }) + expect(transition).toEqual({ state, release: [], settle: [] }) + expect(transition.state).toBe(state) + } + }, + ) + + test.each([ + { kind: "append", completedFrames: 0 }, + { kind: "update", completedFrames: 0 }, + { kind: "snapshot", completedFrames: 0 }, + { kind: "snapshot", completedFrames: 1 }, + { kind: "snapshot", completedFrames: 2 }, + ] as const)( + "discards stale-instance $kind before frame $completedFrames without invalidation", + ({ kind, completedFrames }) => { + const oldFocus = { focusedTaskId: "a", focusedTaskInstanceId: "instance-1" } + const currentFocus = { focusedTaskId: "a", focusedTaskInstanceId: "instance-2" } + const admitted = reduceTranscriptTransport(createTranscriptTransportState(1), { + type: "enqueue", + request: { kind, taskId: "a", taskInstanceId: "instance-1" }, + total: 1, + ...oldFocus, + }) + let state = admitted.state + for (let index = 0; index < completedFrames; index++) { + state = reduceTranscriptTransport(state, { type: "pump", ...oldFocus }).state + state = reduceTranscriptTransport(state, { + type: "settle", + frame: state.inFlight!, + success: true, + }).state + } + const current = reduceTranscriptTransport(state, { + type: "enqueue", + request: { kind: "append", taskId: "a", taskInstanceId: "instance-2" }, + total: 1, + ...currentFocus, + }) + const transition = reduceTranscriptTransport(current.state, { type: "pump", ...currentFocus }) + expect(transition.release).toEqual([admitted.accepted!.id]) + expect(transition.settle).toEqual([{ id: admitted.accepted!.id }]) + expect(transition.post).toEqual({ job: current.accepted, phase: "append", start: 0, count: 0 }) + expect(transition.state.generation).toBe(0) + expect(transition.state.queue).toEqual([]) + }, + ) + + test.each([true, false])("ignores settlement without a physical send (success=%s)", (success) => { + const state = createTranscriptTransportState() + const admitted = reduceTranscriptTransport(state, { + type: "enqueue", + request: { kind: "append", taskId: "a" }, + total: 1, + focusedTaskId: "a", + }) + const frame: TranscriptFrame = { job: admitted.accepted!, phase: "append", start: 0, count: 0 } + const transition = reduceTranscriptTransport(state, { type: "settle", frame, success }) + expect(transition).toEqual({ state, release: [], settle: [] }) + expect(transition.state).toBe(state) + }) + + test.each(["queued", "active"] as const)("discards a stale-generation %s job before sending", (location) => { + const admitted = reduceTranscriptTransport(createTranscriptTransportState(2), { + type: "enqueue", + request: { kind: "snapshot", taskId: "a" }, + total: 3, + focusedTaskId: "a", + }) + let state = admitted.state + if (location === "active") { + state = reduceTranscriptTransport(state, { type: "pump", focusedTaskId: "a" }).state + state = reduceTranscriptTransport(state, { type: "settle", frame: state.inFlight!, success: true }).state + } + // Adversarial reducer input: normal invalidation also releases this work. Keep + // the pre-send guard defensive if stale ownership ever reaches this boundary. + state = { ...state, generation: state.generation + 1 } + const current = reduceTranscriptTransport(state, { + type: "enqueue", + request: { kind: "append", taskId: "a" }, + total: 1, + focusedTaskId: "a", + }) + + const transition = reduceTranscriptTransport(current.state, { type: "pump", focusedTaskId: "a" }) + + expect(transition.release).toEqual([admitted.accepted!.id]) + expect(transition.settle).toEqual([{ id: admitted.accepted!.id }]) + expect(transition.post).toEqual({ job: current.accepted, phase: "append", start: 0, count: 0 }) + expect(transition.state.queue).toEqual([]) + expect(transition.state.active).toEqual({ job: current.accepted, position: 0 }) + }) + + test.each([ + { total: 0, chunks: [] }, + { total: 1, chunks: [{ start: 0, count: 1 }] }, + { total: 2, chunks: [{ start: 0, count: 2 }] }, + { + total: 3, + chunks: [ + { start: 0, count: 2 }, + { start: 2, count: 1 }, + ], + }, + { + total: 5, + chunks: [ + { start: 0, count: 2 }, + { start: 2, count: 2 }, + { start: 4, count: 1 }, + ], + }, + ])("describes exact captured ranges for a $total-message snapshot", ({ total, chunks }) => { + const messages: ClineMessage[] = Array.from({ length: total }, (_, ts) => ({ ts, type: "say" })) + const admitted = reduceTranscriptTransport(createTranscriptTransportState(2), { + type: "enqueue", + request: { kind: "snapshot", taskId: "a" }, + total: messages.length, + focusedTaskId: "a", + }) + let state = admitted.state + const frames: TranscriptFrame[] = [] + for (let index = 0; index < chunks.length + 2; index++) { + const transition = reduceTranscriptTransport(state, { type: "pump", focusedTaskId: "a" }) + expect(transition.post).toBeDefined() + frames.push(transition.post!) + state = reduceTranscriptTransport(transition.state, { + type: "settle", + frame: transition.post!, + success: true, + }).state + } + + expect(state.queue).toEqual([]) + expect(state.active).toBeUndefined() + expect(state.inFlight).toBeUndefined() + expect(frames).toEqual([ + { job: admitted.accepted, phase: "start", start: 0, count: 0 }, + ...chunks.map((range) => ({ job: admitted.accepted, phase: "chunk", ...range })), + { job: admitted.accepted, phase: "end", start: 0, count: 0 }, + ]) + expect(frames.slice(1, -1).map((frame) => transcriptFrameMessage(frame, messages).clineMessages)).toEqual( + chunks.map(({ start, count }) => messages.slice(start, start + count)), + ) + }) +}) + +describe("transcript transport driver", () => { + const message: ClineMessage = { ts: 1, type: "say", text: "initial", images: ["image"] } + + test.each( + (["start", "chunk", "end", "append", "update"] as const).flatMap((phase) => + [true, false].map((success) => ({ phase, success })), + ), + )( + "shutdown detaches held $phase and permits a fresh renderer (late success=$success)", + async ({ phase, success }) => { + const types = { + start: "clineMessagesSnapshotStart", + chunk: "clineMessagesSnapshotChunk", + end: "clineMessagesSnapshotEnd", + append: "clineMessageAppended", + update: "clineMessageUpdated", + } as const + let resolveOld!: () => void + let rejectOld!: (error: Error) => void + const oldPhysical = new Promise((resolve, reject) => { + resolveOld = resolve + rejectOld = reject + }) + const oldPost = vi.fn((frame: ExtensionMessage) => + frame.type === types[phase] ? oldPhysical : Promise.resolve(), + ) + const log = vi.fn() + const transport = new TranscriptTransport( + () => "a", + oldPost, + log, + () => "instance-1", + ) + const resolved = vi.fn() + const rejected = vi.fn() + const active = transport + .enqueue( + { + kind: phase === "append" || phase === "update" ? phase : "snapshot", + taskId: "a", + taskInstanceId: "instance-1", + bumpSeq: true, + }, + [message], + ) + .then(resolved, rejected) + await vi.waitFor(() => + expect(oldPost).toHaveBeenCalledWith(expect.objectContaining({ type: types[phase] })), + ) + const waiting = ["append", "update", "snapshot"].map((kind) => + transport.enqueue( + { kind: kind as "append" | "update" | "snapshot", taskId: "a", taskInstanceId: "instance-1" }, + [message], + ), + ) + const completion = transport["pendingSend"]! + const before = transport["state"] + const oldFrames = oldPost.mock.calls.length + transport.shutdown() + transport.shutdown() + await Promise.all([active, ...waiting]) + expect(resolved).toHaveBeenCalledOnce() + expect(rejected).not.toHaveBeenCalled() + expect(completion.finish).toBeUndefined() + expect(transport["callbacks"]).toBeUndefined() + expect(transport["pendingSend"]).toBeUndefined() + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + expect(transport["state"]).toMatchObject({ + closed: true, + generation: before.generation + 1, + queue: [], + active: undefined, + inFlight: undefined, + }) + expect(transport["state"].nextJobId).toBe(before.nextJobId) + expect(transport.getSequence("a")).toBe(before.sequences.get("a")) + + let releaseNew!: () => void + const newPhysical = new Promise((resolve) => { + releaseNew = resolve + }) + const newPost = vi + .fn<(frame: ExtensionMessage) => Promise>() + .mockReturnValueOnce(newPhysical) + .mockResolvedValue(undefined) + transport.reopen( + () => "a", + newPost, + log, + () => "instance-2", + ) + const recovered = vi.fn() + const fresh = transport + .enqueue({ kind: "snapshot", taskId: "a", taskInstanceId: "instance-2" }, [message]) + .then(recovered) + const delta = transport.enqueue({ kind: "append", taskId: "a", taskInstanceId: "instance-2" }, [message]) + expect(newPost).toHaveBeenCalledOnce() + const newState = transport["state"] + expect(newState.inFlight?.job.id).toBe(before.nextJobId + 1) + expect(newState.inFlight?.job.snapshotId).toBe(`a:${before.nextSnapshotId + 1}`) + // A repeated reopen cannot replace callbacks or reset an already-live barrier. + transport.reopen(() => "b", oldPost, log) + if (success) resolveOld() + else rejectOld(new Error("dead renderer rejected")) + await oldPhysical.catch(() => {}) + await Promise.resolve() + expect(transport["state"]).toBe(newState) + expect(recovered).not.toHaveBeenCalled() + expect(newPost).toHaveBeenCalledOnce() + expect(log).not.toHaveBeenCalled() + releaseNew() + await Promise.all([fresh, delta]) + expect(resolved).toHaveBeenCalledOnce() + expect(recovered).toHaveBeenCalledOnce() + expect(oldPost).toHaveBeenCalledTimes(oldFrames) + expect( + newPost.mock.calls.map(([frame]) => [frame.type, frame.taskInstanceId, frame.clineMessagesSeq]), + ).toEqual([ + ["clineMessagesSnapshotStart", "instance-2", before.sequences.get("a")], + ["clineMessagesSnapshotChunk", "instance-2", before.sequences.get("a")], + ["clineMessagesSnapshotEnd", "instance-2", before.sequences.get("a")], + ["clineMessageAppended", "instance-2", before.sequences.get("a")! + 1], + ]) + }, + ) + + test.each([true, false])( + "shutdown after invalidation settles the detached caller (success=%s)", + async (success) => { + let resolve!: () => void + let reject!: (error: Error) => void + const held = new Promise((yes, no) => { + resolve = yes + reject = no + }) + const log = vi.fn() + const transport = new TranscriptTransport( + () => undefined, + () => held, + log, + ) + const settled = vi.fn() + const active = transport.enqueue({ kind: "snapshot", taskId: undefined }, []).then(settled) + transport.invalidate() + transport.shutdown() + await active + expect(settled).toHaveBeenCalledOnce() + if (success) resolve() + else reject(new Error("late failure")) + await held.catch(() => {}) + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + transport.reopen(() => undefined, post, log) + await transport.enqueue({ kind: "snapshot", taskId: undefined }, []) + expect(post.mock.calls.map(([frame]) => [frame.type, frame.snapshotId, frame.clineMessagesSeq])).toEqual([ + ["clineMessagesSnapshotStart", "none:2", 0], + ["clineMessagesSnapshotEnd", "none:2", 0], + ]) + expect(log).not.toHaveBeenCalled() + }, + ) + + test.each(["append", "update", "snapshot"] as const)( + "drops closed %s before capture and rejects old generations after reopen", + async (kind) => { + const focus = vi.fn(() => "a") + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport(focus, post, vi.fn()) + const oldGeneration = transport.generation + transport.shutdown() + const closedState = transport["state"] + const clone = vi.spyOn(globalThis, "structuredClone") + try { + await transport.enqueue({ kind, taskId: "a" }, [message]) + expect(focus).not.toHaveBeenCalled() + expect(transport["state"]).toBe(closedState) + transport.reopen(focus, post, vi.fn()) + await transport.enqueue({ kind, taskId: "a", generation: oldGeneration }, [message]) + await transport.enqueue({ kind, taskId: "a", generation: closedState.generation }, [message]) + expect(clone).not.toHaveBeenCalled() + expect(post).not.toHaveBeenCalled() + await transport.enqueue({ kind, taskId: "a" }, [message]) + expect(post).toHaveBeenCalled() + } finally { + clone.mockRestore() + } + }, + ) + + test("does not adopt a reopened renderer if capture reenters shutdown", async () => { + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport(() => "a", post, vi.fn()) + await transport.enqueue({ kind: "snapshot", taskId: "a" }, [ + { + ts: 1, + type: "say", + get text() { + transport.shutdown() + transport.reopen(() => "a", post, vi.fn()) + return "obsolete" + }, + }, + ]) + expect(post).not.toHaveBeenCalled() + expect(transport["state"].nextJobId).toBe(0) + await transport.enqueue({ kind: "append", taskId: "a" }, [message]) + expect(post).toHaveBeenCalledOnce() + }) + + test("handles shutdown reentered by a physical post that then throws", async () => { + const log = vi.fn() + const freshPost = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport( + () => "a", + () => { + transport.shutdown() + transport.reopen(() => "a", freshPost, log) + throw new Error("disposed synchronously") + }, + log, + ) + await transport.enqueue({ kind: "snapshot", taskId: "a" }, [message]) + await transport.enqueue({ kind: "append", taskId: "a" }, [message]) + expect(log).not.toHaveBeenCalled() + expect(freshPost).toHaveBeenCalledOnce() + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + }) + + test.each(["append", "update"] as const)( + "rejects empty %s before cloning or admission, then recovers", + async (kind) => { + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const focus = vi.fn(() => "instance-1") + const transport = new TranscriptTransport(() => "a", post, vi.fn(), focus) + const state = transport["state"] + const clone = vi.spyOn(globalThis, "structuredClone") + const payloadSet = vi.spyOn(transport["payloads"], "set") + const callerSet = vi.spyOn(transport["callers"], "set") + // Reading generation would mean the driver has already allocated an admission request. + const readGeneration = vi.fn(() => transport.generation) + const request = { + kind, + taskId: "a", + taskInstanceId: "instance-1", + get generation() { + return readGeneration() + }, + } + try { + await transport.enqueue(request, []) + expect(clone).not.toHaveBeenCalled() + expect(readGeneration).not.toHaveBeenCalled() + expect(focus).not.toHaveBeenCalled() + expect(post).not.toHaveBeenCalled() + expect(payloadSet).not.toHaveBeenCalled() + expect(callerSet).not.toHaveBeenCalled() + expect(transport["state"]).toBe(state) + expect(state).toEqual(createTranscriptTransportState()) + expect(transport.getSequence("a")).toBe(0) + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + + await transport.enqueue(request, [message]) + await transport.enqueue({ kind: "snapshot", taskId: "a", taskInstanceId: "instance-1" }, []) + expect(clone).toHaveBeenCalledTimes(2) + expect(payloadSet.mock.calls.map(([id]) => id)).toEqual([1, 2]) + expect(callerSet.mock.calls.map(([id]) => id)).toEqual([1, 2]) + expect(post.mock.calls.map(([frame]) => frame)).toEqual([ + { + type: kind === "append" ? "clineMessageAppended" : "clineMessageUpdated", + taskId: "a", + taskInstanceId: "instance-1", + clineMessagesSeq: 1, + clineMessage: message, + }, + { + type: "clineMessagesSnapshotStart", + taskId: "a", + taskInstanceId: "instance-1", + clineMessagesSeq: 1, + snapshotId: "a:1", + snapshotTotal: 0, + }, + { + type: "clineMessagesSnapshotEnd", + taskId: "a", + taskInstanceId: "instance-1", + clineMessagesSeq: 1, + snapshotId: "a:1", + snapshotTotal: 0, + }, + ]) + expect(transport["state"].nextJobId).toBe(2) + expect(transport["state"].nextSnapshotId).toBe(1) + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + } finally { + clone.mockRestore() + payloadSet.mockRestore() + callerSet.mockRestore() + } + }, + ) + + test.each(["append", "update", "snapshot"] as const)( + "rejects stale or absent %s instance before cloning", + async (kind) => { + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport( + () => "a", + post, + vi.fn(), + () => "instance-2", + ) + const state = transport["state"] + const clone = vi.spyOn(globalThis, "structuredClone") + try { + for (const taskInstanceId of ["instance-1", undefined]) { + await transport.enqueue({ kind, taskId: "a", taskInstanceId, generation: transport.generation }, [ + message, + ]) + } + expect(clone).not.toHaveBeenCalled() + expect(post).not.toHaveBeenCalled() + expect(transport["state"]).toBe(state) + expect(transport.getSequence("a")).toBe(0) + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + } finally { + clone.mockRestore() + } + }, + ) + + test.each(["append", "update", "snapshot"] as const)( + "rechecks %s instance after cloning without adopting live focus", + async (kind) => { + let instance = "instance-1" + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport( + () => "a", + post, + vi.fn(), + () => instance, + ) + const state = transport["state"] + const reentrant: ClineMessage = { + ts: 1, + type: "say", + get text() { + instance = "instance-2" + return "obsolete" + }, + } + await transport.enqueue({ kind, taskId: "a", taskInstanceId: instance }, [reentrant]) + expect(instance).toBe("instance-2") + expect(transport["state"]).toBe(state) + expect(post).not.toHaveBeenCalled() + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + }, + ) + + test.each( + (["start", "chunk", "end", "append", "update"] as const).flatMap((phase) => + [true, false].flatMap((success) => [true, false].map((invalidate) => ({ phase, success, invalidate }))), + ), + )( + "retains held $phase identity and settles once across replacement (success=$success, invalidate=$invalidate)", + async ({ phase, success, invalidate }) => { + const types = { + start: "clineMessagesSnapshotStart", + chunk: "clineMessagesSnapshotChunk", + end: "clineMessagesSnapshotEnd", + append: "clineMessageAppended", + update: "clineMessageUpdated", + } as const + let instance = "instance-1" + let resolveHeld!: () => void + let rejectHeld!: (error: Error) => void + let notifyStarted!: () => void + const held = new Promise((resolve, reject) => { + resolveHeld = resolve + rejectHeld = reject + }) + const started = new Promise((resolve) => { + notifyStarted = resolve + }) + let physical = 0 + let maximumPhysical = 0 + const post = vi.fn(async (frame: ExtensionMessage) => { + physical++ + maximumPhysical = Math.max(maximumPhysical, physical) + try { + if (frame.taskInstanceId === "instance-1" && frame.type === types[phase]) { + notifyStarted() + await held + } + } finally { + physical-- + } + }) + const log = vi.fn() + const transport = new TranscriptTransport( + () => "a", + post, + log, + () => instance, + ) + const resolved = vi.fn() + const rejected = vi.fn() + const active = transport + .enqueue( + { + kind: phase === "append" || phase === "update" ? phase : "snapshot", + taskId: "a", + taskInstanceId: instance, + }, + [message], + ) + .then(resolved, rejected) + await started + const physicalFrame = transport["state"].inFlight! + const waitingResolved = vi.fn() + const waiting = transport + .enqueue({ kind: "update", taskId: "a", taskInstanceId: instance }, [message]) + .then(waitingResolved) + const before = post.mock.calls.length + instance = "instance-2" + if (invalidate) { + transport.invalidate() + transport.invalidate() + await waiting + expect(transport["payloads"].size).toBe(0) + expect([...transport["callers"].keys()]).toEqual([physicalFrame.job.id]) + } + const recovery = transport.enqueue({ kind: "snapshot", taskId: "a", taskInstanceId: instance }, [message]) + const delta = transport.enqueue({ kind: "append", taskId: "a", taskInstanceId: instance }, [message]) + expect(transport["state"].inFlight).toBe(physicalFrame) + expect(physicalFrame.job.taskInstanceId).toBe("instance-1") + expect(post).toHaveBeenCalledTimes(before) + expect(resolved).not.toHaveBeenCalled() + expect(rejected).not.toHaveBeenCalled() + const failure = new Error("old instance post failed") + if (success) resolveHeld() + else rejectHeld(failure) + await Promise.all([active, waiting, recovery, delta]) + expect(maximumPhysical).toBe(1) + expect(resolved).toHaveBeenCalledTimes(success ? 1 : 0) + expect(rejected.mock.calls).toEqual(success ? [] : [[failure]]) + expect(log.mock.calls).toEqual(success ? [] : [[failure]]) + expect(waitingResolved).toHaveBeenCalledOnce() + expect(post.mock.calls.slice(0, before).every(([frame]) => frame.taskInstanceId === "instance-1")).toBe( + true, + ) + expect(post.mock.calls.slice(before).map(([frame]) => [frame.type, frame.taskInstanceId])).toEqual([ + ["clineMessagesSnapshotStart", "instance-2"], + ["clineMessagesSnapshotChunk", "instance-2"], + ["clineMessagesSnapshotEnd", "instance-2"], + ["clineMessageAppended", "instance-2"], + ]) + expect(transport["callers"].size).toBe(0) + expect(transport["payloads"].size).toBe(0) + }, + ) + + test.each([true, false])( + "retains the sole held caller through repeated invalidation (success=%s)", + async (success) => { + let resolveHeld!: () => void + let rejectHeld!: (error: Error) => void + const held = new Promise((resolve, reject) => { + resolveHeld = resolve + rejectHeld = reject + }) + const post = vi + .fn<(frame: ExtensionMessage) => Promise>() + .mockReturnValueOnce(held) + .mockResolvedValue(undefined) + const log = vi.fn() + const transport = new TranscriptTransport(() => "a", post, log) + const resolved = vi.fn() + const rejected = vi.fn() + const active = transport.enqueue({ kind: "snapshot", taskId: "a" }, [message]).then(resolved, rejected) + const [heldId] = transport["callers"].keys() + + for (let generation = 0; generation < 2; generation++) { + const waiting = transport.enqueue({ kind: "update", taskId: "a" }, [message]) + transport.invalidate() + await waiting + expect([...transport["callers"].keys()]).toEqual([heldId]) + expect(transport["payloads"].size).toBe(0) + expect(post).toHaveBeenCalledOnce() + expect(resolved).not.toHaveBeenCalled() + expect(rejected).not.toHaveBeenCalled() + } + + const recovery = transport.enqueue({ kind: "snapshot", taskId: "a" }, [message]) + const failure = new Error("held post failed") + if (success) resolveHeld() + else rejectHeld(failure) + await Promise.all([active, recovery]) + + expect(resolved).toHaveBeenCalledTimes(success ? 1 : 0) + expect(rejected.mock.calls).toEqual(success ? [] : [[failure]]) + expect(log.mock.calls).toEqual(success ? [] : [[failure]]) + expect(transport["callers"].size).toBe(0) + expect(transport["payloads"].size).toBe(0) + expect(post.mock.calls.slice(1).map(([frame]) => frame.type)).toEqual([ + "clineMessagesSnapshotStart", + "clineMessagesSnapshotChunk", + "clineMessagesSnapshotEnd", + ]) + }, + ) + + test.each(["append", "update", "snapshot"] as const)( + "rejects an unfocused %s before reading the payload or allocating work", + async (kind) => { + const readText = vi.fn(() => "obsolete") + const unread: ClineMessage = { + ts: 1, + type: "say", + get text() { + return readText() + }, + } + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport(() => "a", post, vi.fn()) + const state = transport["state"] + + await transport.enqueue({ kind, taskId: "b" }, [unread]) + + expect(readText).not.toHaveBeenCalled() + expect(post).not.toHaveBeenCalled() + expect(transport["state"]).toBe(state) + expect(transport.getSequence("b")).toBe(0) + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + }, + ) + + test.each(["append", "update"] as const)("rejects a %s without a task scope", async (kind) => { + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport(() => undefined, post, vi.fn()) + const state = transport["state"] + + await transport.enqueue({ kind, taskId: undefined }, [message]) + + expect(post).not.toHaveBeenCalled() + expect(transport["state"]).toBe(state) + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + }) + + test.each(["start", "chunk", "end", "delta"] as const)( + "keeps the physical barrier across rejected held %s and recovers", + async (phase) => { + const type: ExtensionMessage["type"] = + phase === "start" + ? "clineMessagesSnapshotStart" + : phase === "chunk" + ? "clineMessagesSnapshotChunk" + : phase === "end" + ? "clineMessagesSnapshotEnd" + : "clineMessageAppended" + let rejectHeld!: (error: Error) => void + let notifyStarted!: () => void + const held = new Promise((_resolve, reject) => { + rejectHeld = reject + }) + const started = new Promise((resolve) => { + notifyStarted = resolve + }) + let heldOnce = false + let physical = 0 + let maximumPhysical = 0 + const post = vi.fn(async (frame: ExtensionMessage) => { + physical++ + maximumPhysical = Math.max(maximumPhysical, physical) + try { + if (frame.type === type && !heldOnce) { + heldOnce = true + notifyStarted() + await held + } + } finally { + physical-- + } + }) + const log = vi.fn() + const transport = new TranscriptTransport(() => "a", post, log) + const active = transport.enqueue({ kind: phase === "delta" ? "append" : "snapshot", taskId: "a" }, [ + message, + ]) + const rejected = expect(active).rejects.toThrow("held post failed") + await started + const discarded = transport.enqueue({ kind: "update", taskId: "a" }, [message]) + transport.invalidate() + await discarded + const recovery = transport.enqueue({ kind: "snapshot", taskId: "a" }, [message]) + expect(physical).toBe(1) + expect(transport["payloads"].size).toBe(1) + const before = post.mock.calls.length + rejectHeld(new Error("held post failed")) + await Promise.all([rejected, recovery]) + expect(maximumPhysical).toBe(1) + expect(post.mock.calls.slice(before).map(([frame]) => frame.type)).toEqual([ + "clineMessagesSnapshotStart", + "clineMessagesSnapshotChunk", + "clineMessagesSnapshotEnd", + ]) + expect(log).toHaveBeenCalledOnce() + expect(transport["callers"].size).toBe(0) + expect(transport["payloads"].size).toBe(0) + }, + ) + + test("recovers from a synchronous post throw", async () => { + const post = vi + .fn<(frame: ExtensionMessage) => Promise>() + .mockImplementationOnce(() => { + throw new Error("sync failure") + }) + .mockResolvedValue(undefined) + const transport = new TranscriptTransport(() => "a", post, vi.fn()) + await expect(transport.enqueue({ kind: "append", taskId: "a" }, [message])).rejects.toThrow("sync failure") + await transport.enqueue({ kind: "update", taskId: "a" }, [message]) + expect(post.mock.calls.map(([frame]) => frame.clineMessagesSeq)).toEqual([1, 2]) + }) + + test("rejects invalid chunk-size bounds", () => { + for (const size of [0, -1, 1.5, Infinity]) + expect(() => createTranscriptTransportState(size)).toThrow("positive safe integer") + }) + + test("delivers one message per chunk at the minimum valid chunk size", async () => { + const post = vi.fn<(frame: ExtensionMessage) => Promise>().mockResolvedValue(undefined) + const transport = new TranscriptTransport(() => "a", post, vi.fn()) + transport["state"] = createTranscriptTransportState(1) + const second = { ...message, ts: 2, text: "second" } + + await transport.enqueue({ kind: "snapshot", taskId: "a" }, [message, second]) + + const common = { taskId: "a", clineMessagesSeq: 0, snapshotId: "a:1" } + expect(post.mock.calls.map(([frame]) => frame)).toEqual([ + { ...common, type: "clineMessagesSnapshotStart", snapshotTotal: 2 }, + { ...common, type: "clineMessagesSnapshotChunk", snapshotStartIndex: 0, clineMessages: [message] }, + { ...common, type: "clineMessagesSnapshotChunk", snapshotStartIndex: 1, clineMessages: [second] }, + { ...common, type: "clineMessagesSnapshotEnd", snapshotTotal: 2 }, + ]) + expect(transport["payloads"].size).toBe(0) + expect(transport["callers"].size).toBe(0) + }) + + test("does not adopt a newer generation if cloning reenters invalidation", async () => { + const post = vi.fn().mockResolvedValue(undefined) + const transport = new TranscriptTransport(() => "a", post, vi.fn()) + const reentrant: ClineMessage = { + ts: 1, + type: "say", + get text() { + transport.invalidate() + return "obsolete" + }, + } + await transport.enqueue({ kind: "snapshot", taskId: "a", bumpSeq: true }, [reentrant]) + expect(transport.generation).toBe(1) + expect(transport.getSequence("a")).toBe(0) + expect(transport["state"].nextSnapshotId).toBe(0) + expect(post).not.toHaveBeenCalled() + }) +}) diff --git a/src/core/webview/transcriptTransport.ts b/src/core/webview/transcriptTransport.ts new file mode 100644 index 0000000000..a650e4b99a --- /dev/null +++ b/src/core/webview/transcriptTransport.ts @@ -0,0 +1,412 @@ +import type { ClineMessage, ExtensionMessage } from "@roo-code/types" + +export type TranscriptRequest = { + kind: "append" | "update" | "snapshot" + taskId: string | undefined + taskInstanceId?: string + generation?: number + bumpSeq?: boolean +} + +export type TranscriptJob = { + id: number + generation: number + taskId: string | undefined + taskInstanceId: string | undefined + seq: number + kind: TranscriptRequest["kind"] + total: number + /** Snapshot identity is absent on delta descriptors. */ + snapshotId?: string +} + +export type TranscriptFrame = { + job: TranscriptJob + phase: "append" | "update" | "start" | "chunk" | "end" + /** Exact captured-payload range for chunks; both values are zero for other phases. */ + start: number + count: number +} + +/** Payloads and Promise resolvers deliberately live outside the pure protocol state. */ +export type TranscriptTransportState = { + closed: boolean + generation: number + nextJobId: number + nextSnapshotId: number + sequences: ReadonlyMap + chunkSize: number + queue: readonly TranscriptJob[] + active?: { job: TranscriptJob; position: number } + inFlight?: TranscriptFrame +} + +export type TranscriptAction = + | { + type: "enqueue" + request: TranscriptRequest + total: number + focusedTaskId: string | undefined + focusedTaskInstanceId?: string + } + | { type: "invalidate" } + | { type: "shutdown" } + | { type: "reopen" } + | { type: "forget-task"; taskId: string } + | { type: "pump"; focusedTaskId: string | undefined; focusedTaskInstanceId?: string } + | { type: "settle"; frame: TranscriptFrame; success: boolean } + +export type TranscriptTransition = { + state: TranscriptTransportState + accepted?: TranscriptJob + post?: TranscriptFrame + /** Drop all owned payload references, including an invalidated snapshot's unsent suffix. */ + release: number[] + /** Invalidation waits for physical completion; renderer shutdown settles every caller immediately. */ + settle: Array<{ id: number; failed?: boolean }> +} + +export function createTranscriptTransportState(chunkSize = 200): TranscriptTransportState { + if (!Number.isSafeInteger(chunkSize) || chunkSize < 1) { + throw new Error("Transcript chunk size must be a positive safe integer") + } + return { closed: false, generation: 0, nextJobId: 0, nextSnapshotId: 0, sequences: new Map(), chunkSize, queue: [] } +} + +export function isTranscriptRequestCurrent( + state: TranscriptTransportState, + request: TranscriptRequest, + focusedTaskId: string | undefined, + focusedTaskInstanceId?: string, +): boolean { + return ( + !state.closed && + (request.generation ?? state.generation) === state.generation && + request.taskId === focusedTaskId && + request.taskInstanceId === focusedTaskInstanceId && + (request.kind === "snapshot" || request.taskId !== undefined) + ) +} + +/** Shared by the production driver and the exhaustive bounded explorer. No I/O or mutation. */ +export function reduceTranscriptTransport( + state: TranscriptTransportState, + action: TranscriptAction, +): TranscriptTransition { + const result: TranscriptTransition = { state, release: [], settle: [] } + const discard = (job: TranscriptJob) => { + result.release.push(job.id) + if (state.inFlight?.job.id !== job.id) result.settle.push({ id: job.id }) + } + switch (action.type) { + case "enqueue": { + const { request, total, focusedTaskId, focusedTaskInstanceId } = action + const snapshot = request.kind === "snapshot" + if (!snapshot && total === 0) return result + if (!isTranscriptRequestCurrent(state, request, focusedTaskId, focusedTaskInstanceId)) return result + const sequences = new Map(state.sequences) + const seq = request.taskId + ? (sequences.get(request.taskId) ?? 0) + (!snapshot || request.bumpSeq ? 1 : 0) + : 0 + if (request.taskId) sequences.set(request.taskId, seq) + const nextSnapshotId = state.nextSnapshotId + (snapshot ? 1 : 0) + const job: TranscriptJob = { + id: state.nextJobId + 1, + generation: state.generation, + taskId: request.taskId, + taskInstanceId: request.taskInstanceId, + seq, + kind: request.kind, + total, + ...(snapshot ? { snapshotId: `${request.taskId ?? "none"}:${nextSnapshotId}` } : {}), + } + result.accepted = job + result.state = { ...state, sequences, nextSnapshotId, nextJobId: job.id, queue: [...state.queue, job] } + return result + } + case "invalidate": + state.queue.forEach(discard) + if (state.active) discard(state.active.job) + // Never reset inFlight: an already invoked physical send cannot be unsent. + result.state = { ...state, generation: state.generation + 1, queue: [], active: undefined } + return result + case "shutdown": { + if (state.closed) return result + const ids = new Set(state.queue.map((job) => job.id)) + if (state.active) ids.add(state.active.job.id) + if (state.inFlight) ids.add(state.inFlight.job.id) + result.release = [...ids] + result.settle = [...ids].map((id) => ({ id })) + result.state = { + ...state, + closed: true, + generation: state.generation + 1, + queue: [], + active: undefined, + inFlight: undefined, + } + return result + } + case "reopen": + // Preserve sequences and IDs. Work captured before/during closure must never + // acquire the new renderer's generation, even if the task is unchanged. + if (state.closed) result.state = { ...state, closed: false, generation: state.generation + 1 } + return result + case "forget-task": { + const sequences = new Map(state.sequences) + sequences.delete(action.taskId) + result.state = { ...state, sequences } + return result + } + case "pump": { + if (state.closed || state.inFlight) return result + let active = state.active + const queue = [...state.queue] + while (active || queue.length) { + active ??= { job: queue.shift()!, position: 0 } + const { job, position } = active + if ( + job.generation !== state.generation || + job.taskId !== action.focusedTaskId || + job.taskInstanceId !== action.focusedTaskInstanceId + ) { + discard(job) + active = undefined + continue + } + const chunks = Math.ceil(job.total / state.chunkSize) + const phase = + job.kind !== "snapshot" ? job.kind : position === 0 ? "start" : position > chunks ? "end" : "chunk" + const start = phase === "chunk" ? (position - 1) * state.chunkSize : 0 + const frame: TranscriptFrame = { + job, + phase, + start, + count: phase === "chunk" ? Math.min(state.chunkSize, job.total - start) : 0, + } + result.post = frame + result.state = { ...state, queue, active, inFlight: frame } + return result + } + result.state = { ...state, queue, active } + return result + } + case "settle": { + if ( + !state.inFlight || + state.inFlight.job.id !== action.frame.job.id || + state.inFlight.job.generation !== action.frame.job.generation || + state.inFlight.phase !== action.frame.phase || + state.inFlight.start !== action.frame.start + ) + return result + const { job, phase } = state.inFlight + const finished = !action.success || !state.active || phase === "end" || job.kind !== "snapshot" + if (finished) { + result.release.push(job.id) + result.settle.push({ id: job.id, failed: !action.success }) + } + result.state = { + ...state, + inFlight: undefined, + active: finished ? undefined : { job, position: state.active!.position + 1 }, + } + return result + } + } +} + +const transcriptMessageTypes = { + append: "clineMessageAppended", + update: "clineMessageUpdated", + start: "clineMessagesSnapshotStart", + chunk: "clineMessagesSnapshotChunk", + end: "clineMessagesSnapshotEnd", +} as const satisfies Record + +export function transcriptFrameMessage(frame: TranscriptFrame, messages: readonly ClineMessage[]): ExtensionMessage { + const { job, phase } = frame + const common = { + type: transcriptMessageTypes[phase], + taskId: job.taskId, + taskInstanceId: job.taskInstanceId, + clineMessagesSeq: job.seq, + } + if (phase === "append" || phase === "update") { + return { + ...common, + clineMessage: messages[0], + } + } + const snapshot = { ...common, snapshotId: job.snapshotId } + if (phase === "chunk") { + return { + ...snapshot, + snapshotStartIndex: frame.start, + clineMessages: messages.slice(frame.start, frame.start + frame.count), + } + } + return { + ...snapshot, + snapshotTotal: job.total, + } +} + +type TranscriptCallbacks = { + focusedTaskId: () => string | undefined + postMessage: (message: ExtensionMessage) => Promise + onError: (error: unknown) => void + focusedTaskInstanceId: () => string | undefined +} + +type SendCompletion = { finish?: (success: boolean, error?: unknown) => void } + +// The physical Promise retains only this detachable slot, not the driver, provider, +// frame, payload or caller. Keep these handlers outside the driver's lexical scope. +function observePhysicalSend(promise: Promise, completion: SendCompletion): void { + void promise.then( + () => completion.finish?.(true), + (error: unknown) => completion.finish?.(false, error), + ) +} + +/** One physical-send barrier per renderer, preserved across ordinary invalidations. */ +export class TranscriptTransport { + private state = createTranscriptTransportState() + private readonly payloads = new Map() + private readonly callers = new Map void; reject: (error: unknown) => void }>() + private callbacks?: TranscriptCallbacks + private pendingSend?: SendCompletion + + constructor( + focusedTaskId: () => string | undefined, + postMessage: (message: ExtensionMessage) => Promise, + onError: (error: unknown) => void, + focusedTaskInstanceId: () => string | undefined = () => undefined, + ) { + this.callbacks = { focusedTaskId, postMessage, onError, focusedTaskInstanceId } + } + + get closed(): boolean { + return this.state.closed + } + + shutdown(): void { + if (this.pendingSend) this.pendingSend.finish = undefined + this.pendingSend = undefined + this.callbacks = undefined + this.apply({ type: "shutdown" }) + } + + reopen( + focusedTaskId: () => string | undefined, + postMessage: (message: ExtensionMessage) => Promise, + onError: (error: unknown) => void, + focusedTaskInstanceId: () => string | undefined = () => undefined, + ): void { + if (!this.closed) return + this.callbacks = { focusedTaskId, postMessage, onError, focusedTaskInstanceId } + this.apply({ type: "reopen" }) + } + + get generation(): number { + return this.state.generation + } + + getSequence(taskId: string | undefined): number { + // Allow absent scopes in the read-only view; writers still require string task IDs. + const sequences: ReadonlyMap = this.state.sequences + return sequences.get(taskId) ?? 0 + } + + forgetTask(taskId: string): void { + this.apply({ type: "forget-task", taskId }) + } + + invalidate(): number { + this.apply({ type: "invalidate" }) + return this.generation + } + + enqueue(request: TranscriptRequest, messages: readonly ClineMessage[]): Promise { + const callbacks = this.callbacks + if (!callbacks || this.closed) return Promise.resolve() + // An empty delta must not consume a sequence or enter admission at all. + if (request.kind !== "snapshot" && messages.length === 0) return Promise.resolve() + // Guard before deep cloning (and allocating a sequence/ID). A delayed focus sync + // must not traverse a large, already-obsolete transcript. + const capturedRequest = { ...request, generation: request.generation ?? this.generation } + if ( + !isTranscriptRequestCurrent( + this.state, + capturedRequest, + callbacks.focusedTaskId(), + callbacks.focusedTaskInstanceId(), + ) + ) + return Promise.resolve() + // Task mutates message objects AND nested fields while posts are queued. Capture + // the complete value now, together with its sequence, not at physical-send time. + const payload = structuredClone(messages) + const { accepted } = this.apply({ + type: "enqueue", + request: capturedRequest, + total: payload.length, + focusedTaskId: callbacks.focusedTaskId(), + focusedTaskInstanceId: callbacks.focusedTaskInstanceId(), + }) + if (!accepted) return Promise.resolve() + this.payloads.set(accepted.id, payload) + const promise = new Promise((resolve, reject) => this.callers.set(accepted.id, { resolve, reject })) + this.drain() + return promise + } + + private apply(action: TranscriptAction, error?: unknown): TranscriptTransition { + const transition = reduceTranscriptTransport(this.state, action) + this.state = transition.state + for (const id of transition.release) this.payloads.delete(id) + for (const { id, failed } of transition.settle) { + // Admission registers before drain. The reducer settles each caller exactly once, + // retaining a physical-send caller across invalidations until its send settles. + const caller = this.callers.get(id)! + this.callers.delete(id) + if (failed) caller.reject(error) + else caller.resolve() + } + return transition + } + + private drain(): void { + const callbacks = this.callbacks + if (!callbacks) return + const { post } = this.apply({ + type: "pump", + focusedTaskId: callbacks.focusedTaskId(), + focusedTaskInstanceId: callbacks.focusedTaskInstanceId(), + }) + if (post) this.send(post) + } + + private send(frame: TranscriptFrame): void { + const completion: SendCompletion = { + finish: (success, error) => { + completion.finish = undefined + if (this.pendingSend !== completion) return + this.pendingSend = undefined + this.apply({ type: "settle", frame, success }, error) + if (!success) this.callbacks?.onError(error) + this.drain() + }, + } + this.pendingSend = completion + try { + observePhysicalSend( + this.callbacks!.postMessage(transcriptFrameMessage(frame, this.payloads.get(frame.job.id)!)), + completion, + ) + } catch (error) { + completion.finish?.(false, error) + } + } +}