| 1 | import assert from "node:assert/strict"; |
| 2 | import test from "node:test"; |
| 3 | import type { FollowRequest, TranscriptFollowResponse } from "../generated/desktopContract.generated"; |
| 4 | import { installDesktopHostStub } from "./desktopHostStub"; |
| 5 | |
| 6 | Object.defineProperty(globalThis, "window", { configurable: true, value: {} }); |
| 7 | const commands: Record<string, unknown> = {}; |
| 8 | const desktopStub = installDesktopHostStub(commands); |
| 9 | const [{ TranscriptSessionFollower }, { initialState, reducer }, { getTranscriptStore }] = await Promise.all([ |
| 10 | import("../lib/transcriptSessionFollower"), import("../lib/useController"), import("../lib/transcriptStore"), |
| 11 | ]); |
| 12 | function deferred<T>() { |
| 13 | let resolve!: (value: T) => void; |
| 14 | let reject!: (error: unknown) => void; |
| 15 | const promise = new Promise<T>((done, fail) => { resolve = done; reject = fail; }); |
| 16 | return { promise, resolve, reject }; |
| 17 | } |
| 18 | |
| 19 | test("expired settled content requests the owning follower to resynchronize", async () => { |
| 20 | const { canonicalHistoryContent, registerTranscriptContentRecovery } = await import("../lib/canonicalTranscriptBackend"); |
| 21 | let recoveries = 0; |
| 22 | commands.TranscriptContentForTab = async () => ({ stale: true }); |
| 23 | const release = registerTranscriptContentRecovery("expired-content", () => { recoveries++; }); |
| 24 | const ref = { entryId: "m:answer", field: "content", size: 1, chunks: 1, revision: 1, digest: "expired", |
| 25 | transcriptRef: { snapshotId: "expired", recordId: "m:answer", path: ["content"], bytes: 1 } }; |
| 26 | await assert.rejects(canonicalHistoryContent("expired-content", ref, 0), /synchronizing/); |
| 27 | assert.equal(recoveries, 1); |
| 28 | release(); |
| 29 | await assert.rejects(canonicalHistoryContent("expired-content", ref, 0), /synchronizing/); |
| 30 | assert.equal(recoveries, 1, "released owner cannot be restarted by a late content result"); |
| 31 | const delayed = deferred<{ stale: boolean }>(); |
| 32 | commands.TranscriptContentForTab = () => delayed.promise; |
| 33 | const oldRelease = registerTranscriptContentRecovery("expired-content", () => { recoveries++; }); |
| 34 | const staleRead = canonicalHistoryContent("expired-content", ref, 0); |
| 35 | oldRelease(); |
| 36 | const newRelease = registerTranscriptContentRecovery("expired-content", () => { recoveries += 100; }); |
| 37 | delayed.resolve({ stale: true }); |
| 38 | await assert.rejects(staleRead, /synchronizing/); |
| 39 | assert.equal(recoveries, 1, "late stale reference cannot resynchronize a replacement session"); |
| 40 | newRelease(); |
| 41 | }); |
| 42 | |
| 43 | test("reading old pages isolates live output and rejoins the current stable node", () => { |
| 44 | let state: import("../lib/useController").State = { ...initialState, transcriptProtocol: 2, historyHasNewer: true, |
| 45 | items: [{ kind: "user" as const, id: "old-reader", text: "old page" }] }; |
| 46 | const resident = state.items; |
| 47 | state = reducer(state, { type: "event", e: { kind: "text", messageId: "active", text: "full prefix" } }); |
| 48 | assert.equal(state.items, resident); |
| 49 | assert.equal(state.offscreenItems?.find(item => item.id === "m:active")?.kind, "assistant"); |
| 50 | const active = state.offscreenItems!.find(item => item.id === "m:active")!; |
| 51 | state = reducer(state, { type: "history_append", items: [], startTurn: 0, endTurn: 1, totalTurns: 1, hasOlder: true, hasNewer: false }); |
| 52 | assert.equal(state.items.find(item => item.id === "m:active"), active); |
| 53 | assert.equal(state.offscreenItems, undefined); |
| 54 | }); |
| 55 | |
| 56 | test("business replacement preserves authoritative final turn annotations", () => { |
| 57 | const final = { kind: "assistant" as const, id: "m:final", text: "answer", reasoning: "", streaming: false, turnFinal: true, turnDurationMs: 933524, samplingCount: 72, toolCount: 72 }; |
| 58 | const state = reducer({ ...initialState, items: [final] }, { type: "transcript_records", confirmedUsers: [], projection: { |
| 59 | items: [{ ...final, turnFinal: undefined, turnDurationMs: undefined, samplingCount: undefined, toolCount: undefined }], |
| 60 | removeIds: [], startTurn: 0, endTurn: 1, totalTurns: 1, hasOlder: false, hasNewer: false, revision: 2, revisionKnown: true, digest: "cut", |
| 61 | } }); |
| 62 | assert.equal(state.items[0].kind === "assistant" && state.items[0].turnDurationMs, 933524); |
| 63 | }); |
| 64 | async function microtasks() { for (let i = 0; i < 16; i++) await Promise.resolve(); } |
| 65 | |
| 66 | for (const remote of [false, true]) for (const scenario of ["direct", "event-first", "late", "stale", "missing"] as const) { |
| 67 | const lateBinding = scenario !== "direct" && scenario !== "event-first"; |
| 68 | test(`${remote ? "remote" : "local"} offscreen submission confirmation (${scenario})`, async () => { |
| 69 | const tab = `offscreen-${remote}-${scenario}`, path = `/session/${tab}`; |
| 70 | let state = { ...initialState }; |
| 71 | const polls: Array<ReturnType<typeof deferred<TranscriptFollowResponse>>> = []; |
| 72 | const readKey = remote ? "RemoteTranscriptFollowForTab" : "TranscriptFollowForTab"; |
| 73 | commands[readKey] = (_tab: string, request: FollowRequest) => { |
| 74 | if (request.close) return Promise.resolve({ protocolVersion: 2, changes: [], resetRequired: false, subscription: tab }); |
| 75 | if (!request.subscription) return Promise.resolve(initial(tab)); |
| 76 | const poll = deferred<TranscriptFollowResponse>(); polls.push(poll); return poll.promise; |
| 77 | }; |
| 78 | let lookups = 0; |
| 79 | const lookup = deferred<{ status: string; messages: Array<{ role: string; messageId: string }> }>(); |
| 80 | commands[remote ? "RemoteSessionHistoryWindowForTab" : "SessionHistoryWindowForTab"] = (_tab: string, request: { anchor: string; messageId: string }) => { |
| 81 | lookups++; assert.equal(request.anchor, "message"); assert.equal(request.messageId, "sent"); return lookup.promise; |
| 82 | }; |
| 83 | const follower = new TranscriptSessionFollower(tab, path, remote, action => { state = reducer(state, action); }, () => state); |
| 84 | await follower.start(); |
| 85 | try { |
| 86 | state = reducer(state, { type: "user", seq: 0, text: "question", submissionId: "submit" }); |
| 87 | const old = getTranscriptStore().installSlice(tab, path, { |
| 88 | entries: [{ entryId: "m:old", turn: 1, order: 0, message: { role: "user", content: "old page", messageId: "old" }, refs: [] }], |
| 89 | nextCursor: "", newerCursor: "next", hasOlder: false, hasNewer: true, startTurn: 1, endTurn: 1, totalTurns: 100, |
| 90 | revision: 4, digest: "generation", stale: false, |
| 91 | }); |
| 92 | state = reducer(state, { type: "history_replace", ...old }); |
| 93 | const resident = state.items; |
| 94 | if (scenario === "event-first") { |
| 95 | polls.shift()!.resolve({ protocolVersion: 2, subscription: tab, changes: [{ revision: 11, commitSeq: 4, durableSeq: 4, index: 0, |
| 96 | event: { kind: "user_message", messageId: "sent", submissionId: "submit", source: "executor" } }], resetRequired: false }); |
| 97 | await microtasks(); |
| 98 | assert.equal(lookups, 0, "an identity event alone must not initiate a history read"); |
| 99 | assert.ok(state.localSubmissions.submit, "identity alone keeps the echo until the formal record"); |
| 100 | } |
| 101 | polls.shift()!.resolve({ protocolVersion: 2, subscription: tab, changes: [{ revision: scenario === "event-first" ? 12 : 11, firstSeq: 5, commitSeq: 5, durableSeq: 5, index: 0, |
| 102 | records: [{ role: "user", messageId: "sent", submissionId: lateBinding ? undefined : "submit", content: "question" }] }], resetRequired: false }); |
| 103 | await microtasks(); |
| 104 | if (lateBinding) { |
| 105 | assert.ok(state.localSubmissions.submit); |
| 106 | polls.shift()!.resolve({ protocolVersion: 2, subscription: tab, changes: [{ revision: 12, commitSeq: 5, durableSeq: 5, index: 0, |
| 107 | event: { kind: "user_message", messageId: "sent", submissionId: "submit", source: "executor" } }], resetRequired: false }); |
| 108 | await microtasks(); |
| 109 | assert.equal(lookups, 1); |
| 110 | if (scenario === "missing") { |
| 111 | polls.shift()!.resolve({ protocolVersion: 2, subscription: tab, changes: [{ revision: 13, commitSeq: 5, durableSeq: 5, index: 0, |
| 112 | event: { kind: "user_message", messageId: "sent", submissionId: "submit", source: "executor" } }], resetRequired: false }); |
| 113 | await microtasks(); |
| 114 | assert.equal(lookups, 1, "duplicate binding does not create a second in-flight read"); |
| 115 | } |
| 116 | if (scenario === "stale") { |
| 117 | state = reducer(state, { type: "reset" }); |
| 118 | state = reducer(state, { type: "user", seq: 0, text: "replacement", submissionId: "submit" }); |
| 119 | state = { ...state, localSubmissions: { submit: { ...state.localSubmissions.submit, messageId: "sent" } } }; |
| 120 | } |
| 121 | lookup.resolve({ status: "ready", messages: scenario === "missing" ? [] : [{ role: "user", messageId: "sent" }] }); |
| 122 | await microtasks(); |
| 123 | if (scenario === "stale" || scenario === "missing") { |
| 124 | assert.ok(state.localSubmissions.submit, "stale or inconclusive reads must retain the current echo"); |
| 125 | assert.equal(state.localSubmissions.submit.status, scenario === "stale" ? "sending" : "accepted"); |
| 126 | if (scenario === "missing") { |
| 127 | assert.deepEqual(state.items, resident); |
| 128 | polls.shift()!.resolve({ protocolVersion: 2, subscription: tab, changes: [{ revision: 14, firstSeq: 6, commitSeq: 6, durableSeq: 6, index: 0, |
| 129 | records: [{ role: "user", messageId: "unrelated", content: "unrelated" }] }], resetRequired: false }); |
| 130 | await microtasks(); |
| 131 | assert.equal(lookups, 2, "new committed coverage permits one retry of an inconclusive read"); |
| 132 | assert.ok(state.localSubmissions.submit); |
| 133 | } |
| 134 | return; |
| 135 | } |
| 136 | } else assert.equal(lookups, 0, "ordinary formal handoff makes no extra history read"); |
| 137 | assert.equal(state.localSubmissionOrder.length, 0); |
| 138 | assert.deepEqual(state.items, resident); |
| 139 | assert.equal(Object.keys(state.visibleSubmissionHandoffs).length, 0); |
| 140 | state = reducer(state, { type: "history_replace", items: [{ kind: "assistant", id: "m:later", text: "later", reasoning: "", streaming: false }], |
| 141 | startTurn: 101, totalTurns: 101, hasOlder: true, hasNewer: false, revision: 6 }); |
| 142 | assert.equal(state.localSubmissionOrder.length, 0); |
| 143 | } finally { follower.stop(); getTranscriptStore().evictTab(tab); } |
| 144 | }); |
| 145 | } |
| 146 | function initial(subscription: string): TranscriptFollowResponse { |
| 147 | return { |
| 148 | protocolVersion: 2, subscription, changes: [], resetRequired: false, |
| 149 | snapshot: { |
| 150 | protocolVersion: 1, snapshotId: "cut", identity: { sessionId: subscription, runtimeEpoch: "epoch", rewriteEpoch: 0, headId: "" }, |
| 151 | projectionRevision: 10, coveredThroughSeq: 4, durableSeq: 4, records: [], activeRecords: [], activeAttempts: [], |
| 152 | runtime: { status: "completed", pendingEvents: [], samplingCount: 0, toolCount: 0 }, before: 0, hasOlder: false, totalRecords: 1, totalTurns: 1, stale: false, |
| 153 | }, |
| 154 | history: { |
| 155 | status: "ready", snapshotSequence: 4, coverageSequence: 4, generation: "generation", totalTurns: 1, hasOlder: false, hasNewer: false, |
| 156 | messages: [{ messageId: "answer", position: 0, version: 1, role: "assistant", eventSequence: 4, visibleTurn: 1, |
| 157 | preview: "", contentRef: { digest: "canonical-message", bytes: 8192 } }], |
| 158 | }, |
| 159 | }; |
| 160 | } |
| 161 | |
| 162 | for (const remote of [false, true]) for (const during of ["baseline", "baseline rejection", "delta", "retry", "load"] as const) { |
| 163 | test(`${remote ? "remote" : "local"} stopping during ${during} fences current and future followers`, async t => { |
| 164 | t.mock.timers.enable({ apis: ["setTimeout"] }); |
| 165 | desktopStub.emitServiceState({ phase: "ready", generation: "running" }); |
| 166 | const tab = `stopping-${remote}-${during}`; |
| 167 | const requests: FollowRequest[] = []; |
| 168 | const baseline = deferred<TranscriptFollowResponse>(); |
| 169 | const delta = deferred<TranscriptFollowResponse>(); |
| 170 | const readStarted = deferred<void>(); |
| 171 | commands[remote ? "RemoteTranscriptFollowForTab" : "TranscriptFollowForTab"] = (_tab: string, request: FollowRequest) => { |
| 172 | requests.push(request); |
| 173 | if (request.close) return Promise.resolve({ protocolVersion: 2, changes: [] }); |
| 174 | readStarted.resolve(); |
| 175 | if (!request.subscription) return baseline.promise; |
| 176 | return delta.promise; |
| 177 | }; |
| 178 | let state = { ...initialState }; |
| 179 | const follower = new TranscriptSessionFollower(tab, "", remote, action => { state = reducer(state, action); }); |
| 180 | const starting = follower.start(); |
| 181 | if (during !== "load") await readStarted.promise; |
| 182 | if (during === "delta" || during === "retry") { |
| 183 | baseline.resolve(initial(tab)); await starting; |
| 184 | if (during === "retry") { |
| 185 | delta.resolve({ protocolVersion: 2, subscription: tab, changes: [], resetRequired: true }); |
| 186 | await microtasks(); |
| 187 | } |
| 188 | } |
| 189 | const before = requests.length; |
| 190 | desktopStub.emitServiceState({ phase: "stopping", generation: "running" }); |
| 191 | const visible = state; |
| 192 | if (during === "baseline rejection") baseline.reject(new Error("stopping service rejected pending baseline")); |
| 193 | else baseline.resolve(initial(tab)); |
| 194 | delta.resolve({ protocolVersion: 2, subscription: tab, changes: [{ revision: 11, commitSeq: 4, durableSeq: 4, index: 0, event: { kind: "text", messageId: "answer", text: "late" } }], resetRequired: false }); |
| 195 | await starting; await microtasks(); |
| 196 | t.mock.timers.tick(1000); await microtasks(); |
| 197 | const late = new TranscriptSessionFollower(`${tab}-late`, "", remote, () => { throw new Error("stopped service must not publish"); }); |
| 198 | await late.start(); await follower.start(); |
| 199 | assert.equal(requests.length, before, "no cleanup, retry or new baseline after stopping"); |
| 200 | assert.equal(state, visible, "a late response cannot update the retained transcript"); |
| 201 | follower.stop(); late.stop(); |
| 202 | desktopStub.emitServiceState({ phase: "ready", generation: "replacement" }); |
| 203 | getTranscriptStore().evictTab(tab); |
| 204 | }); |
| 205 | } |
| 206 | |
| 207 | for (const remote of [false, true]) test(`${remote ? "remote" : "local"} follower preserves an outer snapshot identity from an older peer`, async () => { |
| 208 | const tab = `outer-record-${remote}`, path = `/session/${tab}`; |
| 209 | const response = initial(tab); |
| 210 | response.snapshot!.records = [ |
| 211 | { id: "view:older:1", order: 0, message: { role: "notice", content: "outer identity" }, refs: [] }, |
| 212 | { id: "tool:older-call", order: 1, message: { role: "tool", messageId: "tool-message", toolCallId: "older-call", toolName: "read_file", content: "result" }, refs: [] }, |
| 213 | { id: "", order: 2, message: { role: "notice", content: "legacy empty identity" }, refs: [] }, |
| 214 | ]; |
| 215 | response.snapshot!.totalRecords = 3; |
| 216 | const key = remote ? "RemoteTranscriptFollowForTab" : "TranscriptFollowForTab"; |
| 217 | const pending = deferred<TranscriptFollowResponse>(); |
| 218 | commands[key] = (_tab: string, request: FollowRequest) => request.close |
| 219 | ? Promise.resolve({ protocolVersion: 2, subscription: tab, changes: [], resetRequired: false }) |
| 220 | : request.subscription ? pending.promise : Promise.resolve(response); |
| 221 | let state = initialState; |
| 222 | const follower = new TranscriptSessionFollower(tab, path, remote, action => { state = reducer(state, action); }); |
| 223 | try { |
| 224 | await follower.start(); |
| 225 | assert.ok(state.items.some(item => item.kind === "notice" && item.text === "outer identity")); |
| 226 | assert.ok(getTranscriptStore().peek(tab, path)?.items.some(item => item.id === "he:view:older:1")); |
| 227 | assert.ok(state.items.some(item => item.kind === "tool" && item.id === "older-call")); |
| 228 | assert.ok(state.items.some(item => item.kind === "notice" && item.text === "legacy empty identity")); |
| 229 | assert.ok(!getTranscriptStore().peek(tab, path)?.items.some(item => item.id === "he:undefined")); |
| 230 | } finally { follower.stop(); getTranscriptStore().evictTab(tab); } |
| 231 | }); |
| 232 | |
| 233 | test("canonical tool history keeps its message identity when a tool call id is also present", async () => { |
| 234 | const tab = "canonical-tool-identity", path = `/session/${tab}`; |
| 235 | const response = initial(tab); |
| 236 | response.history!.messages = [{ |
| 237 | messageId: "tool-message", position: 0, version: 1, role: "tool", eventSequence: 4, visibleTurn: 1, |
| 238 | preview: "result", inline: { id: "tool-message", role: "tool", tool_call_id: "older-call", name: "read_file", content: "result" }, |
| 239 | }]; |
| 240 | commands.TranscriptFollowForTab = (_tab: string, request: FollowRequest) => Promise.resolve(request.close |
| 241 | ? { protocolVersion: 2, subscription: tab, changes: [], resetRequired: false } |
| 242 | : response); |
| 243 | let state = initialState; |
| 244 | const follower = new TranscriptSessionFollower(tab, path, false, action => { state = reducer(state, action); }); |
| 245 | try { |
| 246 | await follower.start(); |
| 247 | assert.ok(getTranscriptStore().peek(tab, path)?.items.some(item => item.kind === "tool" && item.id === "older-call")); |
| 248 | assert.ok(state.items.some(item => item.kind === "tool" && item.id === "older-call")); |
| 249 | } finally { follower.stop(); getTranscriptStore().evictTab(tab); } |
| 250 | }); |
| 251 | |
| 252 | for (const remote of [false, true]) test(`${remote ? "remote" : "local"} malformed snapshot leaves the resident store unchanged`, async () => { |
| 253 | const tab = `invalid-record-${remote}`, path = `/session/${tab}`; |
| 254 | getTranscriptStore().installSlice(tab, path, { |
| 255 | entries: [{ entryId: "m:resident", turn: 1, order: 0, message: { role: "assistant", messageId: "resident", content: "resident" }, refs: [] }], |
| 256 | nextCursor: "", hasOlder: false, newerCursor: "", hasNewer: false, totalTurns: 1, startTurn: 1, endTurn: 1, |
| 257 | revision: 1, revisionKnown: true, digest: "resident", stale: false, |
| 258 | }); |
| 259 | const response = initial(tab); |
| 260 | response.snapshot!.records = [{ id: "", order: 0, message: { role: "notice", content: "invalid" }, |
| 261 | refs: [{ snapshotId: "cut", recordId: "", path: ["content"], bytes: 100 }] }]; |
| 262 | response.snapshot!.totalRecords = 1; |
| 263 | const key = remote ? "RemoteTranscriptFollowForTab" : "TranscriptFollowForTab"; |
| 264 | commands[key] = (_tab: string, request: FollowRequest) => Promise.resolve(request.close |
| 265 | ? { protocolVersion: 2, subscription: tab, changes: [], resetRequired: false } |
| 266 | : response); |
| 267 | const follower = new TranscriptSessionFollower(tab, path, remote, () => undefined); |
| 268 | try { |
| 269 | await assert.rejects(follower.start(), /transcript snapshot content identity missing/); |
| 270 | const resident = getTranscriptStore().peek(tab, path); |
| 271 | assert.equal(resident?.digest, "resident"); |
| 272 | assert.ok(resident?.items.some(item => item.id === "m:resident")); |
| 273 | assert.ok(!resident?.items.some(item => item.id === "he:undefined")); |
| 274 | } finally { follower.stop(); getTranscriptStore().evictTab(tab); } |
| 275 | }); |
| 276 | |
| 277 | test("reducer rejection does not commit a prepared transcript replacement", async () => { |
| 278 | const tab = "reducer-reject", path = `/session/${tab}`; |
| 279 | getTranscriptStore().installSlice(tab, path, { |
| 280 | entries: [{ entryId: "m:resident", turn: 1, order: 0, message: { role: "assistant", messageId: "resident", content: "resident" }, refs: [] }], |
| 281 | nextCursor: "", hasOlder: false, newerCursor: "", hasNewer: false, totalTurns: 1, startTurn: 1, endTurn: 1, |
| 282 | revision: 1, revisionKnown: true, digest: "resident", stale: false, |
| 283 | }); |
| 284 | commands.TranscriptFollowForTab = (_tab: string, request: FollowRequest) => Promise.resolve(request.close |
| 285 | ? { protocolVersion: 2, subscription: tab, changes: [], resetRequired: false } |
| 286 | : initial(tab)); |
| 287 | const follower = new TranscriptSessionFollower(tab, path, false, action => { |
| 288 | if (action.type === "transcript_v2_snapshot") throw new Error("reducer rejected snapshot"); |
| 289 | }); |
| 290 | try { |
| 291 | await assert.rejects(follower.start(), /reducer rejected snapshot/); |
| 292 | const resident = getTranscriptStore().peek(tab, path); |
| 293 | assert.equal(resident?.digest, "resident"); |
| 294 | assert.ok(resident?.items.some(item => item.id === "m:resident")); |
| 295 | } finally { follower.stop(); getTranscriptStore().evictTab(tab); } |
| 296 | }); |
| 297 | |
| 298 | test("snapshot rejects duplicate durable identities and mismatched content references", async () => { |
| 299 | const cases: Array<{ name: string; records: NonNullable<TranscriptFollowResponse["snapshot"]>["records"]; pattern: RegExp }> = [ |
| 300 | { name: "duplicate", records: [ |
| 301 | { id: "same", order: 0, message: { role: "notice", recordId: "same", content: "first" }, refs: [] }, |
| 302 | { id: "same", order: 1, message: { role: "notice", recordId: "same", content: "second" }, refs: [] }, |
| 303 | ], pattern: /duplicate transcript snapshot record identity/ }, |
| 304 | { name: "content-ref", records: [{ id: "owner", order: 0, message: { role: "notice", recordId: "owner", content: "preview" }, |
| 305 | refs: [{ snapshotId: "cut", recordId: "different", path: ["content"], bytes: 100 }] }], pattern: /content identity mismatch/ }, |
| 306 | { name: "embedded-record", records: [{ id: "owner", order: 0, message: { role: "notice", recordId: "different", content: "preview" }, refs: [] }], |
| 307 | pattern: /record identity mismatch/ }, |
| 308 | ]; |
| 309 | for (const fixture of cases) { |
| 310 | const tab = `invalid-${fixture.name}`; |
| 311 | const response = initial(tab); |
| 312 | response.snapshot!.records = fixture.records; |
| 313 | response.snapshot!.totalRecords = fixture.records.length; |
| 314 | commands.TranscriptFollowForTab = (_tab: string, request: FollowRequest) => Promise.resolve(request.close |
| 315 | ? { protocolVersion: 2, subscription: tab, changes: [], resetRequired: false } |
| 316 | : response); |
| 317 | const follower = new TranscriptSessionFollower(tab, "", false, () => undefined); |
| 318 | try { await assert.rejects(follower.start(), fixture.pattern); } |
| 319 | finally { follower.stop(); getTranscriptStore().evictTab(tab); } |
| 320 | } |
| 321 | }); |
| 322 | |
| 323 | for (const remote of [false, true]) { |
| 324 | test(`${remote ? "remote" : "local"} follower retains an empty canonical body reference as a loadable assistant node`, async () => { |
| 325 | const tab = remote ? "follow-ref-remote" : "follow-ref-local"; |
| 326 | let state = initialState; |
| 327 | const requests: FollowRequest[] = []; |
| 328 | const poll = deferred<TranscriptFollowResponse>(); |
| 329 | const read = async (tabId: string, request: FollowRequest): Promise<TranscriptFollowResponse> => { |
| 330 | assert.equal(tabId, tab); requests.push(request); |
| 331 | if (request.close) return { protocolVersion: 2, subscription: tab, changes: [], resetRequired: false }; |
| 332 | if (!request.subscription) return initial(tab); |
| 333 | return poll.promise; |
| 334 | }; |
| 335 | const key = remote ? "RemoteTranscriptFollowForTab" : "TranscriptFollowForTab"; |
| 336 | const wrong = remote ? "TranscriptFollowForTab" : "RemoteTranscriptFollowForTab"; |
| 337 | commands[key] = read; |
| 338 | commands[wrong] = () => { throw new Error("cross-host fallback is forbidden"); }; |
| 339 | commands.SendForTab = () => { throw new Error("history recovery must not invoke the model"); }; |
| 340 | commands.RemoteSendForTab = commands.SendForTab; |
| 341 | const follower = new TranscriptSessionFollower(tab, `/session/${tab}`, remote, action => { state = reducer(state, action); }); |
| 342 | try { |
| 343 | await follower.start(); |
| 344 | const assistant = state.items.find(item => item.kind === "assistant"); |
| 345 | assert.ok(assistant, "empty inline preview with a canonical ref must not disappear"); |
| 346 | assert.equal(assistant.id, "m:answer"); |
| 347 | assert.equal(getTranscriptStore().hasContentReference(tab, "m:answer", "content"), true); |
| 348 | assert.equal(state.running, false); |
| 349 | assert.equal(state.transcriptProtocol, 2); |
| 350 | assert.equal(requests.length, 2, "follow starts only after initial installation"); |
| 351 | } finally { follower.stop(); } |
| 352 | await microtasks(); |
| 353 | assert.ok(requests.some(request => request.close && request.subscription === tab)); |
| 354 | }); |
| 355 | } |
| 356 | |
| 357 | test("stopped session follower cannot install a delayed baseline into its replaced tab", async () => { |
| 358 | const delayed = deferred<TranscriptFollowResponse>(); |
| 359 | const started = deferred<void>(); |
| 360 | const requests: FollowRequest[] = []; |
| 361 | commands.TranscriptFollowForTab = (_tab: string, request: FollowRequest) => { |
| 362 | requests.push(request); |
| 363 | started.resolve(); |
| 364 | return request.close ? Promise.resolve({ protocolVersion: 2, subscription: "stale", changes: [], resetRequired: false }) : delayed.promise; |
| 365 | }; |
| 366 | let state = initialState; |
| 367 | const follower = new TranscriptSessionFollower("stale-tab", "/session/stale", false, action => { state = reducer(state, action); }); |
| 368 | const loading = follower.start(); await started.promise; follower.stop(); delayed.resolve(initial("stale")); await loading; await microtasks(); |
| 369 | assert.equal(state.items.length, 0); |
| 370 | assert.ok(requests.some(request => request.close && request.subscription === "stale")); |
| 371 | }); |
| 372 | |
| 373 | test("stopping before lazy follow startup prevents a backend subscription", async () => { |
| 374 | let requests = 0; |
| 375 | commands.TranscriptFollowForTab = async () => { requests++; return initial("cancelled-load"); }; |
| 376 | const follower = new TranscriptSessionFollower("cancelled-load", "/session/cancelled-load", false, () => { |
| 377 | assert.fail("cancelled module load cannot publish state"); |
| 378 | }); |
| 379 | const loading = follower.start(); |
| 380 | follower.stop(); |
| 381 | await loading; |
| 382 | assert.equal(requests, 0); |
| 383 | }); |
| 384 | |
| 385 | for (const remote of [false, true]) test(`${remote ? "remote" : "local"} active reference is complete before suffix polling`, async () => { |
| 386 | const response = initial(`prefix-${remote}`); |
| 387 | response.snapshot!.runtime.status = "in_progress"; |
| 388 | response.snapshot!.activeAttempts = [{ id: "attempt", messageId: "answer", turnId: "turn", nextIndex: 1 }]; |
| 389 | response.snapshot!.records = [{ id: "m:answer", order: 0, message: { role: "assistant", messageId: "answer", content: "truncated" }, |
| 390 | refs: [{ snapshotId: "cut", recordId: "m:answer", path: ["content"], bytes: 20 }] }]; |
| 391 | const content = deferred<{ data: string; nextOffset: number; done: boolean; stale: boolean }>(); |
| 392 | const poll = deferred<TranscriptFollowResponse>(); |
| 393 | let polls = 0; |
| 394 | commands[remote ? "RemoteTranscriptContentForTab" : "TranscriptContentForTab"] = () => content.promise; |
| 395 | commands[remote ? "RemoteTranscriptFollowForTab" : "TranscriptFollowForTab"] = (_tab: string, request: FollowRequest) => { |
| 396 | if (request.close) return Promise.resolve({ protocolVersion: 2, changes: [], resetRequired: false, subscription: response.subscription }); |
| 397 | if (!request.subscription) return Promise.resolve(response); |
| 398 | polls++; return poll.promise; |
| 399 | }; |
| 400 | let state = initialState; |
| 401 | const follower = new TranscriptSessionFollower(`prefix-${remote}`, "", remote, action => { state = reducer(state, action); }); |
| 402 | const starting = follower.start(); |
| 403 | await microtasks(); |
| 404 | assert.equal(polls, 0); |
| 405 | assert.equal(state.items.length, 0, "partial baseline is not published"); |
| 406 | content.resolve({ data: "complete prefix", nextOffset: 15, done: true, stale: false }); |
| 407 | await starting; |
| 408 | assert.equal(state.live?.text, "complete prefix"); |
| 409 | assert.equal(polls, 1); |
| 410 | follower.stop(); |
| 411 | }); |
| 412 | |
| 413 | test("v2 keeps deferred bodies through subsequent samples and attaches terminal time by backend message identity", () => { |
| 414 | let state: import("../lib/useController").State = { ...initialState, transcriptProtocol: 2, running: true, activeTurnId: "turn", |
| 415 | items: [ |
| 416 | { kind: "assistant" as const, id: "m:deferred", text: "", reasoning: "", streaming: false }, |
| 417 | { kind: "assistant" as const, id: "m:final", text: "answer", reasoning: "thinking", streaming: false }, |
| 418 | ] }; |
| 419 | state = reducer(state, { type: "event", e: { kind: "text", messageId: "next", text: "later sample" } }); |
| 420 | assert.ok(state.items.some(item => item.id === "m:deferred"), "text frames must not remove unloaded body owners"); |
| 421 | state = reducer(state, { type: "event", e: { kind: "turn_done", turnId: "turn" } }); |
| 422 | state = reducer(state, { type: "transcript_runtime", runtime: { turnId: "turn", status: "completed", |
| 423 | finalMessageId: "final", durationMs: 933524, samplingCount: 72, toolCount: 72, pendingEvents: [] } }); |
| 424 | const final = state.items.find(item => item.id === "m:final"); |
| 425 | assert.ok(final?.kind === "assistant"); |
| 426 | assert.equal(final.turnDurationMs, 933524); |
| 427 | assert.equal(final.turnFinal, true); |
| 428 | assert.ok(state.items.some(item => item.id === "m:deferred"), "terminal must retain deferred body owners"); |
| 429 | const later = state.items.find(item => item.id === "m:next"); |
| 430 | assert.ok(later?.kind === "assistant"); |
| 431 | assert.equal(later.turnDurationMs, undefined, "array-tail sample is not the final reply"); |
| 432 | }); |
| 433 |