Merge the questions-and-sessions fix into the launcher branch

Brings in the chat-scoped questions and the turn-scoped question
lifetime. This branch's own changes (the launcher, the listing and
acp.to) are unchanged.

docs/reference/acp-agents.md

@@ -203,10 +203,11 @@ app that starts the agent from the ACP registry passes no arguments, so that's h
 - **A follow-up** in the same session goes to `--to` as a new row whose body names the earlier rows and their states,
   and the app first shows any earlier row that's still open.
 - **A question the team asks you** while a turn is open comes to the app as a permission request, with the ask's own
-  choices. It belongs to the turn when a seat the work reached asked it during the turn, or when its row is tagged
-  with the session (`acp-session:<id>`). Your choice is recorded as your answer, exactly as a click in Slack or the TUI
-  would be, and the seat that asked gets it. Dismissing it records nothing: the question stays open in the TUI and
-  Slack. A question that wants an answer in words is shown with where to answer it.
+  choices, when its row is tagged for that chat (`acp-session:<id>`; the task's row tells the team how). An untagged
+  question from a seat the work reached is named in the chat with where to answer it, because it could be another
+  chat's. Your choice is recorded as your answer, exactly as a click in Slack or the TUI would be, and the seat that
+  asked gets it. Dismissing it, or cancelling the turn, records nothing: the question stays open in the TUI and Slack.
+  A question that wants an answer in words is shown with where to answer it.
 - **Sessions are kept in the queue:** the app can list your earlier sessions (`session/list`) and reopen one
   (`session/load`). Reopening replays each prompt, its work and the reply as they stand now, and a new prompt continues it.
 - **Cancel** stops `rig acp serve` waiting. The team's work carries on, and the reply says so.

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

@@ -99,6 +99,8 @@ export interface AcpAgentOptions {
 }
 
 const ACTIVE_STATES = new Set(["pending", "in-progress", "blocked"]);
+/** How long a cancel waits for the app to settle questions it was showing (ACP has the client answer them cancelled). */
+const QUESTION_SETTLE_MS = 2000;
 const AUTH_REQUIRED = -32000;
 const INVALID_PARAMS = -32602;
 
@@ -115,6 +117,14 @@ interface Session {
 }
 
 type ContentBlock = { type: string; text?: string; uri?: string; name?: string; resource?: { uri?: string; text?: string } };
+
+/** The questions one turn has put to the app. Once the turn has ended, nothing from them reaches the app or the queue. */
+interface TurnQuestions {
+  finished: boolean;
+  /** Questions the app is still showing, by tool call id. */
+  outstanding: Set<string>;
+  tasks: Set<Promise<void>>;
+}
 type Update = Record<string, unknown>;
 
 /** The prompt as plain text for the row's body: text as written, links and embedded text resources named. */
@@ -227,17 +237,20 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
     (await team.queue.sent(team.caller)).filter((row) => row.tags.includes(`${SESSION_TAG}${sessionId}`))
       .sort((a, b) => a.tsCreated.localeCompare(b.tsCreated));
 
-  /** Put one ask's open questions to the person as permission requests, one at a time, and record each choice. */
-  const askPerson = async (team: TeamConnection, sessionId: string, ask: TeamAsk) => {
-    const open = ask.questions.filter((q) => !ask.answered.includes(q.id));
+  /** Put one ask's open questions to the person as permission requests, one at a time, and record each choice. A reply
+   *  that arrives after the turn ended answers nothing here: the question stays open in the TUI and Slack. */
+  const askPerson = async (team: TeamConnection, sessionId: string, ask: TeamAsk, turn: TurnQuestions) => {
+    const emit = (body: Update) => { if (!turn.finished) update(sessionId, body); };
     if (ask.questions.length === 0) {
-      say(sessionId, `\n\n${ask.sourceSession} asks you: ${ask.summary ?? oneLine(ask.body, 200)} (${ask.qitemId}). It wants an answer in words, which this app can't send: answer it in the OpenRig TUI or Slack.\n\n`);
+      emit({ sessionUpdate: "agent_message_chunk", content: { type: "text", text: `\n\n${ask.sourceSession} asks you: ${ask.summary ?? oneLine(ask.body, 200)} (${ask.qitemId}). It wants an answer in words, which this app can't send: answer it in the OpenRig TUI or Slack.\n\n` } });
       return;
     }
-    for (const q of open) {
+    for (const q of ask.questions.filter((question) => !ask.answered.includes(question.id))) {
+      if (turn.finished) return;
       const toolCallId = `${ask.qitemId}#${q.id}`;
-      update(sessionId, { sessionUpdate: "tool_call", toolCallId, kind: "other", status: "pending", title: `${ask.sourceSession} asks: ${q.question}`, content: textContent(ask.body) });
+      emit({ sessionUpdate: "tool_call", toolCallId, kind: "other", status: "pending", title: `${ask.sourceSession} asks: ${q.question}`, content: textContent(ask.body) });
       let outcome: { outcome?: string; optionId?: string } | undefined;
+      turn.outstanding.add(toolCallId);
       try {
         const result = (await peer.request("session/request_permission", {
           sessionId,
@@ -246,16 +259,19 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
         })) as { outcome?: { outcome?: string; optionId?: string } } | undefined;
         outcome = result?.outcome;
       } catch (error) {
-        update(sessionId, { sessionUpdate: "tool_call_update", toolCallId, content: textContent(`This app couldn't show the question (${(error as Error).message}). It's still open: answer it in the OpenRig TUI or Slack.`) });
+        emit({ sessionUpdate: "tool_call_update", toolCallId, content: textContent(`This app couldn't show the question (${(error as Error).message}). It's still open: answer it in the OpenRig TUI or Slack.`) });
         return;
+      } finally {
+        turn.outstanding.delete(toolCallId);
       }
+      if (turn.finished) return;
       const chosen = outcome?.outcome === "selected" ? q.options.find((o) => o.id === outcome?.optionId) : undefined;
       if (!chosen) {
-        update(sessionId, { sessionUpdate: "tool_call_update", toolCallId, content: textContent("Not answered here. The question is still open in the OpenRig TUI and Slack.") });
+        emit({ sessionUpdate: "tool_call_update", toolCallId, content: textContent("Not answered here. The question is still open in the OpenRig TUI and Slack.") });
         return;
       }
       const recorded = await team.queue.answer(ask.qitemId, q.id, chosen.id).catch((error: Error) => ({ ok: false as const, reason: error.message }));
-      update(sessionId, recorded.ok
+      emit(recorded.ok
         ? { sessionUpdate: "tool_call_update", toolCallId, status: "completed", content: textContent(`You answered: ${chosen.label}. ${ask.sourceSession} gets it as your answer.`) }
         : { sessionUpdate: "tool_call_update", toolCallId, status: "failed", content: textContent(`Your choice (${chosen.label}) wasn't recorded: ${recorded.reason}`) });
       log(`session ${sessionId}: ${recorded.ok ? "answered" : "couldn't answer"} ${ask.qitemId} ${q.id}`);
@@ -357,6 +373,7 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
 
     let cancelled = false;
     let wake: () => void = () => {};
+    const questions: TurnQuestions = { finished: false, outstanding: new Set(), tasks: new Set() };
     session.turn = { cancel: () => { cancelled = true; wake(); } };
     let root: string | undefined;
     try {
@@ -370,7 +387,8 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
       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}${FOOTER}${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}. ` +
-        `A question you ask ${team.caller} with choices (rig queue create --destination ${team.caller} --human-questions-file …) appears in the app while this turn is open.`;
+        `A question with choices you ask ${team.caller} while this is open, tagged for this chat, appears in the app as a choice: ` +
+        `rig queue create --destination ${team.caller} --human-questions-file … --tags ${SESSION_TAG}${session.id}.`;
       const turnStarted = new Date().toISOString();
       const qitemId = newQitemId();
       try {
@@ -399,22 +417,42 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
       // ponytail: polls the local daemon for the chain's rows, their transitions and the caller's open asks; the
       // daemon's event stream is the upgrade path if a poll per open turn ever costs too much.
       const follower = new ChainFollower(team.queue, (u) => update(session.id, u), root);
-      const asked = new Set<string>();
+      const seen = new Set<string>();
+      /** Close this turn's questions: any the app still shows are named as open elsewhere, then nothing more comes from them. */
+      const finishQuestions = (why: string) => {
+        for (const toolCallId of questions.outstanding) {
+          update(session.id, { sessionUpdate: "tool_call_update", toolCallId, content: textContent(`${why} The question is still open in the OpenRig TUI and Slack.`) });
+        }
+        questions.finished = true;
+      };
       for (;;) {
         const state = await follower.read();
         sent.tail = state.tail.qitemId;
         if (state.kind === "ended") {
+          finishQuestions("The work closed before this was answered here.");
           say(session.id, state.reply);
           return { stopReason: "end_turn" };
         }
-        // A question belongs to this turn when it names the session, or a seat the work reached asked it during the turn.
         for (const ask of await team.queue.openAsks(team.caller)) {
-          if (asked.has(ask.qitemId)) continue;
-          if (!ask.tags.includes(`${SESSION_TAG}${session.id}`) && !(follower.seats.has(ask.sourceSession) && ask.tsCreated >= turnStarted)) continue;
-          asked.add(ask.qitemId);
-          void askPerson(team, session.id, ask).catch((error: Error) => log(`session ${session.id}: asking about ${ask.qitemId} failed: ${error.message}`));
+          if (seen.has(ask.qitemId)) continue;
+          // An ask tagged for a chat belongs to that chat alone. An untagged one from a seat this work reached might be
+          // another chat's, so it isn't asked here; it's named, with where to answer it.
+          const chats = ask.tags.filter((tag) => tag.startsWith(SESSION_TAG));
+          if (chats.includes(`${SESSION_TAG}${session.id}`)) {
+            seen.add(ask.qitemId);
+            const task: Promise<void> = askPerson(team, session.id, ask, questions)
+              .catch((error: Error) => log(`session ${session.id}: asking about ${ask.qitemId} failed: ${error.message}`))
+              .finally(() => questions.tasks.delete(task));
+            questions.tasks.add(task);
+          } else if (chats.length === 0 && follower.seats.has(ask.sourceSession) && ask.tsCreated >= turnStarted) {
+            seen.add(ask.qitemId);
+            say(session.id, `\n\n${ask.sourceSession}, which this work reached, has a question for you that isn't tagged for this chat (${ask.qitemId}: ${ask.summary ?? oneLine(ask.body, 120)}). Answer it in the OpenRig TUI or Slack.\n\n`);
+          }
         }
         if (cancelled) {
+          // ACP has the app answer the questions it was showing as cancelled; let those answers land before the turn ends.
+          if (questions.tasks.size > 0) await Promise.race([Promise.allSettled([...questions.tasks]), sleep(QUESTION_SETTLE_MS)]);
+          finishQuestions("Not answered here: the turn was cancelled.");
           say(session.id, `Stopped waiting. ${state.tail.qitemId} is still ${state.tail.state} with ${state.tail.destinationSession}: cancelling here doesn't stop the team's work. A new prompt in this session goes to the team as a follow-up and shows where this stands.`);
           return { stopReason: "cancelled" };
         }
@@ -425,6 +463,7 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
       say(session.id, `Lost track of ${root}: ${(error as Error).message}. The item stays with the team; its state is unknown from here.`);
       throw new AcpError(-32603, `Couldn't follow ${root}: ${(error as Error).message}`);
     } finally {
+      questions.finished = true;
       session.turn = undefined;
     }
   });

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

@@ -69,10 +69,11 @@ function connect(overrides: Partial<AcpAgentOptions> = {}, receiver: string | nu
   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);
+  const inbound: Array<Record<string, any>> = [];
+  const client = new StreamJsonRpcPeer(fromAgent, toAgent, (direction, packet) => { if (direction === "in") inbound.push(packet as Record<string, any>); });
   const updates: Array<Record<string, unknown>> = [];
   client.onNotification("session/update", (params) => updates.push((params as { update: Record<string, unknown> }).update));
-  return { client, updates, out };
+  return { client, updates, out, inbound };
 }
 
 async function open(client: StreamJsonRpcPeer): Promise<string> {
@@ -291,7 +292,7 @@ describe("rig acp serve", () => {
     expect(messages(updates).at(-1)).toContain("`rig config set acp.to <member@rig>`, or add `--to <member@rig>`");
   });
 
-  it("a question the working seat asks the caller comes as a permission request, and the choice is recorded as the caller's answer", async () => {
+  it("a question tagged for this chat comes as a permission request and the choice is recorded as the caller's answer; an untagged one is only named", async () => {
     const { client, updates } = connect({}, "asker@team");
     const asked: Array<Record<string, any>> = [];
     client.onRequest("session/request_permission", async (params) => {
@@ -305,8 +306,11 @@ describe("rig acp serve", () => {
     const turn = client.request("session/prompt", prompt(sessionId, "Pick a database for the cache"));
     const task = await until(() => rowsTo("asker@team")[0], "the work row");
     await move("asker@team", `/api/queue/${task.qitem_id}/claim`, {});
+    // Untagged, from the seat working on it: it could be another chat's, so it's named here, never asked.
+    await move("asker@team", "/api/queue/create", { qitemId: "qitem-ask-untagged", destinationSession: CALLER, body: "Which region?", summary: "Region?", nudge: false,
+      humanQuestions: [{ id: "q1", question: "Which region?", options: [{ id: "eu", label: "EU" }, { id: "us", label: "US" }] }] });
     await move("asker@team", "/api/queue/create", { qitemId: "qitem-ask-db", destinationSession: CALLER, body: "Redis is faster; SQLite is already here.", summary: "Which store?", nudge: false,
-      humanQuestions: [{ id: "q1", question: "Which store?", options: [{ id: "a", label: "Redis" }, { id: "b", label: "SQLite", recommended: true }] }] });
+      tags: [`acp-session:${sessionId}`], humanQuestions: [{ id: "q1", question: "Which store?", options: [{ id: "a", label: "Redis" }, { id: "b", label: "SQLite", recommended: true }] }] });
     const answered = await until(() => {
       const row = db.prepare("SELECT human_answers FROM queue_items WHERE qitem_id = 'qitem-ask-db'").get() as { human_answers: string | null };
       return row.human_answers ? JSON.parse(row.human_answers) : undefined;
@@ -328,6 +332,80 @@ describe("rig acp serve", () => {
     expect(replies[0]!.body).toContain("SQLite");
     const unrelated = db.prepare("SELECT state, human_answers FROM queue_items WHERE qitem_id = 'qitem-unrelated-ask'").get() as { state: string; human_answers: string | null };
     expect(unrelated).toEqual({ state: "pending", human_answers: null });
+    expect(messages(updates)).toContain("\n\nasker@team, which this work reached, has a question for you that isn't tagged for this chat (qitem-ask-untagged: Region?). Answer it in the OpenRig TUI or Slack.\n\n");
+    expect(db.prepare("SELECT human_answers FROM queue_items WHERE qitem_id = 'qitem-ask-untagged'").get()).toEqual({ human_answers: null });
+    expect(task.body).toContain(`--tags acp-session:${sessionId}`);
+  });
+
+  it("a question tagged for one chat is asked in that chat only, though another chat's work reached the same seat", async () => {
+    const chatA = connect({}, "shared@team");
+    const chatB = connect({}, "shared@team");
+    const askedIn: Record<string, string[]> = { a: [], b: [] };
+    chatA.client.onRequest("session/request_permission", async (params) => { askedIn.a.push((params as any).toolCall.toolCallId); return { outcome: { outcome: "cancelled" } }; });
+    chatB.client.onRequest("session/request_permission", async (params) => { askedIn.b.push((params as any).toolCall.toolCallId); return { outcome: { outcome: "selected", optionId: "y" } }; });
+    const a = await open(chatA.client);
+    const b = await open(chatB.client);
+    const turnA = chatA.client.request("session/prompt", prompt(a, "Chat A's task"));
+    const turnB = chatB.client.request("session/prompt", prompt(b, "Chat B's task"));
+    const [rowA, rowB] = await until(() => { const rows = rowsTo("shared@team"); return rows.length === 2 ? rows : undefined; }, "both rows");
+    await move("shared@team", "/api/queue/create", { qitemId: "qitem-ask-for-b", destinationSession: CALLER, body: "For B only", summary: "B?", nudge: false,
+      tags: [`acp-session:${b}`], humanQuestions: [{ id: "q1", question: "Ship B?", options: [{ id: "y", label: "Yes" }, { id: "n", label: "No" }] }] });
+    await until(() => (askedIn.b.length > 0 ? true : undefined), "B's question");
+    for (const row of [rowA!, rowB!]) {
+      await move("shared@team", `/api/queue/${row.qitem_id}/claim`, {});
+      await move("shared@team", `/api/queue/${row.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "Done." });
+    }
+    await Promise.all([turnA, turnB]);
+    expect(askedIn).toEqual({ a: [], b: ["qitem-ask-for-b#q1"] });
+    expect(chatA.updates.some((u) => String(u.toolCallId ?? "").startsWith("qitem-ask-for-b"))).toBe(false);
+    expect(messages(chatA.updates).some((m) => m.includes("qitem-ask-for-b"))).toBe(false);
+  });
+
+  it("cancel waits for the app to settle the question it's showing, and nothing about it follows the cancelled answer", async () => {
+    const { client, updates, inbound } = connect({}, "asker3@team");
+    let answerCancelled!: () => void;
+    client.onRequest("session/request_permission", () => new Promise((resolve) => {
+      answerCancelled = () => setTimeout(() => resolve({ outcome: { outcome: "cancelled" } }), 50);
+    }));
+    const sessionId = await open(client);
+    const turn = client.request("session/prompt", prompt(sessionId, "Choose a theme"));
+    await until(() => rowsTo("asker3@team")[0], "the work row");
+    await move("asker3@team", "/api/queue/create", { qitemId: "qitem-ask-theme", destinationSession: CALLER, body: "Light or dark", summary: "Theme?", nudge: false,
+      tags: [`acp-session:${sessionId}`], humanQuestions: [{ id: "q1", question: "Which theme?", options: [{ id: "l", label: "Light" }, { id: "d", label: "Dark" }] }] });
+    await until(() => (answerCancelled ? true : undefined), "the question on screen");
+    // As ACP has it: the app sends the cancel, then answers its open question as cancelled.
+    client.notify("session/cancel", { sessionId });
+    answerCancelled();
+    expect(await turn).toEqual({ stopReason: "cancelled" });
+    await new Promise((resolve) => setTimeout(resolve, 100));
+    const end = inbound.findIndex((p) => p.result?.stopReason === "cancelled");
+    const about = inbound.map((p, i) => ({ p, i })).filter(({ p }) => p.params?.update?.toolCallId === "qitem-ask-theme#q1");
+    expect(about.at(-1)!.i).toBeLessThan(end);
+    expect(about.at(-1)!.p.params.update.content[0].content.text).toBe("Not answered here. The question is still open in the OpenRig TUI and Slack.");
+    expect(updates.filter((u) => u.toolCallId === "qitem-ask-theme#q1")).toHaveLength(2);
+    expect(db.prepare("SELECT human_answers FROM queue_items WHERE qitem_id = 'qitem-ask-theme'").get()).toEqual({ human_answers: null });
+  });
+
+  it("when the work closes with a question still on screen, the app is told before the turn ends, and a later reply answers nothing", async () => {
+    const { client, inbound } = connect({}, "asker4@team");
+    let answerLate!: () => void;
+    client.onRequest("session/request_permission", () => new Promise((resolve) => { answerLate = () => resolve({ outcome: { outcome: "selected", optionId: "x" } }); }));
+    const sessionId = await open(client);
+    const turn = client.request("session/prompt", prompt(sessionId, "Plan the offsite"));
+    const task = await until(() => rowsTo("asker4@team")[0], "the work row");
+    await move("asker4@team", "/api/queue/create", { qitemId: "qitem-ask-venue", destinationSession: CALLER, body: "Two venues", summary: "Venue?", nudge: false,
+      tags: [`acp-session:${sessionId}`], humanQuestions: [{ id: "q1", question: "Which venue?", options: [{ id: "x", label: "Lisbon" }, { id: "z", label: "Oslo" }] }] });
+    await until(() => (answerLate ? true : undefined), "the question on screen");
+    await move("asker4@team", `/api/queue/${task.qitem_id}/claim`, {});
+    await move("asker4@team", `/api/queue/${task.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "Booked without it." });
+    expect(await turn).toEqual({ stopReason: "end_turn" });
+    answerLate();
+    await new Promise((resolve) => setTimeout(resolve, 150));
+    const end = inbound.findIndex((p) => p.result?.stopReason === "end_turn");
+    const about = inbound.map((p, i) => ({ p, i })).filter(({ p }) => p.params?.update?.toolCallId === "qitem-ask-venue#q1");
+    expect(about.at(-1)!.i).toBeLessThan(end);
+    expect(about.at(-1)!.p.params.update.content[0].content.text).toBe("The work closed before this was answered here. The question is still open in the OpenRig TUI and Slack.");
+    expect(db.prepare("SELECT human_answers FROM queue_items WHERE qitem_id = 'qitem-ask-venue'").get()).toEqual({ human_answers: null });
   });
 
   it("a question the caller dismisses stays open and unanswered", async () => {