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);
    },
  },
});