fix(cli): rig acp serve follows a done-form handoff and keeps an unconfirmed create

A row passed on with handoff-and-complete closes as done with closure
reason handed_off_to. Its successor carries the work, so the turn now
follows it as it follows a handed-off row, instead of ending with the
first row's done and never showing the next stage's result.

The adapter now chooses the row id before it creates the row. When the
create's answer is lost, it reads that id back:
- a row that landed carries on as the turn's work;
- no row means nothing was sent;
- a row it can't read is named as unknown and kept in the session, and
  the next prompt names it.

So a retry never silently starts a second task without naming the first.

The reply now names the seat that closed the row from the row's closing
transition. Before, it used the newest transition, so a note written
after the close named its writer as the closer.

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

@@ -16,7 +16,10 @@ export interface TeamRow {
   destinationSession: string;
   handedOffTo: string | null;
   closureReason: string | null;
+  closureTarget: string | null;
   resolution: string | null;
+  /** Who closed the row, from its closing transition (a note written after the close doesn't change it); null while open. */
+  closer: string | null;
 }
 
 export interface TeamTransition {
@@ -29,8 +32,11 @@ export interface TeamTransition {
 
 /** The queue verbs the adapter uses, acting as the caller. */
 export interface TeamQueue {
-  create(row: { destination: string; body: string; summary: string; tags: string[] }): Promise<string>;
+  /** Create the row under the id the adapter chose, so a create whose answer is lost can be read back by that id. */
+  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. */
+  find(qitemId: string): Promise<TeamRow | null>;
   transitions(qitemId: string): Promise<TeamTransition[]>;
   /** The row a handoff of `qitemId` created on this machine, if any. */
   successor(qitemId: string): Promise<string | undefined>;
@@ -84,6 +90,9 @@ export function promptText(prompt: ContentBlock[]): string {
   return parts.join("\n\n").trim();
 }
 
+/** A queue row id in the form the CLI mints (`qitem-<UTC yyyymmddhhmmss>-<16 hex>`). */
+export const newQitemId = () => `qitem-${new Date().toISOString().replace(/\D/g, "").slice(0, 14)}-${randomBytes(8).toString("hex")}`;
+
 function oneLine(text: string, max: number): string {
   const line = text.split("\n").find((l) => l.trim())?.trim() ?? "";
   return line.length > max ? `${line.slice(0, max - 1)}…` : line;
@@ -174,13 +183,31 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
     session.turn = { cancel: () => { cancelled = true; wake(); } };
     let root: string | undefined;
     try {
-      const earlier = await Promise.all(session.sent.map(async (s) => ({ ...s, row: await team.queue.show(s.tail) })));
-      const open = earlier.filter((s) => 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`);
-      const lineage = earlier.length === 0 ? "" : `\nFollows up on ${earlier.map((s) => `${s.root} (${s.tail === s.root ? "" : `reached ${s.tail}, `}${s.row.state})`).join(", ")}.`;
+      // 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 };
+      }));
+      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`);
+      const lineage = earlier.length === 0 ? "" : `\nFollows up on ${earlier.map((s) => `${s.root} (${s.tail === s.root ? "" : `reached ${s.tail}, `}${s.state})`).join(", ")}.`;
       const body = `${text}\n\n---\nSent by ${team.caller} from ${clientName} through rig acp serve (session ${session.id}).${lineage}\n` +
         `The app shows this item's notes and handoffs as progress, and its closing resolution as the answer: close it with a resolution written for ${team.caller}.`;
-      root = await team.queue.create({ destination: team.receiver, body, summary: `ACP: ${oneLine(text, 100)}`, tags: ["acp-serve", `acp-session:${session.id}`] });
+      const qitemId = newQitemId();
+      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 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) {
+          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}`);
+        }
+        log(`session ${session.id}: ${qitemId}'s create wasn't acknowledged, but it reads back`);
+      }
+      root = qitemId;
       const sent = { root, tail: root, closed: false };
       session.sent.push(sent);
       log(`session ${session.id}: sent ${root} to ${team.receiver}`);
@@ -191,12 +218,10 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
       const seen = new Map<string, { lines: string[]; lastId: number; status: string }>();
       for (;;) {
         let tail!: TeamRow;
-        let closing: TeamTransition | undefined;
         for (const qitemId of chain) {
           const row = await team.queue.show(qitemId);
           const transitions = await team.queue.transitions(qitemId);
           tail = row;
-          closing = transitions.at(-1);
           const status = toolStatus(row.state);
           let entry = seen.get(qitemId);
           if (!entry) {
@@ -220,16 +245,18 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
         }
 
         sent.tail = tail.qitemId;
-        if (tail.state === "handed-off") {
+        // Both handoff forms pass the work on: `handoff` leaves the row handed-off, `handoff-and-complete` closes it done
+        // with closure reason handed_off_to. Either way the answer is the successor's, not this row's.
+        if (tail.state === "handed-off" || (tail.state === "done" && tail.closureReason === "handed_off_to")) {
           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 ?? "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. The work goes on there; ask your team for its result.`);
           return { stopReason: "end_turn" };
         }
         if (!ACTIVE_STATES.has(tail.state)) {
           sent.closed = true;
-          const who = closing?.actorSession ?? tail.destinationSession;
+          const who = tail.closer ?? tail.destinationSession;
           // A done row's closure reason says how it ended (canceled is a done close too); the ordinary one goes unsaid.
           const reason = tail.closureReason && tail.closureReason !== "no-follow-on" ? ` (${tail.closureReason})` : "";
           const how = tail.state === "done" ? `done${reason}` : `as ${tail.state}${reason}`;

packages/cli/src/commands/acp.ts

@@ -53,20 +53,26 @@ export function daemonTeamQueue(client: DaemonClient): TeamQueue {
     if (response.status >= 400) throw new Error(`${what} returned HTTP ${response.status}: ${JSON.stringify(response.data)}`);
     return response.data;
   };
+  const face = (data: Record<string, unknown>) => {
+    const row = (data.item ?? data) as Record<string, unknown>;
+    const text = (value: unknown) => (typeof value === "string" ? value : null);
+    return {
+      qitemId: String(row.qitemId), state: String(row.state), destinationSession: String(row.destinationSession),
+      handedOffTo: text(row.handedOffTo), closureReason: text(row.closureReason), closureTarget: text(row.closureTarget), resolution: text(row.resolution),
+      closer: text((row.closingTransition as { actorSession?: unknown } | null | undefined)?.actorSession),
+    };
+  };
   return {
     async create(row) {
-      const qitemId = newId("qitem", 8);
-      ok(await client.post("/api/queue/create", { qitemId, destinationSession: row.destination, body: row.body, summary: row.summary, tags: row.tags }), `queue create for ${row.destination}`);
-      return qitemId;
+      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}`);
     },
     async show(qitemId) {
-      const data = ok(await client.get<Record<string, unknown>>(`/api/queue/${encodeURIComponent(qitemId)}`), `queue show ${qitemId}`);
-      const row = (data.item ?? data) as Record<string, unknown>;
-      const text = (value: unknown) => (typeof value === "string" ? value : null);
-      return {
-        qitemId: String(row.qitemId), state: String(row.state), destinationSession: String(row.destinationSession),
-        handedOffTo: text(row.handedOffTo), closureReason: text(row.closureReason), resolution: text(row.resolution),
-      };
+      return face(ok(await client.get<Record<string, unknown>>(`/api/queue/${encodeURIComponent(qitemId)}`), `queue show ${qitemId}`));
+    },
+    async find(qitemId) {
+      const response = await client.get<Record<string, unknown>>(`/api/queue/${encodeURIComponent(qitemId)}`);
+      if (response.status === 404) return null;
+      return face(ok(response, `queue show ${qitemId}`));
     },
     async transitions(qitemId) {
       const data = ok(await client.get<Array<Record<string, unknown>>>(`/api/queue/${encodeURIComponent(qitemId)}/transitions`), `queue transitions ${qitemId}`);

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

@@ -12,7 +12,7 @@ import { QueueRepository } from "../../daemon/src/domain/queue-repository.js";
 import { queueRoutes } from "../../daemon/src/routes/queue.js";
 import { DaemonClient } from "../src/client.js";
 import { daemonTeamQueue } from "../src/commands/acp.js";
-import { AUTH_METHOD_ID, serveAcpAgent, type AcpAgentOptions, type TeamConnection } from "../src/acp/agent-server.js";
+import { AUTH_METHOD_ID, serveAcpAgent, type AcpAgentOptions, type TeamConnection, type TeamQueue } from "../src/acp/agent-server.js";
 import { JsonRpcRequestError, StreamJsonRpcPeer } from "../src/acp/json-rpc-stdio.js";
 
 const CALLER = "human-ada@external";
@@ -46,13 +46,13 @@ const ends: PassThrough[] = [];
 afterEach(() => { for (const end of ends.splice(0)) end.end(); });
 
 /** A scripted ACP client wired to an in-process rig acp serve. */
-function connect(overrides: Partial<AcpAgentOptions> = {}, receiver: string | null = LEAD) {
+function connect(overrides: Partial<AcpAgentOptions> = {}, receiver: string | null = LEAD, wrap: (queue: TeamQueue) => TeamQueue = (q) => q) {
   const toAgent = new PassThrough();
   const fromAgent = new PassThrough();
   ends.push(toAgent);
   const out: string[] = [];
   fromAgent.on("data", (chunk: Buffer) => out.push(chunk.toString()));
-  const team: TeamConnection = { queue: daemonTeamQueue(new DaemonClient(url(), { sessionName: CALLER })), caller: CALLER, receiver: receiver ?? undefined };
+  const team: TeamConnection = { queue: wrap(daemonTeamQueue(new DaemonClient(url(), { sessionName: CALLER }))), caller: CALLER, receiver: receiver ?? undefined };
   const agent = new StreamJsonRpcPeer(toAgent, fromAgent);
   serveAcpAgent(agent, { connect: async () => team, version: "9.9.9-test", pollMs: 10, closed: agent.closed, ...overrides });
   const client = new StreamJsonRpcPeer(fromAgent, toAgent);
@@ -145,6 +145,73 @@ describe("rig acp serve", () => {
     expect(messages(updates).at(-1)).toBe(`${IMPL} closed ${next.qitem_id} done.\n\nFixed in PR 123: the cookie lacked SameSite=None.`);
   });
 
+  it("a handoff-and-complete passes the work on: the reply is the successor's result, not the first row's done", async () => {
+    const { client, updates } = connect({}, "front@team");
+    const sessionId = await open(client);
+    const turn = client.request("session/prompt", prompt(sessionId, "Upgrade the parser"));
+    const task = await until(() => rowsTo("front@team")[0], "the front row");
+    await move("front@team", `/api/queue/${task.qitem_id}/claim`, {});
+    await move("front@team", `/api/queue/${task.qitem_id}/handoff-and-complete`, { toSession: "back@team", transitionNote: "back owns the parser", nudge: false });
+    expect(db.prepare("SELECT state, closure_reason FROM queue_items WHERE qitem_id = ?").get(task.qitem_id)).toEqual({ state: "done", closure_reason: "handed_off_to" });
+    const next = await until(() => rowsTo("back@team")[0], "the back row");
+    await move("back@team", `/api/queue/${next.qitem_id}/claim`, {});
+    await move("back@team", `/api/queue/${next.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "Parser upgraded to 3.2." });
+    expect(await turn).toEqual({ stopReason: "end_turn" });
+    expect(updates.filter((u) => u.sessionUpdate === "tool_call").map((u) => u.toolCallId)).toEqual([task.qitem_id, next.qitem_id]);
+    expect(messages(updates).at(-1)).toBe(`back@team closed ${next.qitem_id} done.\n\nParser upgraded to 3.2.`);
+  });
+
+  it("the reply names the seat that closed the row: not its owner by default, nor one that wrote a note after the close", async () => {
+    // A slow poll, so the close and the later note both land before the agent reads the row again.
+    const { client, updates } = connect({ pollMs: 500 }, "closer@team");
+    const sessionId = await open(client);
+    const turn = client.request("session/prompt", prompt(sessionId, "Ship the hotfix"));
+    const task = await until(() => rowsTo("closer@team")[0], "the row");
+    // The lead closes its report's row: the closer is neither the row's owner nor the newest writer.
+    await move("closer@team", `/api/queue/${task.qitem_id}/claim`, {});
+    await move("lead@team", `/api/queue/${task.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "Hotfix shipped." });
+    await move("watcher@team", `/api/queue/${task.qitem_id}/update`, { transitionNote: "seen, thanks" });
+    const notes = db.prepare("SELECT actor_session FROM queue_transitions WHERE qitem_id = ? ORDER BY rowid").all(task.qitem_id) as Array<{ actor_session: string }>;
+    expect(notes.at(-1)!.actor_session).toBe("watcher@team");
+    expect(await turn).toEqual({ stopReason: "end_turn" });
+    expect(messages(updates).at(-1)).toBe(`lead@team closed ${task.qitem_id} done.\n\nHotfix shipped.`);
+  });
+
+  it("a create whose answer is lost but whose row landed carries on with that row", async () => {
+    const lostAck = (q: TeamQueue): TeamQueue => ({ ...q, create: async (row) => { await q.create(row); throw new Error("socket hang up"); } });
+    const { client, updates } = connect({}, "acked@team", lostAck);
+    const sessionId = await open(client);
+    const turn = client.request("session/prompt", prompt(sessionId, "Rotate the logs"));
+    const task = await until(() => rowsTo("acked@team")[0], "the row");
+    await move("acked@team", `/api/queue/${task.qitem_id}/claim`, {});
+    await move("acked@team", `/api/queue/${task.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "Rotated." });
+    expect(await turn).toEqual({ stopReason: "end_turn" });
+    expect(rowsTo("acked@team")).toHaveLength(1);
+    expect(messages(updates).at(-1)).toBe(`acked@team closed ${task.qitem_id} done.\n\nRotated.`);
+  });
+
+  it("a create that can't be confirmed either way is named as unknown, and the next prompt names it", async () => {
+    let readbacks = 0;
+    const blind = (q: TeamQueue): TeamQueue => ({
+      ...q,
+      create: async (row) => { await q.create(row); throw new Error("socket hang up"); },
+      find: async (id) => { readbacks += 1; if (readbacks === 1) throw new Error("connect ECONNREFUSED"); return q.find(id); },
+    });
+    const { client, updates } = connect({}, "unsure@team", blind);
+    const sessionId = await open(client);
+    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.`);
+
+    const second = client.request("session/prompt", prompt(sessionId, "Archive March, please"));
+    const followUp = await until(() => rowsTo("unsure@team")[1], "the follow-up row");
+    expect(followUp.body).toContain(`Follows up on ${first.qitem_id} (pending).`);
+    expect(messages(updates)).toContain(`${first.qitem_id} is still pending with unsure@team. Sending this as a follow-up.\n\n`);
+    client.notify("session/cancel", { sessionId });
+    expect(await second).toEqual({ stopReason: "cancelled" });
+  });
+
   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);