Merge #2648 round 2 into the questions-and-sessions stack

Brings in 8aaa4af6 and applies the same change to this branch's create.
An unanswered create stays unconfirmed even when the readback finds
nothing, and the id stays in the lineage. An answered 4xx is a plain
refusal (TeamRefused). A handoff that can't be followed no longer claims
the work goes on. The round-2 cells run here too.

packages/cli/src/acp/agent-server.ts

@@ -58,9 +58,13 @@ export interface SentRow {
   tags: string[];
 }
 
+/** The daemon answered a create and refused it, so nothing was written. Any other failure of a create leaves it unknown. */
+export class TeamRefused extends Error {}
+
 /** The queue verbs the adapter uses, acting as the caller. */
 export interface TeamQueue {
-  /** Create the row under the id the adapter chose, so a create whose answer is lost can be read back by that id. */
+  /** Create the row under the id the adapter chose, so a create whose answer is lost can be read back by that id. Throws
+   *  TeamRefused only for a refusal the daemon answered with. */
   create(row: { qitemId: string; destination: string; body: string; summary: string; tags: string[] }): Promise<void>;
   show(qitemId: string): Promise<TeamRow>;
   /** The row, or null when this machine has no such row; throws when it can't tell. */
@@ -187,7 +191,7 @@ class ChainFollower {
       if (tail.state === "handed-off" || (tail.state === "done" && tail.closureReason === "handed_off_to")) {
         const next = await this.queue.successor(tail.qitemId);
         if (next) { this.chain.push(next); continue; }
-        return { kind: "ended", tail, reply: `${tail.qitemId} was handed to ${tail.handedOffTo ?? tail.closureTarget ?? "another seat"}, whose row isn't on this machine, so this app can't follow it further. The work goes on there; ask your team for its result.` };
+        return { kind: "ended", tail, reply: `${tail.qitemId} was handed to ${tail.handedOffTo ?? tail.closureTarget ?? "another seat"}, whose row isn't on this machine, so this app can't follow it further. Ask your team for its result.` };
       }
       if (ACTIVE_STATES.has(tail.state)) return { kind: "open", tail };
       // A done row's closure reason says how it ended (canceled is a done close too); the ordinary one goes unsaid.
@@ -359,7 +363,7 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
       // Each earlier item as it stands; one whose create was never confirmed may not exist, or can't be read: say which.
       const earlier = await Promise.all(session.sent.map(async (s) => {
         const row = await team.queue.find(s.tail).catch(() => undefined);
-        return { ...s, row, state: row === undefined ? "state unknown" : row === null ? "not on this machine" : row.state };
+        return { ...s, row, state: row === undefined ? "state unknown" : row === null ? "not found here" : row.state };
       }));
       const open = earlier.filter((s) => s.row && ACTIVE_STATES.has(s.row.state));
       if (open.length > 0) say(session.id, `${open.map((s) => `${s.row!.qitemId} is still ${s.row!.state} with ${s.row!.destinationSession}`).join("; ")}. Sending this as a follow-up.\n\n`);
@@ -375,15 +379,17 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
           tags: ["acp-serve", `${SESSION_TAG}${session.id}`, ...(session.cwd ? [`${CWD_TAG}${session.cwd}`] : [])],
         });
       } catch (error) {
-        // The answer was lost, not necessarily the row: read it back by its id before saying what happened.
+        const reason = (error as Error).message;
+        // A refusal the daemon answered with wrote nothing, so it can be said plainly.
+        if (error instanceof TeamRefused) throw new AcpError(-32603, `The team's queue refused the task: ${reason}`);
+        // No answer: the request may still land after any readback, so only a row that reads back settles it.
         const landed = await team.queue.find(qitemId).catch(() => undefined);
-        if (landed === null) throw new AcpError(-32603, `Couldn't send the task (${(error as Error).message}); nothing reached the team.`);
-        if (landed === undefined) {
+        if (!landed) {
           session.sent.push({ root: qitemId, tail: qitemId });
-          say(session.id, `Sending ${qitemId} to ${team.receiver} wasn't confirmed (${(error as Error).message}), and it couldn't be read back: it may or may not have reached the team. Its state is unknown from here; a new prompt in this session names it, and \`rig queue show ${qitemId}\` shows it if it exists.`);
-          throw new AcpError(-32603, `Couldn't confirm ${qitemId}: ${(error as Error).message}`);
+          say(session.id, `Sending ${qitemId} to ${team.receiver} got no answer (${reason}), and ${landed === null ? "it wasn't found on the immediate readback" : "it couldn't be read back"}: delivery is unconfirmed, so it may still reach the team. A new prompt in this session names it, and \`rig queue show ${qitemId}\` shows it if it landed.`);
+          throw new AcpError(-32603, `Couldn't confirm ${qitemId}: ${reason}`);
         }
-        log(`session ${session.id}: ${qitemId}'s create wasn't acknowledged, but it reads back`);
+        log(`session ${session.id}: ${qitemId}'s create got no answer, but it reads back`);
       }
       root = qitemId;
       const sent = { root, tail: root };

packages/cli/src/commands/acp.ts

@@ -9,7 +9,7 @@ import { processIdentity } from "../local-daemon-owner.js";
 import { getOpenRigHome, readOpenRigEnv } from "../openrig-compat.js";
 import { answerCommand, runAcpAssignment, type AcpQueue } from "../acp/connector.js";
 import { findRunProcesses } from "../acp/run-processes.js";
-import { serveAcpAgent, type TeamAsk, type TeamConnection, type TeamQueue } from "../acp/agent-server.js";
+import { serveAcpAgent, TeamRefused, type TeamAsk, type TeamConnection, type TeamQueue } from "../acp/agent-server.js";
 import { StreamJsonRpcPeer } from "../acp/json-rpc-stdio.js";
 import { CLI_VERSION } from "../version.js";
 import { realDeps } from "./daemon.js";
@@ -68,7 +68,10 @@ export function daemonTeamQueue(client: DaemonClient): TeamQueue {
   };
   return {
     async create(row) {
-      ok(await client.post("/api/queue/create", { qitemId: row.qitemId, destinationSession: row.destination, body: row.body, summary: row.summary, tags: row.tags }), `queue create for ${row.destination}`);
+      const response = await client.post("/api/queue/create", { qitemId: row.qitemId, destinationSession: row.destination, body: row.body, summary: row.summary, tags: row.tags });
+      // A 4xx is the daemon's answered refusal, with nothing written; a 5xx leaves the write unknown, like no answer.
+      if (response.status >= 400 && response.status < 500) throw new TeamRefused(`HTTP ${response.status}: ${JSON.stringify(response.data)}`);
+      ok(response, `queue create for ${row.destination}`);
     },
     async show(qitemId) {
       return face(ok(await client.get<Record<string, unknown>>(`/api/queue/${encodeURIComponent(qitemId)}`), `queue show ${qitemId}`));

packages/cli/test/acp-serve.test.ts

@@ -38,7 +38,7 @@ beforeAll(async () => {
   // The caller is a registered person, so the team's asks reach them and their answers record as theirs.
   const registry = { ok: true as const, entities: [{ entityId: "human-ada", class: "human" as const, displayName: "Ada", address: CALLER, connectorBindings: [],
     prefs: { deliveryClass: "B" as const, availability: "available" as const } }] };
-  const repo = new QueueRepository(db, bus, { validateRig: () => true, loadHumanRegistry: () => registry as never });
+  const repo = new QueueRepository(db, bus, { validateRig: (session) => !session.endsWith("@gone-rig"), loadHumanRegistry: () => registry as never });
   const writeContract = new MissionControlWriteContract({ db, eventBus: bus, queueRepo: repo, actionLog: new MissionControlActionLog(db) });
   const app = new Hono();
   app.use("*", async (c, next) => {
@@ -217,7 +217,7 @@ describe("rig acp serve", () => {
     const error = await client.request("session/prompt", prompt(sessionId, "Archive March")).then(() => undefined, (e: unknown) => e);
     const first = rowsTo("unsure@team")[0]!;
     expect((error as JsonRpcRequestError).error).toMatchObject({ code: -32603, message: `Couldn't confirm ${first.qitem_id}: socket hang up` });
-    expect(messages(updates).at(-1)).toBe(`Sending ${first.qitem_id} to unsure@team wasn't confirmed (socket hang up), and it couldn't be read back: it may or may not have reached the team. Its state is unknown from here; a new prompt in this session names it, and \`rig queue show ${first.qitem_id}\` shows it if it exists.`);
+    expect(messages(updates).at(-1)).toBe(`Sending ${first.qitem_id} to unsure@team got no answer (socket hang up), and it couldn't be read back: delivery is unconfirmed, so it may still reach the team. A new prompt in this session names it, and \`rig queue show ${first.qitem_id}\` shows it if it landed.`);
 
     const second = client.request("session/prompt", prompt(sessionId, "Archive March, please"));
     const followUp = await until(() => rowsTo("unsure@team")[1], "the follow-up row");
@@ -227,6 +227,41 @@ describe("rig acp serve", () => {
     expect(await second).toEqual({ stopReason: "cancelled" });
   });
 
+  it("an unanswered create that isn't found on the immediate readback stays unconfirmed, and is named when it lands late", async () => {
+    // The request is still in flight when the agent reads back: it lands only after the readback found nothing.
+    let late: (() => Promise<void>) | undefined;
+    const slow = (q: TeamQueue): TeamQueue => ({ ...q, create: async (row) => {
+      if (late) return q.create(row);
+      late = () => q.create(row);
+      throw new Error("socket hang up");
+    } });
+    const { client, updates } = connect({}, "late@team", slow);
+    const sessionId = await open(client);
+    const error = await client.request("session/prompt", prompt(sessionId, "Tag the release")).then(() => undefined, (e: unknown) => e);
+    expect(rowsTo("late@team")).toHaveLength(0);
+    const id = /Couldn't confirm (qitem-\S+):/.exec((error as JsonRpcRequestError).error.message)![1]!;
+    expect(messages(updates).at(-1)).toBe(`Sending ${id} to late@team got no answer (socket hang up), and it wasn't found on the immediate readback: delivery is unconfirmed, so it may still reach the team. A new prompt in this session names it, and \`rig queue show ${id}\` shows it if it landed.`);
+
+    await late!();
+    expect(rowsTo("late@team")[0]!.qitem_id).toBe(id);
+    const second = client.request("session/prompt", prompt(sessionId, "Tag the release, please"));
+    const followUp = await until(() => rowsTo("late@team")[1], "the follow-up row");
+    expect(followUp.body).toContain(`Follows up on ${id} (pending).`);
+    client.notify("session/cancel", { sessionId });
+    expect(await second).toEqual({ stopReason: "cancelled" });
+  });
+
+  it("a create the daemon answers with a refusal is said plainly, and nothing is kept as sent", async () => {
+    // A seat whose rig has left the topology can't be sent work: the queue answers with a refusal.
+    const { client, updates } = connect({}, "lead@gone-rig");
+    const sessionId = await open(client);
+    const error = await client.request("session/prompt", prompt(sessionId, "Anything")).then(() => undefined, (e: unknown) => e);
+    expect((error as JsonRpcRequestError).error.code).toBe(-32603);
+    expect((error as JsonRpcRequestError).error.message).toMatch(/^The team's queue refused the task: HTTP 4\d\d: /);
+    expect(rowsTo("lead@gone-rig")).toHaveLength(0);
+    expect(messages(updates)).toEqual([]);
+  });
+
   it("cancel stops waiting, never the work; the next prompt is a follow-up that names the open item", async () => {
     const { client, updates } = connect({}, "solo@team");
     const sessionId = await open(client);