feat(cli): rig acp serve asks the person the team's questions and reopens sessions (#2651)

rig acp serve's agent server asks the person: during an open turn it polls the caller's open asks and puts one to the app as session/request_permission only when its tags carry this chat's exact acp-session:<id>; an untagged ask from a seat the turn's work reached is named, never asked. A choice posts the existing Mission Control answer verb, so it is recorded by the same answer contract Slack and the TUI use, with the actor taken from the caller's session header; a refused answer is shown with the daemon's reason. On cancel the turn waits up to 2 s for questions the app is settling, then finishes them before returning cancelled. Sessions are the caller's acp-serve rows tagged acp-session:<id>: session/list groups them, session/load replays each root through the same chain follower a live turn uses, and a later prompt carries the session tag and names the earlier roots. session/load of an unknown session returns JSON-RPC -32602. The answer POST sends OPENRIG_AUTH_BEARER_TOKEN only as its Authorization header. docs: acp-agents.md describes questions and sessions.

docs/reference/acp-agents.md

@@ -199,6 +199,14 @@ rig acp serve --to <member@rig>
   turn ending says only that the turn ended: read the reply for how the work ended.
 - **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, 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.
 - **Work handed to a seat on another machine** can't be followed from here; the reply says where it went.
 
@@ -219,7 +227,6 @@ Other apps take the same command and arguments wherever they configure a custom
 
 **Limits:**
 - One receiver per agent command; the app has no team picker.
-- The team's questions to you, and reopening an earlier session, aren't served: a question addressed to you waits in
-  your usual places (the TUI, Slack).
+- Questions show only while a turn is open; one asked while no turn is open waits in your usual places (the TUI, Slack).
 - Prompts are text. Links and embedded text files are passed on as written; images and audio aren't accepted.
 - Progress is read every 2 seconds while a turn is open (`--poll-seconds`).

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

@@ -2,12 +2,18 @@ import { randomBytes } from "node:crypto";
 import type { JsonRpcPeer } from "./json-rpc-stdio.js";
 
 // OpenRig as an ACP agent (rig acp serve): an ACP app's prompt becomes a queue row to the team's receiver, the row's
-// transitions and handoffs come back as progress, and its closing resolution is the reply. The queue is the only work
-// store; this process keeps nothing but which rows each open session sent. A turn's end_turn says only that the turn
-// ended: the reply says whether the team closed the item, and a cancel stops this process waiting, never the team.
+// transitions and handoffs come back as progress, and its closing resolution is the reply. A question the team asks the
+// caller while it works comes to the app as a permission request, and the choice goes back through the person's answer
+// path. The queue is the only work store: a session is the rows tagged with its id, so it can be listed and reopened.
+// A turn's end_turn says only that the turn ended: the reply says whether the team closed the item, and a cancel stops
+// this process waiting, never the team.
 
 export const ACP_PROTOCOL_VERSION = 1;
 export const AUTH_METHOD_ID = "openrig-local";
+export const SESSION_TAG = "acp-session:";
+export const CWD_TAG = "acp-cwd:";
+/** Where the caller's prompt ends in a row's body; the rest is the note rig acp serve adds for the team. */
+const FOOTER = "\n\n---\nSent by ";
 
 /** One queue row as this adapter reads it. */
 export interface TeamRow {
@@ -30,6 +36,28 @@ export interface TeamTransition {
   actorSession: string;
 }
 
+/** An open row to the caller that asks for a decision, with the choices it offers (none: words only). */
+export interface TeamAsk {
+  qitemId: string;
+  sourceSession: string;
+  tsCreated: string;
+  summary: string | null;
+  body: string;
+  tags: string[];
+  questions: Array<{ id: string; question: string; options: Array<{ id: string; label: string; recommended?: boolean }> }>;
+  answered: string[];
+}
+
+/** A row the caller sent through rig acp serve. */
+export interface SentRow {
+  qitemId: string;
+  tsCreated: string;
+  tsUpdated: string;
+  summary: string | null;
+  body: string;
+  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 {}
 
@@ -44,6 +72,12 @@ export interface TeamQueue {
   transitions(qitemId: string): Promise<TeamTransition[]>;
   /** The row a handoff of `qitemId` created on this machine, if any. */
   successor(qitemId: string): Promise<string | undefined>;
+  /** Open asks addressed to `person`. */
+  openAsks(person: string): Promise<TeamAsk[]>;
+  /** Record the caller's choice on an ask, through the person's answer path; a refusal comes back with its reason. */
+  answer(qitemId: string, questionId: string, optionId: string): Promise<{ ok: true } | { ok: false; reason: string }>;
+  /** Rows the caller sent through rig acp serve. */
+  sent(caller: string): Promise<SentRow[]>;
 }
 
 /** What a session works through: the queue as the caller, and the seat that receives the caller's tasks (when set). */
@@ -65,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;
 
@@ -74,13 +110,23 @@ class AcpError extends Error {
 
 interface Session {
   id: string;
+  cwd: string;
   /** The first row each prompt created, oldest first, with the last row its work reached. */
-  sent: Array<{ root: string; tail: string; closed: boolean }>;
+  sent: Array<{ root: string; tail: string }>;
   turn?: { cancel: () => void };
 }
 
 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. */
 export function promptText(prompt: ContentBlock[]): string {
   const parts: string[] = [];
@@ -108,6 +154,64 @@ const toolStatus = (state: string) =>
 const transitionLine = (t: TeamTransition) =>
   `${t.ts.slice(11, 19)}Z ${t.actorSession} · ${t.state}${t.transitionNote ? `: ${t.transitionNote}` : ""}`;
 
+const textContent = (text: string) => [{ type: "content", content: { type: "text", text } }];
+
+/** Where one prompt's work stands after a read of its rows. */
+type ChainState = { kind: "open"; tail: TeamRow } | { kind: "ended"; tail: TeamRow; reply: string };
+
+/** Follows one prompt's work through its rows: each row is a step, and its new notes are emitted as they appear. */
+class ChainFollower {
+  private readonly chain: string[];
+  private readonly seen = new Map<string, { lines: string[]; lastId: number; status: string }>();
+  /** The seats the work has reached; a question from one of them during the turn belongs to it. */
+  readonly seats = new Set<string>();
+
+  constructor(private readonly queue: TeamQueue, private readonly emit: (update: Update) => void, readonly root: string) {
+    this.chain = [root];
+  }
+
+  async read(): Promise<ChainState> {
+    for (;;) {
+      let tail!: TeamRow;
+      for (const qitemId of this.chain) {
+        const row = await this.queue.show(qitemId);
+        const transitions = await this.queue.transitions(qitemId);
+        tail = row;
+        this.seats.add(row.destinationSession);
+        const status = toolStatus(row.state);
+        let entry = this.seen.get(qitemId);
+        if (!entry) {
+          entry = { lines: [], lastId: 0, status };
+          this.seen.set(qitemId, entry);
+          this.emit({
+            sessionUpdate: "tool_call", toolCallId: qitemId, kind: "other", status,
+            title: qitemId === this.root ? `Sent to ${row.destinationSession} (${qitemId})` : `Handed to ${row.destinationSession} (${qitemId})`,
+            rawInput: { qitemId, destination: row.destinationSession },
+          });
+        }
+        const fresh = transitions.filter((t) => t.transitionId > entry!.lastId);
+        if (fresh.length > 0 || status !== entry.status) {
+          for (const t of fresh) { entry.lines.push(transitionLine(t)); entry.lastId = Math.max(entry.lastId, t.transitionId); }
+          entry.status = status;
+          this.emit({ sessionUpdate: "tool_call_update", toolCallId: qitemId, status, ...(entry.lines.length > 0 ? { content: textContent(entry.lines.join("\n")) } : {}) });
+        }
+      }
+      // 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 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. 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.
+      const reason = tail.closureReason && tail.closureReason !== "no-follow-on" ? ` (${tail.closureReason})` : "";
+      const how = tail.state === "done" ? `done${reason}` : `as ${tail.state}${reason}`;
+      return { kind: "ended", tail, reply: `${tail.closer ?? tail.destinationSession} closed ${tail.qitemId} ${how}.\n\n${tail.resolution?.trim() || "(No resolution was written.)"}` };
+    }
+  }
+}
+
 export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void {
   const sessions = new Map<string, Session>();
   const sleep = options.sleep ?? ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
@@ -124,8 +228,55 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
     return result;
   };
 
-  const update = (sessionId: string, body: Record<string, unknown>) => peer.notify("session/update", { sessionId, update: body });
+  const update = (sessionId: string, body: Update) => peer.notify("session/update", { sessionId, update: body });
   const say = (sessionId: string, text: string) => update(sessionId, { sessionUpdate: "agent_message_chunk", content: { type: "text", text } });
+  const newSessionId = () => `acps-${new Date().toISOString().replace(/\D/g, "").slice(0, 14)}-${randomBytes(4).toString("hex")}`;
+
+  /** Each of the caller's prompts in a session, oldest first, by the rows tagged with its id. */
+  const sessionRoots = async (team: TeamConnection, sessionId: string) =>
+    (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. 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) {
+      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 ask.questions.filter((question) => !ask.answered.includes(question.id))) {
+      if (turn.finished) return;
+      const toolCallId = `${ask.qitemId}#${q.id}`;
+      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,
+          toolCall: { toolCallId, title: `${ask.sourceSession} asks: ${q.question}`, kind: "other", status: "pending" },
+          options: q.options.map((o) => ({ optionId: o.id, name: o.recommended ? `${o.label} (recommended)` : o.label, kind: "allow_once" })),
+        })) as { outcome?: { outcome?: string; optionId?: string } } | undefined;
+        outcome = result?.outcome;
+      } catch (error) {
+        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) {
+        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 }));
+      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}`);
+    }
+  };
 
   peer.onRequest("initialize", async (params) => {
     const p = (params ?? {}) as { clientInfo?: { name?: string; title?: string } };
@@ -133,9 +284,10 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
     return {
       protocolVersion: ACP_PROTOCOL_VERSION,
       agentCapabilities: {
-        loadSession: false,
+        loadSession: true,
         promptCapabilities: { image: false, audio: false, embeddedContext: true },
         mcpCapabilities: { http: false, sse: false },
+        sessionCapabilities: { list: {} },
       },
       authMethods: [{
         id: AUTH_METHOD_ID,
@@ -154,13 +306,50 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
     return {};
   });
 
-  peer.onRequest("session/new", async () => {
+  peer.onRequest("session/new", async (params) => {
     await connected();
-    const id = `acps-${new Date().toISOString().replace(/\D/g, "").slice(0, 14)}-${randomBytes(4).toString("hex")}`;
-    sessions.set(id, { id, sent: [] });
+    const id = newSessionId();
+    sessions.set(id, { id, cwd: String((params as { cwd?: unknown } | undefined)?.cwd ?? ""), sent: [] });
     return { sessionId: id };
   });
 
+  peer.onRequest("session/list", async (params) => {
+    const team = await connected();
+    const cwd = (params as { cwd?: unknown } | undefined)?.cwd;
+    const byId = new Map<string, { sessionId: string; cwd: string; title: string; updatedAt: string }>();
+    for (const row of (await team.queue.sent(team.caller)).sort((a, b) => a.tsCreated.localeCompare(b.tsCreated))) {
+      const sessionId = row.tags.find((t) => t.startsWith(SESSION_TAG))?.slice(SESSION_TAG.length);
+      if (!sessionId) continue;
+      const known = byId.get(sessionId);
+      if (known) { if (row.tsUpdated > known.updatedAt) known.updatedAt = row.tsUpdated; continue; }
+      byId.set(sessionId, {
+        sessionId, cwd: row.tags.find((t) => t.startsWith(CWD_TAG))?.slice(CWD_TAG.length) ?? "",
+        title: oneLine(row.body.split(FOOTER)[0] ?? "", 100), updatedAt: row.tsUpdated,
+      });
+    }
+    const listed = [...byId.values()].filter((s) => typeof cwd !== "string" || s.cwd === cwd).sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
+    return { sessions: listed };
+  });
+
+  peer.onRequest("session/load", async (params) => {
+    const p = (params ?? {}) as { sessionId?: string; cwd?: string };
+    const team = await connected();
+    const sessionId = String(p.sessionId ?? "");
+    const roots = await sessionRoots(team, sessionId);
+    if (roots.length === 0) throw new AcpError(INVALID_PARAMS, `No session ${sessionId} from ${team.caller} on this machine`);
+    const session: Session = { id: sessionId, cwd: String(p.cwd ?? ""), sent: [] };
+    // Replay: each prompt as the caller wrote it, then its work as it stands now.
+    for (const row of roots) {
+      update(sessionId, { sessionUpdate: "user_message_chunk", content: { type: "text", text: row.body.split(FOOTER)[0] ?? "" } });
+      const follower = new ChainFollower(team.queue, (u) => update(sessionId, u), row.qitemId);
+      const state = await follower.read();
+      session.sent.push({ root: row.qitemId, tail: state.tail.qitemId });
+      say(sessionId, state.kind === "ended" ? state.reply : `${state.tail.qitemId} is still ${state.tail.state} with ${state.tail.destinationSession}.`);
+    }
+    sessions.set(sessionId, session);
+    return {};
+  });
+
   peer.onNotification("session/cancel", (params) => {
     const sessionId = (params as { sessionId?: string } | undefined)?.sessionId;
     if (sessionId) sessions.get(sessionId)?.turn?.cancel();
@@ -169,7 +358,7 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
   peer.onRequest("session/prompt", async (params) => {
     const p = (params ?? {}) as { sessionId?: string; prompt?: ContentBlock[] };
     const session = p.sessionId ? sessions.get(p.sessionId) : undefined;
-    if (!session) throw new AcpError(INVALID_PARAMS, `No session ${String(p.sessionId)} in this agent; start one with session/new`);
+    if (!session) throw new AcpError(INVALID_PARAMS, `No session ${String(p.sessionId)} in this agent; start one with session/new or session/load`);
     if (session.turn) throw new AcpError(INVALID_PARAMS, "This session already has a turn running; wait for it or cancel it first");
     const team = await connected();
     const text = promptText(Array.isArray(p.prompt) ? p.prompt : []);
@@ -184,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 {
@@ -195,11 +385,17 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
       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}.`;
+      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 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 {
-        await team.queue.create({ qitemId, destination: team.receiver, body, summary: `ACP: ${oneLine(text, 100)}`, tags: ["acp-serve", `acp-session:${session.id}`] });
+        await team.queue.create({
+          qitemId, destination: team.receiver, body, summary: `ACP: ${oneLine(text, 100)}`,
+          tags: ["acp-serve", `${SESSION_TAG}${session.id}`, ...(session.cwd ? [`${CWD_TAG}${session.cwd}`] : [])],
+        });
       } catch (error) {
         const reason = (error as Error).message;
         // A refusal the daemon answered with wrote nothing, so it can be said plainly.
@@ -207,70 +403,57 @@ export function serveAcpAgent(peer: JsonRpcPeer, options: AcpAgentOptions): void
         // 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) {
-          session.sent.push({ root: qitemId, tail: qitemId, closed: false });
+          session.sent.push({ root: qitemId, tail: qitemId });
           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 got no answer, but it reads back`);
       }
       root = qitemId;
-      const sent = { root, tail: root, closed: false };
+      const sent = { root, tail: root };
       session.sent.push(sent);
       log(`session ${session.id}: sent ${root} to ${team.receiver}`);
 
-      // ponytail: polls the local daemon for the chain's rows and transitions; the daemon's event stream is the upgrade
-      // path if a poll per open turn ever costs too much.
-      const chain: string[] = [root];
-      const seen = new Map<string, { lines: string[]; lastId: number; status: string }>();
+      // 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 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 (;;) {
-        let tail!: TeamRow;
-        for (const qitemId of chain) {
-          const row = await team.queue.show(qitemId);
-          const transitions = await team.queue.transitions(qitemId);
-          tail = row;
-          const status = toolStatus(row.state);
-          let entry = seen.get(qitemId);
-          if (!entry) {
-            entry = { lines: [], lastId: 0, status };
-            seen.set(qitemId, entry);
-            update(session.id, {
-              sessionUpdate: "tool_call", toolCallId: qitemId, kind: "other", status,
-              title: qitemId === root ? `Sent to ${row.destinationSession} (${qitemId})` : `Handed to ${row.destinationSession} (${qitemId})`,
-              rawInput: { qitemId, destination: row.destinationSession },
-            });
-          }
-          const fresh = transitions.filter((t) => t.transitionId > entry!.lastId);
-          if (fresh.length > 0 || status !== entry.status) {
-            for (const t of fresh) { entry.lines.push(transitionLine(t)); entry.lastId = Math.max(entry.lastId, t.transitionId); }
-            entry.status = status;
-            update(session.id, {
-              sessionUpdate: "tool_call_update", toolCallId: qitemId, status,
-              ...(entry.lines.length > 0 ? { content: [{ type: "content", content: { type: "text", text: entry.lines.join("\n") } }] } : {}),
-            });
-          }
-        }
-
-        sent.tail = tail.qitemId;
-        // 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 ?? 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.`);
+        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" };
         }
-        if (!ACTIVE_STATES.has(tail.state)) {
-          sent.closed = true;
-          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}`;
-          say(session.id, `${who} closed ${tail.qitemId} ${how}.\n\n${tail.resolution?.trim() || "(No resolution was written.)"}`);
-          return { stopReason: "end_turn" };
+        for (const ask of await team.queue.openAsks(team.caller)) {
+          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) {
-          say(session.id, `Stopped waiting. ${tail.qitemId} is still ${tail.state} with ${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.`);
+          // 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" };
         }
         await Promise.race([sleep(options.pollMs), new Promise<void>((resolve) => { wake = resolve; })]);
@@ -280,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/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, TeamRefused, 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";
@@ -62,6 +62,10 @@ export function daemonTeamQueue(client: DaemonClient): TeamQueue {
       closer: text((row.closingTransition as { actorSession?: unknown } | null | undefined)?.actorSession),
     };
   };
+  const list = async (query: string, what: string): Promise<Array<Record<string, unknown>>> => {
+    const data = ok(await client.get<unknown>(`/api/queue/list?${query}`), what);
+    return (Array.isArray(data) ? data : ((data as { items?: unknown[] }).items ?? [])) as Array<Record<string, unknown>>;
+  };
   return {
     async create(row) {
       const response = await client.post("/api/queue/create", { qitemId: row.qitemId, destinationSession: row.destination, body: row.body, summary: row.summary, tags: row.tags });
@@ -85,9 +89,33 @@ export function daemonTeamQueue(client: DaemonClient): TeamQueue {
       }));
     },
     async successor(qitemId) {
-      const data = ok(await client.get<unknown>(`/api/queue/list?handedOffFrom=${encodeURIComponent(qitemId)}&limit=1`), `queue successor of ${qitemId}`);
-      const rows = (Array.isArray(data) ? data : ((data as { items?: unknown[] }).items ?? [])) as Array<{ qitemId?: string }>;
-      return rows[0]?.qitemId;
+      return (await list(`handedOffFrom=${encodeURIComponent(qitemId)}&limit=1`, `queue successor of ${qitemId}`))[0]?.qitemId as string | undefined;
+    },
+    async openAsks(person) {
+      const rows = await list(`destinationSession=${encodeURIComponent(person)}&state=pending&limit=500`, `open asks to ${person}`);
+      return rows.filter((row) => row.humanIntent !== "update" && (Array.isArray(row.humanQuestions) || row.humanIntent === "decision")).map((row) => ({
+        qitemId: String(row.qitemId), sourceSession: String(row.sourceSession), tsCreated: String(row.tsCreated),
+        summary: typeof row.summary === "string" ? row.summary : null, body: typeof row.body === "string" ? row.body : "",
+        tags: Array.isArray(row.tags) ? (row.tags as string[]) : [],
+        questions: Array.isArray(row.humanQuestions) ? (row.humanQuestions as TeamAsk["questions"]) : [],
+        answered: Object.keys((row.humanAnswers as Record<string, unknown> | null) ?? {}),
+      }));
+    },
+    async answer(qitemId, questionId, optionId) {
+      // The person's answer path, as the TUI records one: the daemon takes the actor from this client's caller.
+      const bearer = process.env.OPENRIG_AUTH_BEARER_TOKEN;
+      const response = await client.post<Record<string, unknown>>("/api/mission-control/action", { verb: "answer", qitemId, questionId, optionId },
+        bearer ? { headers: { Authorization: `Bearer ${bearer}` } } : undefined);
+      if (response.status < 400) return { ok: true };
+      const data = (response.data ?? {}) as { message?: unknown; error?: unknown };
+      return { ok: false, reason: String(data.message ?? data.error ?? `HTTP ${response.status}`) };
+    },
+    async sent(caller) {
+      const rows = await list(`sourceSession=${encodeURIComponent(caller)}&limit=2000`, `rows from ${caller}`);
+      return rows.filter((row) => Array.isArray(row.tags) && (row.tags as string[]).includes("acp-serve")).map((row) => ({
+        qitemId: String(row.qitemId), tsCreated: String(row.tsCreated), tsUpdated: String(row.tsUpdated),
+        summary: typeof row.summary === "string" ? row.summary : null, body: typeof row.body === "string" ? row.body : "", tags: row.tags as string[],
+      }));
     },
   };
 }
@@ -205,7 +233,8 @@ Guide: rig context get reference/acp-agents.md`)
     .addHelpText("after", `
 An ACP app launches this command and talks to it on stdin and stdout. Each prompt becomes a queue row from you to
 --to. The app shows the row's notes and handoffs as progress, and the closing resolution as the reply. A later
-prompt in the same session goes to --to as a follow-up that names the earlier items.
+prompt in the same session goes to --to as a follow-up that names the earlier items. A question the working seats
+ask you meanwhile comes to the app as a choice, recorded as your answer. The app can list and reopen your sessions.
 
 The turn ends when the team closes the item, or when you cancel. Cancelling stops this process waiting; the team's
 work carries on. A turn that ended doesn't mean the work was accepted: the reply says how the item was closed.

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

@@ -1,4 +1,4 @@
-// rig acp serve against the real queue: the daemon's queueRoutes and QueueRepository over a fully migrated test DB, on
+// rig acp serve against the real queue: the daemon's queue and answer routes over a database with every migration, on
 // loopback, driven by a scripted ACP client over a stream pair. The team's moves go through the same HTTP routes a seat's
 // CLI uses, each as its own seat.
 import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
@@ -6,10 +6,15 @@ import { serve } from "@hono/node-server";
 import { Hono } from "hono";
 import { PassThrough } from "node:stream";
 import type Database from "better-sqlite3";
-import { createFullTestDb } from "../../daemon/test/helpers/test-app.js";
+import { createDb } from "../../daemon/src/db/connection.js";
+import { migrate } from "../../daemon/src/db/migrate.js";
+import { ALL_MIGRATIONS } from "../../daemon/src/db/all-migrations.js";
 import { EventBus } from "../../daemon/src/domain/event-bus.js";
 import { QueueRepository } from "../../daemon/src/domain/queue-repository.js";
 import { queueRoutes } from "../../daemon/src/routes/queue.js";
+import { missionControlRoutes } from "../../daemon/src/routes/mission-control.js";
+import { MissionControlWriteContract } from "../../daemon/src/domain/mission-control/mission-control-write-contract.js";
+import { MissionControlActionLog } from "../../daemon/src/domain/mission-control/mission-control-action-log.js";
 import { DaemonClient } from "../src/client.js";
 import { daemonTeamQueue } from "../src/commands/acp.js";
 import { AUTH_METHOD_ID, serveAcpAgent, type AcpAgentOptions, type TeamConnection, type TeamQueue } from "../src/acp/agent-server.js";
@@ -27,12 +32,21 @@ const ambient = ["OPENRIG_URL", "RIGGED_URL", "OPENRIG_PORT", "OPENRIG_SESSION_N
 
 beforeAll(async () => {
   for (const [key] of ambient) delete process.env[key];
-  db = createFullTestDb();
+  db = createDb();
+  migrate(db, ALL_MIGRATIONS);
   const bus = new EventBus(db);
-  const repo = new QueueRepository(db, bus, { validateRig: (session) => !session.endsWith("@gone-rig") });
+  // 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: (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) => { c.set("eventBus" as never, bus); c.set("queueRepo" as never, repo); await next(); });
+  app.use("*", async (c, next) => {
+    c.set("eventBus" as never, bus); c.set("queueRepo" as never, repo); c.set("missionControlWriteContract" as never, writeContract as never);
+    await next();
+  });
   app.route("/api/queue", queueRoutes());
+  app.route("/api/mission-control", missionControlRoutes());
   await new Promise<void>((resolve) => { server = serve({ fetch: app.fetch, hostname: "127.0.0.1", port: 0 }, (info: { port: number }) => { port = info.port; resolve(); }); });
 });
 
@@ -55,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> {
@@ -95,7 +110,8 @@ describe("rig acp serve", () => {
     expect(init.authMethods).toEqual([expect.objectContaining({ id: AUTH_METHOD_ID, name: "OpenRig on this machine" })]);
     expect(init.authMethods[0].type).toBeUndefined();
     expect(init.agentInfo).toEqual({ name: "openrig", title: "OpenRig", version: "9.9.9-test" });
-    expect(init.agentCapabilities.loadSession).toBe(false);
+    expect(init.agentCapabilities.loadSession).toBe(true);
+    expect(init.agentCapabilities.sessionCapabilities).toEqual({ list: {} });
     expect(connected).toBe(0);
     for (const line of out.join("").split("\n").filter(Boolean)) expect(JSON.parse(line).jsonrpc).toBe("2.0");
   });
@@ -122,7 +138,7 @@ describe("rig acp serve", () => {
     expect(task.state).toBe("pending");
     expect(task.body).toMatch(/^Please fix the login bug\nIt fails on Safari\.\n\n---\nSent by human-ada@external from Zed through rig acp serve \(session acps-/);
     expect(task.body).toContain("close it with a resolution written for human-ada@external");
-    expect(JSON.parse(task.tags)).toEqual(["acp-serve", `acp-session:${sessionId}`]);
+    expect(JSON.parse(task.tags)).toEqual(["acp-serve", `acp-session:${sessionId}`, "acp-cwd:/tmp"]);
 
     await move(LEAD, `/api/queue/${task.qitem_id}/claim`, {});
     await move(LEAD, `/api/queue/${task.qitem_id}/update`, { state: "in-progress", transitionNote: "reproduced on Safari 19" });
@@ -276,6 +292,170 @@ describe("rig acp serve", () => {
     expect(messages(updates).at(-1)).toContain("Add `--to <member@rig>`");
   });
 
+  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) => {
+      asked.push(params as Record<string, any>);
+      return { outcome: { outcome: "selected", optionId: "b" } };
+    });
+    const sessionId = await open(client);
+    // Asked before this turn by a seat the work never reaches: not this turn's question.
+    await move("bystander@team", "/api/queue/create", { qitemId: "qitem-unrelated-ask", destinationSession: CALLER, body: "unrelated", summary: "Unrelated", nudge: false,
+      humanQuestions: [{ id: "q1", question: "Unrelated?", options: [{ id: "y", label: "Yes" }, { id: "n", label: "No" }] }] });
+    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,
+      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;
+    }, "the recorded answer");
+    expect(Object.keys(answered)).toEqual(["q1"]);
+    expect(JSON.stringify(answered)).toContain("b");
+    await move("asker@team", `/api/queue/${task.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "SQLite, as you chose." });
+    expect(await turn).toEqual({ stopReason: "end_turn" });
+
+    expect(asked).toHaveLength(1);
+    expect(asked[0]!.toolCall).toMatchObject({ toolCallId: "qitem-ask-db#q1", title: "asker@team asks: Which store?" });
+    expect(asked[0]!.options).toEqual([{ optionId: "a", name: "Redis", kind: "allow_once" }, { optionId: "b", name: "SQLite (recommended)", kind: "allow_once" }]);
+    const result = updates.filter((u) => u.toolCallId === "qitem-ask-db#q1").at(-1)!;
+    expect(result.status).toBe("completed");
+    expect((result.content as Array<{ content: { text: string } }>)[0]!.content.text).toBe("You answered: SQLite. asker@team gets it as your answer.");
+    // "asker@team gets it as your answer": the answer path's reply row, from the caller to the seat that asked.
+    const replies = db.prepare("SELECT body FROM queue_items WHERE source_session = ? AND destination_session = 'asker@team' AND reply_to = 'qitem-ask-db'").all(CALLER) as Array<{ body: string }>;
+    expect(replies).toHaveLength(1);
+    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 () => {
+    const { client, updates } = connect({}, "asker2@team");
+    client.onRequest("session/request_permission", async () => ({ outcome: { outcome: "cancelled" } }));
+    const sessionId = await open(client);
+    const turn = client.request("session/prompt", prompt(sessionId, "Rename the service"));
+    const task = await until(() => rowsTo("asker2@team")[0], "the work row");
+    await move("asker2@team", "/api/queue/create", { qitemId: "qitem-ask-name", destinationSession: CALLER, body: "Two names", summary: "Name?", nudge: false,
+      tags: [`acp-session:${sessionId}`], humanQuestions: [{ id: "q1", question: "Which name?", options: [{ id: "a", label: "Atlas" }, { id: "b", label: "Beacon" }] }] });
+    await until(() => updates.find((u) => u.toolCallId === "qitem-ask-name#q1" && u.sessionUpdate === "tool_call_update"), "the dismissal");
+    await move("asker2@team", `/api/queue/${task.qitem_id}/claim`, {});
+    await move("asker2@team", `/api/queue/${task.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "Kept the old name." });
+    expect(await turn).toEqual({ stopReason: "end_turn" });
+    const dismissed = updates.filter((u) => u.toolCallId === "qitem-ask-name#q1").at(-1)!;
+    expect(dismissed.status).toBeUndefined();
+    expect((dismissed.content as Array<{ content: { text: string } }>)[0]!.content.text).toBe("Not answered here. The question is still open in the OpenRig TUI and Slack.");
+    expect(db.prepare("SELECT state, human_answers FROM queue_items WHERE qitem_id = 'qitem-ask-name'").get()).toEqual({ state: "pending", human_answers: null });
+  });
+
+  it("a session is listed and reopened from its rows: the prompt, its work and the reply replay, and a follow-up continues it", async () => {
+    const first = connect({}, "keeper@team");
+    const sessionId = await open(first.client);
+    const turn = first.client.request("session/prompt", prompt(sessionId, "Summarize yesterday's incidents"));
+    const task = await until(() => rowsTo("keeper@team")[0], "the row");
+    await move("keeper@team", `/api/queue/${task.qitem_id}/claim`, {});
+    await move("keeper@team", `/api/queue/${task.qitem_id}/update`, { state: "done", closureReason: "no-follow-on", resolution: "Two incidents, both closed." });
+    await turn;
+
+    // A new process, as after the app restarts.
+    const second = connect({}, "keeper@team");
+    await second.client.request("initialize", { protocolVersion: 1, clientCapabilities: {} });
+    const listed = (await second.client.request("session/list", {})) as { sessions: Array<Record<string, unknown>> };
+    expect(listed.sessions.find((s) => s.sessionId === sessionId)).toMatchObject({ cwd: "/tmp", title: "Summarize yesterday's incidents" });
+    expect(((await second.client.request("session/list", { cwd: "/elsewhere" })) as { sessions: unknown[] }).sessions.find((s) => (s as { sessionId: string }).sessionId === sessionId)).toBeUndefined();
+    expect(await second.client.request("session/load", { sessionId, cwd: "/tmp", mcpServers: [] })).toEqual({});
+    expect(second.updates.map((u) => u.sessionUpdate)).toEqual(["user_message_chunk", "tool_call", "tool_call_update", "agent_message_chunk"]);
+    expect((second.updates[0]!.content as { text: string }).text).toBe("Summarize yesterday's incidents");
+    expect(messages(second.updates).at(-1)).toBe(`keeper@team closed ${task.qitem_id} done.\n\nTwo incidents, both closed.`);
+
+    const followUp = second.client.request("session/prompt", prompt(sessionId, "And the week before?"));
+    const next = await until(() => rowsTo("keeper@team")[1], "the follow-up row");
+    expect(next.body).toContain(`Follows up on ${task.qitem_id} (done).`);
+    expect(JSON.parse(next.tags)).toContain(`acp-session:${sessionId}`);
+    second.client.notify("session/cancel", { sessionId });
+    expect(await followUp).toEqual({ stopReason: "cancelled" });
+    const missing = await second.client.request("session/load", { sessionId: "acps-nope", cwd: "/tmp", mcpServers: [] }).then(() => undefined, (e: unknown) => e);
+    expect((missing as JsonRpcRequestError).error.code).toBe(-32602);
+  });
+
   it("an unknown method gets JSON-RPC's method-not-found", async () => {
     const { client } = connect();
     const error = await client.request("session/fork", { sessionId: "x" }).then(() => undefined, (e: unknown) => e);