// CHAT-03 §4, §7 and §9 (#1507): the slash path S1–S7, Pi dialogs P3 and the
// return flow E1–E7, on the in-process fake engine, the library client, the
// Transcript view and the mediated terminal. Every record the controllers
// produce here is checked against the CHAT-01 schema at the end.

import { test, after } from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { join } from "node:path";
import { ConversationClient, OUTCOME_UNKNOWN } from "../src/client.mjs";
import { Transcript, RECONCILE_MARKER } from "../src/transcript.mjs";
import { Terminal, NOT_CONTROLLER, visible } from "../src/terminal.mjs";
import { LATER_LINE_SLASH_INTERPRETED, PREFIXES, TEXT_POLICY } from "../src/text-policy.mjs";
import { LineSplitter, encodeLine } from "../src/framing.mjs";
import { replyToRow } from "../../control-board/src/serve.mjs";
import { REPO, started, receiptState, cleanupAll, tick } from "./harness.mjs";

after(() => cleanupAll());

const PASTE_START = "\x1b[200~";
const PASTE_END = "\x1b[201~";
const PAGE = ["user: hello", "assistant: hi"];
const records = [];

async function until(pred, ms = 4000, what = "condition") {
  const end = Date.now() + ms;
  while (!pred()) {
    if (Date.now() > end) throw new Error(`timed out waiting for ${what}`);
    await tick(5);
  }
}

// A view reads the page once when it attaches to a connected client.
const loaded = (view) => until(() => view.loads >= 1 && !view.loading, 4000, "the first page read");

const prompts = (engine) => engine.commands.filter((c) => c.type === "prompt");

// Every CHAT-01 record a fixture produced, for the schema check.
function collect(h, clients = h.clients) {
  records.push(...h.ctrl.events, ...h.ctrl.requests.values());
  for (const c of clients) {
    records.push(...c.receipts.values());
    for (const p of c.pushes) {
      for (const key of ["binding", "connection", "confirmation", "stop"]) if (p[key]?.version === 2) records.push(p[key]);
    }
  }
}

// A client that follows the conversation through a Transcript from its first
// welcome, as a library user or the terminal would.
async function follower(h) {
  const client = new ConversationClient({ socketPath: h.ctrl.socketPath });
  const view = new Transcript({ client });
  await client.connect();
  await loaded(view);
  return { client, view };
}

async function admitted(client, text) {
  const r = await client.prompt(text);
  assert.equal(r.outcome, "admitted", JSON.stringify(r));
  return r.receipt.id;
}

// Each message exactly once: no line repeats.
function once(lines) {
  assert.equal(new Set(lines).size, lines.length, `a message repeats: ${JSON.stringify(lines)}`);
}

// ---- S: the slash path ------------------------------------------------------

test("S1: `/goal x`, with leading spaces or a tab, refuses text-policy at admission; zero engine bytes", async () => {
  const h = await started();
  try {
    for (const text of ["/goal x", "   /goal x", "\t/goal x", "\n/goal x"]) {
      const before = h.engine.bytes().length;
      const r = await h.client.prompt(text);
      assert.equal(r.refusal, TEXT_POLICY, JSON.stringify(text));
      assert.equal(h.engine.bytes().length, before, `${JSON.stringify(text)} reached the engine`);
    }
    assert.equal(prompts(h.engine).length, 0);
    collect(h);
  } finally {
    await h.close();
  }
});

test("S2: every prefix pinned Pi interprets is refused, from the list the code uses; the rest reach the engine exactly", async () => {
  const h = await started();
  try {
    assert.ok(PREFIXES.some((p) => p.example === "/skill:ms-unslop" && p.interpreted));
    for (const p of PREFIXES) {
      const before = h.engine.bytes().length;
      const r = await h.client.prompt(p.example);
      if (p.interpreted) {
        assert.equal(r.refusal, TEXT_POLICY, p.prefix);
        assert.equal(h.engine.bytes().length, before, `${p.prefix} reached the engine`);
      } else {
        assert.equal(r.outcome, "admitted", `${p.prefix}: ${JSON.stringify(r)}`);
        await receiptState(h.client, r.receipt.id, "finished");
        assert.equal(prompts(h.engine).at(-1).message, p.example);
      }
    }
    assert.equal(prompts(h.engine).length, PREFIXES.filter((p) => !p.interpreted).length);
    collect(h);
  } finally {
    await h.close();
  }
});

test("S3: `/goal` on the second line is pinned from the source: Pi checks only index 0, so it is admitted and sent exactly", async () => {
  const h = await started();
  try {
    const text = "please note\n/goal x";
    const r = await h.client.prompt(text);
    if (LATER_LINE_SLASH_INTERPRETED) {
      assert.equal(r.refusal, TEXT_POLICY);
    } else {
      assert.equal(r.outcome, "admitted", JSON.stringify(r));
      await receiptState(h.client, r.receipt.id, "finished");
      assert.equal(prompts(h.engine).at(-1).message, text);
    }
    collect(h);
  } finally {
    await h.close();
  }
});

test("S4: a `/` left in the composer is cleared when control transfers and returns; the next submit sends only the new text", async () => {
  const h = await started({ clients: 2 });
  try {
    const [mine, other] = h.clients;
    const term = new Terminal({ client: mine });
    await term.key("/");
    assert.equal(term.composer, "/");
    await until(() => other.binding?.controllerGeneration === h.ctrl.binding.controllerGeneration, 2000, "the binding push");
    assert.equal((await other.takeover()).outcome, "transferred");
    await until(() => !mine.isController, 2000, "the transfer to reach the terminal");
    assert.equal(term.composer, "", "the composer clears on transfer");
    await term.key("\x14");
    await until(() => mine.isController, 2000, "control back");
    await term.key("hello");
    await term.key("\r");
    await until(() => prompts(h.engine).length === 1, 2000, "the prompt");
    assert.equal(prompts(h.engine)[0].message, "hello");
    const line = h.engine.bytes().toString("utf8").split("\n").find((l) => l.includes('"type":"prompt"'));
    assert.equal(JSON.parse(line).message, "hello", "the engine's exact bytes");
    assert.equal(term.composer, "");
    await receiptState(mine, term.lastReceipt.id, "finished");
    collect(h);
  } finally {
    await h.close();
  }
});

test("S5: an observer terminal gets a paste then Enter, as send-message.sh does: not admitted: controller, nothing sent", async () => {
  const h = await started({ clients: 2 });
  try {
    const observer = h.clients[1];
    let sends = 0;
    const send = observer.send.bind(observer);
    observer.send = (...a) => {
      if (a[0].command.operation !== "observe") sends += 1;
      return send(...a);
    };
    const term = new Terminal({ client: observer });
    const before = h.engine.bytes().length;
    await term.key(`${PASTE_START}run the deploy\n/goal x${PASTE_END}`);
    assert.equal(term.composer, "run the deploy\n/goal x", "a paste is literal and never submits by itself");
    const r = await term.submit();
    await term.key("\r");
    assert.deepEqual(r, { sent: false, refusal: "controller" });
    assert.equal(term.status, NOT_CONTROLLER);
    assert.ok(term.frame.includes(NOT_CONTROLLER));
    assert.equal(sends, 0, "the client sent nothing but its page read");
    assert.equal(h.engine.bytes().length, before, "zero engine bytes");
    assert.equal(h.ctrl.requests.size, 0);
    collect(h);
  } finally {
    await h.close();
  }
});

test("S6: a mediated-shaped registration (no tmux) passed to the board's replyToRow: 409 no tmux session; exec never runs", () => {
  const calls = [];
  const exec = (...a) => {
    calls.push(a);
    return { status: 0, stdout: "", stderr: "" };
  };
  const index = { sessions: [{ project: "proj", agent: "mediated-seat", registered: { pid: process.pid, alive: true, tmux: null, mediated: { socket: "/nonexistent/control.sock" } } }] };
  const r = replyToRow({ index, key: "proj/mediated-seat", text: "hello", exec });
  assert.equal(r.status, 409);
  assert.match(r.body.error, /no tmux session/);
  assert.equal(calls.length, 0);
});

test("S7: ESC, bracketed-paste markers and U+2028/U+2029 travel as one JSON string; the engine receives the exact text in one record", async () => {
  const h = await started();
  try {
    const text = `a\x1b[31mred\x1b[0m ${PASTE_START}pasted${PASTE_END} b
c
d\r\ne`;
    const id = await admitted(h.client, text);
    await receiptState(h.client, id, "finished");
    assert.equal(prompts(h.engine).length, 1);
    assert.equal(prompts(h.engine)[0].message, text);
    const lines = [];
    const split = new LineSplitter((l) => lines.push(l));
    split.push(h.engine.bytes());
    const records = lines.filter((l) => l.includes('"type":"prompt"'));
    assert.equal(records.length, 1, "no record split");
    assert.equal(JSON.parse(records[0]).message, text);
    collect(h);
  } finally {
    await h.close();
  }
});

// ---- P3: Pi dialogs ---------------------------------------------------------

test("P3: a Pi confirm, select, input or editor dialog is shown disabled with a reason and never answered", async () => {
  const h = await started();
  try {
    const term = new Terminal({ client: h.client });
    for (const method of ["confirm", "select", "input", "editor"]) h.engine.dialog(method);
    await until(() => term.dialogs.length === 4, 2000, "four dialogs");
    for (const d of term.dialogs) {
      assert.equal(d.disabled, true);
      assert.match(d.reason, /not answered/);
      assert.ok(term.frame.some((l) => l.startsWith(`[dialog ${d.method} disabled: `)), d.method);
    }
    await tick(100);
    assert.equal(h.engine.commands.filter((c) => c.type === "extension_ui_response").length, 0, "no response is sent");
    assert.equal(h.ctrl.evidence.dialogs.length, 4);
    collect(h);
  } finally {
    await h.close();
  }
});

// ---- E: the return flow -----------------------------------------------------

test("E1: send, ack, user, toolCall, toolResult, final answer: shown once, no refresh, draft and reading position kept", async () => {
  const h = await started({ clients: 2 });
  try {
    const lib = new Transcript({ client: h.client });
    await loaded(lib);
    const term = new Terminal({ client: h.clients[1], rows: 8 });
    await loaded(term.transcript);
    assert.equal(term.transcript.reconcile, false, "an idle engine gives a quiet cut");
    await term.key("half-typed draft");
    term.scroll = 1;
    h.engine.script([{ tool: { id: "call-e1", name: "bash", args: { cmd: "ls" }, result: "file.txt" } }, { text: "final answer", stop: "stop" }]);
    const id = await admitted(h.client, "list the files");
    await receiptState(h.client, id, "finished");
    await until(() => term.transcript.events.some((e) => e.type === "run-settled"), 2000, "the settle");
    const want = [...PAGE, "user: list the files", 'assistant: [tool-call bash {"cmd":"ls"}]', "tool: [tool-result] file.txt", "assistant: final answer"];
    assert.deepEqual(term.transcript.lines(), want);
    assert.deepEqual(lib.lines(), want);
    assert.equal(term.transcript.loads, 1, "no refresh");
    assert.equal(lib.loads, 1, "no refresh");
    assert.equal(term.composer, "half-typed draft", "the draft is kept");
    assert.equal(term.scroll, 1, "the reading position is kept");
    collect(h);
  } finally {
    await h.close();
  }
});

test("E2: U+2028, U+2029 inside JSON strings and CRLF line ends each parse as one record, on the splitter and through the controller", async () => {
  const lines = [];
  const split = new LineSplitter((l) => lines.push(l));
  const a = JSON.stringify({ type: "x", text: "one
two
three" });
  const b = JSON.stringify({ type: "y", text: "crlf" });
  const bytes = Buffer.from(`${a}\r\n${b}\n`);
  for (let i = 0; i < bytes.length; i += 7) split.push(bytes.subarray(i, i + 7));
  assert.deepEqual(lines.map((l) => JSON.parse(l).type), ["x", "y"]);
  assert.equal(JSON.parse(lines[0]).text, "one
two
three");

  const h = await started();
  try {
    const text = "line
sep
para";
    h.engine.script([
      { raw: JSON.stringify({ type: "message_start", message: { role: "assistant", content: [] } }) + "\r\n" },
      { raw: JSON.stringify({ type: "message_end", message: { role: "assistant", content: [{ type: "text", text }], stopReason: "stop" } }) + "\r\n" },
    ]);
    const id = await admitted(h.client, "go");
    await receiptState(h.client, id, "finished");
    const end = h.ctrl.events.find((e) => e.type === "message-end" && e.role === "assistant");
    assert.equal(end.content[0].text, text);
    assert.equal(Object.keys(h.ctrl.evidence.unknownEvents).length, 0);
    assert.equal(h.ctrl.evidence.dropped.lines, 0);
    collect(h);
  } finally {
    await h.close();
  }
});

// Synthetic events for the Transcript alone.
const target = { conversation: "pi-c", branch: "main", execution: "exec-1", controllerGeneration: 1 };
const ev = (sequence, type, fields = {}, streamEpoch = "E1") => ({
  version: 2, kind: "event", id: `${streamEpoch}.${sequence}`, target, sequence, streamEpoch, request: null, entry: null, type, contentIndex: null,
  updateMode: "none", content: [], visibility: "permitted-visible", createdAt: "2026-10-04T12:00:00.000Z", stop: null, message: null, part: null, lastPart: null, role: null, ...fields,
});
const text = (t, block = 0) => ({ type: "text", text: t, block, fragment: 0, lastFragment: true });
const end = (sequence, message, part, lastPart, content, epoch) => ev(sequence, "message-end", { message, role: "assistant", entry: `${message}.e`, part, lastPart, updateMode: "replace", content }, epoch);

test("E3: a multipart final, two blocks, null request correlation and duplicate delivery", async () => {
  const v = new Transcript();
  v.load([], { streamEpoch: "E1", fromSequence: 1, quiet: true });
  const stream = [
    ev(1, "message-start", { message: "m1", role: "assistant" }),
    end(2, "m1", 0, false, [text("part zero")]),
    end(3, "m1", 1, true, [text("part one", 1)]),
    ev(4, "message-start", { message: "m2", role: "assistant" }),
    end(5, "m2", 0, true, [text("block A", 0), text("block B", 1)]),
  ];
  for (const e of stream) {
    assert.equal(v.event(e), null);
    assert.equal(v.event(structuredClone(e)), null, "an identical repeat is dropped");
  }
  assert.ok(stream.every((e) => e.request === null));
  assert.deepEqual(v.lines(), ["assistant: part zero\npart one", "assistant: block A\nblock B"]);
  assert.equal(v.reconcile, false);
  assert.equal(v.event({ ...stream[4], content: [text("changed")] }), "conflict", "a repeated ID with other bytes reconciles");
  assert.equal(v.reconcile, true);

  // Live: a final message of 70 blocks crosses the 64-block part limit.
  const h = await started();
  try {
    const blocks = Array.from({ length: 70 }, (_, i) => ({ type: "text", text: `b${i}` }));
    h.engine.script([{ emit: { type: "message_start", message: { role: "assistant", content: [] } } }, { emit: { type: "message_end", message: { role: "assistant", content: blocks, stopReason: "stop" } } }]);
    const { view } = await follower(h);
    const id = await admitted(h.client, "many blocks");
    await receiptState(h.client, id, "finished");
    const ends = h.ctrl.events.filter((e) => e.type === "message-end" && e.role === "assistant");
    assert.deepEqual(ends.map((e) => [e.part, e.lastPart]), [[0, false], [1, true]]);
    await until(() => view.events.some((e) => e.type === "run-settled"), 2000, "the settle");
    assert.equal(view.lines().at(-1), "assistant: " + blocks.map((b) => b.text).join("\n"));
    collect(h);
  } finally {
    await h.close();
  }
});

test("E4: a page read after message_end but before its entry is persisted: marker at the seam, re-read after run-settled, each message once", async () => {
  const h = await started();
  try {
    h.engine.holdPersist();
    h.engine.script([{ text: "one", stop: "stop" }, { pause: "p1" }, { text: "two", stop: "stop" }]);
    h.engine.arm("p1");
    const id = await admitted(h.client, "count");
    await h.engine.waitPaused("p1");
    const { client, view } = await follower(h);
    assert.equal(view.seam.replay, "unavailable");
    assert.equal(view.seam.quiet, false);
    assert.equal(view.reconcile, true);
    assert.deepEqual(view.lines(), [...PAGE, RECONCILE_MARKER], "the unpersisted messages are missing, and the seam says so");
    h.engine.flushPersist();
    h.engine.resume("p1");
    await receiptState(h.client, id, "finished");
    await until(() => view.loads >= 2 && !view.loading && !view.reconcile, 4000, "the re-read after run-settled");
    assert.equal(view.lastReload, "settled");
    const lines = view.lines();
    assert.deepEqual(lines, [...PAGE, "user: count", "assistant: one", "assistant: two"]);
    once(lines);
    client.close();
    collect(h);
  } finally {
    await h.close();
  }
});

test("E4: a gap or a new epoch also reconciles; nothing is concatenated across a gap", async () => {
  const entry = (id, role, t) => ({ version: 2, kind: "entry", id: `n.${id}:0`, message: id, role, content: [text(t)], part: 0, lastPart: true });
  let page = { entries: [], seam: { replay: "unavailable", streamEpoch: "E1", fromSequence: 1, quiet: true } };
  const listeners = new Set();
  const stub = {
    observes: 0,
    on: (fn) => listeners.add(fn),
    observe: async () => {
      stub.observes += 1;
      return { outcome: "observing", data: { page: { entries: page.entries, hasMore: false, nextCursor: null }, seam: page.seam } };
    },
  };
  const push = (event) => listeners.forEach((fn) => fn({ type: "push", kind: "event", event }));
  const v = new Transcript({ client: stub });
  await v.reload("initial");
  push(ev(1, "message-start", { message: "m1", role: "assistant" }));
  push(ev(2, "text-delta", { message: "m1", role: "assistant", content: [text("hel")] }));
  page = { entries: [entry("x1", "assistant", "hello world")], seam: { replay: "unavailable", streamEpoch: "E1", fromSequence: 5, quiet: true } };
  push(ev(4, "text-delta", { message: "m1", role: "assistant", content: [text("WORLD")] }));
  assert.equal(v.reason, "gap");
  assert.ok(!v.lines().some((l) => l.includes("WORLD")), "no concatenation across the gap");
  await until(() => !v.loading && !v.reconcile, 2000, "the re-read after the gap");
  assert.deepEqual(v.lines(), ["assistant: hello world"]);

  page = { entries: page.entries, seam: { replay: "unavailable", streamEpoch: "E2", fromSequence: 2, quiet: true } };
  push(ev(1, "message-start", { message: "n1", role: "assistant" }, "E2"));
  assert.equal(v.reason, "epoch");
  push(ev(2, "message-start", { message: "n2", role: "assistant" }, "E2"));
  push(end(3, "n2", 0, true, [text("after the epoch")], "E2"));
  await until(() => !v.loading && !v.reconcile, 2000, "the re-read after the epoch");
  const lines = v.lines();
  assert.deepEqual(lines, ["assistant: hello world", "assistant: after the epoch"]);
  once(lines);
  assert.equal(stub.observes, 3, "one read per reconcile, none repeated");
});

test("E5: an unknown native event gives no client event; evidence records its type and bytes; the terminal count goes up", async () => {
  const h = await started();
  try {
    const term = new Terminal({ client: h.client });
    const events = h.client.events.length;
    const line = { type: "brand_new_event", payload: "x".repeat(40) };
    h.engine.emit(line);
    await until(() => term.unknownCount === 1, 2000, "the count");
    h.engine.emit({ type: "another_new_one" });
    await until(() => term.unknownCount === 2, 2000, "the second count");
    assert.equal(h.client.events.length, events, "no client event");
    assert.equal(h.ctrl.evidence.unknownEvents.brand_new_event.count, 1);
    assert.equal(h.ctrl.evidence.unknownEvents.brand_new_event.bytes, Buffer.byteLength(JSON.stringify(line)));
    assert.match(term.frame[0], /unknown events: 2/);
    collect(h);
  } finally {
    await h.close();
  }
});

test("E6: a tool result delayed across a pause and a reconnect is reconciled without a manual refresh", async () => {
  const h = await started();
  try {
    h.engine.script([{ tool: { id: "call-e6", name: "bash", args: {}, result: "late result", hold: true } }, { pause: "p2" }, { text: "after", stop: "stop" }]);
    h.engine.arm("tool:call-e6");
    h.engine.arm("p2");
    const { client, view } = await follower(h);
    const id = await admitted(h.client, "slow tool");
    await h.engine.waitPaused("tool:call-e6");
    await until(() => view.lines().some((l) => l.includes("[tool bash running]")), 2000, "the running call");
    client.close();
    await until(() => client.closed, 2000, "the disconnect");
    h.engine.resume("tool:call-e6");
    await h.engine.waitPaused("p2");
    await client.connect();
    await until(() => view.loads >= 2 && !view.loading, 2000, "the read on reconnect");
    h.engine.resume("p2");
    await receiptState(h.client, id, "finished");
    await until(() => view.lastReload === "settled" && !view.loading && !view.reconcile, 4000, "the settled re-read");
    const lines = view.lines();
    assert.deepEqual(lines, [...PAGE, "user: slow tool", "assistant: [tool-call bash {}]", "tool: [tool-result] late result", "assistant: after"]);
    once(lines);
    assert.notEqual(view.lastReload, "manual");
    client.close();
    collect(h);
  } finally {
    await h.close();
  }
});

test("E7: the terminal renders the same stream as the library client, as observer and then as controller, and submits only as controller", async () => {
  const h = await started({ clients: 2 });
  try {
    const lib = new Transcript({ client: h.client });
    await loaded(lib);
    const term = new Terminal({ client: h.clients[1], rows: 40 });
    await loaded(term.transcript);
    const body = () => term.frame.slice(1, 1 + term.transcript.lines().join("\n").split("\n").length);
    const id = await admitted(h.client, "first");
    await receiptState(h.client, id, "finished");
    await until(() => term.transcript.events.some((e) => e.type === "run-settled"), 2000, "settle at the terminal");
    assert.deepEqual(term.transcript.lines(), lib.lines());
    term.render();
    assert.deepEqual(body(), lib.lines().join("\n").split("\n").map(visible));
    await term.key("blocked\r");
    assert.equal(term.status, NOT_CONTROLLER);
    assert.equal(prompts(h.engine).length, 1);
    term.composer = "";
    await term.key("\x14");
    await until(() => h.clients[1].isController, 2000, "terminal control");
    await term.key("second\r");
    await until(() => prompts(h.engine).length === 2, 2000, "the terminal's prompt");
    assert.equal(prompts(h.engine)[1].message, "second");
    await receiptState(h.client, term.lastReceipt.id, "finished");
    await until(() => term.transcript.events.filter((e) => e.type === "run-settled").length === 2 && lib.events.filter((e) => e.type === "run-settled").length === 2, 2000, "both settles");
    assert.deepEqual(term.transcript.lines(), lib.lines());
    assert.deepEqual(term.transcript.lines().slice(-2), ["user: second", "assistant: ok"]);
    collect(h);
  } finally {
    await h.close();
  }
});

test("terminal: engine control characters are made visible; a lost connection refuses submit", async () => {
  const h = await started();
  try {
    const term = new Terminal({ client: h.client });
    h.engine.script([{ text: "\x1b]52;c;ZXZpbA==\x07 bell‮", stop: "stop" }]);
    await term.key("go\r");
    await receiptState(h.client, term.lastReceipt.id, "finished");
    await until(() => term.transcript.lines().some((l) => l.includes("bell")), 2000, "the reply");
    const shown = term.frame.find((l) => l.includes("bell"));
    assert.equal(shown, "assistant: ^[]52;c;ZXZpbA==^G bell<U+202E>");
    assert.ok(!term.frame.join("").includes("\x1b"), "no raw ESC in the frame");
    h.client.close();
    await until(() => h.client.closed, 2000, "the disconnect");
    assert.match(term.status, /disconnected/);
    term.composer = "x";
    const r = await term.submit();
    assert.equal(r.reply.refusal, "channel");
    assert.equal(prompts(h.engine).length, 1);
  } finally {
    await h.close();
  }
});

test("terminal: outcome unknown is shown as such, with no resend offer, and nothing is resent", async () => {
  const listeners = new Set();
  let calls = 0;
  const stub = {
    closed: false, connection: { id: "conn-1" }, binding: { state: "active", controllerConnection: "conn-1" }, isController: true,
    on: (fn) => listeners.add(fn),
    observe: async () => ({ outcome: "refused:channel", refusal: "channel" }),
    prompt: async () => {
      calls += 1;
      return { outcome: "outcome-unknown", refusal: null, display: OUTCOME_UNKNOWN, reason: "disconnected", request: null, receipt: null };
    },
  };
  const term = new Terminal({ client: stub });
  await term.key("held\r");
  assert.equal(term.status, `prompt: ${OUTCOME_UNKNOWN}`);
  for (const fn of listeners) fn({ type: "outcome-unknown", request: "req-1", operation: "prompt", reason: "disconnected", display: OUTCOME_UNKNOWN });
  for (const fn of listeners) fn({ type: "push", kind: "receipt", receipt: { id: "receipt-1", state: "admitted" }, outcomeUnknown: true });
  assert.ok(term.notices.includes(`prompt: ${OUTCOME_UNKNOWN}`));
  assert.ok(term.notices.includes(`prompt receipt-1: ${OUTCOME_UNKNOWN}`));
  await term.key("\r");
  assert.equal(calls, 1, "an empty Enter sends nothing; nothing is resent");
  assert.equal(term.composer, "");
  assert.ok(!term.frame.some((l) => /resend|retry/i.test(l)));
});

// A controller-side stub that records every prompt.
function promptStub() {
  const sent = [];
  const stub = {
    closed: false, connection: { id: "conn-1" }, binding: { state: "active", controllerConnection: "conn-1" }, isController: true,
    on: () => {},
    observe: async () => ({ outcome: "refused:channel", refusal: "channel" }),
    prompt: async (text) => (sent.push(text), { outcome: "admitted", refusal: null, receipt: { id: `receipt-${sent.length}`, state: "admitted" } }),
  };
  return { stub, sent };
}

test("terminal: text after Enter in the same input chunk starts the next message; it never joins the one submitted", async () => {
  {
    const { stub, sent } = promptStub();
    const term = new Terminal({ client: stub });
    await term.key("first\rsecond\r");
    assert.deepEqual(sent, ["first", "second"]);
    assert.equal(term.composer, "");
  }
  {
    const { stub, sent } = promptStub();
    const term = new Terminal({ client: stub });
    await term.key("abc\rdef");
    assert.deepEqual(sent, ["abc"]);
    assert.equal(term.composer, "def", "the rest waits in the composer");
  }
});

test("terminal: a paste-start marker split right after its ESC still opens the paste; the Enter inside it never submits", async () => {
  for (const cut of [1, 2, 3, 4, 5]) {
    const { stub, sent } = promptStub();
    const term = new Terminal({ client: stub });
    const input = `${PASTE_START}a\rb${PASTE_END}`;
    await term.key(input.slice(0, cut));
    await term.key(input.slice(cut));
    assert.deepEqual(sent, [], `cut ${cut}: nothing sent`);
    assert.equal(term.composer, "a\rb", `cut ${cut}: the paste is literal`);
  }
  // A lone Escape has no action: it waits, then is consumed with what follows.
  const { stub, sent } = promptStub();
  const term = new Terminal({ client: stub });
  await term.key("x\x1b");
  await term.key("\r");
  assert.deepEqual(sent, [], "ESC then CR is Alt-Enter, a newline");
  assert.equal(term.composer, "x\n");
});

test("terminal: invisible and bidi characters are made visible; head, status and notice lines stay one line", () => {
  assert.equal(visible("a\u061cb\u200bc\u2060d\ufeffe\u{e0041}f"), "a<U+061C>b<U+200B>c<U+2060>d<U+FEFF>e<U+E0041>f");
  assert.equal(visible("\u{1F469}\u200d\u{1F4BB}"), "\u{1F469}\u200d\u{1F4BB}", "ZWJ sequences pass");
  const { stub } = promptStub();
  const term = new Terminal({ client: stub });
  term.status = "one\ntwo";
  term.notices.push("three\nfour");
  term.render();
  assert.ok(term.frame.includes("one^Jtwo"), "status");
  assert.ok(term.frame.includes("three^Jfour"), "notice");
});

test("every record these fixtures produced is a valid CHAT-01 record (E5: no record fails the schema)", () => {
  assert.ok(records.length > 100, `only ${records.length} records`);
  const kinds = [...new Set(records.map((r) => r.kind))].sort();
  for (const k of ["binding", "connection", "event", "receipt", "request"]) assert.ok(kinds.includes(k), `no ${k} records`);
  const script = `
import json, sys
from jsonschema import Draft202012Validator, FormatChecker
s = json.load(open(sys.argv[1]))
bad = []
for i, r in enumerate(json.load(sys.stdin)):
    v = Draft202012Validator({"$defs": s["$defs"], "$ref": "#/$defs/" + r["kind"]}, format_checker=FormatChecker())
    for e in v.iter_errors(r):
        bad.append(f"{i} {r['kind']}: {e.message[:200]}")
        break
print(json.dumps(bad))
`;
  const run = spawnSync("python3", ["-c", script, join(REPO, "docs", "plans", "chat-01", "contracts.schema.json")], { input: JSON.stringify(records), encoding: "utf8", maxBuffer: 1 << 30 });
  assert.equal(run.status, 0, run.stderr);
  assert.deepEqual(JSON.parse(run.stdout), [], `kinds checked: ${kinds.join(", ")}`);
});