test: cover conditional switch refusal, receipt conflict and bounded record polling

packages/daemon/test/expected-conversation.test.ts

@@ -1,3 +1,4 @@
+import fs from "node:fs";
 import { afterEach, expect, it, vi } from "vitest";
 import { Hono } from "hono";
 import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs";
@@ -22,7 +23,7 @@ import { transportRoutes } from "../src/routes/transport.js";
 const A = "11111111-1111-4111-8111-111111111111", B = "22222222-2222-4222-8222-222222222222";
 const cleanup: Array<() => unknown> = [];
 afterEach(() => { vi.restoreAllMocks(); for (const close of cleanup.splice(0).reverse()) close(); });
-function fixture(runtime: "claude-code" | "codex" = "claude-code") {
+function fixture(runtime: "claude-code" | "codex" = "claude-code", guarded = false) {
   const root = mkdtempSync(join(tmpdir(), "expected-conversation-")); cleanup.push(() => rmSync(root, { recursive: true, force: true }));
   const db = createDb(); migrate(db, ALL_MIGRATIONS); cleanup.push(() => db.close());
   const repo = new RigRepository(db), rig = repo.createRig("expected"), node = repo.addNode(rig.id, "team.worker", { runtime });
@@ -32,18 +33,22 @@ function fixture(runtime: "claude-code" | "codex" = "claude-code") {
   registry.updateBinding(node.id, { tmuxSession: sessionName, tmuxPane: "%1" });
   new SeatIdentityStore(db).upsert({ nodeId: node.id, verdict: "verified", evidenceSource: "pane_process", reason: null,
     evidence: { registeredPane: "%1", observedPid: 20, observedPanePid: 10, observedCommand: runtime, matchedLayer: 1 }, sessionName, observedAt: new Date().toISOString() });
-  let pasted = false;
+  let pasted = false, staged = "", composer = "";
+  const frame = () => ({ screen: ["Ready", "────────────────────", `❯ ${composer}`, "────────────────────", "? for shortcuts", ""].join("\n"), cursor: { x: 2 + composer.length, y: 2, width: 240, height: 60 }, inMode: false });
   let current = A, afterLoad: (() => void) | undefined, beforeKeys: (() => void) | undefined;
   const commands: string[] = [];
   const tmux = new TmuxAdapter(async command => {
-    commands.push(command); if (command.includes("load-buffer")) afterLoad?.(); if (command.includes("paste-buffer")) pasted = true; return "";
-  }, { writeFile: async () => {}, unlink: async () => {}, tmpName: () => join(root, "paste"), bufferName: () => "expected-test" });
+    commands.push(command); if (command.includes("load-buffer")) afterLoad?.(); if (command.includes("paste-buffer")) { pasted = true; composer = staged; } if (command.includes("send-keys") && command.includes("Enter")) composer = ""; return "";
+  }, { writeFile: async (_path, text) => { staged = text; }, unlink: async () => {}, tmpName: () => join(root, "paste"), bufferName: () => "expected-test" });
   vi.spyOn(tmux, "getPanePid").mockImplementation(async () => { if (pasted) beforeKeys?.(); return 10; });
   vi.spyOn(tmux, "getPaneCommand").mockResolvedValue(runtime === "codex" ? "codex" : "claude");
   vi.spyOn(tmux, "listPanes").mockResolvedValue([{ id: "%1", index: 0, cwd: root, width: 120, height: 40, active: true }]);
   vi.spyOn(tmux, "probeSession").mockResolvedValue({ state: "present" } as never);
   vi.spyOn(tmux, "isPaneInMode").mockResolvedValue(false);
-  vi.spyOn(tmux, "capturePaneContent").mockResolvedValue("❯ ");
+  vi.spyOn(tmux, "capturePaneContent").mockImplementation(async () => guarded ? frame().screen : "❯ ");
+  vi.spyOn(tmux, "captureComposerSnapshot").mockImplementation(async () => frame());
+  const guard = guarded ? new SeatDeliveryGuard(db, name => resolveGuardTarget(db, name)) : undefined;
+  if (guard) tmux.deliveryGuard = guard;
   const rows = () => [{ pid: 10, ppid: 1, command: "zsh" }, { pid: 20, ppid: 10,
     command: runtime === "codex" ? `codex resume ${current}` : `claude --session-id ${current}`, startedAt: "Thu Oct 29 00:00:00 2026" }];
   vi.spyOn(ProcessCensus.prototype, "list").mockImplementation(async () => rows());
@@ -63,7 +68,7 @@ function fixture(runtime: "claude-code" | "codex" = "claude-code") {
   }); app.route("/api/sessions", sessionAdminRoutes); app.route("/api/transport", transportRoutes());
   const expected = { nodeId: node.id, sessionName, runtime, nativeId: A };
   const readUrl = () => `/api/sessions/${sessionName}/generation-record-expected?` + new URLSearchParams({ expectedNodeId: node.id, expectedRuntime: runtime, expectedNativeId: A, sinceBytes: "0" });
-  return { root, db, registry, node, tmux, transport, app, expected, commands, store, records, readUrl, processes: rows,
+  return { root, db, registry, node, tmux, guard, transport, app, expected, commands, store, records, readUrl, processes: rows,
     replace: () => { current = B; }, onLoad: (fn: () => void) => { afterLoad = fn; }, beforeKeys: (fn: () => void) => { beforeKeys = fn; } };
 }
 
@@ -135,3 +140,85 @@ it("legacy sends retain their ordinary behavior", async () => {
   const f = fixture(); const result = await f.transport.send(f.expected.sessionName, "ordinary input");
   expect(result.ok, JSON.stringify(result)).toBe(true); expect(f.commands.some(command => command.includes("paste-buffer"))).toBe(true);
 });
+
+
+it("rejects a conditional runtime-switch send without retaining input and requires expectation on retry", async () => {
+  const f = fixture();
+  f.db.prepare(`INSERT INTO seat_runtime_switches (id,node_id,canonical_session,staging_session,predecessor_pane,successor_pane,
+    predecessor_generation,successor_generation,from_runtime,to_runtime,phase,reason,plan_json,deadline_at)
+    VALUES ('switch-original',?,?,'staging-original','%1','%2',?,'next-generation','claude-code','codex','preparing','test','{}','2099-01-01T00:00:00Z')`)
+    .run(f.node.id, f.expected.sessionName, f.registry.currentOccupantTenure(f.node.id)!.generationUuid);
+  const post = (path: string, expectedIdentity?: unknown) => f.app.request(path, { method: "POST", headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ session: f.expected.sessionName, text: "for original A", deliveryId: "switch-held", expectedIdentity }) });
+  const held = await post("/api/transport/send-expected", f.expected), body = await held.json();
+  expect(held.status).toBe(409); expect(body).toMatchObject({ ok: false, sent: false, inputPhase: "not_reached", reason: "expected_identity_unavailable" });
+  expect(body.outcome).not.toBe("retained");
+  expect(f.db.prepare("SELECT * FROM outbox_entries WHERE outbox_id='switch-held'").get()).toBeUndefined();
+  const ordinary = await post("/api/transport/send"); expect(await ordinary.json()).toMatchObject({ outcome: "retained", reason: "runtime_switch_in_progress" });
+  f.db.prepare("UPDATE seat_runtime_switches SET phase='finished',committed_at=? WHERE id='switch-original'").run(new Date().toISOString()); f.replace();
+  const retry = await post("/api/transport/send-expected", f.expected);
+  expect(retry.status).toBe(409); expect(await retry.json()).toMatchObject({ reason: "expected_identity_mismatch", sent: false });
+  expect((await post("/api/transport/send-expected")).status).toBe(400);
+  expect(f.commands.some(command => command.includes("paste-buffer"))).toBe(false);
+});
+
+it("a conflicting ID request cannot fail the first pending draft-aware input", async () => {
+  const f = fixture("claude-code", true); await f.guard!.set(f.node.id, { mode: "draft-aware" }, "test", "protect input");
+  expect(f.transport.guardedDelivery).toBeDefined();
+  let entered!: () => void, release!: () => void;
+  const atResolver = new Promise<void>(resolve => { entered = resolve; }), gate = new Promise<void>(resolve => { release = resolve; });
+  vi.mocked(f.tmux.getPanePid).mockImplementationOnce(async () => { entered(); await gate; return 10; });
+  const first = f.transport.send(f.expected.sessionName, "first input", { expectedIdentity: f.expected, deliveryId: "concurrent-original" } as never);
+  await atResolver;
+  let conflict, pending;
+  try {
+    conflict = await f.transport.send(f.expected.sessionName, "different input", { expectedIdentity: f.expected, deliveryId: "concurrent-original" } as never);
+    pending = f.db.prepare("SELECT delivery_state FROM outbox_entries WHERE outbox_id='concurrent-original'").get();
+  } finally { release(); }
+  const firstResult = await first;
+  expect(conflict?.reason).toBe("delivery_identity_conflict"); expect(pending).toEqual({ delivery_state: "pending" });
+  expect(firstResult).toMatchObject({ ok: true, checkedIdentity: f.expected });
+  expect(f.db.prepare("SELECT delivery_state FROM outbox_entries WHERE outbox_id='concurrent-original'").get()).toEqual({ delivery_state: "delivered" });
+});
+
+it("conditional Claude polling stops reading at the first native identity in a large record", async () => {
+  const f = fixture();
+  const record = JSON.stringify({ type: "user", sessionId: A, message: { content: "original" } }) + "\n" + " ".repeat(2 * 1024 * 1024);
+  writeFileSync(f.records.get(A)!, record);
+  const url = f.readUrl().replace("sinceBytes=0", `sinceBytes=${Buffer.byteLength(record)}`);
+  const reads = vi.spyOn(fs, "readSync");
+  expect((await f.app.request(url)).status).toBe(200);
+  reads.mockClear();
+  expect((await f.app.request(url)).status).toBe(200);
+  const requested = (reads.mock.calls as unknown as unknown[][]).reduce<number>((sum, call) => sum + (typeof call[3] === "number" ? call[3] : 0), 0);
+  expect(requested).toBeLessThanOrEqual(128 * 1024);
+});
+
+it("conditional read maps a filesystem errno to documented unavailable", async () => {
+  const f = fixture(); vi.spyOn(fs, "readSync").mockImplementationOnce(() => { throw Object.assign(new Error("record disappeared"), { code: "ENOENT" }); });
+  const response = await f.app.request(f.readUrl());
+  expect(response.status).toBe(409); expect(await response.json()).toMatchObject({ error: "expected_identity_unavailable" });
+});
+
+it("expected input preserves caller cancellation before typing", async () => {
+  const f = fixture("claude-code", true), controller = new AbortController(); controller.abort();
+  const result = await f.transport.send(f.expected.sessionName, "for A", { expectedIdentity: f.expected, signal: controller.signal } as never);
+  expect(result).toMatchObject({ ok: false, sent: false, reason: "caller_cancelled" });
+  expect(f.commands.some(command => command.includes("paste-buffer") || command.includes("send-keys"))).toBe(false);
+});
+
+it("expected input keeps typing protection ahead of input", async () => {
+  const f = fixture("claude-code", true); await f.guard!.set(f.node.id, true, "test", "hold input");
+  expect(f.transport.guardedDelivery).toBeDefined();
+  const result = await f.transport.send(f.expected.sessionName, "for A", { expectedIdentity: f.expected } as never);
+  expect(result).toMatchObject({ ok: true, sent: false, outcome: "retained" });
+  expect(f.commands.some(command => command.includes("paste-buffer") || command.includes("send-keys"))).toBe(false);
+});
+
+it("expected input does not answer a native selection prompt", async () => {
+  const f = fixture("codex", true);
+  vi.mocked(f.tmux.capturePaneContent).mockResolvedValue("Codex update available.\n\n› 1. Update now\n  2. Skip this version\n\n  gpt-5.5 high · Context [████ ] · ~/project");
+  const result = await f.transport.send(f.expected.sessionName, "for A", { expectedIdentity: f.expected, waitForIdleMs: 100 } as never);
+  expect(result).toMatchObject({ ok: false, sent: false, reason: "target_needs_input" });
+  expect(f.commands.some(command => command.includes("paste-buffer") || command.includes("send-keys"))).toBe(false);
+});