fix(cli): rig acp serve keeps an unanswered create unconfirmed, even when the readback finds nothing

A create that got no answer can still land after the readback, so an
immediate "not found" no longer reads as "nothing reached the team". The
chosen id stays in the session's lineage, the reply says delivery is
unconfirmed, and the next prompt names the id with its state.

A refusal the daemon answered with (a 4xx) is different: nothing was
written. The adapter now raises TeamRefused for it, and the agent says
the queue refused the task. A 5xx stays unknown, like no answer.

A handoff that can't be followed from this machine no longer claims
"the work goes on there". It says only that this app can't follow it,
and to ask the team for its result.

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

@@ -30,9 +30,13 @@ export interface TeamTransition {
   actorSession: 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. */
@@ -186,7 +190,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`);
@@ -197,15 +201,17 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
       try {
         await team.queue.create({ qitemId, destination: team.receiver, body, summary: `ACP: ${oneLine(text, 100)}`, tags: ["acp-serve", `acp-session:${session.id}`] });
       } 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, closed: false });
-          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, closed: false };
@@ -251,7 +257,7 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
           const next = await team.queue.successor(tail.qitemId);
           if (next) { chain.push(next); continue; }
           sent.closed = true;
-          say(session.id, `${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.`);
+          say(session.id, `${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.`);
           return { stopReason: "end_turn" };
         }
         if (!ACTIVE_STATES.has(tail.state)) {

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 TeamConnection, type TeamQueue } from "../acp/agent-server.js";
+import { serveAcpAgent, TeamRefused, 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";
@@ -64,7 +64,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

@@ -29,7 +29,7 @@ beforeAll(async () => {
   for (const [key] of ambient) delete process.env[key];
   db = createFullTestDb();
   const bus = new EventBus(db);
-  const repo = new QueueRepository(db, bus, { validateRig: () => true });
+  const repo = new QueueRepository(db, bus, { validateRig: (session) => !session.endsWith("@gone-rig") });
   const app = new Hono();
   app.use("*", async (c, next) => { c.set("eventBus" as never, bus); c.set("queueRepo" as never, repo); await next(); });
   app.route("/api/queue", queueRoutes());
@@ -202,7 +202,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");
@@ -212,6 +212,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);