| 1 | import assert from "node:assert/strict"; |
| 2 | import test from "node:test"; |
| 3 | import { initialState, reducer, type Item } from "../lib/useController"; |
| 4 | import { ChatSource } from "../lib/chatViewSource"; |
| 5 | import { orderedLocalSubmissions } from "../lib/localSubmissionState"; |
| 6 | import type { TranscriptSnapshot } from "../lib/transcriptProtocol"; |
| 7 | |
| 8 | const user = (messageId: string, submissionId?: string): Item => ({ kind: "user", id: `m:${messageId}`, messageId, submissionId, text: "question" }); |
| 9 | const projection = (items: Item[]) => ({ items, removeIds: [], startTurn: 1, endTurn: 1, totalTurns: 1, |
| 10 | hasOlder: false, hasNewer: false, revision: 1, revisionKnown: true, digest: "cut" }); |
| 11 | const sent = () => reducer({ ...initialState, transcriptProtocol: 2 }, { type: "user", seq: 0, text: "question", submissionId: "send" }); |
| 12 | const event = (messageId = "durable") => ({ type: "event" as const, e: { kind: "user_message" as const, source: "executor" as const, submissionId: "send", messageId } }); |
| 13 | const input = { running: true, hydrating: false, hasOlder: false, loadingOlder: false }; |
| 14 | |
| 15 | test("a batched bound handoff reserves its displayed key before a conflicting canonical row", () => { |
| 16 | const source = new ChatSource("batched-conflict"); |
| 17 | let state = sent(); |
| 18 | source.update({ ...input, items: [], localSubmissions: orderedLocalSubmissions(state) }); |
| 19 | const original = source.getOrderSnapshot().find(key => source.getNodeSnapshot(key)?.kind === "user"); |
| 20 | state = reducer(state, event("bound")); |
| 21 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], |
| 22 | projection: projection([user("conflict", "send"), user("bound", "send")]) }); |
| 23 | source.update({ ...input, items: state.items, localSubmissions: orderedLocalSubmissions(state), visibleSubmissionHandoffs: state.visibleSubmissionHandoffs }); |
| 24 | const rows = source.getOrderSnapshot().map(key => source.getNodeSnapshot(key)).filter(node => node?.kind === "user"); |
| 25 | assert.equal(rows.find(node => node?.item.messageId === "bound")?.key, original); |
| 26 | assert.equal(rows.find(node => node?.item.messageId === "conflict")?.key, "m:conflict"); |
| 27 | source.dispose(); |
| 28 | }); |
| 29 | |
| 30 | test("offscreen formal confirmation retires an echo without changing the reader page", () => { |
| 31 | const old = user("old"); |
| 32 | const state = { ...sent(), items: [old], historyHasNewer: true }; |
| 33 | const action = { type: "transcript_records" as const, projection: { ...projection([old]), hasNewer: true }, |
| 34 | confirmedUsers: [{ messageId: "durable", submissionId: "send" }] }; |
| 35 | const next = reducer(state, action); |
| 36 | assert.equal(next.localSubmissionOrder.length, 0); |
| 37 | assert.equal(next.items[0], old); |
| 38 | assert.equal(next.localSubmissionSendRevision, state.localSubmissionSendRevision); |
| 39 | }); |
| 40 | |
| 41 | for (const order of ["erc", "ecr", "rec", "rce", "cer", "cre"]) test(`handoff permutation ${order}`, () => { |
| 42 | let state = sent(); |
| 43 | for (const step of order) { |
| 44 | state = step === "e" ? reducer(state, event()) : step === "c" ? reducer(state, { type: "send_confirmed", submissionId: "send" }) |
| 45 | : reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable", "send")]) }); |
| 46 | assert.equal(state.items.filter(x => x.kind === "user").length + state.localSubmissionOrder.length, 1); |
| 47 | } |
| 48 | assert.equal(state.items[0].id, "m:durable"); |
| 49 | }); |
| 50 | |
| 51 | test("records without submission identity are reconciled when the binding event arrives later", () => { |
| 52 | let state = reducer(sent(), { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable")]) }); |
| 53 | state = reducer(state, event()); |
| 54 | assert.equal(state.localSubmissionOrder.length, 0); |
| 55 | }); |
| 56 | |
| 57 | test("batched binding and formal install preserve the mounted display key", () => { |
| 58 | const source = new ChatSource("batched-handoff"); |
| 59 | let state = sent(); |
| 60 | source.update({ ...input, items: [], localSubmissions: orderedLocalSubmissions(state) }); |
| 61 | const before = source.getOrderSnapshot(); |
| 62 | state = reducer(reducer(state, event()), { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable")]) }); |
| 63 | const next = { ...input, items: state.items, localSubmissions: orderedLocalSubmissions(state), |
| 64 | visibleSubmissionHandoffs: (state as typeof state & { visibleSubmissionHandoffs: Record<string, { submissionId: string }> }).visibleSubmissionHandoffs }; |
| 65 | source.update(next); |
| 66 | assert.deepEqual(source.getOrderSnapshot(), before); |
| 67 | source.dispose(); |
| 68 | }); |
| 69 | |
| 70 | test("a bound echo cannot be rebound or hidden by a different message sharing the submission", () => { |
| 71 | let state = reducer(sent(), event("first")); |
| 72 | state = reducer(state, event("other")); |
| 73 | assert.equal(state.localSubmissions.send.messageId, "first"); |
| 74 | const source = new ChatSource("conflict"); |
| 75 | source.update({ ...input, items: [user("other", "send")], localSubmissions: orderedLocalSubmissions(state) }); |
| 76 | assert.equal(source.getOrderSnapshot().filter(key => source.getNodeSnapshot(key)?.kind === "user").length, 2); |
| 77 | source.dispose(); |
| 78 | }); |
| 79 | |
| 80 | test("a reclaimed anchor does not insert a pending bubble before unrelated history", () => { |
| 81 | const source = new ChatSource("missing-anchor"); |
| 82 | const state = reducer({ ...initialState, items: [user("old-anchor")] }, { type: "user", seq: 0, text: "question", submissionId: "send" }); |
| 83 | source.update({ ...input, items: [user("unrelated")], localSubmissions: orderedLocalSubmissions(state) }); |
| 84 | assert.equal(source.getOrderSnapshot().includes("submission:send"), false); |
| 85 | source.dispose(); |
| 86 | }); |
| 87 | |
| 88 | test("a session-start echo stays outside a window that no longer contains the start", () => { |
| 89 | const source = new ChatSource("reclaimed-start"); |
| 90 | source.update({ ...input, items: [user("later")], hasOlder: true, historyStartTurn: 500, |
| 91 | localSubmissions: orderedLocalSubmissions(sent()) }); |
| 92 | assert.equal(source.getOrderSnapshot().includes("submission:send"), false); |
| 93 | source.dispose(); |
| 94 | }); |
| 95 | |
| 96 | test("late management acknowledgement removes its own echo after request acceptance", () => { |
| 97 | let state = reducer(sent(), { type: "send_confirmed", submissionId: "send" }); |
| 98 | state = reducer(state, { type: "user", seq: 1, text: "next", submissionId: "next" }); |
| 99 | state = reducer(state, { type: "management_confirmed", submissionId: "send" }); |
| 100 | assert.equal(state.localSubmissions.send, undefined); |
| 101 | assert.ok(state.localSubmissions.next); |
| 102 | assert.equal(state.pendingSubmissionId, "next"); |
| 103 | assert.equal(state.running, true); |
| 104 | }); |
| 105 | |
| 106 | test("old rejection cannot stop a newer accepted submission awaiting its turn identity", () => { |
| 107 | let state = reducer(sent(), { type: "turn_submit_unknown", submissionId: "send", error: "timeout" }); |
| 108 | state = reducer(state, { type: "user", seq: 1, text: "next", submissionId: "next" }); |
| 109 | state = reducer(state, { type: "send_confirmed", submissionId: "next" }); |
| 110 | state = reducer(state, { type: "transcript_connection", status: "connected" }); |
| 111 | state = reducer(state, { type: "turn_submit_unknown", submissionId: "send", error: "late timeout" }); |
| 112 | assert.equal(state.transcriptConnection, "connected", "old timeout cannot disconnect the newer submission"); |
| 113 | state = reducer(state, { type: "send_failed", submissionId: "send", error: "late rejection" }); |
| 114 | assert.equal(state.localSubmissions.send.status, "failed"); |
| 115 | assert.equal(state.localSubmissions.next.status, "accepted"); |
| 116 | assert.equal(state.running, true); |
| 117 | }); |
| 118 | |
| 119 | for (const status of ["send_failed", "turn_submit_unknown"] as const) test(`${status} converges on a late formal record`, () => { |
| 120 | let state = reducer(sent(), { type: status, submissionId: "send", error: "fixture" }); |
| 121 | assert.equal(state.localSubmissions.send.status, status === "send_failed" ? "failed" : "unknown"); |
| 122 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable", "send")]) }); |
| 123 | assert.equal(state.localSubmissionOrder.length, 0); |
| 124 | state = reducer(state, { type: status, submissionId: "send", error: "late" }); |
| 125 | assert.equal(state.localSubmissionOrder.length, 0); |
| 126 | }); |
| 127 | |
| 128 | test("remove plus upsert is atomic and reclaimed rows cannot be revived by a content patch", () => { |
| 129 | let state = reducer(sent(), { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable", "send")]) }); |
| 130 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: { ...projection([user("durable", "send")]), removeIds: ["m:durable"] } }); |
| 131 | assert.equal(state.items.length, 1); |
| 132 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: { ...projection([]), removeIds: ["m:durable"] } }); |
| 133 | assert.equal(Object.keys(state.visibleSubmissionHandoffs).length, 0); |
| 134 | state = reducer(state, { type: "history_items_patch", patches: { "m:durable": user("durable", "send") } }); |
| 135 | assert.equal(state.items.length, 0); |
| 136 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable", "send")]) }); |
| 137 | assert.equal(state.items.length, 1); |
| 138 | assert.equal(Object.keys(state.visibleSubmissionHandoffs).length, 0); |
| 139 | }); |
| 140 | |
| 141 | test("a delayed user record precedes its already streaming answer", () => { |
| 142 | let state = reducer(sent(), { type: "event", e: { kind: "text", messageId: "answer", turnId: "turn", text: "answer" } }); |
| 143 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([{ ...user("durable", "send"), turnId: "turn" }]) }); |
| 144 | assert.deepEqual(state.items.map(item => item.id), ["m:durable", "m:answer"]); |
| 145 | }); |
| 146 | |
| 147 | test("a start-of-session user record precedes streamed output without a turn id", () => { |
| 148 | let state = reducer(sent(), { type: "event", e: { kind: "text", messageId: "answer", text: "answer" } }); |
| 149 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable", "send")]) }); |
| 150 | assert.deepEqual(state.items.map(item => item.id), ["m:durable", "m:answer"]); |
| 151 | }); |
| 152 | |
| 153 | test("a later turn's user record stays after its anchor and before the new live output", () => { |
| 154 | let state = reducer(sent(), { type: "transcript_records", confirmedUsers: [], projection: projection([user("first", "send")]) }); |
| 155 | state = reducer(state, { type: "user", seq: 1, submissionId: "next", text: "next question" }); |
| 156 | state = reducer(state, { type: "event", e: { kind: "text", messageId: "answer", text: "already streaming" } }); |
| 157 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([user("first", "send"), user("second", "next")]) }); |
| 158 | assert.deepEqual(state.items.map(item => item.id), ["m:first", "m:second", "m:answer"]); |
| 159 | }); |
| 160 | |
| 161 | test("a reclaimed submission anchor does not place the user row before unrelated history", () => { |
| 162 | let state = reducer({ ...initialState, items: [user("old")] }, { type: "user", seq: 1, submissionId: "send", text: "new" }); |
| 163 | state = { ...state, items: [user("other")], transcriptProjectedIds: ["m:other"], historyHasOlder: true, historyStartTurn: 4 }; |
| 164 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([user("other"), user("new", "send")]) }); |
| 165 | assert.deepEqual(state.items.map(item => item.id), ["m:other", "m:new"]); |
| 166 | }); |
| 167 | |
| 168 | test("formal output published ahead of its user moves behind that user", () => { |
| 169 | const state = reducer(sent(), { type: "transcript_records", confirmedUsers: [], projection: projection([ |
| 170 | { kind: "assistant", id: "m:answer", text: "answer", reasoning: "", streaming: false, turnId: "turn" }, |
| 171 | { ...user("durable", "send"), turnId: "turn" }, |
| 172 | ]) }); |
| 173 | assert.deepEqual(state.items.map(item => item.id), ["m:durable", "m:answer"]); |
| 174 | }); |
| 175 | |
| 176 | test("a projection update without a turn id keeps the resident turn id", () => { |
| 177 | let state = reducer(sent(), { type: "event", e: { kind: "text", messageId: "answer", turnId: "turn", text: "answer" } }); |
| 178 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([ |
| 179 | { kind: "assistant", id: "m:answer", text: "answer", reasoning: "", streaming: false }, |
| 180 | ]) }); |
| 181 | const assistant = state.items.find(item => item.id === "m:answer"); |
| 182 | assert.equal(assistant?.kind === "assistant" ? assistant.turnId : undefined, "turn"); |
| 183 | }); |
| 184 | |
| 185 | test("a later event stamps the live answer that arrived without a turn id", () => { |
| 186 | let state = reducer(sent(), { type: "event", e: { kind: "text", messageId: "answer", text: "answer" } }); |
| 187 | assert.equal(state.items.find(item => item.kind === "assistant")?.turnId, undefined); |
| 188 | state = reducer(state, { type: "event", e: { kind: "stream_attempt", messageId: "answer", turnId: "turn", streamAttempt: { id: "sa", action: "begin", attempt: 1, max: 1 } } }); |
| 189 | assert.equal(state.items.find(item => item.id === "m:answer")?.turnId, "turn"); |
| 190 | }); |
| 191 | |
| 192 | test("a stream batch stamps the active turn onto the live answer", () => { |
| 193 | let state = reducer(sent(), { type: "event", e: { kind: "turn_started", status: "in_progress" } }); |
| 194 | assert.equal(state.items.find(item => item.kind === "assistant")?.turnId, undefined); |
| 195 | state = reducer({ ...state, activeTurnId: "turn" }, { type: "stream_batch", segments: [{ kind: "reasoning", delta: "想" }] }); |
| 196 | assert.equal(state.items.find(item => item.kind === "assistant")?.turnId, "turn"); |
| 197 | assert.equal(state.live?.reasoning, "想"); |
| 198 | }); |
| 199 | |
| 200 | test("reset clears echoes and presentation mappings", () => { |
| 201 | let state = reducer(sent(), { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable", "send")]) }); |
| 202 | state = reducer(state, { type: "user", seq: 1, submissionId: "pending", text: "pending" }); |
| 203 | state = reducer(state, { type: "reset" }); |
| 204 | assert.equal(state.localSubmissionOrder.length, 0); |
| 205 | assert.equal(Object.keys(state.visibleSubmissionHandoffs).length, 0); |
| 206 | }); |
| 207 | |
| 208 | test("late unknown transport result cannot downgrade terminal delivery evidence", () => { |
| 209 | const state = reducer(sent(), { type: "event", e: { kind: "turn_done", submissionId: "send", checkpointTurn: 1 } }); |
| 210 | assert.equal(reducer(state, { type: "turn_submit_unknown", submissionId: "send", error: "late timeout" }), state); |
| 211 | }); |
| 212 | |
| 213 | test("a batch hands off once and retains all distinct formal identities", () => { |
| 214 | const state = reducer(sent(), { type: "transcript_records", confirmedUsers: [], |
| 215 | projection: projection([user("first", "send"), user("second", "send")]) }); |
| 216 | assert.deepEqual(state.items.map(item => item.id), ["m:first", "m:second"]); |
| 217 | assert.deepEqual(state.visibleSubmissionHandoffs, { first: { submissionId: "send" } }); |
| 218 | const source = new ChatSource("multiple-formal"); |
| 219 | source.update({ ...input, items: [], localSubmissions: orderedLocalSubmissions(sent()) }); |
| 220 | source.update({ ...input, items: state.items, visibleSubmissionHandoffs: state.visibleSubmissionHandoffs }); |
| 221 | assert.deepEqual(source.getOrderSnapshot().filter(key => source.getNodeSnapshot(key)?.kind === "user"), ["submission:send", "m:second"]); |
| 222 | source.dispose(); |
| 223 | }); |
| 224 | |
| 225 | test("legacy installs preserve handoffs for other resident formal messages", () => { |
| 226 | let state = reducer({ ...sent(), transcriptProtocol: undefined }, event()); |
| 227 | assert.deepEqual(state.visibleSubmissionHandoffs, { durable: { submissionId: "send" } }); |
| 228 | state = reducer(state, { type: "event", e: { kind: "user_message", source: "executor", messageId: "another", text: "other" } }); |
| 229 | assert.deepEqual(state.visibleSubmissionHandoffs, { durable: { submissionId: "send" } }); |
| 230 | }); |
| 231 | |
| 232 | test("binding and admission cannot erase a failed echo before its formal confirmation", () => { |
| 233 | let state = reducer(sent(), { type: "send_failed", submissionId: "send", error: "rejected" }); |
| 234 | state = reducer(state, event()); |
| 235 | state = reducer(state, { type: "turn_admitted", submissionId: "send", turnId: "turn" }); |
| 236 | assert.equal(state.localSubmissions.send.status, "failed"); |
| 237 | state = reducer(state, { type: "transcript_records", confirmedUsers: [], projection: projection([user("durable")]) }); |
| 238 | assert.equal(state.localSubmissionOrder.length, 0); |
| 239 | }); |
| 240 | |
| 241 | test("same-session recovery retains pending echoes, a different snapshot owner clears them", () => { |
| 242 | const snapshot: TranscriptSnapshot = { protocolVersion: 1, snapshotId: "snapshot", identity: { |
| 243 | sessionId: "session-a", runtimeEpoch: "runtime", rewriteEpoch: 0, headId: "" }, projectionRevision: 1, |
| 244 | coveredThroughSeq: 1, records: [], activeRecords: [], activeAttempts: [], runtime: { status: "completed", pendingEvents: [] }, |
| 245 | before: 0, hasOlder: false, totalRecords: 0, totalTurns: 0, stale: false }; |
| 246 | let state = reducer(initialState, { type: "transcript_snapshot", snapshot }); |
| 247 | state = reducer(state, { type: "user", seq: 0, submissionId: "send", text: "question" }); |
| 248 | state = reducer(state, { type: "transcript_snapshot", snapshot }); |
| 249 | assert.ok(state.localSubmissions.send); |
| 250 | const offscreenSnapshot = { ...snapshot, totalRecords: 1, records: [{ id: "m:durable", order: 0, refs: [], |
| 251 | message: { role: "user" as const, messageId: "durable", submissionId: "send", content: "question" } }] }; |
| 252 | const confirmed = reducer(state, { type: "transcript_v2_snapshot", snapshot: offscreenSnapshot, projection: projection([user("old")]) }); |
| 253 | assert.equal(confirmed.localSubmissionOrder.length, 0, "snapshot confirmations precede window filtering"); |
| 254 | assert.deepEqual(confirmed.items.map(item => item.id), ["m:old"]); |
| 255 | assert.equal(Object.keys(confirmed.visibleSubmissionHandoffs).length, 0); |
| 256 | const generation = state.sessionGen; |
| 257 | state = reducer(state, { type: "transcript_snapshot", snapshot: { ...snapshot, identity: { ...snapshot.identity, sessionId: "session-b" } } }); |
| 258 | assert.equal(state.localSubmissionOrder.length, 0); |
| 259 | assert.equal(Object.keys(state.visibleSubmissionHandoffs).length, 0); |
| 260 | assert.equal(state.pendingSubmissionId, undefined); |
| 261 | assert.equal(state.sessionGen, generation + 1); |
| 262 | assert.equal(state.running, false); |
| 263 | }); |
| 264 |