| 1 | // Run: tsx src/__tests__/stream-delta-batch.test.ts |
| 2 | // |
| 3 | // A frame's worth of token deltas must become ONE stream_batch action per tab |
| 4 | // (one reducer pass, one liveStore notification), with segment order kept so |
| 5 | // the reasoning→text boundary completes reasoning exactly as per-delta |
| 6 | // delivery would. |
| 7 | |
| 8 | import { initialState, reducer } from "../lib/useController"; |
| 9 | import { coalesceStreamDeltas } from "../lib/streamDeltaBatch"; |
| 10 | import type { StreamDeltaEntry } from "../lib/streamDeltaBatch"; |
| 11 | import type { WireEvent } from "../lib/types"; |
| 12 | |
| 13 | let passed = 0; |
| 14 | let failed = 0; |
| 15 | |
| 16 | function eq(a: unknown, b: unknown, label: string) { |
| 17 | if (a === b) { |
| 18 | process.stdout.write(` PASS ${label}\n`); |
| 19 | passed += 1; |
| 20 | } else { |
| 21 | process.stdout.write(` FAIL ${label}: expected ${JSON.stringify(b)}, got ${JSON.stringify(a)}\n`); |
| 22 | failed += 1; |
| 23 | } |
| 24 | } |
| 25 | |
| 26 | const text = (tabId: string, t: string): StreamDeltaEntry => ({ tabId, e: { kind: "text", text: t } as WireEvent }); |
| 27 | const reasoning = (tabId: string, t: string): StreamDeltaEntry => ({ tabId, e: { kind: "reasoning", text: t } as WireEvent }); |
| 28 | |
| 29 | // --- consecutive same-kind deltas merge into ordered segments, one batch per tab --- |
| 30 | { |
| 31 | const batches = coalesceStreamDeltas([ |
| 32 | reasoning("a", "我"), reasoning("a", "需要"), reasoning("a", "检查"), reasoning("a", "一下"), |
| 33 | text("a", "发现"), text("a", "问题"), |
| 34 | text("b", "other"), |
| 35 | text("a", "如下"), |
| 36 | reasoning("a", "再想"), |
| 37 | ]); |
| 38 | eq(batches.length, 2, "one stream_batch per tab"); |
| 39 | eq(batches[0].tabId, "a", "first-seen tab comes first"); |
| 40 | eq(batches[0].segments.length, 3, "tab a: reasoning, text, reasoning segments"); |
| 41 | eq(batches[0].segments[0].kind, "reasoning", "segment order preserved"); |
| 42 | eq(batches[0].segments[0].delta, "我需要检查一下", "reasoning run concatenates in order"); |
| 43 | eq(batches[0].segments[1].delta, "发现问题如下", "text run spans the other tab's interleave"); |
| 44 | eq(batches[0].segments[2].delta, "再想", "reasoning after text stays a separate segment"); |
| 45 | eq(batches[1].segments[0].delta, "other", "tab b keeps its own batch"); |
| 46 | } |
| 47 | |
| 48 | // --- legacy deltas carried in the reasoning field still coalesce --- |
| 49 | { |
| 50 | const batches = coalesceStreamDeltas([ |
| 51 | { tabId: "a", e: { kind: "reasoning", reasoning: "le" } as WireEvent }, |
| 52 | { tabId: "a", e: { kind: "reasoning", reasoning: "gacy" } as WireEvent }, |
| 53 | ]); |
| 54 | eq(batches[0].segments.length, 1, "legacy-field deltas merge"); |
| 55 | eq(batches[0].segments[0].delta, "legacy", "legacy payload concatenates"); |
| 56 | } |
| 57 | |
| 58 | // --- reducer equivalence: one stream_batch equals per-delta dispatch --- |
| 59 | { |
| 60 | const deltas = [reasoning("a", "th"), reasoning("a", "inking"), text("a", "an"), text("a", "swer"), reasoning("a", "more")]; |
| 61 | let perDelta = { ...initialState, running: true, turnActive: true }; |
| 62 | for (const { e } of deltas) perDelta = reducer(perDelta, { type: "event", e }); |
| 63 | |
| 64 | let batched = { ...initialState, running: true, turnActive: true }; |
| 65 | const batches = coalesceStreamDeltas(deltas); |
| 66 | eq(batches.length, 1, "single tab folds to a single action"); |
| 67 | for (const b of batches) batched = reducer(batched, { type: "stream_batch", segments: b.segments } as never); |
| 68 | |
| 69 | eq(batched.live?.reasoning, perDelta.live?.reasoning, "reasoning matches per-delta result"); |
| 70 | eq(batched.live?.text, perDelta.live?.text, "text matches per-delta result"); |
| 71 | eq(batched.live?.reasoningComplete, perDelta.live?.reasoningComplete, "reasoning reopened by trailing segment"); |
| 72 | eq(batched.live?.reasoningComplete, false, "trailing reasoning segment leaves reasoning open"); |
| 73 | eq(batched.currentAssistant, perDelta.currentAssistant, "assistant identity matches"); |
| 74 | } |
| 75 | |
| 76 | // --- the reasoning→text boundary completes reasoning inside one batch --- |
| 77 | { |
| 78 | let s = { ...initialState, running: true, turnActive: true }; |
| 79 | s = reducer(s, { type: "stream_batch", segments: [{ kind: "reasoning", delta: "想" }, { kind: "text", delta: "答" }] } as never); |
| 80 | eq(s.live?.reasoningComplete, true, "text segment after reasoning completes it"); |
| 81 | eq(s.live?.reasoningCompletedAt !== undefined, true, "completion is timestamped"); |
| 82 | } |
| 83 | |
| 84 | // --- preamble parity: discarded turns drop deltas; retry clears --- |
| 85 | { |
| 86 | const discarding = { ...initialState, discardTurn: true }; |
| 87 | const after = reducer(discarding, { type: "stream_batch", segments: [{ kind: "text", delta: "x" }] } as never); |
| 88 | eq(after, discarding, "discardTurn swallows a stream_batch like per-delta events"); |
| 89 | |
| 90 | let retrying: typeof initialState = { ...initialState, running: true, turnActive: true, retry: { attempt: 1, max: 3, observedAt: 1 } }; |
| 91 | retrying = reducer(retrying, { type: "stream_batch", segments: [{ kind: "text", delta: "x" }] } as never); |
| 92 | eq(retrying.retry, undefined, "stream_batch clears the retry indicator like per-delta events"); |
| 93 | } |
| 94 | |
| 95 | process.stdout.write(`\n${passed} passed, ${failed} failed\n`); |
| 96 | if (failed > 0) process.exit(1); |
| 97 |