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

});