import type { V2ClientMsg, V2EventRecord, V2ServerMsg } from "../src/lib/protocol-v2";
const SCRIPT = [
{
seq: 1, session_id: "default", lane: "input", created_ms: 1760000000000,
payload: {
source: "operator", author: "dawn", text: "read HANDOFF.md, remove the dead runtime-v1 code, and tell me what survived.",
provenance: { origin_kind: "human", origin: "human", reporter_id: "dawn", reporter: "dawn", observed_ms: 1760000000000 },
},
},
{
seq: 2, session_id: "default", lane: "lifecycle", created_ms: 1760000012000,
payload: { kind: "daemon.worker_attached", details: "worker klbr-fixture-07 attached to runtime-v2", data: { pid: 4812, transport: "local", protocol_version: 2 } },
},
{
seq: 3, session_id: "default", lane: "scratchpad", created_ms: 1760000024000,
payload: { text: "i should read the handoff before touching the crate. archaeology is cheaper when somebody already left a map.", reasoning: "locate the handoff, extract its explicit invariants, then verify each claimed dead module against references rather than trusting the prose blindly.", execution_id: "reason-1", kind: "thought", step: 1, tokens: 184, active: true },
},
{
seq: 4, session_id: "default", lane: "cell", created_ms: 1760000036000,
payload: { type: "started", execution_id: "cell-handoff", generation: "gen-1", revision: "rev-a", code: "sed -n '1,240p' HANDOFF.md" },
},
{
seq: 5, session_id: "default", lane: "cell", created_ms: 1760000048000,
payload: { type: "completed", execution_id: "cell-handoff", state: "ok", output: [
{ stream: "stdout", text: "handoff: runtime-v2 owns scheduling and ipc\nlegacy candidates: src/v1.rs, src/pump.rs, src/compat.rs\nrelease input: /nix/store/9p1k8v0b4n2c7m6x5z3q0r8s1t4u6w9y-klbr-runtime-handoff/share/klbr/HANDOFF.md\n" },
] },
},
{
seq: 6, session_id: "default", lane: "scratchpad", created_ms: 1760000060000,
payload: { text: "the handoff is specific enough to test. the three files are suspects, not convictions.", reasoning: "the ownership boundary is clear: runtime-v2 is authoritative, but public re-exports and migration glue may still make compat.rs live.", execution_id: "reason-1", kind: "accumulated", step: 1, tokens: 331, duration_ms: 2760, active: false },
},
{
seq: 7, session_id: "default", lane: "lifecycle", created_ms: 1760000072000,
payload: { kind: "hook.decision", details: "pre-edit policy allowed repository-local inspection", data: { hook: "before_cell", decision: "allow", policy: "workspace-read", request: "trace runtime-v1 references" }, step: 2 },
},
{
seq: 8, session_id: "default", lane: "input", created_ms: 1760000084000,
payload: {
source: "ci/watch", author: "checks", text: "external observation: klbr-runtime last passed on linux; macos does not compile the sandbox backend.",
provenance: { origin_kind: "external", origin: "external", reporter_id: "ci-watch", reporter: "ci/watch", source_refs: ["build:8841"], observed_ms: 1760000084000 },
},
},
{
seq: 9, session_id: "default", lane: "lifecycle", created_ms: 1760000096000,
payload: { kind: "capability.health", data: { health: [
{ id: "cap-rust", name: "rust toolchain", scope: "workspace", status: "healthy", target: "klbr-runtime", evidence: "rustc 1.89.0; cargo metadata ok", observed_ms: 1760000096000, valid_until_ms: 1760003696000, origin: "system", reporter: "fixture-daemon" },
{ id: "cap-linux-sandbox", name: "linux sandbox", scope: "host", status: "degraded", target: "local-macos", evidence: "cross-host validation required", impact: "sandbox tests deferred", observed_ms: 1760000096000, origin: "system", reporter: "fixture-daemon" },
] } },
},
{
seq: 10, session_id: "default", lane: "lifecycle", created_ms: 1760000108000,
payload: { kind: "work.updated", data: { work_items: [
{ id: "work-dead-code", title: "remove runtime-v1 dead code", purpose: "make runtime-v2 the only scheduling path", status: "executing", target: "klbr-runtime", origin: "human", reporter: "dawn", selection_reason: "handoff names an obsolete compatibility layer", dependencies: ["work-trace-refs"], created_ms: 1759999700000, age_ms: 408000 },
{ id: "work-trace-refs", title: "trace references", purpose: "prove each deletion is closed over callers", status: "completed", target: "klbr-runtime", origin: "agent", reporter: "klbr", created_ms: 1759999800000, age_ms: 308000 },
] } },
},
{
seq: 11, session_id: "default", lane: "scratchpad", created_ms: 1760000120000,
payload: { text: "now for the reference graph. filenames lie for free, while rust's module tree at least charges compiler errors.", reasoning: "search module declarations, use sites, feature gates, tests, and generated references. deletion is valid only if the surviving api does not name these types and no cfg-only edge keeps them reachable. compat.rs looks suspicious because the handoff calls it migration glue rather than dead code.", execution_id: "reason-2", kind: "thought", step: 3, tokens: 512, active: true },
},
{
seq: 12, session_id: "default", lane: "cell", created_ms: 1760000132000,
payload: { type: "started", execution_id: "cell-refs", generation: "gen-1", revision: "rev-a", code: "rg -n 'mod (v1|pump|compat)|crate::(v1|pump|compat)' crates/klbr-runtime" },
},
{
seq: 13, session_id: "default", lane: "cell", created_ms: 1760000144000,
payload: { type: "completed", execution_id: "cell-refs", state: "ok", output: [
{ stream: "stdout", text: "┌───────────────┬──────┬──────────────┐\n│ module │ refs │ verdict │\n├───────────────┼──────┼──────────────┤\n│ v1.rs │ 0 │ remove │\n│ pump.rs │ 0 │ remove │\n│ compat.rs │ 2 │ keep facade │\n└───────────────┴──────┴──────────────┘\n" },
{ stream: "result", text: "{ removable: [\"v1.rs\", \"pump.rs\"], retained: \"compat.rs\" }" },
] },
},
{
seq: 14, session_id: "default", lane: "scratchpad", created_ms: 1760000156000,
payload: { text: "two clean deletions, one false positive. compat.rs still translates persisted v1 envelopes during restore, so removing it would be a data migration disguised as tidying.", reasoning: "keep compat.rs and narrow its visibility; remove v1.rs and pump.rs plus their module declarations.", execution_id: "reason-2", kind: "accumulated", step: 3, tokens: 690, duration_ms: 4180, active: false },
},
{
seq: 15, session_id: "default", lane: "cell", created_ms: 1760000168000,
payload: { type: "started", execution_id: "cell-edit", generation: "gen-1", revision: "rev-a", code: "rm crates/klbr-runtime/src/{v1,pump}.rs && python3 dev/remove_modules.py crates/klbr-runtime/src/lib.rs v1 pump" },
},
{
seq: 16, session_id: "default", lane: "cell", created_ms: 1760000180000,
payload: { type: "completed", execution_id: "cell-edit", state: "ok", output: [
{ stream: "stdout", text: "removed 184 lines across 3 files\nleft compat.rs crate-private\n" },
{ stream: "result", text: "{ changed: 3, deleted: 2, net_lines: -184 }" },
] },
},
{
seq: 17, session_id: "default", lane: "scratchpad", created_ms: 1760000192000,
payload: { text: "the patch is pleasantly boring. i still need the compiler to disagree with me at machine speed.", reasoning: "run the focused suite first. if it passes, inspect all-target failures separately because the local host cannot validate the linux sandbox path. do not broaden the deletion just because the diff looks symmetric.", execution_id: "reason-3", kind: "thought", step: 4, tokens: 402, active: true },
},
{
seq: 18, session_id: "default", lane: "cell", created_ms: 1760000204000,
payload: { type: "started", execution_id: "cell-tests", generation: "gen-1", revision: "rev-b", code: "cargo test -p klbr-runtime --lib" },
},
{
seq: 19, session_id: "default", lane: "scratchpad", created_ms: 1760000216000,
payload: { text: "tests are still running; the linker has entered its contemplative phase.", reasoning: "wait for the focused receipt before deciding whether the candidate can advance.", execution_id: "reason-3", kind: "prose", step: 4, tokens: 488, active: true },
},
{
seq: 20, session_id: "default", lane: "cell", created_ms: 1760000228000,
payload: { type: "completed", execution_id: "cell-tests", state: "ok", output: [
{ stream: "stdout", text: "running 27 tests\ntest restore::v1_envelope_uses_compat ... ok\ntest scheduler::runtime_v2_only ... ok\ntest result: ok. 27 passed; 0 failed; finished in 4.81s\n" },
] },
},
{
seq: 21, session_id: "default", lane: "lifecycle", created_ms: 1760000240000,
payload: { kind: "candidate.updated", details: "focused tests passed", data: { candidate: { candidate_id: "cand-runtime-v2-cleanup", work_item_id: "work-dead-code", base_revision: "a82ce91", target_revision: "9c".repeat(60), status: "testing", epoch: 7, test_receipts: [{ suite_id: "klbr-runtime-lib", passed: true, duration_ms: 4810 }] } } },
},
{
seq: 22, session_id: "default", lane: "input", created_ms: 1760000252000,
payload: {
source: "agent/work", author: "klbr", text: "queued follow-up: run all-target checks, but keep host-specific failures out of the deletion patch.",
provenance: { origin_kind: "agent", origin: "agent", reporter_id: "klbr-agent", reporter: "klbr", source_refs: ["work-dead-code"], observed_ms: 1760000252000 },
},
},
{
seq: 23, session_id: "default", lane: "cell", created_ms: 1760000264000,
payload: { type: "started", execution_id: "cell-all-targets", generation: "gen-1", revision: "rev-b", code: "python3 dev/check_all_targets.py klbr-runtime" },
},
{
seq: 24, session_id: "default", lane: "cell", created_ms: 1760000276000,
payload: { type: "completed", execution_id: "cell-all-targets", state: "error", error: "all-target check exited with status 1", output: [
{ stream: "stderr", text: "Traceback (most recent call last):\n File \"dev/check_all_targets.py\", line 44, in <module>\n subprocess.run(command, check=True)\n File \"subprocess.py\", line 571, in run\n raise CalledProcessError(retcode, process.args)\nsubprocess.CalledProcessError: Command 'cargo check -p klbr-runtime --all-targets' returned non-zero exit status 101." },
] },
},
{
seq: 25, session_id: "default", lane: "speech", created_ms: 1760000288000,
payload: { type: "failed", effect_id: "speech-rejected-1", text: "i deleted compat.rs too. symmetry demanded a sacrifice.", status: "rejected", reason: "delivery review rejected an inaccurate claim", provenance: { origin_kind: "agent", origin: "agent", reporter_id: "klbr-agent", reporter: "klbr", observed_ms: 1760000288000 } },
},
{
seq: 26, session_id: "default", lane: "scratchpad", created_ms: 1760000300000,
payload: { text: "the failure is the known macos sandbox cfg, not a missing symbol from this patch. focused tests cover the changed closure; linux validation remains explicit instead of magically becoming green through optimism.", reasoning: "classify the failed receipt as host capability evidence. preserve compat.rs, report the focused pass, and do not claim the all-target suite passed.", execution_id: "reason-3", kind: "accumulated", step: 5, tokens: 811, duration_ms: 6940, active: false },
},
{
seq: 27, session_id: "default", lane: "cell", created_ms: 1760000312000,
payload: { type: "started", execution_id: "cell-linux", generation: "gen-1", revision: "rev-b", code: "await remote_linux_validation(\"cand-runtime-v2-cleanup\")" },
},
{
seq: 28, session_id: "default", lane: "cell", created_ms: 1760000324000,
payload: { type: "completed", execution_id: "cell-linux", state: "yielded", output: [
{ stream: "result", text: "{ wait: \"linux-validation\", ticket: \"linux-8842\", retry_after_seconds: 8 }" },
] },
},
{
seq: 29, session_id: "default", lane: "lifecycle", created_ms: 1760000336000,
payload: { kind: "runtime.yielded", seconds: 8, reason: "waiting for linux validation receipt", details: "ticket linux-8842 retained by scheduler", step: 6, data: { wait_for: "capability:linux-sandbox", ticket: "linux-8842" } },
},
{
seq: 30, session_id: "default", lane: "cell", created_ms: 1760000348000,
payload: { type: "started", execution_id: "cell-docs", generation: "gen-1", revision: "rev-b", code: "cargo doc -p klbr-runtime --no-deps" },
},
{
seq: 31, session_id: "default", lane: "cell", created_ms: 1760000360000,
payload: { type: "completed", execution_id: "cell-docs", state: "interrupted", error: "interrupted by operator", output: [
{ stream: "stderr", text: "documenting klbr-runtime\ninterrupt received; rustdoc terminated\n" },
] },
},
{
seq: 32, session_id: "default", lane: "lifecycle", created_ms: 1760000372000,
payload: { kind: "context.compacted", details: "collapsed repository inspection and test receipts", data: { before_tokens: 18422, after_tokens: 6910, retained: ["handoff invariants", "reference graph", "test receipts"] } },
},
{
seq: 33, session_id: "default", lane: "lifecycle", created_ms: 1760000384000,
payload: { kind: "candidate.activated", details: "candidate adopted at epoch 7", data: { epoch: 7, candidate: { candidate_id: "cand-runtime-v2-cleanup", work_item_id: "work-dead-code", base_revision: "a82ce91", target_revision: "d19f82c", status: "active", epoch: 7, test_receipts: [{ suite_id: "klbr-runtime-lib", passed: true, duration_ms: 4810 }] }, adoption: { actor: "fixture-supervisor", release_path: "/tmp/klbr/releases/runtime-v2-cleanup", status: "active" } } },
},
{
seq: 34, session_id: "default", lane: "speech", created_ms: 1760000396000,
payload: { type: "held", effect_id: "speech-held-1", text: "the patch is ready; linux coverage is still pending on ticket linux-8842.", status: "held", reason: "awaiting delivery review", provenance: { origin_kind: "agent", origin: "agent", reporter_id: "klbr-agent", reporter: "klbr", observed_ms: 1760000396000 } },
},
{
seq: 35, session_id: "default", lane: "speech", created_ms: 1760000408000,
payload: { type: "sent", effect_id: "speech-confirmed-1", event_id: 35, status: "confirmed", text: "done.\n\n- removed `v1.rs` and `pump.rs` (184 lines)\n- kept `compat.rs`; restore still needs it for persisted v1 envelopes\n- passed `cargo test -p klbr-runtime --lib` (27/27)\n- left linux sandbox validation on `linux-8842`\n\nthe all-target macos failure is host-shaped, not patch-shaped. annoyingly, the distinction matters.", provenance: { origin_kind: "agent", origin: "agent", reporter_id: "klbr-agent", reporter: "klbr", observed_ms: 1760000408000 } },
},
{
seq: 36, session_id: "default", lane: "lifecycle", created_ms: 1760000420000,
payload: {
kind: "context.receipt", details: "context budget: 61400 of 200000 tokens",
data: {
type: "context_receipt",
consumed_tokens: 61400,
total_budget_tokens: 200000,
context_receipt: {
receipt_id: "rcpt-1", session_id: "default", timestamp_ms: 1760000420000,
total_budget_tokens: 200000, consumed_tokens: 61400, guidance_status: "ok",
included_spans: [], omitted_items: [],
},
},
},
},
{
seq: 37, session_id: "default", lane: "lifecycle", created_ms: 1760000432000,
payload: {
kind: "model.usage", details: "request usage for turn 1",
data: {
type: "request_usage", purpose: "execution", step: 1,
request_sha256: "4a82ce91d19f82c0", message_count: 4, response_event_id: 6,
route: { url: "http://127.0.0.1:8000/v1", model: "claude-3-7-sonnet" },
usage_reported: true,
usage: {
prompt_tokens: 61400, completion_tokens: 184, total_tokens: 61584,
cached_prompt_tokens: 23300, cache_write_tokens: 38100, uncached_prompt_tokens: 38100,
tokens_reported: true,
prompt_processing_ms: 320, generation_ms: 2190, bridge_processing_ms: 10,
},
},
},
},
{
seq: 38, session_id: "default", lane: "lifecycle", created_ms: 1760000444000,
payload: {
kind: "model.usage", details: "request usage for turn 2",
data: {
type: "request_usage", purpose: "execution", step: 2,
request_sha256: "b9c0211a76f284e3", message_count: 8, response_event_id: 14,
route: { url: "http://127.0.0.1:8000/v1", model: "claude-3-7-sonnet" },
usage_reported: true,
usage: {
prompt_tokens: 78200, completion_tokens: 331, total_tokens: 78531,
cached_prompt_tokens: 29700, cache_write_tokens: 48500, uncached_prompt_tokens: 48500,
tokens_reported: true,
prompt_processing_ms: 410, generation_ms: 3940, bridge_processing_ms: 14,
},
},
},
},
{
seq: 39, session_id: "default", lane: "lifecycle", created_ms: 1760000456000,
payload: {
kind: "context.receipt", details: "context budget: 88100 of 200000 tokens",
data: {
type: "context_receipt",
consumed_tokens: 88100,
total_budget_tokens: 200000,
context_receipt: {
receipt_id: "rcpt-2", session_id: "default", timestamp_ms: 1760000456000,
total_budget_tokens: 200000, consumed_tokens: 88100, guidance_status: "ok",
included_spans: [], omitted_items: [],
},
},
},
},
{
seq: 40, session_id: "default", lane: "lifecycle", created_ms: 1760000468000,
payload: {
kind: "model.usage", details: "unmeasured fallback usage",
data: {
type: "request_usage", purpose: "execution", step: 3,
request_sha256: "2d774a1e941f32a0", message_count: 12, response_event_id: 26,
route: { url: "http://127.0.0.1:8000/v1", model: "claude-3-7-sonnet" },
usage_reported: false,
usage: {
prompt_tokens: 88100, completion_tokens: 0, total_tokens: 88100,
tokens_reported: false,
prompt_processing_ms: 120, generation_ms: 0, bridge_processing_ms: 5,
},
},
},
},
{
seq: 41, session_id: "default", lane: "lifecycle", created_ms: 1760000480000,
payload: {
kind: "model.usage", details: "request usage for turn 3",
data: {
type: "request_usage", purpose: "execution", step: 4,
request_sha256: "71e54bc8310d4812", message_count: 16, response_event_id: 35,
route: { url: "http://127.0.0.1:8000/v1", model: "claude-3-7-sonnet" },
usage_reported: true,
usage: {
prompt_tokens: 102400, completion_tokens: 420, total_tokens: 102820,
cached_prompt_tokens: 38900, cache_write_tokens: 63500, uncached_prompt_tokens: 63500,
tokens_reported: true,
prompt_processing_ms: 520, generation_ms: 5000, bridge_processing_ms: 18,
},
},
},
},
] satisfies V2EventRecord[];
/**
* `normal` replays the full session and keeps trickling. A real operator refusal is reachable without a
* mode of its own: activate a revision twice and the second one is refused by the daemon, because the
* fixture advances its epoch on success just as the real activator does.
*/
type Mode = "normal" | "empty" | "disconnected" | "error" | "gap" | "bulk";
type SocketData = {
mode: Mode;
/** `?live=off` stops the synthetic trickle so the replay can be read without new cells arriving. */
live: boolean;
sessionId: string;
/** How many turns `mode=bulk` should generate. */
bulkTurns: number;
epoch: number;
nextSeq: number;
liveCycle: number;
closed: boolean;
timer?: ReturnType<typeof setTimeout>;
chunkTimers: Array<ReturnType<typeof setTimeout>>;
};
/**
* A windowed history is what the console asks for: the newest records, bounded by the session's own fold
* and by the size asked for. The mirror of the daemon's `plan_replay`: the fold is a durable
* `lcm_frontier` record, and everything at or below what it drained through is carried by its summary.
*/
function foldOf(records: V2EventRecord[]): number | null {
const fold = [...records].reverse().find((record) => record.lane === "lifecycle" && (record.payload as { kind?: string }).kind === "lcm_frontier");
if (!fold) return null;
const data = (fold.payload as { data?: { drained_through_seq?: number } }).data;
return data?.drained_through_seq ?? fold.seq - 1;
}
function windowFloor(records: V2EventRecord[], limit: number | null | undefined): number {
const newest = records.at(-1)?.seq ?? 0;
const fold = foldOf(records);
const floorByLimit = limit && limit > 0 ? newest - limit + 1 : 1;
const floorByFold = fold === null ? 1 : fold + 1;
return Math.max(1, floorByFold, floorByLimit);
}
/**
* A session large enough to be worth virtualizing: `?mode=bulk` replays this instead of the
* hand-written script. It uses the same record shapes, many more of them, and cell outputs of wildly
* different heights — so dynamic measurement is exercised rather than assumed, and the console has a
* transcript where the difference between "render everything" and "render the fold" is measurable.
*/
function bulkScript(turns: number): V2EventRecord[] {
const records: V2EventRecord[] = [];
let seq = 0;
let at = 1760000000000;
const push = (lane: V2EventRecord["lane"], payload: unknown): void => {
seq += 1;
at += 4000;
records.push({ seq, session_id: "default", lane, created_ms: at, payload } as V2EventRecord);
};
for (let t = 0; t < turns; t++) {
// A fold: the runtime installs a context frontier when it has summarized a prefix, and the durable
// record of it says which record it drained through. That record is the boundary a window opens at.
if (t > 0 && t % 40 === 0) {
push("lifecycle", {
kind: "lcm_frontier",
details: `folded ${t} turns of session context`,
data: {
epoch: 1 + t / 40,
drained_through_seq: seq,
summary: `turns 1 to ${t}: the runtime crate and its focused suite`,
source_children: [`bulk-${t}`],
created_ms: at,
},
});
}
push("input", {
source: t % 5 === 0 ? "ci/watch" : "operator",
author: "dawn",
text: `turn ${t + 1}: ${t % 3 === 0 ? "check the runtime crate and report only what the compiler agreed with." : "keep going; the previous step left a loose end in the scheduler."}`,
provenance: { origin_kind: "human", origin: "human", reporter_id: "dawn", reporter: "dawn", observed_ms: at },
});
push("scratchpad", {
text: `thinking about step ${t + 1} of the session. the cheap explanation is that the harness is stale; the expensive one is that the code is.`,
reasoning: "read the receipt before believing the summary, then decide whether the difference is host-shaped or patch-shaped.",
execution_id: `reason-bulk-${t}`, kind: t % 2 === 0 ? "thought" : "accumulated", step: t + 1,
tokens: 120 + (t % 400), duration_ms: 500 + (t % 3000), active: t % 4 === 0,
});
const cells = 2 + (t % 5);
for (let c = 0; c < cells; c++) {
const executionId = `bulk-${t}-${c}`;
push("cell", { type: "started", execution_id: executionId, generation: "gen-1", revision: `rev-${t % 7}`, code: `cargo test -p klbr-runtime --lib -- bulk_${t}_${c}` });
// one tall cell per turn, so most rows are short and a few are enormous — and past the
// console's own 400-line truncation point, so the expander is a state that can actually be reached
const lines = c === 0 ? 520 : 1 + ((t * 7 + c * 13) % 9);
const body = Array.from({ length: lines }, (_, i) => `test scheduler::bulk_${t}_${c}::case_${i} ... ${i % 11 === 0 ? "ignored" : "ok"}`).join("\n");
push("cell", {
type: "completed", execution_id: executionId,
state: c % 4 === 3 ? "error" : "ok", exit_code: c % 4 === 3 ? 101 : 0, duration_ms: 120 + ((t * 31 + c * 7) % 4000),
output: [
{ stream: "stdout", text: `${body}\n` },
{ stream: "result", text: `{ passed: ${lines - 1}, failed: ${c % 4 === 3 ? 1 : 0} }` },
],
});
}
push("speech", {
type: t % 4 === 3 ? "held" : "sent", effect_id: `speech-bulk-${t}`, event_id: 1000 + t,
status: t % 4 === 3 ? "held" : "confirmed",
reason: t % 4 === 3 ? "awaiting delivery review" : undefined,
text: `turn ${t + 1}: the focused suite is the only thing that speaks for the whole closure. everything else is a guess with better manners.`,
provenance: { origin_kind: "agent", origin: "agent", reporter_id: "klbr-agent", reporter: "klbr", observed_ms: at },
});
push("lifecycle", { kind: "model.usage", details: `request usage for turn ${t + 1}`, data: { type: "request_usage", purpose: "execution", step: t + 1, message_count: 12 + (t % 30), usage: { prompt_tokens: 1024 * (1 + (t % 40)), completion_tokens: 200 + (t % 600), total_tokens: 1200 + (t % 900) } } });
}
return records;
}
const modes = new Set<Mode>(["normal", "empty", "disconnected", "error", "gap", "bulk"]);
/** Generated once per size: two `bulk` clients with the same size must get the same session. */
const bulkCache = new Map<number, V2EventRecord[]>();
function bulkFor(turns: number): V2EventRecord[] {
const cached = bulkCache.get(turns);
if (cached) return cached;
const script = bulkScript(turns);
bulkCache.set(turns, script);
return script;
}
const latestScriptSeq = SCRIPT.at(-1)?.seq ?? 0;
function send(ws: Bun.ServerWebSocket<SocketData>, message: V2ServerMsg): void {
if (ws.data.closed) return;
ws.send(JSON.stringify(message));
}
function clearLive(ws: Bun.ServerWebSocket<SocketData>): void {
if (ws.data.timer) clearTimeout(ws.data.timer);
for (const timer of ws.data.chunkTimers) clearTimeout(timer);
ws.data.timer = undefined;
ws.data.chunkTimers = [];
}
function scheduleLive(ws: Bun.ServerWebSocket<SocketData>): void {
clearLive(ws);
const delay = 1000 + Math.floor(Math.random() * 2001);
const executionId = `fixture-live-${ws.data.liveCycle++}`;
const deltas = ["checking one more reference… ", "0 callers remain; ", "the compiler also seems bored.\n"];
ws.data.chunkTimers = deltas.map((delta, index) => setTimeout(() => {
send(ws, { type: "live_chunk", session_id: ws.data.sessionId, lane: "cell", execution_id: executionId, stream: "stdout", delta });
}, Math.floor(delay * (index + 1) / (deltas.length + 1))));
ws.data.timer = setTimeout(() => {
const record: V2EventRecord = {
seq: ws.data.nextSeq++, session_id: ws.data.sessionId, lane: "cell", created_ms: Date.now(),
payload: { type: "completed", execution_id: executionId, state: "ok", output: [{ stream: "result", text: "{ references: 0, stable: true }" }] },
};
send(ws, { type: "event", record });
scheduleLive(ws);
}, delay);
}
function sync(ws: Bun.ServerWebSocket<SocketData>, message: Extract<V2ClientMsg, { type: "subscribe" }>): void {
clearLive(ws);
ws.data.sessionId = message.session_id;
if (ws.data.mode === "disconnected") return;
if (ws.data.mode === "empty") {
send(ws, { type: "sync_completed", session_id: message.session_id, latest_seq: 0 });
return;
}
const script = ws.data.mode === "bulk" ? bulkFor(ws.data.bulkTurns) : SCRIPT;
const records = ws.data.mode === "gap" ? script.filter((record) => record.seq !== 18) : script;
const earliest = records[0]?.seq ?? 0;
const latestSeq = records.at(-1)?.seq ?? 0;
const fold = foldOf(records);
// No cursor at all is a live-only subscription: nothing to send, the trickle alone is the answer.
if (message.after_seq === null || message.after_seq === undefined) {
if (ws.data.live) scheduleLive(ws);
return;
}
// A client with a cursor is asking for the gap in front of it and gets all of it. A client that holds
// nothing is asking for what it can hold: the newest window, bounded by the fold and by its limit.
const resuming = message.after_seq > 0;
const from = resuming ? message.after_seq + 1 : windowFloor(records, message.limit);
const replay = records.filter((record) => record.seq >= from && record.seq <= latestSeq);
ws.data.nextSeq = latestSeq + 1;
send(ws, {
type: "sync_started", session_id: message.session_id,
from_seq: replay[0]?.seq ?? from, to_seq: latestSeq, earliest_seq: earliest, fold_seq: fold,
});
for (const fixture of replay) {
send(ws, { type: "event", record: { ...fixture, session_id: message.session_id } });
}
send(ws, {
type: "sync_completed", session_id: message.session_id,
latest_seq: latestSeq, earliest_seq: earliest,
});
if (ws.data.mode === "error") {
send(ws, { type: "error", code: "fixture", message: "fixture mode requested a post-sync daemon error" });
} else if (ws.data.live) {
scheduleLive(ws);
}
}
/**
* The answer to "load earlier history": the records behind a loaded window, in the same replay shape a
* subscribe uses. It does not touch the live trickle — reading the past is not what starts the future.
*/
function history(ws: Bun.ServerWebSocket<SocketData>, message: Extract<V2ClientMsg, { type: "history" }>): void {
const script = ws.data.mode === "bulk" ? bulkFor(ws.data.bulkTurns) : SCRIPT;
const records = ws.data.mode === "gap" ? script.filter((record) => record.seq !== 18) : script;
const earliest = records[0]?.seq ?? 0;
const behind = records.filter((record) => record.seq < message.before_seq);
const page = behind.slice(Math.max(0, behind.length - Math.max(1, message.limit)));
send(ws, {
type: "sync_started", session_id: message.session_id,
from_seq: page[0]?.seq ?? message.before_seq, to_seq: page.at(-1)?.seq ?? message.before_seq,
earliest_seq: earliest, fold_seq: foldOf(records),
});
for (const fixture of page) {
send(ws, { type: "event", record: { ...fixture, session_id: message.session_id } });
}
send(ws, {
type: "sync_completed", session_id: message.session_id,
latest_seq: records.at(-1)?.seq ?? 0, earliest_seq: earliest,
});
}
function parseClientMessage(raw: string): V2ClientMsg | null {
try {
const value: unknown = JSON.parse(raw);
return value !== null && typeof value === "object" && "type" in value ? value as V2ClientMsg : null;
} catch {
return null;
}
}
function operatorAck(ws: Bun.ServerWebSocket<SocketData>, message: V2ClientMsg): void {
if (message.type === "activate") {
if (message.expected_epoch !== ws.data.epoch) {
send(ws, { type: "operator_ack", action: "activate", session_id: message.session_id, success: false, message: `epoch changed: expected ${message.expected_epoch}, current ${ws.data.epoch}` });
return;
}
ws.data.epoch += 1;
send(ws, { type: "operator_ack", action: "activate", session_id: message.session_id, success: true, message: `activated ${message.release_path} at epoch ${ws.data.epoch}` });
return;
}
if (["interrupt", "reset_kernel", "stop", "drain", "compact"].includes(message.type)) {
const sessionId = "session_id" in message ? message.session_id : ws.data.sessionId;
send(ws, { type: "operator_ack", action: message.type, session_id: sessionId, success: true, message: `${message.type.replace("_", " ")} accepted by fixture supervisor` });
}
}
const PORT = Number(Bun.env.PORT ?? 8765);
const HOSTNAME = Bun.env.HOSTNAME_OVERRIDE ?? "127.0.0.1";
Bun.serve<SocketData>({
port: PORT,
hostname: HOSTNAME,
fetch(request, server) {
const url = new URL(request.url);
if (url.pathname !== "/ws") {
return new Response(`klbr fixture websocket lives at ws://${HOSTNAME}:${PORT}/ws?mode=normal|empty|disconnected|error|gap\n`, { status: 404 });
}
const requestedMode = url.searchParams.get("mode") as Mode | null;
const mode = requestedMode && modes.has(requestedMode) ? requestedMode : "normal";
const live = url.searchParams.get("live") !== "off";
const bulkTurns = Math.max(1, Math.min(2000, Number(url.searchParams.get("turns") ?? 120) || 120));
const upgraded = server.upgrade(request, { data: { mode, live, bulkTurns, sessionId: "default", epoch: 7, nextSeq: latestScriptSeq + 1, liveCycle: 1, closed: false, chunkTimers: [] } });
return upgraded ? undefined : new Response("websocket upgrade required", { status: 400 });
},
websocket: {
open(ws) {
if (ws.data.mode === "disconnected") {
ws.data.timer = setTimeout(() => ws.close(1012, "fixture reconnect cycle"), 1500);
}
},
message(ws, incoming) {
const raw = typeof incoming === "string" ? incoming : new TextDecoder().decode(incoming);
const message = parseClientMessage(raw);
if (!message) {
send(ws, { type: "error", code: "fixture", message: "malformed client message" });
return;
}
switch (message.type) {
case "subscribe": sync(ws, message); break;
case "history": history(ws, message); break;
case "ping": send(ws, { type: "pong", timestamp: message.timestamp }); break;
case "send_input": {
const record: V2EventRecord = {
seq: ws.data.nextSeq++, session_id: message.session_id, lane: "input", created_ms: Date.now(),
payload: { source: message.source, text: message.text, author: "dawn" },
};
send(ws, { type: "event", record });
send(ws, { type: "ack", session_id: message.session_id, seq: record.seq, client_request_id: message.client_request_id });
break;
}
case "session_list":
send(ws, { type: "session_list", sessions: [[ws.data.sessionId, { state: "running", epoch: ws.data.epoch, latest_seq: ws.data.nextSeq - 1, fixture: true }]] });
break;
case "session_status":
send(ws, { type: "session_status", session_id: message.session_id, state: { phase: "running", epoch: ws.data.epoch, latest_seq: ws.data.nextSeq - 1, worker: "klbr-fixture-07" } });
break;
case "interrupt":
case "reset_kernel":
case "stop":
case "drain":
case "compact":
case "activate": operatorAck(ws, message); break;
case "ack": break;
}
},
close(ws) {
ws.data.closed = true;
clearLive(ws);
},
},
});