| 1 | import assert from "node:assert/strict"; |
| 2 | import test from "node:test"; |
| 3 | import type { Change, FollowRequest, Snapshot, TranscriptFollowResponse } from "../generated/desktopContract.generated"; |
| 4 | import { TranscriptFollowClient, type FollowConsumer, type TranscriptFollowClientOptions } from "../lib/transcriptFollowClient"; |
| 5 | import type { TranscriptDiagnostic } from "../lib/transcriptDiagnostics"; |
| 6 | import { addBreadcrumb, snapshotBreadcrumbs } from "../lib/breadcrumbs"; |
| 7 | |
| 8 | function deferred<T>() { |
| 9 | let resolve!: (value: T) => void; |
| 10 | let reject!: (reason?: unknown) => void; |
| 11 | const promise = new Promise<T>((done, fail) => { resolve = done; reject = fail; }); |
| 12 | return { promise, resolve, reject }; |
| 13 | } |
| 14 | async function microtasks() { for (let i = 0; i < 16; i++) await Promise.resolve(); } |
| 15 | function baseline(subscription = "subscription", overrides: Partial<Snapshot> = {}): TranscriptFollowResponse { |
| 16 | return { protocolVersion: 2, subscription, changes: [], resetRequired: false, snapshot: { |
| 17 | protocolVersion: 1, snapshotId: "cut", identity: { sessionId: "session", headId: "", runtimeEpoch: "epoch", rewriteEpoch: 0 }, |
| 18 | projectionRevision: 10, coveredThroughSeq: 4, durableSeq: 4, |
| 19 | records: [{ id: "m:answer", order: 0, message: { role: "assistant", messageId: "answer", content: "prefix" }, refs: [] }], |
| 20 | activeRecords: [], activeAttempts: [], runtime: { status: "in_progress", pendingEvents: [], samplingCount: 0, toolCount: 0 }, |
| 21 | before: 0, hasOlder: false, totalRecords: 1, totalTurns: 1, stale: false, ...overrides, |
| 22 | } }; |
| 23 | } |
| 24 | function suffix(changes: Change[], resetRequired = false): TranscriptFollowResponse { |
| 25 | return { protocolVersion: 2, subscription: "subscription", changes, resetRequired }; |
| 26 | } |
| 27 | function frame(revision: number, text: string): Change { |
| 28 | return { revision, commitSeq: 4, durableSeq: 4, index: 0, event: { kind: "text", messageId: "answer", text } } as Change; |
| 29 | } |
| 30 | function transport(options: TranscriptFollowClientOptions = {}) { |
| 31 | const requests: FollowRequest[] = []; |
| 32 | const pending: ReturnType<typeof deferred<TranscriptFollowResponse>>[] = []; |
| 33 | const read = (request: FollowRequest) => { |
| 34 | requests.push(request); |
| 35 | if (request.close) return Promise.resolve(suffix([])); |
| 36 | const next = deferred<TranscriptFollowResponse>(); pending.push(next); return next.promise; |
| 37 | }; |
| 38 | return { requests, pending, client: new TranscriptFollowClient(read, options) }; |
| 39 | } |
| 40 | function consumer(): FollowConsumer & { text: string; installed: number; delivered: Change[]; states: string[] } { |
| 41 | return { |
| 42 | text: "", installed: 0, delivered: [], states: [], |
| 43 | install(response) { this.installed++; this.text = response.snapshot!.records[0]?.message.content ?? ""; }, |
| 44 | changes(changes) { this.delivered.push(...changes); for (const change of changes) this.text += change.event?.text ?? ""; }, |
| 45 | connection(state) { this.states.push(state); }, |
| 46 | }; |
| 47 | } |
| 48 | |
| 49 | test("follow installs the complete baseline before asking for its suffix", async () => { |
| 50 | const io = transport(); const view = consumer(); const installed = deferred<void>(); |
| 51 | const original = view.install.bind(view); |
| 52 | view.install = async response => { await installed.promise; await original(response); }; |
| 53 | const starting = io.client.start(view); |
| 54 | io.pending.shift()!.resolve(baseline()); |
| 55 | await microtasks(); |
| 56 | assert.equal(io.requests.length, 1, "suffix must wait until installation commits"); |
| 57 | installed.resolve(); await starting; |
| 58 | assert.equal(view.text, "prefix"); |
| 59 | assert.deepEqual(io.requests[1], { subscription: "subscription", afterRevision: 10 }); |
| 60 | io.pending.shift()!.resolve(suffix([frame(11, " suffix")])); await microtasks(); |
| 61 | assert.equal(view.text, "prefix suffix"); io.client.stop(); |
| 62 | }); |
| 63 | |
| 64 | test("follow acknowledges display revisions independently and ignores duplicate frames", async () => { |
| 65 | const io = transport(); const view = consumer(); const starting = io.client.start(view); |
| 66 | io.pending.shift()!.resolve(baseline()); await starting; |
| 67 | io.pending.shift()!.resolve(suffix([frame(11, " first"), frame(11, " first"), frame(12, " second")])); |
| 68 | await microtasks(); |
| 69 | assert.equal(view.text, "prefix first second"); assert.equal(view.delivered.length, 2); |
| 70 | assert.deepEqual(io.requests[io.requests.length - 1], { subscription: "subscription", afterRevision: 12 }); |
| 71 | io.pending.shift()!.resolve(suffix([{ revision: 13, firstSeq: 5, commitSeq: 8, durableSeq: 4, index: 0 }])); |
| 72 | await microtasks(); assert.equal(view.delivered[view.delivered.length - 1]?.commitSeq, 8); io.client.stop(); |
| 73 | }); |
| 74 | |
| 75 | test("settlement pairs with the stable attempt and commit despite delivery order and duplication", async () => { |
| 76 | const io = transport(); const view = consumer(); const starting = io.client.start(view); |
| 77 | io.pending.shift()!.resolve(baseline("subscription", { activeAttempts: [{ id: "attempt", messageId: "answer", turnId: "turn", nextIndex: 1 }] })); |
| 78 | await starting; |
| 79 | const commit: Change = { revision: 11, firstSeq: 5, commitSeq: 5, durableSeq: 4, index: 0, records: [{ messageId: "answer", role: "assistant", content: "complete" }] }; |
| 80 | const end: Change = { revision: 12, commitSeq: 5, durableSeq: 5, index: 1, attemptId: "attempt", resultSeq: 5, resultKind: "message/complete", |
| 81 | event: { kind: "stream_attempt", messageId: "answer", streamAttempt: { id: "attempt", action: "commit" } } } as Change; |
| 82 | io.pending.shift()!.resolve(suffix([end, commit, end])); await microtasks(); |
| 83 | assert.deepEqual(view.delivered.map(change => change.revision), [11, 12]); |
| 84 | assert.equal(view.states[view.states.length - 1], "connected"); io.client.stop(); |
| 85 | }); |
| 86 | |
| 87 | for (const failure of ["overflow", "revision gap", "business gap", "sampling gap", "settlement mismatch"] as const) { |
| 88 | test(`follow ${failure} preserves visible content and only requests a new baseline`, async t => { |
| 89 | t.mock.timers.enable({ apis: ["setTimeout"] }); |
| 90 | const diagnostics: TranscriptDiagnostic[] = []; |
| 91 | const io = transport({ onDiagnostic: (event, visible) => { if (visible) diagnostics.push(event); } }); |
| 92 | const view = consumer(); const starting = io.client.start(view); |
| 93 | io.pending.shift()!.resolve(baseline()); await starting; |
| 94 | const broken = failure === "overflow" ? suffix([], true) |
| 95 | : failure === "revision gap" ? suffix([frame(12, "bad")]) |
| 96 | : failure === "business gap" ? suffix([{ revision: 11, firstSeq: 6, commitSeq: 6, durableSeq: 4, index: 0 }]) |
| 97 | : failure === "settlement mismatch" ? suffix([{ revision: 11, commitSeq: 4, durableSeq: 4, index: 0, attemptId: "unknown", resultSeq: 4, resultKind: "message/complete", event: { kind: "stream_attempt", messageId: "different", streamAttempt: { id: "unknown", action: "commit" } } } as Change]) |
| 98 | : suffix([{ ...frame(11, "bad"), attemptId: "missing-attempt", index: 3 }]); |
| 99 | io.pending.shift()!.resolve(broken); await microtasks(); |
| 100 | assert.equal(view.text, "prefix"); assert.equal(view.delivered.length, 0); |
| 101 | assert.equal(view.states[view.states.length - 1], "disconnected"); |
| 102 | assert.ok(io.requests.some(request => request.close && request.subscription === "subscription")); |
| 103 | const expectedReason = failure === "overflow" ? "resync_required" |
| 104 | : failure === "revision gap" ? "revision_gap" |
| 105 | : failure === "business gap" ? "business_gap" |
| 106 | : failure === "sampling gap" ? "sampling_gap" |
| 107 | : "settlement_identity_mismatch"; |
| 108 | assert.deepEqual( |
| 109 | diagnostics[0] && { event: diagnostics[0].event, stage: diagnostics[0].stage, reason: diagnostics[0].reason }, |
| 110 | { event: "failure", stage: "delta_validate", reason: expectedReason }, |
| 111 | ); |
| 112 | t.mock.timers.tick(1000); await microtasks(); |
| 113 | assert.deepEqual(io.requests[io.requests.length - 1], {}, "recovery issues a read, never a model submission"); |
| 114 | assert.equal(view.text, "prefix", "content remains visible while replacement is pending"); |
| 115 | io.pending.shift()!.resolve(baseline("replacement", { projectionRevision: 20, coveredThroughSeq: 8 })); |
| 116 | await microtasks(); assert.equal(view.installed, 2); assert.equal(view.states[view.states.length - 1], "connected"); |
| 117 | assert.ok(io.requests.every(request => Object.keys(request).every(key => ["subscription", "afterRevision", "close"].includes(key)))); |
| 118 | io.client.stop(); |
| 119 | }); |
| 120 | } |
| 121 | |
| 122 | test("identical transcript failures are summarized every 30 seconds and recovery is recorded once", async t => { |
| 123 | t.mock.timers.enable({ apis: ["setTimeout"] }); |
| 124 | let now = 0; |
| 125 | const diagnostics: Array<{ event: TranscriptDiagnostic; visible: boolean }> = []; |
| 126 | const io = transport({ now: () => now, onDiagnostic: (event, visible) => diagnostics.push({ event, visible }) }); |
| 127 | const view = consumer(); |
| 128 | const starting = io.client.start(view); |
| 129 | io.pending.shift()!.resolve(baseline()); |
| 130 | await starting; |
| 131 | |
| 132 | io.pending.shift()!.reject(new Error("transport unavailable")); |
| 133 | await microtasks(); |
| 134 | now = 1_000; t.mock.timers.tick(1000); await microtasks(); |
| 135 | io.pending.shift()!.reject(new Error("transport unavailable")); |
| 136 | await microtasks(); |
| 137 | now = 2_000; t.mock.timers.tick(1000); await microtasks(); |
| 138 | io.pending.shift()!.reject(new Error("transport unavailable")); |
| 139 | await microtasks(); |
| 140 | now = 32_000; t.mock.timers.tick(1000); await microtasks(); |
| 141 | io.pending.shift()!.reject(new Error("transport unavailable")); |
| 142 | await microtasks(); |
| 143 | now = 33_000; t.mock.timers.tick(1000); await microtasks(); |
| 144 | io.pending.shift()!.resolve(baseline("replacement", { projectionRevision: 20 })); |
| 145 | await microtasks(); |
| 146 | assert.ok(!diagnostics.some(({ event }) => event.event === "recovered"), "a baseline alone does not prove following recovered"); |
| 147 | io.pending.shift()!.resolve(suffix([])); |
| 148 | await microtasks(); |
| 149 | |
| 150 | assert.deepEqual( |
| 151 | diagnostics.map(({ event, visible }) => ({ type: event.event, stage: event.stage, failures: event.failures, visible })), |
| 152 | [ |
| 153 | { type: "failure", stage: "delta_read", failures: 1, visible: true }, |
| 154 | { type: "failure", stage: "baseline_read", failures: 2, visible: true }, |
| 155 | { type: "failure", stage: "baseline_read", failures: 3, visible: false }, |
| 156 | { type: "summary", stage: "baseline_read", failures: 4, visible: true }, |
| 157 | { type: "recovered", stage: "baseline_read", failures: 4, visible: true }, |
| 158 | ], |
| 159 | ); |
| 160 | io.client.stop(); |
| 161 | }); |
| 162 | |
| 163 | test("persistent delta failure retains its segment across successful baseline retries", async t => { |
| 164 | t.mock.timers.enable({ apis: ["setTimeout"] }); |
| 165 | let now = 0; |
| 166 | const diagnostics: TranscriptDiagnostic[] = []; |
| 167 | const io = transport({ now: () => now, onDiagnostic: (event, visible) => { if (visible) diagnostics.push(event); } }); |
| 168 | const starting = io.client.start(consumer()); |
| 169 | io.pending.shift()!.resolve(baseline()); await starting; |
| 170 | addBreadcrumb("test.close", "preserve navigation and close context"); |
| 171 | for (let i = 0; i < 32; i++) { |
| 172 | io.pending.shift()!.resolve(suffix([frame(12, "gap")])); await microtasks(); |
| 173 | now += 1000; t.mock.timers.tick(1000); await microtasks(); |
| 174 | io.pending.shift()!.resolve(baseline(`retry-${i}`)); await microtasks(); |
| 175 | } |
| 176 | assert.deepEqual(diagnostics.map(event => [event.event, event.failures]), [["failure", 1], ["summary", 31]]); |
| 177 | assert.ok(snapshotBreadcrumbs().some(crumb => crumb.cat === "test.close")); |
| 178 | io.pending.shift()!.resolve(suffix([frame(11, "valid")])); await microtasks(); |
| 179 | const recovery = diagnostics[diagnostics.length - 1]; |
| 180 | assert.deepEqual([recovery.event, recovery.failures, recovery.durationMs], ["recovered", 32, 32_000]); |
| 181 | io.client.stop(); |
| 182 | }); |
| 183 | |
| 184 | for (const capability of ["absent", "throws", "rejects"] as const) { |
| 185 | test(`diagnostic host capability ${capability} cannot interrupt follow recovery`, async t => { |
| 186 | t.mock.timers.enable({ apis: ["setTimeout"] }); |
| 187 | const previous = Object.getOwnPropertyDescriptor(globalThis, "window"); |
| 188 | const native = capability === "absent" ? {} : { |
| 189 | recordRendererDiagnostic: () => { |
| 190 | if (capability === "throws") throw new Error("diagnostic transport broken"); |
| 191 | return Promise.reject(new Error("diagnostic write failed")); |
| 192 | }, |
| 193 | }; |
| 194 | Object.defineProperty(globalThis, "window", { configurable: true, value: { reasonixDesktop: { native } } }); |
| 195 | const io = transport(); const view = consumer(); |
| 196 | try { |
| 197 | const starting = io.client.start(view); |
| 198 | io.pending.shift()!.resolve(baseline()); await starting; |
| 199 | io.pending.shift()!.reject(new Error("original transport failure")); await microtasks(); |
| 200 | t.mock.timers.tick(1000); await microtasks(); |
| 201 | assert.deepEqual(io.requests[io.requests.length - 1], {}, "the original failure still schedules recovery"); |
| 202 | io.pending.shift()!.resolve(baseline("recovered")); await microtasks(); |
| 203 | io.pending.shift()!.resolve(suffix([frame(11, " recovered")])); await microtasks(); |
| 204 | assert.equal(view.text, "prefix recovered"); |
| 205 | assert.equal(view.states[view.states.length - 1], "connected"); |
| 206 | } finally { |
| 207 | io.client.stop(); |
| 208 | if (previous) Object.defineProperty(globalThis, "window", previous); |
| 209 | else Reflect.deleteProperty(globalThis, "window"); |
| 210 | } |
| 211 | }); |
| 212 | } |
| 213 | |
| 214 | for (const outcome of ["late baseline", "install resolves", "install rejects"] as const) { |
| 215 | test(`service stopping suppresses cleanup when ${outcome}`, async () => { |
| 216 | const io = transport(); const view = consumer(); const install = deferred<void>(); |
| 217 | if (outcome !== "late baseline") view.install = () => install.promise; |
| 218 | const starting = io.client.start(view); |
| 219 | if (outcome !== "late baseline") { io.pending.shift()!.resolve(baseline()); await microtasks(); } |
| 220 | io.client.stop(false, "service_stopping"); |
| 221 | if (outcome === "late baseline") io.pending.shift()!.resolve(baseline()); |
| 222 | else if (outcome === "install resolves") install.resolve(); |
| 223 | else install.reject(new Error("late content read rejected")); |
| 224 | if (outcome === "install rejects") await assert.rejects(starting, /late content read rejected/); |
| 225 | else await starting; |
| 226 | assert.ok(!io.requests.some(request => request.close)); |
| 227 | assert.ok(!view.states.includes("connected")); |
| 228 | }); |
| 229 | } |
| 230 | |
| 231 | test("remote baseline transport failures retain their channel and stable classification", async () => { |
| 232 | const diagnostics: TranscriptDiagnostic[] = []; |
| 233 | const io = transport({ transport: "remote", onDiagnostic: (event, visible) => { if (visible) diagnostics.push(event); } }); |
| 234 | const starting = io.client.start(consumer()); |
| 235 | io.pending.shift()!.reject(new Error("remote response body must stay private")); |
| 236 | await assert.rejects(starting, /remote response body/); |
| 237 | assert.deepEqual( |
| 238 | diagnostics.map(event => ({ transport: event.transport, stage: event.stage, reason: event.reason })), |
| 239 | [{ transport: "remote", stage: "baseline_read", reason: "transport_rejected" }], |
| 240 | ); |
| 241 | }); |
| 242 | |
| 243 | test("stopped generations ignore delayed suffixes and close their subscription", async () => { |
| 244 | const io = transport(); const oldView = consumer(); const starting = io.client.start(oldView); |
| 245 | io.pending.shift()!.resolve(baseline()); await starting; |
| 246 | const oldSuffix = io.pending.shift()!; io.client.stop(); |
| 247 | const newView = consumer(); const restarting = io.client.start(newView); |
| 248 | io.pending.shift()!.resolve(baseline("next", { identity: { sessionId: "next", headId: "", runtimeEpoch: "epoch-next", rewriteEpoch: 0 } })); |
| 249 | await restarting; |
| 250 | oldSuffix.resolve(suffix([frame(11, " stale")])); await microtasks(); |
| 251 | assert.equal(oldView.text, "prefix"); assert.equal(newView.text, "prefix"); |
| 252 | assert.ok(io.requests.some(request => request.close && request.subscription === "subscription")); io.client.stop(); |
| 253 | }); |
| 254 | |
| 255 | test("service shutdown stops in-flight delivery without sending a cleanup RPC", async () => { |
| 256 | const diagnostics: TranscriptDiagnostic[] = []; |
| 257 | const io = transport({ onDiagnostic: (event, visible) => { if (visible) diagnostics.push(event); } }); |
| 258 | const view = consumer(); const starting = io.client.start(view); |
| 259 | io.pending.shift()!.resolve(baseline()); await starting; |
| 260 | const delayed = io.pending.shift()!; |
| 261 | io.client.stop(false, "service_stopping"); |
| 262 | delayed.resolve(suffix([frame(11, " stale")])); |
| 263 | await microtasks(); |
| 264 | |
| 265 | assert.equal(view.text, "prefix"); |
| 266 | assert.ok(!io.requests.some(request => request.close), "a stopping service must not receive subscription cleanup"); |
| 267 | assert.deepEqual( |
| 268 | diagnostics.map(event => ({ event: event.event, stage: event.stage, reason: event.reason })), |
| 269 | [{ event: "stopped", stage: "none", reason: "service_stopping" }], |
| 270 | ); |
| 271 | }); |
| 272 | |
| 273 | test("stopping before baseline arrives closes the late subscription without installation", async () => { |
| 274 | const io = transport(); const view = consumer(); const starting = io.client.start(view); |
| 275 | const late = io.pending.shift()!; io.client.stop(); late.resolve(baseline()); await starting; await microtasks(); |
| 276 | assert.equal(view.installed, 0); |
| 277 | assert.ok(io.requests.some(request => request.close && request.subscription === "subscription")); |
| 278 | }); |
| 279 | |
| 280 | test("stopping during asynchronous installation releases the newly opened subscription", async () => { |
| 281 | const io = transport(); const installed = deferred<void>(); const view = consumer(); |
| 282 | view.install = async () => installed.promise; |
| 283 | const starting = io.client.start(view); io.pending.shift()!.resolve(baseline()); await microtasks(); |
| 284 | io.client.stop(); installed.resolve(); await starting; await microtasks(); |
| 285 | assert.ok(io.requests.some(request => request.close && request.subscription === "subscription"), "cancelled install leaked its subscription"); |
| 286 | assert.ok(!view.states.includes("connected")); |
| 287 | }); |
| 288 | |
| 289 | test("a stopped generation's cleanup policy cannot affect a replacement subscription", async () => { |
| 290 | const io = transport(); |
| 291 | const oldStart = io.client.start(consumer()); |
| 292 | const oldBaseline = io.pending.shift()!; |
| 293 | io.client.stop(false, "service_stopping"); |
| 294 | const newStart = io.client.start(consumer()); |
| 295 | io.pending.shift()!.resolve(baseline("new-service")); await newStart; |
| 296 | oldBaseline.resolve(baseline("old-service")); await oldStart; |
| 297 | io.client.stop(); |
| 298 | assert.deepEqual(io.requests.filter(request => request.close), [{ subscription: "new-service", close: true }]); |
| 299 | }); |
| 300 |