// The runner against a real broker on a real socket, with a fake adapter.
// The broker runs in this process; the runner is a child, as the host
// starts it (minus the PID namespace, which packages/seat tests).

import { test } from "node:test";
import assert from "node:assert/strict";
import { spawn, spawnSync } from "node:child_process";
import { once } from "node:events";
import { existsSync, mkdirSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs";
import { open } from "node:fs/promises";
import { createServer } from "node:net";
import { join } from "node:path";
import { Store } from "../../bus/src/store.mjs";
import { Broker, BusError } from "../../bus/src/broker.mjs";
import { serve } from "../../bus/src/server.mjs";
import { EXIT, founderCheck, landingRequest, stateBlock, turnRequest, unslopCounts, unslopLog, unslopNotes } from "../src/runner.mjs";
import { typedTools } from "../src/tools.mjs";
import { resolvedFor, scratch } from "./helpers.mjs";

const RUNNER = new URL("../src/runner.mjs", import.meta.url).pathname;
const FAKE = new URL("./fixtures/fake-adapter.mjs", import.meta.url).pathname;
const businesses = {
  demo: {
    id: "demo",
    human: "jason",
    arbiters: { technical: "pm", delivery: "pm" },
    launch: { by: "pm", instances: ["coder"], models:{}, limits:{} },
    roles: {
      pm: { authority: { withinRole: ["message.send", "role.launch"], crossRole: [] } },
      coder: { authority: { withinRole: ["message.send"], crossRole: [] } },
      // Row 132 (#1628): the chief of staff's grants and the assistant. Since row 139 (Q169) the
      // chief may end the assistant; with no thread run holding a claim the end refuses not-held.
      // Row 155 (#1649) adds row 141's instance.name for set_name.
      chief: { definition: "chief-of-staff", authority: { withinRole: ["message.send", "role.end", "organization.change", "instance.name"], crossRole: [] } },
      assistant: { definition: "assistant", authority: { withinRole: ["message.send"], crossRole: [] } },
    },
  },
};

test("founderCheck: founder variables, then a needed service without a usable token", () => {
  const policy = { needs: ["gitea"], credentials: [{ service: "gitea", state: "valid" }] };
  assert.equal(founderCheck(policy, {}), null);
  assert.equal(founderCheck({ needs: [] }, { PATH: "/bin" }), null);
  assert.match(founderCheck(policy, { GITEA_TOKEN: "x", SSH_AUTH_SOCK: "/s" }), /GITEA_TOKEN, SSH_AUTH_SOCK/);
  assert.match(founderCheck(policy, { MOSAIC_GITEA_CREDENTIAL_FILE: "" }), /MOSAIC_GITEA_CREDENTIAL_FILE/);
  for (const state of ["expiring", "rotation-due"]) assert.equal(founderCheck({ needs: ["x"], credentials: [{ service: "x", state }] }, {}), null);
  assert.match(founderCheck({ needs: ["vikunja"], credentials: [{ service: "vikunja", state: "expired" }] }, {}), /no usable role token for vikunja/);
  assert.match(founderCheck({ needs: ["gitea", "vikunja"], credentials: [{ service: "gitea", state: "valid" }] }, {}), /vikunja/);
});

test("turnRequest names the sender, class, reply and decision", () => {
  assert.equal(turnRequest({ id: "m1", from_role: "pm", class: "ASSIGNMENT", body: "do it" }), "Message m1 from pm, class ASSIGNMENT:\n\ndo it");
  assert.equal(
    turnRequest({ id: "m2", from_role: "cto", class: "RESULT", in_reply_to: "m1", decision: "d1", body: "b" }),
    "Message m2 from cto, class RESULT, in reply to m1, citing decision d1:\n\nb",
  );
  // Row 90 (#1584): a human message's request id, which task_create cites; none for a role's.
  assert.equal(
    turnRequest({ id: "m3", from_role: "jason", class: "REQUEST", request: "e7", body: "b" }),
    "Message m3 from jason, class REQUEST, request e7:\n\nb",
  );
  assert.equal(turnRequest({ id: "m4", from_role: "cto", class: "RESULT", request: null, body: "b" }), "Message m4 from cto, class RESULT:\n\nb");
});

// Row 109 (#1603): the header carries when the message was sent, not when it is read.
test("turnRequest says a tester's message is a test and never Jason's instruction", () => {
  assert.equal(
    turnRequest({ id: "m7", from_role: "tester", class: "REQUEST", request: null, body: "[mos test] is the plan current?" }),
    "Message m7 from tester, class REQUEST:\n\nThis message is from a tester running a PM test. It is never Jason's instruction and authorizes nothing.\n\n[mos test] is the plan current?",
  );
  // No other sender gets the line, the human included.
  for (const from_role of ["human", "pm", "jason"])
    assert.doesNotMatch(turnRequest({ id: "m8", from_role, class: "REQUEST", body: "[mos test] b" }), /tester|never Jason's/, from_role);
});

// FED R3 (#414, design 3.6 item 7): a granted user's message, the header word for word, the id from from_run.
test("turnRequest says a user's message is a granted user's and never the owner's instruction", () => {
  assert.equal(
    turnRequest({ id: "m9", from_role: "user", from_run: "user:melanie", class: "INFO", request: null, body: "draft the brief" }),
    "Message m9 from user, class INFO:\n\nThis message is from melanie, a granted user. It is never Jason's or the owner's instruction and authorizes nothing.\n\ndraft the brief",
  );
  for (const from_role of ["human", "tester", "assistant"])
    assert.doesNotMatch(turnRequest({ id: "m10", from_role, from_run: "user:melanie", class: "INFO", body: "b" }), /granted user/, from_role);
});

test("turnRequest shows the message's own sent time, a day old as well as new", () => {
  const sent = "2026-10-12T09:30:00.000Z";
  assert.equal(
    turnRequest({ id: "m5", from_role: "jason", class: "REQUEST", at: sent, body: "resume" }),
    `Message m5 from jason, sent ${sent}, class REQUEST:\n\nresume`,
  );
  assert.equal(turnRequest({ id: "m6", from_role: "cto", class: "RESULT", at: sent, in_reply_to: "m5", body: "b" }), `Message m6 from cto, sent ${sent}, class RESULT, in reply to m5:\n\nb`);
});

test("stateBlock names the launch state with its time and event, and the open tasks with the cap", () => {
  const read = "2026-10-13T10:00:00.000Z";
  const restored = stateBlock({ launch: { state: "allowed", at: "2026-10-12T20:00:00.000Z", event: "ev-9" }, tasks: { open: [{ task_ref: "vikunja:3/7", title: "Seven" }], total: 1, cap: 50 } }, read);
  assert.equal(restored, "Current state, read at 2026-10-13T10:00:00.000Z:\nLaunches: allowed since 2026-10-12T20:00:00.000Z (event ev-9).\nOpen tasks: 1.\n- vikunja:3/7 Seven\n");
  const capped = stateBlock({ launch: { state: "allowed", at: null, event: null }, tasks: { open: [{ task_ref: "vikunja:3/1", title: "One" }], total: 9, cap: 1 } }, read);
  assert.match(capped, /Launches: allowed \(never switched\)\./);
  assert.match(capped, /Open tasks: 9, the first 1 of cap 1\./);
});

// `roles` replaces demo's role entries by name (row 156 round 2 gives the coder git.merge.protected).
async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy, usage = null, limits = null, roles = null, tester = false, processing = null } = {}) {
  const dir = scratch(t);
  const store = new Store(dir);
  let demo = launchBy ? { ...businesses.demo, launch: { ...businesses.demo.launch, by: launchBy } } : businesses.demo;
  if (tester) demo = { ...demo, vars: { ...demo.vars, "bus.tester": true } };
  if (roles) demo = { ...demo, roles: { ...demo.roles, ...roles } };
  const broker = new Broker({ store, businesses: { demo }, organization, usage, limits, ...(processing ? { processing } : {}) });
  const brokerSocket = join(dir, "bus.sock");
  const server = await serve({ broker, path: brokerSocket });
  t.after(async () => {
    await server.close().catch(() => {});
    store.close();
  });
  const runDir = join(dir, "run");
  const bundle = join(runDir, "bundle");
  mkdirSync(bundle, { recursive: true, mode: 0o700 });
  const adapter = join(dir, "adapter.sh");
  writeFileSync(adapter, `exec '${process.execPath}' '${FAKE}'\n`);
  writeFileSync(join(bundle, "prompt.md"), "# test\n");
  writeFileSync(join(bundle, "policy.json"), JSON.stringify({ harness: "pi", workspace: join(dir, "ws"), tools: [], typed: ["send_message", "launch"], network: "none", needs: [], credentials: [], ...policy }));
  writeFileSync(join(bundle, "tools.json"), JSON.stringify(tools ?? [
    { name: "send_message", verb: "message.send", action: "message.send" },
    { name: "launch", verb: "launch", action: "role.launch" },
  ]));
  mkdirSync(join(dir, "ws"));
  const launches = [];
  const launchSocket = join(dir, "launch.sock");
  const launchServer = createServer((s) => {
    s.once("data", (b) => {
      const r = JSON.parse(b);
      launches.push(r);
      s.end(JSON.stringify(r.instance === "coder" ? { ok: false, error: "capacity-full" } : { ok: true, result: { run: "new-run" } }) + "\n");
    });
  });
  await new Promise((r) => launchServer.listen(launchSocket, r));
  t.after(() => launchServer.close());
  writeFileSync(join(runDir, "session.json"), JSON.stringify({
    business: "demo",
    instance: as,
    run: `${as}-run`,
    launchedBy: "pm",
    harness: "pi",
    provider: "fake",
    model: "fake-model",
    brokerSocket,
    launchSocket,
    workspace: join(dir, "ws"),
    sessionDir: join(runDir, "session"),
    adapter,
    bundle: { prompt: join(bundle, "prompt.md"), policy: join(bundle, "policy.json"), tools: join(bundle, "tools.json") },
    toolSocket: join(runDir, "tools.sock"),
    turnMarker: join(runDir, "turn.done"),
    turnEnd: "marker",
    harnessEnv: {},
    pollInterval: 50,
    turnTimeout: 20,
    ...session,
  }));
  const cap = broker.bindLaunch({ business: "demo", role: as, run: `${as}-run`, harness: "pi" });
  const pm = broker.bindLaunch({ business: "demo", role: "pm", run: "pm-run", harness: "pi" });
  const call = (c, verb, args = {}) => broker.request(c, { verb, args });
  call(pm, "role.claim");
  const human = broker.bindHuman({ business: "demo", human: "jason", via: "cli", outsideAgent: true });
  return { dir, runDir, store, broker, server, cap, pm, human, call, launches, pids: join(dir, "pids") };
}

// A runner that hasn't exited after a minute is killed, so a test that
// expected an exit fails on the code instead of hanging.
function startRunner(runDir, cap, env = {}, input = cap === null ? "" : JSON.stringify({ cap }) + "\n") {
  const child = spawn(process.execPath, [RUNNER, runDir], {
    stdio: ["pipe", "ignore", "pipe"],
    env: { PATH: process.env.PATH, ...env },
  });
  let stderr = "";
  child.stderr.on("data", (b) => (stderr += b));
  child.stdin.on("error", () => {});
  child.stdin.end(input);
  const clock = setTimeout(() => child.kill("SIGKILL"), 60_000);
  const exited = once(child, "exit").then(([code, signal]) => {
    clearTimeout(clock);
    return { code, signal, stderr };
  });
  return { child, exited, stderr: () => stderr };
}

async function until(fn, ms = 15_000) {
  const end = Date.now() + ms;
  for (;;) {
    const v = fn();
    if (v) return v;
    if (Date.now() > end) throw new Error("timed out waiting");
    await new Promise((r) => setTimeout(r, 25));
  }
}

const replies = (ctx) => () => {
  const got = ctx.call(ctx.pm, "message.receive");
  ctx.inbox.push(...got);
  return ctx.inbox.length ? ctx.inbox : null;
};

test("a message becomes a turn, the answer goes back as a RESULT, SIGTERM releases and exits 0", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" });
  const [reply] = await until(replies(ctx));
  assert.equal(reply.class, "RESULT");
  assert.equal(reply.from_role, "coder");
  assert.equal(reply.from_run, "coder-run");
  assert.equal(reply.in_reply_to, sent.id);
  const sentAt = ctx.store.get("SELECT at FROM messages WHERE id=?", sent.id).at;
  assert.match(reply.body, new RegExp(`^echo: Current state, read at \\S+:\nLaunches: allowed \\(never switched\\)\\.\nOpen tasks: 0\\.\n\nMessage ${sent.id} from pm, sent ${sentAt}, class REQUEST:\n\nhello$`));
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  const err = join(ctx.runDir, "turns", "0001.stderr");
  assert.equal(statSync(err).mode & 0o777, 0o600);
  assert.equal(readFileSync(err, "utf8"), "fake adapter: fake-model\n");
  assert.equal(statSync(join(ctx.runDir, "tools.sock")).mode & 0o777, 0o600);
  r.child.kill("SIGTERM");
  const out = await r.exited;
  assert.equal(out.code, EXIT.stopped, out.stderr);
  assert.match(out.stderr, /runner: demo\/coder claimed by run coder-run/);
  assert.equal(ctx.store.get("SELECT op FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").op, "release");
  assert.ok(!existsSync(join(ctx.runDir, "tools.sock")));
});

test("agent replies to RESULT end unless they are REQUEST; ordinary INFO and human and tester follow-ups still get answers", async t => {
  const ctx = await setup(t, { tester: true });
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const seed = ctx.call(ctx.pm, "message.send", { to: "coder", body: "first question", class: "REQUEST" });
  const anchor = await until(() => ctx.store.get("SELECT id FROM messages WHERE from_run='coder-run' AND in_reply_to=? AND class='RESULT'", seed.id));
  const info = ctx.call(ctx.pm, "message.send", { to: "coder", body: "noted", class: "INFO", in_reply_to: anchor.id });
  const ordinary = ctx.call(ctx.pm, "message.send", { to: "coder", body: "ordinary relay", class: "INFO" });
  const linked = ctx.call(ctx.pm, "message.send", { to: "coder", body: "linked relay", class: "INFO", in_reply_to: seed.id });
  const request = ctx.call(ctx.pm, "message.send", { to: "coder", body: "follow-up", class: "REQUEST", in_reply_to: anchor.id });
  const human = ctx.call(ctx.human, "message.send", { to: "coder", body: "human follow-up", in_reply_to: anchor.id });
  const tester = ctx.broker.bindTester({ business: "demo" });
  const testMessage = ctx.call(tester, "message.send", { to: "coder", body: "tester follow-up", in_reply_to: anchor.id });
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", testMessage.id));
  const answers = ctx.store.all("SELECT in_reply_to,class FROM messages WHERE from_run='coder-run' ORDER BY seq");
  assert.deepEqual(answers.map(m => [m.in_reply_to, m.class]), [[seed.id, "RESULT"], [ordinary.id, "RESULT"], [linked.id, "RESULT"], [request.id, "RESULT"], [human.id, "RESULT"], [testMessage.id, "RESULT"]]);
  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE in_reply_to=?", info.id).n, 0);
});

test("managed empty answer formatting does not mark model output", async t => {
  const ctx = await setup(t, {session: {proxyFirstTurn: true, turnEnd: "exit"}});
  writeFileSync(join(ctx.runDir, "proxy-first-turn-enabled"), "enabled\n");
  writeFileSync(join(ctx.dir, "adapter.sh"), "printf '\\n\\n'\n");
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, "message.send", {to:"coder", body:"empty", class:"REQUEST"});
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  assert.equal(existsSync(join(ctx.runDir, "proxy-first-output")), false);
});

test("SIGKILL of the model child keeps base failure receipts through the real adapter", async t => {
  const records = [];
  for (const managed of [false, true]) {
    const ctx = await setup(t, {session:{harness:"claude-code", turnEnd:"exit", adapter:new URL("../../../adapters/claude/adapter.sh", import.meta.url).pathname}});
    const sessionFile = join(ctx.runDir, "session.json");
    const session = JSON.parse(readFileSync(sessionFile, "utf8"));
    const bin = join(ctx.dir, "model-fixture");
    const pidFile = join(ctx.dir, "model.pid");
    writeFileSync(bin, `#!${process.execPath}\nrequire('node:fs').writeFileSync(${JSON.stringify(pidFile)}, String(process.pid));\nif (process.argv.includes('stream-json')) console.log(JSON.stringify({type:'stream_event',event:{type:'content_block_delta',delta:{type:'text_delta',text:'partial'}}}));\nsetInterval(()=>{}, 1000);\n`, {mode:0o755});
    session.harnessEnv = {MOSAIC_CLAUDE_BIN:bin, MOSAIC_THINKING:"low", MOSAIC_CLAUDE_SETTINGS:session.bundle.policy, MOSAIC_CLAUDE_MCP_CONFIG:session.bundle.tools};
    writeFileSync(sessionFile, JSON.stringify(session));
    if (managed) writeFileSync(join(ctx.runDir, "proxy-first-turn-enabled"), "enabled\n");
    ctx.inbox = [];
    const r = startRunner(ctx.runDir, ctx.cap);
    t.after(() => r.child.kill("SIGKILL"));
    await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
    const sent = ctx.call(ctx.pm, "message.send", {to:"coder", body:"kill fixture", class:"REQUEST"});
    await until(() => existsSync(pidFile));
    const pid = Number(readFileSync(pidFile, "utf8"));
    assert.ok(pid > 1);
    process.kill(pid, "SIGKILL");
    const [reply] = await until(replies(ctx));
    await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
    const stderr = readFileSync(join(ctx.runDir, "turns/0001.stderr"), "utf8");
    records.push({body:reply.body, klass:reply.class, stderr, log:r.stderr()});
    r.child.kill("SIGTERM"); await r.exited;
  }
  for (const record of records) {
    assert.equal(record.klass, "RESULT");
    assert.equal(record.body, "The turn failed: adapter exited 137");
    assert.match(record.stderr, /Killed|SIGKILL/);
    assert.match(record.log, /failed: adapter exited 137/);
  }
  assert.equal(records[1].body, records[0].body);
});

// Claude's private stream reader owns model activity; final result stdout alone
// (even a synthetic success) must not cancel the model watchdog. Other adapters
// still need the stdout fallback when no streaming producer is installed.
for (const [harness, exit, activity] of [['claude-code', 1, false], ['claude-code', 0, false], ['pi', 0, true], ['mock', 0, true]]) {
  test(`stdout activity fallback: ${harness}, adapter exit ${exit}`, async t => {
    const ctx = await setup(t, { session: { harness, turnEnd: 'exit',
      harnessEnv: activity ? {} : { MOSAIC_MODEL_ACTIVITY_SOURCE: 'adapter' },
    } });
    ctx.inbox = [];
    writeFileSync(join(ctx.dir, 'adapter.sh'), `printf 'API Error: 400 synthetic-result\\n'\nexit ${exit}\n`);
    // An ambient declaration cannot suppress the fallback for other adapters.
    const r = startRunner(ctx.runDir, ctx.cap, { MOSAIC_MODEL_ACTIVITY_SOURCE: 'adapter' });
    t.after(async () => { r.child.kill('SIGTERM'); await r.exited; });
    const reply = await ask(ctx);
    if (exit) assert.match(reply.body, /The turn failed: adapter exited 1/);
    else assert.equal(reply.body, 'API Error: 400 synthetic-result');
    const marker = join(ctx.runDir, 'proxy-first-output');
    assert.equal(existsSync(marker), activity);
    if (activity) assert.equal(readFileSync(marker, 'utf8'), 'output\n');
  });
}

// Row 109 (#1603): each turn reads the state when it starts. The fake adapter echoes the request,
// so a reply shows the block the turn was given.
const ask = async (ctx, body = "what now") => {
  const before = ctx.inbox.length;
  ctx.call(ctx.pm, "message.send", { to: "coder", body, class: "REQUEST" });
  await until(() => replies(ctx)() && ctx.inbox.length > before);
  return ctx.inbox.at(-1);
};
const snapshot = (i, over = {}, read = "2026-10-12T12:00:00.000Z") => ({
  task_ref: `vikunja:3/${i}`,
  updated: read,
  digest: "d".repeat(64),
  fields: { project_id: 3, title: `Task ${i}`, description: "", done: false, bucket: 11, ...over },
  via: "board",
  read_at: read,
});

test("launches revoked and then restored: the turn's block reads allowed and gives the restore time", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  ctx.call(ctx.human, "launch.revoke");
  assert.match((await ask(ctx)).body, /\nLaunches: revoked since \S+ \(event \S+\)\.\n/);
  ctx.call(ctx.human, "launch.restore");
  const restored = ctx.store.get("SELECT id,at FROM events WHERE kind='launch.restored'");
  assert.ok((await ask(ctx)).body.includes(`\nLaunches: allowed since ${restored.at} (event ${restored.id}).\n`));
});

test("a task closed on the tracker is missing from the next turn's open list", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  ctx.broker.recordTask({ business: "demo", snapshots: [snapshot(1), snapshot(2)] });
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const first = (await ask(ctx)).body;
  assert.match(first, /\nOpen tasks: 2\.\n- vikunja:3\/1 Task 1\n- vikunja:3\/2 Task 2\n/);
  ctx.broker.recordTask({ business: "demo", snapshots: [snapshot(2, { done: true }, "2026-10-12T13:00:00.000Z")] });
  const second = (await ask(ctx)).body;
  assert.match(second, /\nOpen tasks: 1\.\n- vikunja:3\/1 Task 1\n\nMessage /);
  assert.ok(!second.includes("vikunja:3/2"));
});

test("a message read a day after it was sent shows the time it was sent", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "resume the launch", class: "REQUEST" });
  const sentAt = ctx.store.get("SELECT at FROM messages WHERE id=?", sent.id).at;
  // The broker's clock jumps a day before the runner reads: the header keeps the original time.
  const now = Date.now;
  Date.now = () => now() + 86_400_000;
  t.after(() => (Date.now = now));
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const [reply] = await until(replies(ctx));
  assert.ok(reply.body.includes(`Message ${sent.id} from pm, sent ${sentAt}, class REQUEST:`));
  assert.ok(Date.parse(reply.at) - Date.parse(sentAt) >= 86_000_000);
});

// Row 115 (#1609): the human closing the role's decision gives the role a turn. The runner is unchanged;
// the header cites the decision and the close's human.input answer as the request.
test("a decision the human resolves becomes a turn citing the decision and the answer", async (t) => {
  const ctx = await setup(t);
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const d = ctx.call(ctx.cap, "decision.raise", {
    action: "deploy",
    target: "release-1",
    question: "Ship release-1?",
    options: [{ key: "yes", text: "Allow" }, { key: "no", text: "Decline" }],
    recommendation: "yes",
    blocking: false,
  });
  assert.equal(d.route_to, "human");
  ctx.call(ctx.human, "decision.resolve", { id: d.id, choice: "yes" });
  const m = ctx.store.get("SELECT id,at FROM messages WHERE class='DECISION' AND decision=?", d.id);
  const answer = ctx.store.get("SELECT id FROM events WHERE kind='human.input' AND subject=?", d.id).id;
  const [reply] = await until(() => {
    const got = ctx.call(ctx.human, "message.receive");
    return got.length ? got : null;
  });
  assert.equal(reply.from_role, "coder");
  assert.equal(reply.class, "RESULT");
  assert.equal(reply.in_reply_to, m.id);
  assert.ok(
    reply.body.endsWith(
      `\nMessage ${m.id} from human, sent ${m.at}, class DECISION, citing decision ${d.id}, request ${answer}:\n\n` +
        `Decision ${d.id} on deploy was resolved by jason: yes (Allow).`,
    ),
    reply.body,
  );
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", m.id));
});

test("a failed state read fails the turn: no adapter runs, the sender hears why, the message is read", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const request = ctx.broker.request.bind(ctx.broker);
  ctx.broker.request = (cap, req) => {
    if (req.verb === "state") throw new BusError("state-unavailable");
    return request(cap, req);
  };
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "plan the week", class: "REQUEST" });
  const [reply] = await until(replies(ctx));
  assert.equal(reply.class, "RESULT");
  assert.equal(reply.in_reply_to, sent.id);
  assert.equal(reply.body, "The turn failed: could not read the current state: state-unavailable");
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  assert.ok(!existsSync(join(ctx.runDir, "turns", "0001.stderr")));
  assert.match(r.stderr(), /turn 1 for message \S+ failed: could not read the current state: state-unavailable/);
});

test("a SIGTERM before the claim stops the runner with exit 0 and no claim", async (t) => {
  const ctx = await setup(t);
  // session.json becomes a FIFO: main opens it right after its handlers go
  // in, so the open on this side returns once a SIGTERM would be caught.
  const file = join(ctx.runDir, "session.json");
  const body = readFileSync(file);
  rmSync(file);
  assert.equal(spawnSync("mkfifo", ["-m", "600", file]).status, 0);
  const child = spawn(process.execPath, [RUNNER, ctx.runDir], { stdio: ["pipe", "ignore", "pipe"], env: { PATH: process.env.PATH } });
  t.after(() => child.kill("SIGKILL"));
  let stderr = "";
  child.stderr.on("data", (b) => (stderr += b));
  const exited = once(child, "exit");
  const fifo = await open(file, "w");
  child.kill("SIGTERM");
  await fifo.write(body);
  await fifo.close();
  child.stdin.end(JSON.stringify({ cap: ctx.cap }) + "\n");
  const [code, signal] = await exited;
  assert.deepEqual([code, signal], [EXIT.stopped, null], stderr);
  assert.match(stderr, /runner: stopped before claim/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});

test("typed tools carry the runner's capability; launch goes to the host's launch socket", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL send_message {"to":"pm","body":"side note","class":"INFO"}' });
  await until(() => replies(ctx)() && ctx.inbox.length >= 2);
  const [info, result] = ctx.inbox;
  assert.equal(info.class, "INFO");
  assert.equal(info.body, "side note");
  assert.equal(info.from_run, "coder-run");
  assert.equal(JSON.parse(result.body).id, info.id);
  ctx.inbox.length = 0;
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL launch {"instance":"coder"}' });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL launch {"instance":"reviewer"}' });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL resolve_decision {"id":"x"}' });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: 'TOOL send_message {"to":"nobody","body":"x"}' });
  await until(() => replies(ctx)() && ctx.inbox.length >= 4);
  assert.deepEqual(ctx.inbox.map((m) => m.body), ["refused: capacity-full", '{"run":"new-run"}', "refused: unknown-tool", "refused: unknown-role"]);
  assert.deepEqual(ctx.launches, [{ cap: ctx.cap, instance: "coder" }, { cap: ctx.cap, instance: "reviewer" }]);
});

// Row 132 (#1628, REQ-ROLE-2, REQ-COS-1, REQ-LIMIT-1): the chief's two tools reach the broker by their verbs
// with the arguments the session gave, and a refusal comes back with the broker's code.
// Row 134 (#1669): organization_revision reaches the broker's organization.revision and answers the revision alone.
test("end_run, organization_revision and change_organization reach the broker with the tool's arguments, and refusals keep their codes", async (t) => {
  const written = [];
  const organization = (...args) => {
    written.push(args);
    if (args[2].revision === "e".repeat(64)) throw Object.assign(new Error("stale"), { code: "revision-conflict" });
    return args[1] === "limits" ? { revision: "c".repeat(64), limits: { total: 7 }, tokens: {} } : { revision: "b".repeat(64) };
  };
  const tools = typedTools({ resolved: resolvedFor(t, "chief") }).filter((x) => ["end_run", "organization_revision", "change_organization"].includes(x.name));
  assert.deepEqual(tools.map((x) => x.name), ["end_run", "organization_revision", "change_organization"]);
  const ctx = await setup(t, { as: "chief", organization, tools, policy: { typed: tools.map((x) => x.name) } });
  ctx.inbox = [];
  ctx.call(ctx.broker.bindLaunch({ business: "demo", role: "coder", run: "coder-run", harness: "pi" }), "role.claim");
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const send = (name, args) => ctx.call(ctx.pm, "message.send", { to: "chief", body: `TOOL ${name} ${JSON.stringify(args)}` });
  const change = { revision: "a".repeat(64), limits: { total: 3 } };
  send("end_run", { role: "coder", reason: "stuck on the same test for an hour" });
  send("end_run", { role: "chief", reason: "my own run" });
  send("end_run", { role: "assistant", reason: "the user's seat" });
  send("organization_revision", {});
  send("organization_revision", { surprise: 1 });
  send("change_organization", change);
  send("change_organization", { ...change, surprise: 1 });
  send("change_organization", { ...change, revision: "e".repeat(64) });
  await until(() => replies(ctx)() && ctx.inbox.length >= 8);
  assert.deepEqual(ctx.inbox.map((m) => m.body), [
    '{"revoked":true}',
    "refused: role-end-refused",
    "refused: not-held",
    '{"revision":"' + "c".repeat(64) + '"}',
    "refused: invalid-request",
    '{"revision":"' + "b".repeat(64) + '"}',
    "refused: invalid-request",
    "refused: revision-conflict",
  ]);
  assert.deepEqual({ ...ctx.store.get("SELECT op, reason, by FROM role_claims WHERE role='coder' ORDER BY rowid DESC LIMIT 1") }, { op: "revoke", reason: "stuck on the same test for an hour", by: "chief-run" });
  // The writer saw the tool's arguments and the chief's name; the unknown key never reached it.
  assert.deepEqual(written.map((w) => [w[0], w[1], w[2], w[3]]), [
    ["demo", "limits", {}, "chief-of-staff.chief"],
    ["demo", "limits.set", change, "chief-of-staff.chief"],
    ["demo", "limits.set", { ...change, revision: "e".repeat(64) }, "chief-of-staff.chief"],
  ]);
});

// Row 155 (#1649, REQ-MSG-2): set_name reaches name.set with the session's arguments, and the broker's
// refusals come back with their codes.
test("set_name reaches the broker's name.set with the tool's arguments, and refusals keep their codes", async (t) => {
  const tools = typedTools({ resolved: resolvedFor(t, "chief") }).filter((x) => x.name === "set_name");
  assert.deepEqual(tools.map((x) => x.name), ["set_name"]);
  const ctx = await setup(t, { as: "chief", tools, policy: { typed: tools.map((x) => x.name) } });
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const send = (args) => ctx.call(ctx.pm, "message.send", { to: "chief", body: `TOOL set_name ${JSON.stringify(args)}` });
  send({ instance: "coder", name: "rex" });
  send({ instance: "coder", name: "pm" });
  await until(() => replies(ctx)() && ctx.inbox.length >= 2);
  assert.deepEqual(ctx.inbox.map((m) => m.body), [
    '{"instance":"coder","name":"rex"}',
    "refused: name-is-instance",
  ]);
  assert.deepEqual({ ...ctx.store.get("SELECT instance, name, by FROM instance_names ORDER BY seq DESC LIMIT 1") }, { instance: "coder", name: "rex", by: "chief-run" });
});

test("a RESULT gets no automatic reply; failed turns reply with the reason", async (t) => {
  const ctx = await setup(t, { session: { turnTimeout: 1 } });
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const first = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hi" });
  const res = ctx.call(ctx.pm, "message.send", { to: "coder", body: "thanks", class: "RESULT", in_reply_to: first.id });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "FAIL" });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "NOMARKER" });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "SLEEP" });
  await until(() => replies(ctx)() && ctx.inbox.length >= 4);
  assert.deepEqual(ctx.inbox.map((m) => m.body.replace(/^echo: .*/s, "echo")), [
    "echo",
    "The turn failed: adapter exited 3",
    "The turn failed: pi exited before the turn ended (no turn marker)",
    "The turn failed: turn passed its 1 s wall clock",
  ]);
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", res.id));
});

// Row 146 (#1640): the bus's notice to launch.by is a RESULT from _bus. The runner takes one turn
// on it, sends no automatic reply and marks it read. Any other class would send a reply to _bus,
// which the bus refuses; the runner would log the failure and skip the read, leaving the notice
// unread for good.
test("the bus's launch notice is one turn with no reply, and is not taken again", async (t) => {
  const ctx = await setup(t, { as: "chief", launchBy: "chief" });
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='chief' AND op='claim'"));
  const sent = ctx.call(ctx.human, "message.send", { to: "coder", body: "is the build green?" });
  assert.equal(sent.waiting.told, true);
  const notice = ctx.store.get("SELECT * FROM messages WHERE from_role='_bus' AND in_reply_to=?", sent.id);
  assert.equal(notice.to_role, "chief");
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", notice.id));
  // A later message is one more turn, so the notice wasn't taken again in between.
  const later = ctx.call(ctx.pm, "message.send", { to: "chief", body: "status?" });
  const [reply] = await until(replies(ctx));
  assert.equal(reply.in_reply_to, later.id);
  assert.deepEqual(readdirSync(join(ctx.runDir, "turns")).sort(), ["0001.stderr", "0002.stderr"]);
  assert.equal(ctx.store.get("SELECT count(*) n FROM deliveries WHERE message=? AND op='delivered'", notice.id).n, 1);
  assert.equal(ctx.store.get("SELECT count(*) n FROM deliveries WHERE message=? AND op='read'", notice.id).n, 1);
  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE in_reply_to=? AND from_run='chief-run'", notice.id).n, 0);
  assert.doesNotMatch(r.stderr(), /reply to .* failed/);
});

test("SIGTERM during a turn kills the turn's process group and still exits 0", async (t) => {
  // A long wall clock, so only the stop path can end the turn in time.
  const ctx = await setup(t, { session: { turnTimeout: 600 } });
  const r = startRunner(ctx.runDir, ctx.cap, { FAKE_PIDS: ctx.pids });
  t.after(() => r.child.kill("SIGKILL"));
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "SLEEP" });
  const pid = Number(await until(() => existsSync(ctx.pids) && readFileSync(ctx.pids, "utf8").trim()));
  const started = Date.now();
  r.child.kill("SIGTERM");
  const out = await Promise.race([r.exited, new Promise((res) => setTimeout(() => res({ code: "still running" }), 8000))]);
  assert.equal(out.code, EXIT.stopped, out.stderr);
  assert.ok(Date.now() - started < 8000);
  await until(() => {
    try {
      process.kill(pid, 0);
      return false;
    } catch {
      return true;
    }
  }, 8000);
});

test("founder credentials stop before the claim (20)", async (t) => {
  const ctx = await setup(t);
  const out = await startRunner(ctx.runDir, ctx.cap, { GH_TOKEN: "x" }).exited;
  assert.equal(out.code, EXIT.founder);
  assert.match(out.stderr, /REQ-CRED-2.*GH_TOKEN/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
  const ctx2 = await setup(t, { policy: { needs: ["gitea"], credentials: [{ service: "gitea", state: "expired" }] } });
  const out2 = await startRunner(ctx2.runDir, ctx2.cap).exited;
  assert.equal(out2.code, EXIT.founder);
  assert.match(out2.stderr, /no usable role token for gitea/);
});

test("a refused claim exits 21; an ended run's capability exits 22", async (t) => {
  const ctx = await setup(t);
  const other = ctx.broker.bindLaunch({ business: "demo", role: "coder", run: "other-run", harness: "pi" });
  ctx.call(other, "role.claim");
  const out = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(out.code, EXIT.claim, out.stderr);
  assert.match(out.stderr, /role.claim refused: role-already-held/);
  const ctx2 = await setup(t);
  ctx2.broker.endLaunch({ business: "demo", run: "coder-run", reason: "stopped", exitCode: 0 });
  assert.equal((await startRunner(ctx2.runDir, ctx2.cap).exited).code, EXIT.session);
});

test("the launch ending under a running session exits 22", async (t) => {
  const ctx = await setup(t);
  const r = startRunner(ctx.runDir, ctx.cap);
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  ctx.broker.endLaunch({ business: "demo", run: "coder-run", reason: "killed", exitCode: null });
  const out = await r.exited;
  assert.equal(out.code, EXIT.session, out.stderr);
  assert.match(out.stderr, /session stopped working: unauthenticated/);
});

test("a broker that stays unreachable exits 23 after brokerRetries polls", async (t) => {
  const ctx = await setup(t, { session: { brokerRetries: 3 } });
  const r = startRunner(ctx.runDir, ctx.cap);
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  await ctx.server.close();
  const out = await r.exited;
  assert.equal(out.code, EXIT.broker, out.stderr);
  assert.match(out.stderr, /broker unreachable after 3 tries/);
});

test("a broker that is down at the claim exits 23, not 21", async (t) => {
  const ctx = await setup(t);
  await ctx.server.close();
  const out = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(out.code, EXIT.broker, out.stderr);
  assert.match(out.stderr, /role.claim got no answer from the broker/);
});

test("no capability, or a malformed one, on stdin exits 2", async (t) => {
  const ctx = await setup(t);
  assert.equal((await startRunner(ctx.runDir, null).exited).code, EXIT.usage);
  assert.equal((await startRunner(ctx.runDir, "not-hex").exited).code, EXIT.usage);
  assert.equal((await startRunner(join(ctx.dir, "nope"), ctx.cap).exited).code, EXIT.usage);
  // A valid capability inside an over-long line: the 4096-byte cap refuses it.
  const long = await startRunner(ctx.runDir, ctx.cap, {}, JSON.stringify({ cap: ctx.cap, pad: "x".repeat(5000) }) + "\n").exited;
  assert.equal(long.code, EXIT.usage, long.stderr);
  assert.match(long.stderr, /stdin too large/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});

test("a missing or malformed policy exits 2 before the claim", async (t) => {
  const ctx = await setup(t);
  const policy = join(ctx.runDir, "bundle", "policy.json");
  writeFileSync(policy, "{");
  const bad = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(bad.code, EXIT.usage, bad.stderr);
  assert.match(bad.stderr, /^runner: /m);
  rmSync(policy);
  const missing = await startRunner(ctx.runDir, ctx.cap).exited;
  assert.equal(missing.code, EXIT.usage, missing.stderr);
  assert.match(missing.stderr, /runner: ENOENT/);
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});


test("REQ-LIMIT-4 runner brackets its actual harness turn before the reply and read receipt", async t => {
  const calls = [];
  let ctx;
  ctx = await setup(t, { usage: { collect: async (session, phase, turn) => {
    calls.push({ phase, turn, run: session.run, started: existsSync(join(ctx.runDir, "turns/0001.stderr")) });
    return { recorded: true };
  } } });
  ctx.inbox = [];
  const runner = startRunner(ctx.runDir, ctx.cap);
  t.after(() => runner.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "account this turn", class: "REQUEST" });
  await until(replies(ctx));
  runner.child.kill("SIGTERM");
  assert.equal((await runner.exited).code, 0);
  assert.deepEqual(calls, [
    { phase: 'begin', turn: sent.id, run: 'coder-run', started: false },
    { phase: 'end', turn: sent.id, run: 'coder-run', started: true },
  ]);
});

// Row 125 (#1619, REQ-LIMIT-4, decision 150 Q4): a run at its limit stops at the end of its turn.
// The turn's RESULT and read receipt land first; the next receive is refused not-holder, the
// runner exits as the session gone, and the claim row names the limit.
test("REQ-LIMIT-4 a runner at its limit stops at the end of its turn: RESULT and read first, then exit 22 and a limit-check revoke", async t => {
  const asked = [];
  const hit = { path: ["tokens", "instances", "coder"], limit: 100, count: 120, unknownTurns: 0 };
  const ctx = await setup(t, {
    usage: { collect: async () => ({ recorded: true }) },
    limits: { mustStop: async (business, subject) => {
      asked.push({ business, subject });
      return { stop: true, check: "token-limit", hit };
    } },
  });
  ctx.inbox = [];
  const runner = startRunner(ctx.runDir, ctx.cap);
  t.after(() => runner.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "one turn", class: "REQUEST" });
  const out = await runner.exited;
  assert.equal(out.code, EXIT.session, out.stderr);
  assert.match(out.stderr, /runner: session stopped working: not-holder/);
  const [reply] = await until(replies(ctx));
  assert.equal(reply.class, "RESULT");
  assert.equal(reply.in_reply_to, sent.id);
  assert.ok(ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  assert.deepEqual(asked, [{ business: "demo", subject: { instance: "coder", provider: null, model: null, account: null } }]);
  const row = ctx.store.get("SELECT op, by, reason, holder_run FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1");
  assert.deepEqual({ ...row }, { op: "revoke", by: "limit-check", holder_run: "coder-run",
    reason: "limit: token-limit tokens.instances.coder limit 100 count 120" });
  // The RESULT and the read receipt are both older than the revoke.
  const revokedAt = ctx.store.get("SELECT at FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").at;
  assert.ok(ctx.store.get("SELECT at FROM messages WHERE id=?", reply.id).at <= revokedAt);
});

// Row 156 (#1651): the session file's landing. The adapter records each turn's request, so a test
// sees what the turn was given and in which order.
test("landingRequest is the fixed template, with the decision and task ref and nothing else", () => {
  assert.equal(
    landingRequest({ decision: "d1", task_ref: "vikunja:3/7" }),
    "Landing from the host, decision d1, task vikunja:3/7:\n\nYour merge decision d1 for vikunja:3/7 is approved and unused. Land it now with land_work, task_ref vikunja:3/7 and decision d1. The reply only says the landing started; the gate's outcome comes later as a RESULT from the bus. If land_work refuses decision-consumed, the landing already happened.",
  );
});

function recording(ctx) {
  const file = join(ctx.dir, "requests");
  writeFileSync(join(ctx.dir, "adapter.sh"), `printf '%s\\0' "$MOSAIC_REQUEST" >> '${file}'\nexec '${process.execPath}' '${FAKE}'\n`);
  return () => (existsSync(file) ? readFileSync(file, "utf8").split("\0").slice(0, -1) : []);
}

test("each landing entry is one turn before any message, with the state block and the fixed request; it sends nothing and reads nothing", async (t) => {
  const landing = [{ decision: "d1", task_ref: "vikunja:3/7" }, { decision: "d2", task_ref: "vikunja:3/8" }];
  const ctx = await setup(t, { session: { landing } });
  const requests = recording(ctx);
  ctx.inbox = [];
  // The message waits from before the start; both landing turns still come first.
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" });
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const [reply] = await until(replies(ctx));
  assert.equal(reply.in_reply_to, sent.id);
  const got = requests();
  assert.equal(got.length, 3);
  for (const [i, e] of landing.entries())
    assert.match(got[i], new RegExp(`^Current state, read at \\S+:\nLaunches: allowed \\(never switched\\)\\.\nOpen tasks: 0\\.\n\n${landingRequest(e).replace(/[.*+?^${}()|[\]\\/]/g, "\\$&")}$`));
  assert.match(got[2], new RegExp(`Message ${sent.id} from pm, sent \\S+, class REQUEST:\n\nhello$`));
  r.child.kill("SIGTERM");
  const out = await r.exited;
  assert.equal(out.code, EXIT.stopped, out.stderr);
  assert.match(out.stderr, /runner: landing turn 1 for decision d1 \(vikunja:3\/7\): ok\n[\s\S]*runner: landing turn 2 for decision d2 \(vikunja:3\/8\): ok\n/);
  // Each landing turn asks usage.begin and usage.end with its decision id. These decisions aren't
  // held in this broker, so both refuse not-recipient (land.test.mjs has a held one); a message
  // turn with no usage configured refuses usage-not-configured instead.
  for (const [i, e] of landing.entries())
    assert.match(out.stderr, new RegExp(`runner: usage baseline unavailable: not-recipient\\n(?:(?!runner: landing turn)[\\s\\S])*?runner: usage outcome unavailable: not-recipient\\n(?:(?!runner: landing turn)[\\s\\S])*?runner: landing turn ${i + 1} for decision ${e.decision} `));
  // The landing turns sent nothing and marked nothing read: one RESULT, one read, both for the message.
  assert.deepEqual(ctx.store.all("SELECT in_reply_to FROM messages WHERE from_run='coder-run'").map((m) => m.in_reply_to), [sent.id]);
  assert.deepEqual(ctx.store.all("SELECT message FROM deliveries WHERE op='read'").map((d) => d.message), [sent.id]);
  assert.deepEqual(readdirSync(join(ctx.runDir, "turns")).sort(), ["0001.stderr", "0002.stderr", "0003.stderr"]);
});

test("an empty landing takes no landing turn: the message is turn 1", async (t) => {
  const ctx = await setup(t, { session: { landing: [] } });
  const requests = recording(ctx);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" });
  await until(replies(ctx));
  r.child.kill("SIGTERM");
  const out = await r.exited;
  assert.equal(out.code, EXIT.stopped, out.stderr);
  assert.doesNotMatch(out.stderr, /landing/);
  const got = requests();
  assert.equal(got.length, 1);
  assert.match(got[0], new RegExp(`Message ${sent.id} from pm`));
});

// Row 156 round 2 (Filbert's F1 in 29033): while the runner is in a landing turn, the launcher's
// landing pass must not stop it for a second decision resolved meanwhile. The pass stops a run only
// when landRelease answers released (packages/cli/tests/launcher.test.mjs has the launcher's side),
// so this holds a real landing turn open on a real broker and asks landRelease as the pass does.
test("a run in a landing turn isn't released for a second held decision until its landing turns end and it has read its messages", async (t) => {
  const MERGE = "git.merge.protected";
  const ctx = await setup(t, { roles: { coder: { authority: { withinRole: ["message.send"], crossRole: [MERGE] } } } });
  const merge = (target) => ({
    action: MERGE, target, question: `Merge ${target}?`,
    options: [{ key: "yes", text: "Land it" }, { key: "no", text: "Not yet" }], recommendation: "yes", blocking: false,
  });
  // d1 is raised and resolved before the run starts, by an earlier run of the instance.
  const earlier = ctx.broker.bindLaunch({ business: "demo", role: "coder", run: "coder-0", harness: "pi" });
  ctx.call(earlier, "role.claim");
  const d1 = ctx.call(earlier, "decision.raise", merge("vikunja:1/2"));
  assert.equal(d1.route_to, "pm");
  ctx.call(ctx.pm, "decision.resolve", { id: d1.id, choice: "yes" });
  ctx.broker.endLaunch({ business: "demo", run: "coder-0", reason: "exited" });
  const file = join(ctx.runDir, "session.json");
  writeFileSync(file, JSON.stringify({ ...JSON.parse(readFileSync(file, "utf8")), landing: [{ decision: d1.id, task_ref: "vikunja:1/2" }] }));
  // The adapter's first turn, the landing turn, waits for the go file.
  const started = join(ctx.dir, "started");
  const go = join(ctx.dir, "go");
  writeFileSync(join(ctx.dir, "adapter.sh"), `touch '${started}'\nwhile [ ! -e '${go}' ]; do sleep 0.05; done\nexec '${process.execPath}' '${FAKE}'\n`);
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => existsSync(started));
  // d2 is resolved during the landing turn, so the landing pass lists it for the running holder.
  const d2 = ctx.call(ctx.cap, "decision.raise", merge("vikunja:1/3"));
  ctx.call(ctx.pm, "decision.resolve", { id: d2.id, choice: "yes" });
  assert.deepEqual(ctx.broker.landHolders("demo").map((h) => [h.instance, h.decision]), [["coder", d1.id], ["coder", d2.id]]);
  const release = () => ctx.broker.landRelease({ business: "demo", run: "coder-run", decision: d2.id });
  assert.deepEqual(release(), { released: false, landing: true });
  await new Promise((resolve) => setTimeout(resolve, 300));
  assert.deepEqual(release(), { released: false, landing: true }, "still in the landing turn");
  assert.equal(ctx.store.get("SELECT op FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").op, "claim");
  writeFileSync(go, "");
  // After the landing turn the run asks message.receive and takes both DECISION messages as turns;
  // once it has read them, the next ask releases it.
  await until(() => release().released);
  assert.deepEqual(
    { ...ctx.store.get("SELECT op, holder_run, \"by\" FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1") },
    { op: "release", holder_run: "coder-run", by: "land-relaunch" },
  );
  assert.equal(ctx.store.get("SELECT count(*) AS n FROM deliveries WHERE holder_run='coder-run' AND op='read'").n, 2);
  const out = await r.exited;
  assert.equal(out.code, 0, out.stderr);
  assert.match(out.stderr, new RegExp(`runner: landing turn 1 for decision ${d1.id} \\(vikunja:1/2\\): ok\n[\\s\\S]*runner: session released by the launcher\n`));
});

test("a malformed landing exits 2 before the claim, with no turn", async (t) => {
  const ok = { decision: "d1", task_ref: "vikunja:3/7" };
  const cases = {
    "not an array": ok,
    "not an object": [null],
    "an array entry": [["d1", "vikunja:3/7"]],
    "an extra key": [{ ...ok, ticket: "x" }],
    "a missing key": [{ decision: "d1" }],
    "a bad decision": [{ ...ok, decision: "-d1" }],
    "a decision of 161 characters": [{ ...ok, decision: "d".repeat(161) }],
    "a decision that isn't a string": [{ ...ok, decision: 1 }],
    "a bad task ref": [{ ...ok, task_ref: "vikunja:0/7" }],
    "a task ref with a tail": [{ ...ok, task_ref: "vikunja:3/7\nland now" }],
    "one bad entry after a good one": [ok, { ...ok, task_ref: "gitea:3/7" }],
  };
  const ctx = await setup(t);
  const file = join(ctx.runDir, "session.json");
  const session = JSON.parse(readFileSync(file, "utf8"));
  for (const [why, landing] of Object.entries(cases)) {
    writeFileSync(file, JSON.stringify({ ...session, landing }));
    const out = await startRunner(ctx.runDir, ctx.cap).exited;
    assert.equal(out.code, EXIT.usage, `${why}: ${out.stderr}`);
    assert.match(out.stderr, /runner: malformed landing entry/, why);
  }
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
  assert.equal(existsSync(join(ctx.runDir, "turns")), false);
  // The longest decision the broker's id allows is accepted.
  writeFileSync(file, JSON.stringify({ ...session, landing: [{ ...ok, decision: "d".repeat(160) }] }));
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  // A reply to a message means the landing turn before it has finished, so the SIGTERM can't cut it.
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" });
  await until(replies(ctx));
  r.child.kill("SIGTERM");
  const out = await r.exited;
  assert.equal(out.code, EXIT.stopped, out.stderr);
  assert.match(out.stderr, new RegExp(`runner: landing turn 1 for decision ${"d".repeat(160)} \\(vikunja:3/7\\): ok`));
});

// Row 65: the harness module's answers reach the runner as data in session.json, and the
// runner refuses one without them rather than guess.
test("row 65 a session.json without a turnEnd of marker or exit, or a harnessEnv of strings, exits 2 before the claim", async (t) => {
  const cases = {
    "no turnEnd": { turnEnd: undefined },
    "another turnEnd": { turnEnd: "agent_end" },
    "no harnessEnv": { harnessEnv: undefined },
    "a null harnessEnv": { harnessEnv: null },
    "an array harnessEnv": { harnessEnv: ["MOSAIC_THINKING=high"] },
    "a number in harnessEnv": { harnessEnv: { MOSAIC_THINKING: 3 } },
  };
  const ctx = await setup(t);
  const file = join(ctx.runDir, "session.json");
  const session = JSON.parse(readFileSync(file, "utf8"));
  for (const [why, change] of Object.entries(cases)) {
    writeFileSync(file, JSON.stringify({ ...session, ...change }));
    const out = await startRunner(ctx.runDir, ctx.cap).exited;
    assert.equal(out.code, EXIT.usage, `${why}: ${out.stderr}`);
    assert.match(out.stderr, /runner: session\.json has no turnEnd of marker or exit, or no harnessEnv of strings/, why);
  }
  assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined);
});

test("row 65 harnessEnv reaches the adapter, and an exit harness needs no turn marker", async (t) => {
  const ctx = await setup(t, { session: { turnEnd: "exit", harnessEnv: { MOSAIC_THINKING: "xhigh" } } });
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "ENV MOSAIC_THINKING" });
  ctx.call(ctx.pm, "message.send", { to: "coder", body: "NOMARKER" });
  await until(() => replies(ctx)() && ctx.inbox.length >= 2);
  assert.deepEqual(ctx.inbox.map((m) => m.body.trim()), ["env: xhigh", "half an answer"]);
});

// Row 66 (#1559): the harness's unslop check logs to <run>/unslop.jsonl. The fake adapter's
// UNSLOP body writes the events a real check would; pi-session and claude-session tests run the real
// extension and hooks.
const tell = (detail, source = "ms-unslop:7") => ({ ev: "flagged", clean: false, charsChecked: 40, findings: [{ rule: "word", source, detail, count: 1 }] });
const unslopAsk = (ctx, events, cls = "REQUEST") => ctx.call(ctx.pm, "message.send", { to: "coder", body: `UNSLOP ${JSON.stringify(events)}`, class: cls });

test("row 66 the adapter gets the run's unslop log, and a flagged reply's note rides that reply's RESULT, once", async (t) => {
  const ctx = await setup(t, { session: { harnessEnv: { MOSAIC_UNSLOP_LOG: "/elsewhere.jsonl" } } });
  ctx.inbox = [];
  // A thread run's sessionDir is elsewhere and the run has no session/; the log needs none.
  rmSync(join(ctx.runDir, "session"), { recursive: true, force: true });
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  const env = await ask(ctx, "ENV MOSAIC_UNSLOP_LOG");
  assert.equal(env.body, `env: ${join(ctx.runDir, "unslop.jsonl")}`);
  assert.equal(existsSync(join(ctx.runDir, "session")), false);
  unslopAsk(ctx, [{ ev: "checked", clean: true, charsChecked: 10 }, tell('banned word "delve" x1')]);
  await until(() => replies(ctx)() && ctx.inbox.length >= 2);
  assert.equal(ctx.inbox[1].body, 'unslop: 2\n\n[unslop: 1 reply of this session flagged since the last note: banned word "delve" x1. The session gets a correction notice on its next turn.]');
  const clean = await ask(ctx, "ENV MOSAIC_MODEL");
  assert.equal(clean.body, "env: fake-model");
  assert.match(r.stderr(), /runner: unslop: a reply flagged: banned word "delve" x1\n/);
});

test("row 66 a turn with no reply holds its note for the next reply; a broken check is told once", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  // A RESULT takes a turn and gets no reply, so its flagged reply waits.
  const res = unslopAsk(ctx, [tell('banned word "tapestry" x1', "ms-unslop:7")], "RESULT");
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", res.id));
  assert.equal(replies(ctx)(), null);
  unslopAsk(ctx, [{ ev: "broken_lists", reason: "lists.json invalid: empty file" }, tell('filler phrase "in order to" x1', "ms-unslop:23")]);
  await until(() => replies(ctx)() && ctx.inbox.length >= 1);
  assert.equal(ctx.inbox[0].body, "unslop: 2\n\n[unslop: the check is broken, so this session's replies go unchecked: lists.json invalid: empty file]\n" +
    '[unslop: 2 replies of this session flagged since the last note: banned word "tapestry" x1; filler phrase "in order to" x1. The session gets a correction notice on its next turn.]');
  // The next broken event isn't told again.
  unslopAsk(ctx, [{ ev: "skipped_broken", reason: "lists.json invalid: empty file" }]);
  await until(() => replies(ctx)() && ctx.inbox.length >= 2);
  assert.equal(ctx.inbox[1].body, "unslop: 1");
  assert.equal(r.stderr().match(/runner: unslop: the check is broken/g).length, 1);
  const counts = unslopCounts(unslopLog(ctx.runDir));
  assert.deepEqual(counts, { checked: 2, flagged: 2, patterns: { "ms-unslop:7": 1, "ms-unslop:23": 1 }, notices: 0, broken: "lists.json invalid: empty file" });
});

test("row 66 unslopNotes keeps a note until sent(), reads only whole lines, and unslopCounts reads a log after the session", (t) => {
  const dir = scratch(t);
  const path = join(dir, "unslop.jsonl");
  const notes = unslopNotes(path);
  notes.read();
  assert.equal(notes.note(), "");
  writeFileSync(path, JSON.stringify(tell('banned word "pivotal" x1')) + "\n" + '{"ev":"flagged","findings":[{"detail":"half');
  notes.read();
  const first = notes.note();
  assert.match(first, /1 reply of this session flagged since the last note: banned word "pivotal" x1\./);
  // A send that failed calls no sent(), so the note is still there.
  assert.equal(notes.note(), first);
  notes.sent();
  assert.equal(notes.note(), "");
  // The half line completes; it is read once, from where the last read stopped.
  writeFileSync(path, readFileSync(path, "utf8") + ' x","source":"ms-unslop:9"}]}\nnot json\n[1]\n' + JSON.stringify({ ev: "notice_injected" }) + "\n");
  notes.read();
  assert.match(notes.note(), /1 reply of this session flagged since the last note: half x\./);
  assert.deepEqual(unslopCounts(path), { checked: 2, flagged: 2, patterns: { "ms-unslop:7": 1, "ms-unslop:9": 1 }, notices: 1, broken: null });
  assert.deepEqual(unslopCounts(join(dir, "none.jsonl")), { checked: 0, flagged: 0, patterns: {}, notices: 0, broken: null });
  // A note is bounded however many findings a reply had.
  const many = unslopNotes(join(dir, "many.jsonl"));
  writeFileSync(join(dir, "many.jsonl"), Array.from({ length: 50 }, (_, i) => JSON.stringify(tell(`banned word "w${i}" x1`.padEnd(40, "x")))).join("\n") + "\n");
  many.read();
  assert.ok(many.note().length <= 600, many.note().length);
  assert.match(many.note(), /^\n\n\[unslop: 50 replies .*\.\.\.\. The session gets a correction notice on its next turn\.\]$/s);
  // Both parts together are bounded too, and a reason the runner makes itself is cut like one the
  // log gives: here the log path is a directory with a long name, so the read fails with EISDIR.
  const deep = join(dir, "d".repeat(250));
  mkdirSync(deep);
  const both = unslopNotes(deep);
  both.read();
  assert.match(both.note(), /^\n\n\[unslop: the check is broken, so this session's replies go unchecked: the runner can't read .*\.\.\.\]$/);
  assert.equal(both.note().length, 2 + 71 + 300);
  const both2 = unslopNotes(join(dir, "both.jsonl"));
  writeFileSync(join(dir, "both.jsonl"), [JSON.stringify({ ev: "broken_lists", reason: "b".repeat(400) }), ...Array.from({ length: 50 }, (_, i) => JSON.stringify(tell(`banned word "w${i}" x1`.padEnd(40, "x"))))].join("\n") + "\n");
  both2.read();
  assert.equal(both2.note().length, 600);
  assert.match(both2.note(), /^\n\n\[unslop: the check is broken, so this session's replies go unchecked: b{297}\.\.\.\]\n\[unslop: 50 replies .*\.\.\.\. The session gets a correction notice on its next turn\.\]$/s);
});

test("row 66 a clipped answer with both note parts stays under the bus's body limit, and the note under 600", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const events = [{ ev: "hook_failed", reason: "r".repeat(400) }, ...Array.from({ length: 30 }, (_, i) => tell(`banned word "w${i}" x1`.padEnd(60, "x")))];
  ctx.call(ctx.pm, "message.send", { to: "coder", body: `UNSLOPLONG ${JSON.stringify(events)}`, class: "REQUEST" });
  await until(() => replies(ctx)() && ctx.inbox.length >= 1);
  const body = ctx.inbox[0].body;
  const at = body.indexOf("\n[answer cut at 32000 characters]");
  assert.equal(at, 32000);
  const note = body.slice(at + "\n[answer cut at 32000 characters]".length);
  assert.ok(note.length <= 600, note.length);
  assert.ok(body.length <= 32768, body.length);
  assert.match(note, /^\n\n\[unslop: the check is broken, so this session's replies go unchecked: r+\.\.\.\]\n\[unslop: 30 replies .*\.\.\.\. The session gets a correction notice on its next turn\.\]$/s);
});

test("row 66 a refused send keeps the note: it rides the next RESULT the session sends", async (t) => {
  const ctx = await setup(t);
  ctx.inbox = [];
  // The broker refuses the session's first RESULT once, as it would a body it can't take.
  const request = ctx.broker.request.bind(ctx.broker);
  let refused = 0;
  ctx.broker.request = (cap, r) => {
    if (r.verb === "message.send" && r.args?.class === "RESULT" && refused++ === 0) throw new BusError("invalid-request");
    return request(cap, r);
  };
  const r = startRunner(ctx.runDir, ctx.cap);
  t.after(() => r.child.kill("SIGKILL"));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = unslopAsk(ctx, [tell('banned word "delve" x1')]);
  await until(() => r.stderr().includes(`runner: reply to ${sent.id} failed: invalid-request\n`));
  // The refused message was delivered and isn't read; its reply never reached the bus.
  assert.ok(!ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  const next = await ask(ctx, "ENV MOSAIC_MODEL");
  assert.equal(next.body, 'env: fake-model\n\n[unslop: 1 reply of this session flagged since the last note: banned word "delve" x1. The session gets a correction notice on its next turn.]');
  assert.equal(ctx.inbox.length, 1);
});

test('empty length continues in the same session, records usage and relays the answer once', async t => {
  const calls = [];
  let ctx;
  ctx = await setup(t, { usage: { collect: async (_session, phase, turn) => { const file = join(ctx.runDir, "fixture-count"); calls.push({phase, turn, attempts: existsSync(file) ? Number(readFileSync(file, "utf8")) : 0}); return {recorded:true}; } } });
  const fixture = new URL('./fixtures/citation-adapter.mjs', import.meta.url).pathname;
  writeFileSync(join(ctx.dir, 'adapter.sh'), `exec '${process.execPath}' '${fixture}'\n`);
  writeFileSync(join(ctx.runDir, 'fixture-steps.json'), JSON.stringify([
    {marker:{stopReason:'length',hasText:false,hasToolCall:false}}, {text:'recovered',marker:{stopReason:'stop',hasText:true,hasToolCall:false}},
  ]));
  const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, 'message.send', {to:'coder',body:'work',class:'REQUEST'});
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  const rows = ctx.store.all("SELECT * FROM messages WHERE from_role='coder' AND to_role='pm'");
  assert.equal(rows.length,1); assert.equal(rows[0].class,'RESULT'); assert.equal(rows[0].body,'recovered'); assert.equal(rows[0].in_reply_to,sent.id);
  assert.equal(Number(readFileSync(join(ctx.runDir,'fixture-count'),'utf8')),2);
  const requests = readFileSync(join(ctx.runDir,'fixture-requests.jsonl'),'utf8').trim().split('\n').map(JSON.parse);
  assert.match(requests[1], /output limit.*thinking/i); assert.match(requests[1], /answer or call a tool/i);
  assert.deepEqual(calls, [{phase:'begin',turn:sent.id,attempts:0},{phase:'end',turn:sent.id,attempts:2}]);
});


test('empty length exhaustion records one usage bracket, one chief notice and one trail event', async t => {
  const calls = []; let ctx;
  ctx = await setup(t, { usage: { collect: async (_s, phase, turn) => { const file = join(ctx.runDir, 'fixture-count'); calls.push({phase,turn,attempts:existsSync(file) ? Number(readFileSync(file,'utf8')) : 0}); return {recorded:true}; } } });
  const fixture = new URL('./fixtures/citation-adapter.mjs', import.meta.url).pathname;
  writeFileSync(join(ctx.dir,'adapter.sh'), `exec '${process.execPath}' '${fixture}'\n`);
  writeFileSync(join(ctx.runDir,'fixture-steps.json'), JSON.stringify(Array(3).fill({marker:{stopReason:'length',hasText:false,hasToolCall:false}})));
  const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm,'message.send',{to:'coder',body:'work',class:'REQUEST'});
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'",sent.id));
  assert.deepEqual(calls,[{phase:'begin',turn:sent.id,attempts:0},{phase:'end',turn:sent.id,attempts:3}]);
  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='coder' AND to_role='pm'").n,0);
  const notice = ctx.store.all("SELECT * FROM messages WHERE from_role='coder' AND to_role='chief'");
  assert.equal(notice.length,1); assert.equal(notice[0].class,'INFO'); assert.equal(notice[0].in_reply_to,sent.id);
  assert.match(notice[0].body,/output limit.*thinking/i);
  assert.equal(ctx.store.get("SELECT count(*) n FROM events WHERE subject=? AND kind='action.allowed'",sent.id).n,1);
});

test('chief does not reply to the failure notice, but replies to ordinary agent and prefixed human INFO', async t => {
  const { LENGTH_FAILURE_NOTICE_PREFIX } = await import('../src/failure-notice.mjs');
  const ctx = await setup(t,{as:'chief'});
  const coder = ctx.broker.bindLaunch({business:'demo',role:'coder',run:'coder-notifier',harness:'pi'});
  ctx.call(coder,'role.claim');
  const r = startRunner(ctx.runDir,ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='chief' AND op='claim'"));
  const seed = ctx.call(ctx.pm,'message.send',{to:'coder',body:'work',class:'REQUEST'});
  const notice = ctx.call(coder,'message.send',{to:'chief',body:LENGTH_FAILURE_NOTICE_PREFIX+'fixture',class:'INFO',in_reply_to:seed.id});
  const ordinary = ctx.call(coder,'message.send',{to:'chief',body:'ordinary information',class:'INFO',in_reply_to:seed.id});
  const human = ctx.call(ctx.human,'message.send',{to:'chief',body:LENGTH_FAILURE_NOTICE_PREFIX+'human information',class:'INFO',in_reply_to:seed.id});
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'",human.id));
  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='chief' AND in_reply_to=?",notice.id).n,0);
  for (const id of [ordinary.id,human.id]) assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='chief' AND class='RESULT' AND in_reply_to=?",id).n,1);
});


test('stop interrupts a length continuation and still closes usage', async t => {
  const calls = []; let ctx;
  ctx = await setup(t,{usage:{collect:async (_s,phase,turn) => { calls.push({phase,turn}); return {recorded:true}; }}});
  const fixture = new URL('./fixtures/citation-adapter.mjs',import.meta.url).pathname;
  writeFileSync(join(ctx.dir,'adapter.sh'),`exec '${process.execPath}' '${fixture}'\n`);
  writeFileSync(join(ctx.runDir,'fixture-steps.json'),JSON.stringify([{marker:{stopReason:'length',hasText:false,hasToolCall:false}},{sleep:true}]));
  const r = startRunner(ctx.runDir,ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm,'message.send',{to:'coder',body:'work',class:'REQUEST'});
  await until(() => existsSync(join(ctx.runDir,'fixture-count')) && readFileSync(join(ctx.runDir,'fixture-count'),'utf8') === '2');
  r.child.kill('SIGTERM'); assert.equal((await r.exited).code,EXIT.stopped);
  assert.deepEqual(calls,[{phase:'begin',turn:sent.id},{phase:'end',turn:sent.id}]);
  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='coder'").n,0);
});

test('managed event monitor exit 1 with empty length marker continues', async t => {
  const ctx = await setup(t);
  const fixture = new URL('./fixtures/citation-adapter.mjs',import.meta.url).pathname;
  writeFileSync(join(ctx.dir,'adapter.sh'),`exec '${process.execPath}' '${fixture}'\n`);
  writeFileSync(join(ctx.runDir,'proxy-first-turn-enabled'),'enabled\n');
  writeFileSync(join(ctx.runDir,'fixture-steps.json'),JSON.stringify([{marker:{stopReason:'length',hasText:false,hasToolCall:false},exitCode:1},{text:'recovered'}]));
  const r = startRunner(ctx.runDir,ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm,'message.send',{to:'coder',body:'work',class:'REQUEST'});
  const reply = await until(() => ctx.store.get("SELECT * FROM messages WHERE in_reply_to=? AND class='RESULT'",sent.id));
  assert.equal(reply.body,'recovered'); assert.equal(readFileSync(join(ctx.runDir,'fixture-count'),'utf8'),'2');
});

test('failure notice recognition excludes bus notices and non-agent senders', async () => {
  const { LENGTH_FAILURE_NOTICE_PREFIX } = await import('../src/failure-notice.mjs');
  const { isLengthFailureNotice } = await import('../src/runner.mjs');
  const notice = {class:'INFO',from_role:'coder',body:LENGTH_FAILURE_NOTICE_PREFIX+'fixture'};
  assert.equal(isLengthFailureNotice(notice),true);
  for (const from_role of ['human','tester','user','_bus']) assert.equal(isLengthFailureNotice({...notice,from_role}),false,from_role);
  assert.equal(isLengthFailureNotice({...notice,class:'REQUEST'}),false);
  assert.equal(isLengthFailureNotice({...notice,body:'ordinary information'}),false);
});

// Sol review r1 (#1852): the continuation checks, every attempt inside the m.id usage bracket.
const lengthAdapter = (ctx, steps) => {
  const fixture = new URL('./fixtures/citation-adapter.mjs', import.meta.url).pathname;
  writeFileSync(join(ctx.dir, 'adapter.sh'), `exec '${process.execPath}' '${fixture}'\n`);
  writeFileSync(join(ctx.runDir, 'fixture-steps.json'), JSON.stringify(steps));
};
const emptyLength = {marker:{stopReason:'length',hasText:false,hasToolCall:false}};
const attemptsOf = ctx => existsSync(join(ctx.runDir, 'fixture-count')) ? Number(readFileSync(join(ctx.runDir, 'fixture-count'), 'utf8')) : 0;
// The second state read is the first continuation's check; the first is the message's own read.
const onSecondState = (ctx, act) => {
  let reads = 0;
  const request = ctx.broker.request.bind(ctx.broker);
  ctx.broker.request = (cap, req) => { if (req.verb === 'state' && ++reads === 2) act(); return request(cap, req); };
};

test('a processing halt after an empty length turn starts no continuation, closes usage and leaves the message unread', async t => {
  let halted = false; const calls = []; let ctx;
  const processing = { read: () => ({ version: 1, halted, generation: '00000000-0000-4000-8000-000000000000' }) };
  ctx = await setup(t, { processing, usage: { collect: async (_s, phase, turn) => { calls.push({ phase, turn, attempts: attemptsOf(ctx) }); return { recorded: true }; } } });
  lengthAdapter(ctx, [emptyLength, { text: 'must not run' }]);
  onSecondState(ctx, () => { halted = true; });
  const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, 'message.send', { to: 'coder', body: 'work', class: 'REQUEST' });
  await until(() => /left unread: processing is halted/.test(r.stderr()));
  r.child.kill('SIGTERM'); await r.exited;
  assert.equal(attemptsOf(ctx), 1);
  assert.deepEqual(calls, [{ phase: 'begin', turn: sent.id, attempts: 0 }, { phase: 'end', turn: sent.id, attempts: 1 }]);
  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='coder'").n, 0);
  assert.equal(ctx.store.get("SELECT count(*) n FROM deliveries WHERE message=? AND op='read'", sent.id).n, 0);
  assert.equal(ctx.store.get("SELECT count(*) n FROM events WHERE subject=? AND kind='action.allowed'", sent.id).n, 0);
});

test('a failed continuation check after an empty length turn sends no RESULT and records the failure', async t => {
  const calls = []; let ctx;
  ctx = await setup(t, { usage: { collect: async (_s, phase, turn) => { calls.push({ phase, turn }); return { recorded: true }; } } });
  lengthAdapter(ctx, [emptyLength, { text: 'must not run' }]);
  onSecondState(ctx, () => { throw new BusError('state-unavailable'); });
  const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, 'message.send', { to: 'coder', body: 'work', class: 'REQUEST' });
  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
  assert.equal(attemptsOf(ctx), 1);
  assert.deepEqual(calls, [{ phase: 'begin', turn: sent.id }, { phase: 'end', turn: sent.id }]);
  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE from_role='coder' AND to_role='pm'").n, 0);
  const notice = ctx.store.all("SELECT * FROM messages WHERE from_role='coder' AND to_role='chief'");
  assert.equal(notice.length, 1); assert.match(notice[0].body, /on the turn\. No answer was sent\.$/);
  assert.equal(ctx.store.get("SELECT count(*) n FROM events WHERE subject=? AND kind='action.allowed'", sent.id).n, 1);
  assert.match(r.stderr(), /continuation check for .* failed: state-unavailable/);
});

test('an attempt that throws still closes the message usage bracket once', async t => {
  const calls = []; let ctx;
  ctx = await setup(t, { usage: { collect: async (_s, phase, turn) => { calls.push({ phase, turn }); return { recorded: true }; } } });
  lengthAdapter(ctx, [emptyLength, { text: 'must not run' }]);
  // runTurn opens turns/0002.stderr with "wx": an existing file throws before the adapter starts.
  onSecondState(ctx, () => writeFileSync(join(ctx.runDir, 'turns', '0002.stderr'), 'preexisting'));
  const r = startRunner(ctx.runDir, ctx.cap); t.after(() => r.child.kill('SIGKILL'));
  await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'"));
  const sent = ctx.call(ctx.pm, 'message.send', { to: 'coder', body: 'work', class: 'REQUEST' });
  const exit = await r.exited;
  assert.notEqual(exit.code, 0);
  assert.deepEqual(calls, [{ phase: 'begin', turn: sent.id }, { phase: 'end', turn: sent.id }]);
  assert.equal(attemptsOf(ctx), 1);
});

test('output policy is required and validated for new sessions while legacy metadata remains readable', async () => {
  const { harnessOk } = await import('../src/runner.mjs');
  const legacy = { turnEnd: 'marker', harnessEnv: {} };
  assert.equal(harnessOk(legacy), true);
  for (const outputPolicy of ['adapter', 'non-newline-stdout', 'stdout']) {
    assert.equal(harnessOk({ ...legacy, outputPolicyVersion: 1, outputPolicy }), true);
  }
  for (const session of [
    { ...legacy, outputPolicyVersion: 1 },
    { ...legacy, outputPolicyVersion: 1, outputPolicy: 'unknown' },
    { ...legacy, outputPolicyVersion: 2, outputPolicy: 'stdout' },
    { ...legacy, outputPolicy: 'stdout' },
  ]) assert.equal(harnessOk(session), false);
});

test('non-newline stdout policy only suppresses blank output for monitored proxy turns', async t => {
  for (const managed of [false, true]) {
    const ctx = await setup(t, { session: { turnEnd: 'exit', outputPolicyVersion: 1, outputPolicy: 'non-newline-stdout' } });
    if (managed) writeFileSync(join(ctx.runDir, 'proxy-first-turn-enabled'), 'enabled\n');
    ctx.inbox = [];
    writeFileSync(join(ctx.dir, 'adapter.sh'), "printf '\\n\\n'\n");
    const r = startRunner(ctx.runDir, ctx.cap);
    t.after(async () => { r.child.kill('SIGTERM'); await r.exited; });
    await ask(ctx);
    assert.equal(existsSync(join(ctx.runDir, 'proxy-first-output')), !managed);
    r.child.kill('SIGTERM'); await r.exited;
  }
});