import { describe, test, expect, beforeEach } from "bun:test";
import { get } from "svelte/store";
import { StreamingClientV2, WINDOW_LIMIT, HISTORY_PAGE } from "./streaming-v2";
import type { ServerMsgV2 } from "./protocol-v2";
import fixture from "../../tests/fixtures/lifecycle_diagnostics.json";
// Mock WebSocket
class MockWebSocket {
public static readonly CONNECTING = 0;
public static readonly OPEN = 1;
public static readonly CLOSING = 2;
public static readonly CLOSED = 3;
public static instances: MockWebSocket[] = [];
public sent: string[] = [];
public readyState: number = MockWebSocket.CONNECTING;
public url: string;
public onopen: (() => void) | null = null;
public onmessage: ((event: { data: string }) => void) | null = null;
public onerror: (() => void) | null = null;
public onclose: (() => void) | null = null;
constructor(url: string) {
this.url = url;
MockWebSocket.instances.push(this);
setTimeout(() => {
this.readyState = MockWebSocket.OPEN;
if (this.onopen) this.onopen();
}, 10);
}
send(data: string) {
this.sent.push(data);
}
close() {
this.readyState = MockWebSocket.CLOSED;
if (this.onclose) this.onclose();
}
emitMessage(msg: ServerMsgV2) {
if (this.onmessage) {
this.onmessage({ data: JSON.stringify(msg) });
}
}
}
(globalThis as any).WebSocket = MockWebSocket;
describe("StreamingClientV2 sequence cursor & lane algebra", () => {
let client: StreamingClientV2;
let ws: MockWebSocket;
beforeEach(async () => {
MockWebSocket.instances = [];
client = new StreamingClientV2("ws://127.0.0.1:8765", "test-session-" + Math.random().toString(36).slice(2, 7));
client.clearSessionState();
client.connect();
await new Promise((resolve) => setTimeout(resolve, 30));
ws = MockWebSocket.instances[MockWebSocket.instances.length - 1];
});
test("a client that holds nothing asks for a window, not for a session", () => {
expect(ws.sent.length).toBeGreaterThanOrEqual(1);
const subMsg = JSON.parse(ws.sent[0]);
expect(subMsg.type).toBe("subscribe");
// Nothing held is `after_seq: 0` with a `limit`: the newest window, anchored at the daemon's own fold,
// and at most that many records of it. Zero without a limit is what this console used to ask for, and
// a daemon that does not know about windows still answers it with everything.
expect(subMsg.after_seq).toBe(0);
expect(subMsg.limit).toBe(WINDOW_LIMIT);
expect(get(client.timelineStore).length).toBe(0);
});
test("in-order event ingestion updates cursor and stores", () => {
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "scratchpad",
payload: {
kind: "thought",
thoughts: "analyzing problem (˘ω˘)",
active: false,
tokens: 42,
},
created_ms: 1000,
},
});
const conn = get(client.connectionStore);
expect(conn.lastSeenSeq).toBe(1);
const scratchpad = get(client.scratchpadStore);
expect(scratchpad.length).toBe(1);
expect(scratchpad[0].seq).toBe(1);
expect(scratchpad[0].thoughts).toBe("analyzing problem (˘ω˘)");
const lastSent = JSON.parse(ws.sent[ws.sent.length - 1]);
expect(lastSent.type).toBe("ack");
expect(lastSent.seq).toBe(1);
});
test("deduplicates duplicate events", () => {
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "scratchpad",
payload: { thoughts: "first", active: false },
created_ms: 1000,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "scratchpad",
payload: { thoughts: "duplicate attempt", active: false },
created_ms: 1000,
},
});
const scratchpad = get(client.scratchpadStore);
expect(scratchpad.length).toBe(1);
expect(scratchpad[0].thoughts).toBe("first");
});
test("ingests sparse sequences [1, 3] and displays both events without gap buffering", () => {
// Session filtered stream: event 2 belongs to another session
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "input",
payload: { source: "operator", text: "message 1" },
created_ms: 1001,
},
});
expect(get(client.connectionStore).lastSeenSeq).toBe(1);
expect(get(client.timelineStore).map((x) => x.seq)).toEqual([1]);
// Sparse seq 3 arrives immediately
ws.emitMessage({
type: "event",
record: {
seq: 3,
session_id: "test",
lane: "input",
payload: { source: "operator", text: "message 3" },
created_ms: 1003,
},
});
// Both events must be visible immediately - no gap blocking!
expect(get(client.connectionStore).lastSeenSeq).toBe(3);
const visible = get(client.timelineStore).map((x) => x.seq);
expect(visible).toEqual([1, 3]);
// Sync completion acknowledges latest seq 3
ws.emitMessage({
type: "sync_completed",
session_id: "test",
latest_seq: 3,
});
// Replay duplicate of event 3 is safely deduplicated
ws.emitMessage({
type: "event",
record: {
seq: 3,
session_id: "test",
lane: "input",
payload: { source: "operator", text: "message 3 duplicate" },
created_ms: 1003,
},
});
expect(get(client.timelineStore).map((x) => x.seq)).toEqual([1, 3]);
expect(client.eventBuffer.size).toBe(0);
});
test("integrates out-of-order events into chronological sorted timeline", () => {
// Arrival order: seq 4, then seq 3, then seq 5, then seq 1, then seq 2
ws.emitMessage({
type: "event",
record: {
seq: 4,
session_id: "test",
lane: "speech",
payload: { effect_id: "e4", status: "confirmed", text: "step 4" },
created_ms: 4000,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 3,
session_id: "test",
lane: "cell",
payload: { execution_id: "c3", stream: "stdout", text: "step 3" },
created_ms: 3000,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 5,
session_id: "test",
lane: "lifecycle",
payload: { kind: "completed" },
created_ms: 5000,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "input",
payload: { source: "operator", text: "start" },
created_ms: 1000,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 2,
session_id: "test",
lane: "scratchpad",
payload: { thoughts: "step 2 reasoning" },
created_ms: 2000,
},
});
// All 5 events must be in timeline, strictly sorted by seq
expect(get(client.connectionStore).lastSeenSeq).toBe(5);
const timeline = get(client.timelineStore);
expect(timeline.length).toBe(5);
expect(timeline.map((t) => t.seq)).toEqual([1, 2, 3, 4, 5]);
expect(timeline.map((t) => t.lane)).toEqual(["input", "scratchpad", "cell", "speech", "lifecycle"]);
});
test("strict speech delivery status validation", () => {
// 1. Confirmed speech
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "speech",
payload: {
effect_id: "eff_conf",
status: "confirmed",
text: "confirmed delivery",
},
created_ms: 100,
},
});
// 2. Failed delivery - must NEVER default to confirmed
ws.emitMessage({
type: "event",
record: {
seq: 2,
session_id: "test",
lane: "speech",
payload: {
effect_id: "eff_fail",
status: "failed",
text: "failed unsent message",
},
created_ms: 200,
},
});
// 3. Held delivery
ws.emitMessage({
type: "event",
record: {
seq: 3,
session_id: "test",
lane: "speech",
payload: {
effect_id: "eff_held",
status: "held",
reason: "supervisor review",
text: "held message",
},
created_ms: 300,
},
});
// 4. Unknown/malformed status - must NEVER default to confirmed
ws.emitMessage({
type: "event",
record: {
seq: 4,
session_id: "test",
lane: "speech",
payload: {
effect_id: "eff_unknown",
status: "corrupted_status_code",
text: "unverified message",
},
created_ms: 400,
},
});
const speech = get(client.speechStore);
expect(speech.length).toBe(4);
expect(speech[0].receipt.status).toBe("confirmed");
expect(speech[1].receipt.status).toBe("failed");
expect(speech[1].receipt.status).not.toBe("confirmed");
expect(speech[1].receipt.reason).toBeTruthy();
expect(speech[2].receipt.status).toBe("held");
expect(speech[2].receipt.reason).toBe("supervisor review");
expect(speech[3].receipt.status).toBe("held");
expect(speech[3].receipt.status).not.toBe("confirmed");
expect(speech[3].receipt.reason).toContain("corrupted_status_code");
});
test("strictly separates scratchpad, cell, speech, and input lanes", () => {
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "input",
payload: { source: "operator", text: "please run diagnostics" },
created_ms: 100,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 2,
session_id: "test",
lane: "scratchpad",
payload: { thoughts: "i will run cargo check", active: false },
created_ms: 200,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 3,
session_id: "test",
lane: "cell",
payload: {
execution_id: "cargo_check",
stream: "stderr",
text: "warning: unused variable",
status: "ok",
truncated: true,
exit_code: 0,
},
created_ms: 300,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 4,
session_id: "test",
lane: "speech",
payload: {
effect_id: "eff_1",
status: "held",
reason: "review needed",
text: "unverified output",
},
created_ms: 400,
},
});
ws.emitMessage({
type: "event",
record: {
seq: 5,
session_id: "test",
lane: "speech",
payload: {
effect_id: "eff_2",
status: "confirmed",
text: "diagnostics finished with 1 warning!",
},
created_ms: 500,
},
});
const scratchpad = get(client.scratchpadStore);
const cells = get(client.cellStore);
const speech = get(client.speechStore);
const timeline = get(client.timelineStore);
expect(scratchpad.length).toBe(1);
expect(cells.length).toBe(1);
expect(speech.length).toBe(2);
expect(timeline.length).toBe(5);
expect(speech[0].receipt.status).toBe("held");
expect(speech[0].receipt.reason).toBe("review needed");
expect(speech[1].receipt.status).toBe("confirmed");
expect(cells[0].outputs[0].stream).toBe("stderr");
expect(cells[0].truncated).toBe(true);
expect(cells[0].exit_code).toBe(0);
});
test("handles live chunks streaming into active entries", () => {
ws.emitMessage({
type: "live_chunk",
session_id: "test",
lane: "scratchpad",
stream: "thought",
delta: "deep thought chunk 1... ",
});
let sp = get(client.scratchpadStore);
expect(sp.length).toBe(1);
expect(sp[0].active).toBe(true);
expect(sp[0].thoughts).toBe("deep thought chunk 1... ");
ws.emitMessage({
type: "live_chunk",
session_id: "test",
lane: "scratchpad",
stream: "thought",
delta: "chunk 2 (˘ω˘)",
});
sp = get(client.scratchpadStore);
expect(sp[0].thoughts).toBe("deep thought chunk 1... chunk 2 (˘ω˘)");
ws.emitMessage({
type: "event",
record: {
seq: 1,
session_id: "test",
lane: "scratchpad",
payload: {
thoughts: sp[0].thoughts,
active: false,
tokens: 55,
},
created_ms: 1000,
},
});
sp = get(client.scratchpadStore);
expect(sp.length).toBe(1);
expect(sp[0].active).toBe(false);
expect(sp[0].seq).toBe(1);
});
test("client send methods format messages correctly", () => {
// sendInput
client.sendInput("hello world", "custom-source");
const inputMsg = JSON.parse(ws.sent[ws.sent.length - 1]);
expect(inputMsg.type).toBe("send_input");
expect(inputMsg.source).toBe("custom-source");
expect(inputMsg.text).toBe("hello world");
// interrupt
client.interrupt("exec_123");
const interruptMsg = JSON.parse(ws.sent[ws.sent.length - 1]);
expect(interruptMsg.type).toBe("interrupt");
expect(interruptMsg.execution_id).toBe("exec_123");
// resetKernel
client.resetKernel();
const resetMsg = JSON.parse(ws.sent[ws.sent.length - 1]);
expect(resetMsg.type).toBe("reset_kernel");
});
test("sync_started and sync_completed manage isSyncing flag", () => {
ws.emitMessage({
type: "sync_started",
session_id: "test",
from_seq: 1,
to_seq: 10,
});
expect(get(client.connectionStore).isSyncing).toBe(true);
ws.emitMessage({
type: "sync_completed",
session_id: "test",
latest_seq: 10,
});
expect(get(client.connectionStore).isSyncing).toBe(false);
expect(get(client.connectionStore).lastSeenSeq).toBe(10);
});
test("a page that starts with nothing accepts whatever the daemon replays for it", () => {
const sessionId = "reload-test-session";
const serverId = "local-server";
const reloadedClient = new StreamingClientV2("ws://127.0.0.1:8765", sessionId, serverId);
// Nothing is remembered between pages, so there is nothing a fresh page could suppress a replay with.
expect(get(reloadedClient.timelineStore).length).toBe(0);
// Incoming events 1 and 3 arrive
reloadedClient.handleServerMsg({
type: "event",
record: {
seq: 1,
session_id: sessionId,
lane: "input",
payload: { source: "operator", text: "hello reloaded" },
created_ms: 1001,
},
});
reloadedClient.handleServerMsg({
type: "event",
record: {
seq: 3,
session_id: sessionId,
lane: "input",
payload: { source: "operator", text: "message 3 reloaded" },
created_ms: 1003,
},
});
const timeline = get(reloadedClient.timelineStore);
expect(timeline.length).toBe(2);
expect(timeline.map((t) => t.seq)).toEqual([1, 3]);
});
test("handles live speech deltas and updates speechStore and timelineStore", () => {
client.handleServerMsg({
type: "live_chunk",
session_id: "test",
lane: "speech",
delta: "hello live speech",
});
const speech = get(client.speechStore);
expect(speech.length).toBe(1);
expect(speech[0].receipt.text).toBe("hello live speech");
const timeline = get(client.timelineStore);
expect(timeline.length).toBe(1);
expect(timeline[0].lane).toBe("speech");
});
test("sendInput includes client_request_id and clears it on ack", () => {
const reqId = client.sendInput("hello with request id", "operator");
expect(reqId).toBeDefined();
expect(reqId.startsWith("req_")).toBe(true);
const inputMsg = JSON.parse(ws.sent[ws.sent.length - 1]);
expect(inputMsg.type).toBe("send_input");
expect(inputMsg.client_request_id).toBe(reqId);
// Retrying with same request ID reuses it
const retriedId = client.sendInput("hello with request id", "operator", reqId);
expect(retriedId).toBe(reqId);
// Receiving Ack clears pending request
client.handleServerMsg({
type: "ack",
session_id: "test",
seq: 1,
client_request_id: reqId,
});
});
test("nothing is remembered between loads: a reload asks the daemon again", async () => {
const sessionId = "reload-" + Math.random().toString(36).slice(2, 7);
const first = new StreamingClientV2("ws://test", sessionId, "reload-server");
first.handleServerMsg({
type: "event",
record: { seq: 1, session_id: sessionId, lane: "input", payload: { source: "operator", text: "hello" }, created_ms: 1000 },
});
first.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 1, earliest_seq: 1 });
expect(get(first.timelineStore).length).toBe(1);
// A second client for the same session is a page that was reloaded. It holds nothing, and it says
// so: the window is the daemon's to hand out, and a remembered cursor could only be ahead of the
// records a fresh page does not have.
const reloaded = new StreamingClientV2("ws://test", sessionId, "reload-server");
expect(get(reloaded.timelineStore).length).toBe(0);
expect(get(reloaded.connectionStore).hasEarlier).toBeUndefined();
reloaded.connect();
await new Promise((resolve) => setTimeout(resolve, 30));
const sub = JSON.parse(MockWebSocket.instances[MockWebSocket.instances.length - 1].sent[0]);
expect(sub.after_seq).toBe(0);
expect(sub.limit).toBe(WINDOW_LIMIT);
});
test("the daemon's floor is what makes the console offer to load more", () => {
const sessionId = "floor-" + Math.random().toString(36).slice(2, 7);
const floorClient = new StreamingClientV2("ws://test", sessionId);
for (const seq of [100, 101, 102]) {
floorClient.handleServerMsg({
type: "event",
record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq },
});
}
// A daemon that says nothing about its history gets no load-more row: the console cannot page what
// it cannot measure, and a button that does nothing is worse than no button.
floorClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 102 });
expect(get(floorClient.connectionStore).hasEarlier).toBe(false);
// One that names a floor below the window gets one.
floorClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 102, earliest_seq: 1 });
expect(get(floorClient.connectionStore).hasEarlier).toBe(true);
// A floor at the window means the window is the whole session.
floorClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 102, earliest_seq: 100 });
expect(get(floorClient.connectionStore).hasEarlier).toBe(false);
});
test("the fold is the daemon's fact about the session, reported with every range", () => {
const sessionId = "fold-" + Math.random().toString(36).slice(2, 7);
const foldClient = new StreamingClientV2("ws://test", sessionId);
expect(get(foldClient.connectionStore).foldSeq).toBeUndefined();
foldClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 401, to_seq: 900, earliest_seq: 1, fold_seq: 400 });
expect(get(foldClient.connectionStore).foldSeq).toBe(400);
// A later range that mentions no fold leaving the fact alone: a session does not become unfolded.
foldClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 380, to_seq: 400, earliest_seq: 1 });
expect(get(foldClient.connectionStore).foldSeq).toBe(400);
// A page behind the fold carries it too, and the view is the one that decides whether to mention it.
foldClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 300, to_seq: 379, earliest_seq: 1, fold_seq: 400 });
expect(get(foldClient.connectionStore).foldSeq).toBe(400);
});
test("the floor arrives with the range, not only at the end of the replay", () => {
const sessionId = "floor-early-" + Math.random().toString(36).slice(2, 7);
const early = new StreamingClientV2("ws://test", sessionId);
// A replay states what it is sending and what is under it in the same breath, so the row that
// offers the rest of the history is there before the reader is placed, not after.
early.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 401, to_seq: 900, earliest_seq: 1 });
expect(get(early.connectionStore).isSyncing).toBe(true);
expect(get(early.connectionStore).hasEarlier).toBe(true);
early.handleServerMsg({
type: "event",
record: { seq: 401, session_id: sessionId, lane: "input", payload: { source: "operator", text: "first in the window" }, created_ms: 1401 },
});
// A floor at the window's own start is not history to fetch.
early.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 401, earliest_seq: 401 });
expect(get(early.connectionStore).hasEarlier).toBe(false);
});
test("load earlier sends a history request for the records behind the window", () => {
const sessionId = "page-" + Math.random().toString(36).slice(2, 7);
const pageClient = new StreamingClientV2("ws://test", sessionId, "page-server");
pageClient.connect();
// Nothing loaded, no floor known: there is nothing to ask for yet.
expect(pageClient.loadEarlier()).toBe(false);
pageClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 500, to_seq: 501, earliest_seq: 1 });
for (const seq of [500, 501]) {
pageClient.handleServerMsg({
type: "event",
record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq },
});
}
// While the window is arriving, a page behind a floor that is still moving is refused.
expect(pageClient.loadEarlier()).toBe(false);
pageClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 501, earliest_seq: 1 });
expect(get(pageClient.connectionStore).isSyncing).toBe(false);
const ws = MockWebSocket.instances[MockWebSocket.instances.length - 1];
ws.readyState = MockWebSocket.OPEN;
ws.sent = [];
expect(pageClient.loadEarlier()).toBe(true);
const req = JSON.parse(ws.sent[ws.sent.length - 1]);
expect(req).toEqual({ type: "history", session_id: sessionId, before_seq: 500, limit: HISTORY_PAGE });
expect(get(pageClient.connectionStore).loadingEarlier).toBe(true);
// One page at a time: a second press while the first is in flight is refused rather than queued.
expect(pageClient.loadEarlier()).toBe(false);
});
test("a history page prepends, keeps the live cursor, and a re-delivered page is not doubled", () => {
const sessionId = "prepend-" + Math.random().toString(36).slice(2, 7);
const histClient = new StreamingClientV2("ws://test", sessionId);
for (const seq of [500, 501, 502]) {
histClient.handleServerMsg({
type: "event",
record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq },
});
}
const cursorBefore = get(histClient.connectionStore).lastSeenSeq;
expect(cursorBefore).toBe(502);
// The daemon answers a history request with the same replay shape it uses for a subscribe.
histClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 498, to_seq: 499 });
for (const seq of [498, 499]) {
histClient.handleServerMsg({
type: "event",
record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq },
});
}
histClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 502, earliest_seq: 1 });
const seqs = get(histClient.timelineStore).map((t) => t.seq);
expect(seqs).toEqual([498, 499, 500, 501, 502]);
// A page behind the window is history, not an advance: the live cursor is where the live stream
// left it, whatever the daemon said its latest record is.
expect(get(histClient.connectionStore).lastSeenSeq).toBe(502);
expect(get(histClient.connectionStore).loadingEarlier).toBe(false);
// The same page again: every record is already held, so the timeline is untouched. A retried or
// overlapping page must be idempotent, and a set of seen sequences is what makes it so.
for (const seq of [498, 499]) {
histClient.handleServerMsg({
type: "event",
record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `w${seq}` }, created_ms: 1000 + seq },
});
}
expect(get(histClient.timelineStore).map((t) => t.seq)).toEqual([498, 499, 500, 501, 502]);
});
test("history arriving mid-stream is never merged into the live cell it arrives beside", () => {
const sessionId = "live-history-" + Math.random().toString(36).slice(2, 7);
const mixClient = new StreamingClientV2("ws://test", sessionId);
// a live execution streaming into its entry
mixClient.handleServerMsg({
type: "event",
record: {
seq: 900,
session_id: sessionId,
lane: "cell",
payload: { execution_id: "cell-live", code: "cargo test", status: "running", stream: "stdout", text: "live" },
created_ms: 9000,
},
});
const live = get(mixClient.cellStore).find((e) => e.id === "cell-live");
expect(live?.status).toBe("running");
// The reader pages back, and the daemon replays a finished execution from before the window.
mixClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 100, to_seq: 101 });
mixClient.handleServerMsg({
type: "event",
record: {
seq: 101,
session_id: sessionId,
lane: "cell",
payload: { execution_id: "cell-old", code: "cargo fmt", status: "ok", stream: "result", text: "42", exit_code: 0 },
created_ms: 1010,
},
});
mixClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 900, earliest_seq: 1 });
// The old record is its own cell, and the streaming one is exactly as it was.
const ids = get(mixClient.cellStore).map((e) => e.id).sort();
expect(ids).toEqual(["cell-live", "cell-old"]);
expect(get(mixClient.cellStore).find((e) => e.id === "cell-live")?.status).toBe("running");
expect(get(mixClient.timelineStore).map((t) => t.seq)).toEqual([101, 900]);
});
test("lifecycle diagnostics update healthStore, workItemsStore, and candidateStore while strictly preserving lane separation", () => {
const sessionId = "diag-test-" + Math.random().toString(36).slice(2, 7);
const client = new StreamingClientV2("ws://test", sessionId);
// Ingest lifecycle event with capability health, work items, and candidate
client.handleServerMsg({
type: "event",
record: {
seq: 1,
session_id: sessionId,
lane: "lifecycle",
payload: {
kind: "checkup_completed",
details: "health probe executed",
health: [
{
id: "h1",
name: "context_assembly",
scope: "guidance",
status: "healthy",
observed_ms: 1000,
valid_until_ms: 5000,
origin: "checker",
},
{
id: "h2",
name: "model_adapter",
scope: "llm",
status: "degraded",
impact: "high latency",
observed_ms: 1000,
origin: "agent",
}
],
work_items: [
{
id: "w1",
title: "repair adapter",
purpose: "latency reduction",
status: "selected",
origin: "agent",
selection_reason: "reliability repair outranks low-urgency feature",
created_ms: 1000,
}
],
candidate: {
candidate_id: "cand-123",
base_revision: "rev-1",
target_revision: "rev-2",
status: "pinned",
epoch: 2,
test_receipts: [
{ suite_id: "g01_g26", passed: true, duration_ms: 120 }
]
}
},
created_ms: 1000,
},
});
// Check stores are updated
const health = get(client.healthStore);
expect(health.length).toBe(2);
expect(health[0].name).toBe("context_assembly");
expect(health[1].status).toBe("degraded");
const work = get(client.workItemsStore);
expect(work.length).toBe(1);
expect(work[0].status).toBe("selected");
expect(work[0].selection_reason).toContain("reliability repair");
const cand = get(client.candidateStore);
expect(cand).not.toBeNull();
expect(cand?.candidate_id).toBe("cand-123");
expect(cand?.epoch).toBe(2);
// Strict lane separation: lifecycle event MUST NOT pollute speech, scratchpad, or cell lanes!
expect(get(client.speechStore).length).toBe(0);
expect(get(client.scratchpadStore).length).toBe(0);
expect(get(client.cellStore).length).toBe(0);
// Timeline has the lifecycle event
const timeline = get(client.timelineStore);
expect(timeline.length).toBe(1);
expect(timeline[0].lane).toBe("lifecycle");
// clearSessionState resets diagnostic stores
client.clearSessionState();
expect(get(client.healthStore).length).toBe(0);
expect(get(client.workItemsStore).length).toBe(0);
expect(get(client.candidateStore)).toBeNull();
});
test("speech and input items preserve provenance metadata badges", () => {
const sessionId = "prov-test-" + Math.random().toString(36).slice(2, 7);
const client = new StreamingClientV2("ws://test", sessionId);
// Operator input with human provenance
client.handleServerMsg({
type: "event",
record: {
seq: 1,
session_id: sessionId,
lane: "input",
payload: {
source: "operator",
text: "hello",
provenance: {
origin: "human",
reporter: "operator",
observed_ms: 1000,
},
},
created_ms: 1000,
},
});
// Agent speech with agent provenance
client.handleServerMsg({
type: "event",
record: {
seq: 2,
session_id: sessionId,
lane: "speech",
payload: {
type: "sent",
effect_id: "eff-1",
text: "acknowledged",
event_id: 1,
status: "confirmed",
provenance: {
origin: "agent",
reporter: "klbr",
observed_ms: 1005,
},
},
created_ms: 1005,
},
});
const timeline = get(client.timelineStore);
expect(timeline.length).toBe(2);
const inputItem = timeline[0];
expect(inputItem.lane).toBe("input");
if (inputItem.lane === "input") {
expect(inputItem.data.provenance?.origin).toBe("human");
}
const speechItem = timeline[1];
expect(speechItem.lane).toBe("speech");
if (speechItem.lane === "speech") {
expect(speechItem.data.receipt.provenance?.origin).toBe("agent");
}
});
test("consumes exact Rust LifecyclePayload {kind, data} envelope from shared fixture", () => {
const sessionId = "host-envelope-" + Math.random().toString(36).slice(2, 7);
const client = new StreamingClientV2("ws://test", sessionId);
// Feed real Rust-serialized server message envelope
client.handleServerMsg(fixture.v2_server_msg as ServerMsgV2);
const health = get(client.healthStore);
expect(health.length).toBe(1);
expect(health[0].id).toBe("guidance");
expect(health[0].status).toBe("degraded");
expect(health[0].origin_kind).toBe("checker");
const work = get(client.workItemsStore);
expect(work.length).toBe(1);
expect(work[0].id).toBe("repair1");
expect(work[0].status).toBe("pending");
expect(work[0].origin_kind).toBe("agent");
const candidate = get(client.candidateStore);
expect(candidate).not.toBeNull();
expect(candidate?.candidate_id).toBe("c1");
expect(candidate?.status).toBe("testing");
// Strict lane purity: diagnostics must NEVER enter speech
expect(get(client.speechStore).length).toBe(0);
expect(get(client.scratchpadStore).length).toBe(0);
expect(get(client.cellStore).length).toBe(0);
});
test("a replayed window is what rebuilds the diagnostics a page lost", async () => {
const sessionId = "window-diag-" + Math.random().toString(36).slice(2, 7);
const windowClient = new StreamingClientV2("ws://test", sessionId);
// This is what a fresh page receives: the newest window of the session, lifecycle records included.
windowClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 9, to_seq: 12 });
windowClient.handleServerMsg({
type: "event",
record: { seq: 10, session_id: sessionId, lane: "lifecycle", payload: fixture.host_payload, created_ms: 1010 },
});
windowClient.handleServerMsg({
type: "event",
record: { seq: 12, session_id: sessionId, lane: "input", payload: { source: "operator", text: "and then?" }, created_ms: 1012 },
});
windowClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 12, earliest_seq: 9 });
const health = get(windowClient.healthStore);
expect(health.length).toBe(2);
expect(health[0].id).toBe("guidance");
expect(health[0].status).toBe("degraded");
expect(health[1].id).toBe("context_assembly");
expect(health[1].status).toBe("healthy");
const work = get(windowClient.workItemsStore);
expect(work.length).toBe(1);
expect(work[0].id).toBe("repair1");
expect(work[0].status).toBe("pending");
expect(get(windowClient.candidateStore)?.candidate_id).toBe("c1");
expect(get(windowClient.candidateStore)?.epoch).toBe(2);
expect(get(windowClient.connectionStore).lastSeenSeq).toBe(12);
expect(get(windowClient.speechStore).length).toBe(0);
// The daemon named a floor of 9 below a window that starts at 10, so there is a record behind it
// and the console offers to fetch it. Sequences are sparse, so that floor is a bound rather than a
// promise: a page that comes back empty is what ends the offering.
expect(get(windowClient.connectionStore).hasEarlier).toBe(true);
// The reader presses: the daemon answers with an empty page, because there is no record 9 either.
// A page the console asked for that brings back nothing is the end of the history, and the offering
// retires. Without that, the button would be a promise to return the same nothing forever.
windowClient.connect();
await new Promise((resolve) => setTimeout(resolve, 30));
const socket = MockWebSocket.instances[MockWebSocket.instances.length - 1];
socket.readyState = MockWebSocket.OPEN;
expect(windowClient.loadEarlier()).toBe(true);
windowClient.handleServerMsg({ type: "sync_started", session_id: sessionId, from_seq: 9, to_seq: 9 });
windowClient.handleServerMsg({ type: "sync_completed", session_id: sessionId, latest_seq: 12, earliest_seq: 9 });
expect(get(windowClient.connectionStore).hasEarlier).toBe(false);
expect(windowClient.loadEarlier()).toBe(false);
});
test("incremental lifecycle events (capability.health, work.updated, candidate.updated) update stores cleanly", () => {
const sessionId = "incremental-" + Math.random().toString(36).slice(2, 7);
const client = new StreamingClientV2("ws://test", sessionId);
// 1. Single health observation
client.handleServerMsg({
type: "event",
record: fixture.single_health_event as any,
});
const health = get(client.healthStore);
expect(health.length).toBe(1);
expect(health[0].id).toBe("db_integrity");
expect(health[0].status).toBe("healthy");
expect(health[0].origin_kind).toBe("system");
// 2. Single work item
client.handleServerMsg({
type: "event",
record: fixture.single_work_event as any,
});
const work = get(client.workItemsStore);
expect(work.length).toBe(1);
expect(work[0].id).toBe("maint_vacuum");
expect(work[0].status).toBe("selected");
expect(work[0].origin_kind).toBe("system");
// 3. Candidate update
client.handleServerMsg({
type: "event",
record: fixture.single_candidate_event as any,
});
const candidate = get(client.candidateStore);
expect(candidate?.candidate_id).toBe("cand_v2");
expect(candidate?.status).toBe("active");
// Speech is still 0
expect(get(client.speechStore).length).toBe(0);
});
test("malformed diagnostic records are strictly rejected at boundary and never touch stores", () => {
const sessionId = "neg-" + Math.random().toString(36).slice(2, 7);
const client = new StreamingClientV2("ws://test", sessionId);
// Malformed health in lifecycle
expect(() =>
client.handleServerMsg({
type: "event",
record: {
seq: 1,
session_id: sessionId,
lane: "lifecycle",
payload: { kind: "capability.checked", data: { health: "invalid-string" } },
created_ms: 1000,
},
})
).toThrow(/health must be an array/);
// Diagnostics in speech lane
expect(() =>
client.handleServerMsg({
type: "event",
record: {
seq: 2,
session_id: sessionId,
lane: "speech",
payload: { text: "bad speech", health: [{ id: "h1" }] },
created_ms: 1001,
},
})
).toThrow(/diagnostics must not enter speech lane/);
// Stores remain empty
expect(get(client.healthStore).length).toBe(0);
expect(get(client.workItemsStore).length).toBe(0);
expect(get(client.candidateStore)).toBeNull();
expect(get(client.speechStore).length).toBe(0);
});
test("rejections during sync are tracked in connection state across sync_completed", () => {
const sessionId = "rej-" + Math.random().toString(36).slice(2, 7);
const client = new StreamingClientV2("ws://test", sessionId);
expect(get(client.connectionStore).rejectedCount).toBe(0);
expect(get(client.connectionStore).lastRejection).toBeUndefined();
client.handleServerMsg({ type: "sync_started", session_id: sessionId });
expect(get(client.connectionStore).isSyncing).toBe(true);
expect(() =>
client.handleServerMsg({
type: "event",
record: {
seq: 5,
session_id: sessionId,
lane: "lifecycle",
payload: { kind: "capability.checked", data: { health: "invalid-string" } },
created_ms: 1005,
},
})
).toThrow(/health must be an array/);
const connMid = get(client.connectionStore);
expect(connMid.rejectedCount).toBe(1);
expect(connMid.lastRejection?.seq).toBe(5);
expect(connMid.lastRejection?.reason).toMatch(/health must be an array/);
client.handleServerMsg({
type: "sync_completed",
session_id: sessionId,
latest_seq: 10,
});
const connAfter = get(client.connectionStore);
expect(connAfter.isSyncing).toBe(false);
expect(connAfter.lastSeenSeq).toBe(10);
expect(connAfter.rejectedCount).toBe(1);
expect(connAfter.lastRejection?.seq).toBe(5);
expect(connAfter.lastRejection?.reason).toMatch(/health must be an array/);
});
});