fix(queue): reuse human answers across Slack and Steering retries (#1896)

Retrying an answer from Slack or Steering/TUI after an interrupted close now reuses the first committed reply, finishes the request once, and names where the original answer went. Before, the two surfaces could each create a reply and wake the asker twice.

While an ask is still pending, the first committed answer wins and a second typed message is treated as a retry. Once it closes, a later typed Slack message is ordinary conversation again, and a replay of the typed event that produced the answer reuses it. Legacy reply ids and attribution stay readable; wrong-person replies stay correspondence; reactions and recovered late messages stay separate. No schema, setting or error code is added.

One narrower case remains for a follow-up: if a button or Steering answer's close is interrupted, a typed retry then completes it, and that retry's seen record is also lost, a replay of the typed retry can still arrive as a second reply.

packages/daemon/src/domain/gateway/slack/inbound.ts

@@ -17,7 +17,7 @@ import type { InboundQueuePort } from "./queue-access.js";
 import { createHash } from "node:crypto";
 import { ADMITTED_EVENT_TYPES } from "./capabilities.js";
 import { escapeSlackText, parseQuestionAction, redactSecrets } from "./message.js";
-import { formatHumanAnswers, unansweredQuestions, type RecordHumanAnswerResult } from "../../human-questions.js";
+import { humanAnswerText, formatHumanAnswers, unansweredQuestions, type RecordHumanAnswerResult } from "../../human-questions.js";
 import { BotLimiter, DEFAULT_MAX_PER_HOUR, DEFAULT_COOLDOWN_MINUTES, type BotLookup, type NamedBot } from "./bot-inbox.js";
 import { envelopeChannel, type EnvelopeForwarder } from "./forwarding.js";
 
@@ -329,9 +329,19 @@ export class InboundRouter {
       const { summary, body } = this.summaryOf(ev, transfer, route.correlationQitemId);
       // A person's top-level message (not a reply in a thread) becomes the root its answer lands under.
       const mintsRoot = Boolean(this.deps.mintThread && ev.channel && (!ev.thread_ts || ev.thread_ts === ev.ts));
+      const correlated = route.correlationQitemId ? this.deps.queue.getQitem?.(route.correlationQitemId) : null;
+      const askPending = correlated?.state === "pending";
+      const closed = correlated?.state === "done" || correlated?.state === "handed-off";
+      const committedAnswer = (askPending || closed) && !ev.recoveredThreadReply && route.correlationQitemId
+        ? this.deps.queue.getHumanAnswer?.(route.correlationQitemId, who.source) : null;
+      // The generated trailing footer records this event's channel and ts, separately from the person's text.
+      // A crash after close but before seen-mark must still reuse that event's committed answer.
+      const replayedAnswer = closed && committedAnswer?.body.endsWith(body.slice(body.lastIndexOf("\n\n---\n"))) ? committedAnswer : null;
+      // A different event after close remains conversation; only the pending ask coalesces other retries.
+      const earlierAnswer = askPending ? committedAnswer : replayedAnswer;
       let qitemId: string;
       try {
-        qitemId = await this.deps.queue.createQitem({
+        qitemId = replayedAnswer ? replayedAnswer.qitemId : await this.deps.queue.createQitem({
           qitemId: this.inboundQitemId(ev),
           source: who.source, // the REGISTERED human's canonical ref (human-class), never a raw platform id
           destination: route.destination,
@@ -340,18 +350,19 @@ export class InboundRouter {
           summary,
           body,
           ...(route.correlationQitemId ? { inReplyTo: route.correlationQitemId } : {}),
+          ...(askPending && !ev.recoveredThreadReply ? { humanAnswer: true } : {}),
         });
       } catch (e) {
         this.deps.log?.(`qitem create failed ts=${ts}: ${(e as Error).message}`);
         return { landed: false, reason: "create_failed" };
       }
       let replyResolution: "resolved" | "resolved-uncertain" | "already-resolved" | "not-applicable" | undefined;
-      if (route.correlationQitemId && this.deps.resolveHumanReply && !ev.recoveredThreadReply) {
+      if (!closed && route.correlationQitemId && this.deps.resolveHumanReply && !ev.recoveredThreadReply) {
         try {
           replyResolution = await this.deps.resolveHumanReply({
             qitemId: route.correlationQitemId,
             actorSession: who.source,
-            decision: String(ev.text ?? "").trim() || "[file reply]",
+            decision: earlierAnswer ? humanAnswerText(earlierAnswer.body) : String(ev.text ?? "").trim() || "[file reply]",
             messageTs: ev.ts,
           });
         } catch (e) {
@@ -371,6 +382,12 @@ export class InboundRouter {
           return { landed: false, qitemId, reason: "resolve_failed", correlationQitemId: route.correlationQitemId };
         }
       }
+      if (earlierAnswer && ev.channel && ev.thread_ts) {
+        try {
+          await this.deps.acknowledgeAnswer?.({ channel: ev.channel, threadTs: ev.thread_ts,
+            text: escapeSlackText(`Already answered; the original answer went to ${route.destination} as ${earlierAnswer.qitemId}. No new reply was sent.`) });
+        } catch (e) { this.deps.log?.(`answer acknowledgement failed: ${(e as Error).message}`); }
+      }
       if (mintsRoot) {
         try {
           this.deps.mintThread!({ threadTs: ts, channel: ev.channel!, human: who.source, seat: route.destination, conversationId: qitemId });
@@ -469,6 +486,22 @@ export class InboundRouter {
       if (live) await acknowledge(CLICK_REPLIES.unmapped);
       return { status: "ignored", reason: "unmapped-message" };
     }
+    const existing = this.deps.queue.getHumanAnswer?.(qitemId, who.source);
+    if (existing) {
+      try {
+        const resolution = await this.deps.resolveHumanReply?.({ qitemId, actorSession: who.source, decision: humanAnswerText(existing.body) });
+        if (resolution === "not-applicable") {
+          if (live) await acknowledge(CLICK_REPLIES.notClosed);
+          return { status: "refused", reason: "resolve-not-applicable" };
+        }
+        if (live) await acknowledge(escapeSlackText(`Already answered; the original answer went to ${route.destination} as ${existing.qitemId}. No new reply was sent.`));
+        return { status: "accepted", reason: "answers-complete" };
+      } catch (e) {
+        this.deps.log?.(`answer continuation failed qitem=${qitemId}: ${(e as Error).message}`);
+        if (live) await acknowledge("Your answer is recorded, but closing its request failed. OpenRig will retry.");
+        return { status: "handler-failed", reason: "answer-continuation-failed" };
+      }
+    }
     const recorded = this.deps.recordHumanAnswer?.({ qitemId, actorSession: who.source, ...picked });
     if (!recorded || recorded.status !== "recorded") {
       const reason = recorded?.reason ?? "answers-unavailable";
@@ -497,6 +530,7 @@ export class InboundRouter {
         priority: "routine",
         tags: [...route.tags ?? ["founder-slack", "inbound"], "human-answer"],
         inReplyTo: qitemId,
+        humanAnswer: true,
         summary: `Founder via Slack: answered ${lines.length === 1 ? "1 question" : `${lines.length} questions`}`,
         body: `${lines.join("\n")}\n\n---\nAnswers (question id → option id): ${JSON.stringify(recorded.answers)}\nIn reply to: ${qitemId} (its humanAnswers field holds the same)\nRouted by openrig slack-inbound (button click).`,
       });

packages/daemon/src/domain/gateway/slack/queue-access.ts

@@ -59,11 +59,15 @@ export interface CreateQitemInput {
   tags?: string[];
   /** #822 — the ask whose OpenRig thread this message, answer, or reaction came from. */
   inReplyTo?: string;
+  /** A current typed/button answer, never a reaction or a recovered late message. */
+  humanAnswer?: boolean;
 }
 
 /** What the inbound router needs: land a durable qitem, get its id (or throw). */
 export interface InboundQueuePort {
   createQitem(input: CreateQitemInput): Promise<string>;
+  getQitem?(qitemId: string): QueueItem | null;
+  getHumanAnswer?(qitemId: string, actorSession: string): { qitemId: string; body: string } | null;
 }
 
 /** What the outbound driver needs: the current human-alert set, full items included. */
@@ -138,8 +142,13 @@ export function makeQueuePorts(
   opts: { loadHumanRegistry?: () => LoadResult } = {},
 ): InboundQueuePort & OutboundQueuePort {
   return {
+    getQitem: (qitemId) => queueRepo.getById(qitemId),
+    getHumanAnswer: (qitemId, actorSession) => queueRepo.getHumanAnswer(qitemId, actorSession),
     async createQitem(input: CreateQitemInput): Promise<string> {
-      const created = await queueRepo.create({
+      const ask = input.inReplyTo ? queueRepo.getById(input.inReplyTo) : null;
+      const directAnswer = input.humanAnswer && ask && ask.destinationSession === input.source && ask.humanIntent !== "update"
+        && queueRepo.isPersonAddress(input.source);
+      const createInput = {
         qitemId: input.qitemId,
         sourceSession: input.source,
         destinationSession: input.destination,
@@ -148,7 +157,10 @@ export function makeQueuePorts(
         priority: (input.priority ?? "routine") as never,
         tags: input.tags ?? ["founder-slack", "inbound"],
         inboundReplyTo: input.inReplyTo ?? null,
-      });
+      };
+      const created = directAnswer && input.inReplyTo
+        ? await queueRepo.createHumanAnswer({ ...createInput, inboundReplyTo: input.inReplyTo })
+        : await queueRepo.create(createInput);
       return (created as unknown as { qitemId: string }).qitemId;
     },
     async listHumanAlerts(filter: AlertFilterOpts): Promise<QueueItem[]> {

packages/daemon/src/domain/human-questions.ts

@@ -108,3 +108,9 @@ export function formatHumanAnswers(questions: readonly HumanQuestion[], answers:
 export function unansweredQuestions(questions: readonly HumanQuestion[], answers: HumanAnswers): HumanQuestion[] {
   return questions.filter((q) => ownAnswer(answers, q.id) == null);
 }
+
+/** Return the person's original answer, without the attribution footer used by both surfaces. */
+export function humanAnswerText(body: string): string {
+  const footer = body.lastIndexOf("\n\n---\n");
+  return (footer < 0 ? body : body.slice(0, footer)).trim();
+}

packages/daemon/src/domain/mission-control/mission-control-write-contract.ts

@@ -49,8 +49,7 @@ import {
   type MissionControlVerb,
 } from "./mission-control-action-log.js";
 import { isHumanSeatSession } from "../human-route-enforcer.js";
-import { createHash } from "node:crypto";
-import { formatHumanAnswers, unansweredQuestions, type HumanAnswers } from "../human-questions.js";
+import { humanAnswerText, formatHumanAnswers, unansweredQuestions, type HumanAnswers } from "../human-questions.js";
 
 export class MissionControlWriteContractError extends Error {
   constructor(
@@ -104,7 +103,7 @@ export interface MissionControlActionResult {
   notifyResult: string | null;
   auditedAt: string;
   /** For `answer`: recorded with questions still open, or answered (returned to the asker, and the ask closed). */
-  answer?: { outcome: "recorded" | "answered"; unanswered: string[] };
+  answer?: { outcome: "recorded" | "answered"; unanswered: string[]; message?: string };
 }
 
 interface WriteContractDeps {
@@ -494,26 +493,35 @@ export class MissionControlWriteContract {
    * (`decision`) through resolveDirectHumanReply, placed under the first unanswered question. Once it's answered, the
    * asking seat gets one reply row and then the ask closes once, the order of Slack's own continuation, so a retry never
    * leaves an ask closed with its answer not returned; the reply row's id comes from the ask, so a retry finds the same
-   * row. With neither a choice nor words, a complete question set whose return didn't finish is sent again, with exactly
-   * its recorded answers. A retry in other words than the reply already returned is refused, never closed on. An ask
-   * that's already answered records, sends and closes nothing more.
+   * row, including a return from Slack before the shared ID existed. A committed return wins over a later attempt's
+   * words: finish its close and tell the person where their original answer went. No second reply or wake is created.
    */
   private async answerAsk(input: MissionControlActionInput): Promise<MissionControlActionResult> {
     const ask = this.queueRepo.getById(input.qitemId);
     if (!ask) {
       throw new MissionControlWriteContractError("qitem_not_found", `qitem ${input.qitemId} not found`, { qitemId: input.qitemId });
     }
-    if (ask.state !== "pending") {
-      throw new MissionControlWriteContractError("qitem_already_terminal",
-        `qitem ${input.qitemId} is ${ask.state}, not waiting on an answer (it may already be answered), so nothing was recorded or sent.`,
-        { qitemId: input.qitemId, state: ask.state });
-    }
     if (ask.destinationSession !== input.actorSession || ask.humanIntent === "update" || !this.queueRepo.isPersonAddress(ask.destinationSession)) {
       throw new MissionControlWriteContractError("answer_not_recorded",
         `qitem ${input.qitemId} isn't an ask to ${input.actorSession}, so ${input.actorSession} can't answer it.`, { qitemId: input.qitemId });
     }
     const before = snapshotQitem(ask);
     const questions = ask.humanQuestions ?? [];
+    const existing = this.queueRepo.getHumanAnswer(input.qitemId, input.actorSession);
+    if (existing && (ask.state === "pending" || ask.state === "done")) {
+      const decision = questions.length && unansweredQuestions(questions, ask.humanAnswers ?? {}).length === 0
+        ? formatHumanAnswers(questions, ask.humanAnswers ?? {}).join("; ") : humanAnswerText(existing.body);
+      if (ask.state === "pending" && !this.queueRepo.resolveDirectHumanReply({ qitemId: input.qitemId, actorSession: input.actorSession, decision })) {
+        throw new MissionControlWriteContractError("answer_not_closed", `The answer is already in ${existing.qitemId}, but ${input.qitemId} didn't close.`, { qitemId: input.qitemId });
+      }
+      return this.recordAnswer(input, before, decision, { outcome: "answered", unanswered: [],
+        message: `Already answered; the original answer went to ${existing.destinationSession} as ${existing.qitemId}. No new reply was sent.` }, existing.qitemId);
+    }
+    if (ask.state !== "pending") {
+      throw new MissionControlWriteContractError("qitem_already_terminal",
+        `qitem ${input.qitemId} is ${ask.state}, not waiting on an answer (it may already be answered), so nothing was recorded or sent.`,
+        { qitemId: input.qitemId, state: ask.state });
+    }
     const chose = input.questionId !== undefined || input.optionId !== undefined;
     const words = input.decision?.trim() ?? "";
     // A complete set whose return didn't finish (it failed, or the process stopped): sending again returns exactly those
@@ -544,11 +552,10 @@ export class MissionControlWriteContract {
       const earlier = formatHumanAnswers(questions, ask.humanAnswers ?? {});
       reply = `${words}\n\n---\n${earlier.length ? `Earlier answers: ${earlier.join("; ")}\n` : ""}In reply to: ${input.qitemId}${questions.length ? " (its humanAnswers field holds these words under its first unanswered question)" : ""}`;
     }
-    const replyId = `qitem-human-answer-${createHash("sha256").update(input.qitemId).digest("hex").slice(0, 20)}`;
-    let returned: Awaited<ReturnType<QueueRepository["create"]>>;
+    let returned: QueueItem;
     try {
-      returned = await this.queueRepo.create({
-        qitemId: replyId, sourceSession: input.actorSession, destinationSession: ask.sourceSession,
+      returned = await this.queueRepo.createHumanAnswer({
+        sourceSession: input.actorSession, destinationSession: ask.sourceSession,
         body: `${reply}\nAnswered by ${input.actorSession} in OpenRig.`,
         summary: `${input.actorSession} answered: ${ask.summary ?? input.qitemId}`,
         tags: ["human-answer", ...(ask.tags ?? []).filter((t) => /^(project|mission|slice):/.test(t))],
@@ -560,16 +567,11 @@ export class MissionControlWriteContract {
         + (chose || resume ? "Its recorded answers stay on the row; open it again to send them." : "Nothing was recorded; answer again once that's fixed."),
         { qitemId: input.qitemId });
     }
-    // The reply row is keyed by the ask, so this is an earlier attempt's row in other words: the asker holds those. Close
-    // only on words that went back; this answer wasn't sent.
-    if (returned.createWarning?.code === "qitem_body_not_saved") {
-      throw new MissionControlWriteContractError("answer_not_closed",
-        `An earlier answer already went back to ${ask.sourceSession} as ${replyId}, in other words, so this one wasn't sent and ${input.qitemId} wasn't closed with it. `
-        + `Answer again with the same words as ${replyId} to close it, or Chat with the asker.`,
-        { qitemId: input.qitemId, replyQitemId: replyId });
-    }
-    const decision = chose || resume ? formatHumanAnswers(questions, this.queueRepo.getById(input.qitemId)?.humanAnswers ?? {}).join("; ") : words;
-    if (!this.queueRepo.resolveDirectHumanReply({ qitemId: input.qitemId, actorSession: input.actorSession, decision })) {
+    const replyId = returned.qitemId;
+    const decision = chose || resume
+      ? formatHumanAnswers(questions, this.queueRepo.getById(input.qitemId)?.humanAnswers ?? {}).join("; ") : humanAnswerText(returned.body);
+    if (!this.queueRepo.resolveDirectHumanReply({ qitemId: input.qitemId, actorSession: input.actorSession, decision })
+      && this.queueRepo.getById(input.qitemId)?.state !== "done") {
       throw new MissionControlWriteContractError("answer_not_closed",
         `The answer went back to ${ask.sourceSession} as ${replyId}, but qitem ${input.qitemId} didn't close: it changed meanwhile.`,
         { qitemId: input.qitemId, replyQitemId: replyId });

packages/daemon/src/domain/queue-repository.ts

@@ -1651,6 +1651,30 @@ export class QueueRepository {
     }
   }
 
+  /** First committed return of a direct ask, including pre-shared-ID Slack replies. */
+  getHumanAnswer(qitemId: string, actorSession: string): QueueItem | null {
+    const ask = this.getById(qitemId);
+    if (!this.hasReplyToColumn || !ask || ask.destinationSession !== actorSession
+      || ask.humanIntent === "update" || !this.isPersonAddress(actorSession)) return null;
+    const row = this.db.prepare(`SELECT qitem_id FROM queue_items
+      WHERE reply_to = ? AND source_session = ? AND destination_session = ?
+        AND (EXISTS (SELECT 1 FROM json_each(queue_items.tags) WHERE value = 'human-answer')
+          OR (qitem_id LIKE 'qitem-slack-inbound-%' AND COALESCE(summary, '') NOT LIKE '[Recovered after gap] %'))
+      ORDER BY ts_created, rowid LIMIT 1`).get(qitemId, actorSession, ask.sourceSession) as { qitem_id: string } | undefined;
+    return row ? this.getById(row.qitem_id) : null;
+  }
+
+  /** The lookup and create's insert run synchronously before its first delivery await.
+   * Every new surface uses the same ID; old IDs and their original bodies remain intact. */
+  async createHumanAnswer(input: QueueCreateInput & { inboundReplyTo: string }): Promise<QueueItem> {
+    const existing = this.getHumanAnswer(input.inboundReplyTo, input.sourceSession);
+    if (existing) return existing;
+    return this.create({ ...input,
+      qitemId: `qitem-human-answer-${createHash("sha256").update(input.inboundReplyTo).digest("hex").slice(0, 20)}`,
+      tags: [...new Set([...(input.tags ?? []), "human-answer"])],
+    });
+  }
+
   async create(input: QueueCreateInput): Promise<QueueItem & {
     /** Response-only: the retry's changed body was not written. Never stored on the row. */
     createWarning?: { code: "qitem_body_not_saved"; message: string };

packages/daemon/test/cross-surface-interrupted-answer.test.ts

@@ -0,0 +1,217 @@
+// Test-only: real inbound router, Mission Control HTTP route/contract and migrated SQLite.
+// Transport records nudge attempts; it is not a model-consumption or Slack-network proof.
+import { afterEach, describe, expect, it, vi } from "vitest";
+import Database from "better-sqlite3";
+import { Hono } from "hono";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { migrate } from "../src/db/migrate.js";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { QueueRepository } from "../src/domain/queue-repository.js";
+import { MissionControlWriteContract } from "../src/domain/mission-control/mission-control-write-contract.js";
+import { MissionControlActionLog } from "../src/domain/mission-control/mission-control-action-log.js";
+import { missionControlRoutes } from "../src/routes/mission-control.js";
+import { InboundRouter } from "../src/domain/gateway/slack/inbound.js";
+import { makeQueuePorts } from "../src/domain/gateway/slack/queue-access.js";
+import { makeHumanReplyResolver } from "../src/domain/gateway/slack/slack-subsystem.js";
+import { SeenStore, DeadLetterStore } from "../src/domain/gateway/slack/state-store.js";
+
+const PERSON = "human-test@external", ASKER = "requester@test-rig";
+const registry = { ok: true as const, entities: [{ entityId: "human-test", class: "human" as const, displayName: "Test person", address: PERSON,
+  connectorBindings: [{ kind: "slack" as const, connectorRef: "primary", secretsRef: "unused", role: "primary" as const, handle: "UTEST" }], prefs: { deliveryClass: "A" as const } }] };
+const questions = [{ id: "choice", question: "Proceed?", options: [{ id: "yes", label: "Yes" }, { id: "no", label: "No" }] }];
+
+const closes: Array<() => void> = [];
+afterEach(() => {
+  for (const close of closes.splice(0).reverse()) close();
+});
+
+async function fixture() {
+  const home = mkdtempSync(join(tmpdir(), "cross-answer-"));
+  const dbPath = join(home, "queue.sqlite");
+  const db = new Database(dbPath); migrate(db, ALL_MIGRATIONS); const bus = new EventBus(db);
+  const acknowledgements: string[] = [];
+  const nudges: Array<{ session: string; text: string }> = [];
+  const repo = new QueueRepository(db, bus, { validateRig: () => true, loadHumanRegistry: () => registry,
+    transport: { send: async (session, text) => { nudges.push({ session, text }); return { ok: true, verified: true, sent: true }; } } });
+  const contract = new MissionControlWriteContract({ db, eventBus: bus, queueRepo: repo, actionLog: new MissionControlActionLog(db) });
+  const ask = await repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "Synthetic one-question ask", summary: "Proceed?",
+    humanIntent: "decision", humanAlert: true, humanQuestions: questions, evidenceRef: "/fixture/decision", nudge: false });
+  const resolveHumanReply = vi.fn(makeHumanReplyResolver(repo, contract));
+  const seen = new SeenStore(join(home, "seen.jsonl"));
+  const router = new InboundRouter({ queue: makeQueuePorts(repo), destination: ASKER, seen,
+    deadLetter: new DeadLetterStore(join(home, "dead.jsonl")), resolveSender: () => ({ admitted: true, source: PERSON }),
+    resolveRoute: () => ({ destination: ASKER, correlationQitemId: ask.qitemId }),
+    acknowledgeAnswer: async ({ text }) => { acknowledgements.push(text); },
+    recordHumanAnswer: input => repo.recordHumanAnswer(input), resolveHumanReply });
+  const app = new Hono(); app.use("*", async (c, next) => { c.set("queueRepo" as never, repo as never); c.set("eventBus" as never, bus as never); c.set("missionControlWriteContract" as never, contract as never); await next(); });
+  app.route("/api/mission-control", missionControlRoutes());
+  const slack = () => router.routeAction({ type: "block_actions", user: { id: "UTEST" }, channel: { id: "CTEST" },
+    container: { type: "message", message_ts: "1.1", channel_id: "CTEST" }, message: { ts: "1.1" },
+    actions: [{ type: "button", block_id: "or-q:choice", action_id: "or-opt:yes", value: "yes", action_ts: "2.2" }] });
+  const steering = async (input: Record<string, string> = { questionId: "choice", optionId: "yes" }) => {
+    const response = await app.request("/api/mission-control/action", { method: "POST", headers: { "Content-Type": "application/json", "X-OpenRig-Session": input.actorSession ?? PERSON },
+      body: JSON.stringify({ verb: "answer", qitemId: ask.qitemId, actorSession: PERSON, ...input }) });
+    return { status: response.status, body: await response.json() };
+  };
+  const snapshot = () => {
+    const replies = repo.list({ limit: 100 }).filter(item => item.replyTo === ask.qitemId);
+    const resolved = repo.listTransitions(ask.qitemId).filter(t => t.ownerNotificationKind === "human-decision-resolved");
+    const readback = new Database(dbPath, { readonly: true });
+    let committedReplyCount: number;
+    try { committedReplyCount = (readback.prepare("SELECT count(*) n FROM queue_items WHERE reply_to = ?").get(ask.qitemId) as { n: number }).n; }
+    finally { readback.close(); }
+    return { committedReplyCount, askState: repo.getById(ask.qitemId)?.state, answers: repo.getById(ask.qitemId)?.humanAnswers,
+      replies: replies.map(item => ({ id: item.qitemId, replyTo: item.replyTo, source: item.sourceSession, destination: item.destinationSession, body: item.body })),
+      replyCount: replies.length, resolutionCount: resolved.length, nudgeCount: nudges.filter(n => n.session === ASKER).length };
+  };
+  closes.push(() => { db.close(); rmSync(home, { recursive: true, force: true }); });
+  return { repo, ask, slack, steering, snapshot, acknowledgements, resolveHumanReply, seen,
+    typedSlack: (text: string, recoveredThreadReply = false, ts = "3.3", channel = "CTEST") => router.route({ type: "message", user: "UTEST", channel, thread_ts: "1.1", ts, text, recoveredThreadReply }) };
+}
+
+describe("completed human answer interrupted before ask close", () => {
+  it.each(["slack", "steering"] as const)("%s first, then the other surface: one durable reply and one owner nudge", async first => {
+    const f = await fixture();
+    // This is the shared close seam called only AFTER the real first surface wrote its reply row.
+    // The one-shot throw leaves the durable reply and complete answers, with the ask still pending.
+    let interruptedAt: ReturnType<typeof f.snapshot> | null = null;
+    const original = f.repo.resolveDirectHumanReply.bind(f.repo);
+    vi.spyOn(f.repo, "resolveDirectHumanReply").mockImplementationOnce(input => {
+      interruptedAt = f.snapshot(); throw new Error("fixture interruption after durable reply, before ask close");
+    }).mockImplementation(original);
+    await (first === "slack" ? f.slack() : f.steering());
+    const afterFirst = f.snapshot();
+    const secondResult = first === "slack" ? await f.steering() : await f.slack();
+    if (first === "slack") expect(secondResult).toMatchObject({ status: 200, body: { answer: { message: expect.stringContaining("Already answered") } } });
+    else expect(f.acknowledgements.at(-1)).toContain("Already answered");
+    const afterSecond = f.snapshot();
+    expect(interruptedAt).toMatchObject({ committedReplyCount: 1, askState: "pending", replyCount: 1, resolutionCount: 0, nudgeCount: 1 });
+    expect(afterFirst).toMatchObject({ askState: "pending", replyCount: 1, resolutionCount: 0, nudgeCount: 1 });
+    expect(afterSecond).toMatchObject({ committedReplyCount: 1, askState: "done", replyCount: 1, resolutionCount: 1, nudgeCount: 1 });
+  });
+
+  it.each(["slack", "steering"] as const)("%s typed answer interrupted at close, then retried on the same surface: one return", async surface => {
+    const f = await fixture();
+    const answer = surface === "slack" ? () => f.typedSlack("Original words") : () => f.steering({ decision: "Original words" });
+    let interruptedAt: ReturnType<typeof f.snapshot> | null = null;
+    const original = f.repo.resolveDirectHumanReply.bind(f.repo);
+    vi.spyOn(f.repo, "resolveDirectHumanReply").mockImplementationOnce(() => {
+      interruptedAt = f.snapshot(); throw new Error("fixture interruption after durable reply, before ask close");
+    }).mockImplementation(original);
+    await answer();
+    const beforeRetry = f.snapshot();
+    expect(interruptedAt).toMatchObject({ committedReplyCount: 1, askState: "pending", replyCount: 1, resolutionCount: 0, nudgeCount: 1 });
+    expect(beforeRetry).toMatchObject({ askState: "pending", replyCount: 1, resolutionCount: 0, nudgeCount: 1 });
+    const retry = await answer();
+    if (surface === "steering") expect(retry).toMatchObject({ status: 200, body: { answer: { message: expect.stringContaining("Already answered") } } });
+    else expect(retry).toMatchObject({ disposition: "accepted" });
+    expect(f.snapshot()).toMatchObject({ committedReplyCount: 1, askState: "done", replyCount: 1, resolutionCount: 1, nudgeCount: 1,
+      replies: beforeRetry.replies, answers: { choice: { kind: "typed-reply", text: "Original words" } } });
+  });
+
+  it("without interruption, an answer closes once and the other surface creates no second reply", async () => {
+    const f = await fixture(); await f.slack(); await f.steering(); const result = f.snapshot();
+    expect(result).toMatchObject({ askState: "done", replyCount: 1, resolutionCount: 1, nudgeCount: 1 });
+  });
+
+  it.each(["qitem-slack-answers-old", "qitem-human-answer-old", "qitem-slack-inbound-old"])("reuses %s with its original attribution and words", async qitemId => {
+    const f = await fixture();
+    const body = "Yes, but preserve this\n\n---\nquoted divider\n\n---\nSource: original surface";
+    await f.repo.create({ qitemId, sourceSession: PERSON, destinationSession: ASKER, inboundReplyTo: f.ask.qitemId,
+      body, tags: qitemId.includes("inbound") ? ["inbound"] : ["human-answer"] });
+    const result = await f.steering({ decision: "Different words must not replace the committed answer" });
+    expect(result).toMatchObject({ status: 200, body: { createdQitemId: qitemId, answer: { message: expect.stringContaining(qitemId) } } });
+    expect(f.repo.getById(qitemId)?.body).toBe(body);
+    expect(f.repo.getById(f.ask.qitemId)).toMatchObject({ state: "done", humanAnswers: {
+      choice: { kind: "typed-reply", text: "Yes, but preserve this\n\n---\nquoted divider" } } });
+    expect(f.snapshot()).toMatchObject({ committedReplyCount: 1, resolutionCount: 1, nudgeCount: 1 });
+    expect((await f.steering({ decision: "retry" })).body).toMatchObject({ createdQitemId: qitemId });
+    expect(f.snapshot().nudgeCount).toBe(1);
+  });
+
+  it.each(["slack", "steering", "button-after-typed"] as const)("keeps the first typed answer when %s returns before an interrupted close", async first => {
+    const f = await fixture();
+    vi.spyOn(f.repo, "resolveDirectHumanReply").mockReturnValueOnce(false);
+    if (first === "slack") await f.typedSlack("Original words");
+    else await f.steering({ decision: "Original words" });
+    if (first === "slack") await f.steering({ decision: "Later words" });
+    else if (first === "steering") await f.typedSlack("Later words");
+    else await f.slack(); // A later button must not replace the earlier typed answer.
+    expect(f.snapshot()).toMatchObject({ committedReplyCount: 1, replyCount: 1, resolutionCount: 1, nudgeCount: 1,
+      answers: { choice: { kind: "typed-reply", text: "Original words" } } });
+  });
+
+  it.each(["slack", "steering"] as const)("a closed ask answered in %s delivers a later typed Slack follow-up as conversation", async first => {
+    const f = await fixture();
+    if (first === "slack") await f.typedSlack("Yes, ship it.");
+    else await f.steering({ decision: "Yes, ship it." });
+    expect(f.snapshot()).toMatchObject({ askState: "done", replyCount: 1, resolutionCount: 1, nudgeCount: 1 });
+    const resolveCalls = f.resolveHumanReply.mock.calls.length;
+    expect(await f.typedSlack("Also: use the release branch", false, "4.4")).toMatchObject({ disposition: "accepted" });
+    const after = f.snapshot();
+    expect(after).toMatchObject({ askState: "done", committedReplyCount: 2, replyCount: 2, resolutionCount: 1, nudgeCount: 2 });
+    const followup = after.replies.find(row => row.body.startsWith("Also: use the release branch"))!;
+    expect(followup).toBeDefined();
+    expect(f.repo.getById(followup.id)?.tags).not.toContain("human-answer");
+    expect(f.resolveHumanReply).toHaveBeenCalledTimes(resolveCalls);
+    expect(f.acknowledgements.some(text => text.startsWith("Already answered"))).toBe(false);
+  });
+
+  it("the same Slack event replayed after close and before its seen mark returns the original answer once", async () => {
+    const f = await fixture();
+    const mark = vi.spyOn(f.seen, "mark").mockImplementationOnce(() => { throw new Error("fixture interruption after close, before seen append"); });
+    await expect(f.typedSlack("Original words")).rejects.toThrow("before seen append");
+    mark.mockRestore();
+    const beforeReplay = f.snapshot();
+    expect(beforeReplay).toMatchObject({ committedReplyCount: 1, askState: "done", replyCount: 1, resolutionCount: 1, nudgeCount: 1 });
+    expect(f.seen.load().has("CTEST:3.3")).toBe(false);
+    const resolveCalls = f.resolveHumanReply.mock.calls.length;
+    const replay = await f.typedSlack("Original words");
+    expect(f.snapshot()).toMatchObject({ committedReplyCount: 1, askState: "done", replyCount: 1, resolutionCount: 1, nudgeCount: 1, replies: beforeReplay.replies });
+    expect(replay).toMatchObject({ disposition: "accepted", qitemId: beforeReplay.replies[0]!.id });
+    expect(f.resolveHumanReply).toHaveBeenCalledTimes(resolveCalls);
+    expect(f.seen.load().has("CTEST:3.3")).toBe(true);
+    expect(await f.typedSlack("Original words")).toMatchObject({ disposition: "ignored", reason: "dup" });
+    expect(f.snapshot()).toMatchObject({ replyCount: 1, resolutionCount: 1, nudgeCount: 1 });
+  });
+
+  it("the same timestamp in another channel is a different follow-up event", async () => {
+    const f = await fixture(); await f.typedSlack("Original words");
+    expect(await f.typedSlack("Context from another channel", false, "3.3", "COTHER")).toMatchObject({ disposition: "accepted" });
+    expect(f.snapshot()).toMatchObject({ committedReplyCount: 2, replyCount: 2, resolutionCount: 1, nudgeCount: 2 });
+  });
+
+  it("a wrong person cannot reuse the answer, and a recovered late message remains separate", async () => {
+    const f = await fixture(); await f.steering();
+    const wrong = await f.steering({ actorSession: "someone-else@external", decision: "retry" });
+    expect(wrong).toMatchObject({ status: 400, body: { error: "answer_not_recorded" } });
+    const late = await f.typedSlack("Recovered late context", true);
+    expect(late).toMatchObject({ disposition: "accepted" });
+    const replies = f.repo.list({ limit: 100 }).filter(row => row.replyTo === f.ask.qitemId);
+    expect(replies).toHaveLength(2);
+    expect(replies.find(row => row.body.startsWith("Recovered late context"))?.tags).not.toContain("human-answer");
+    expect(f.snapshot().resolutionCount).toBe(1);
+  });
+
+  it("simultaneous surface attempts return the same answer and close once", async () => {
+    const f = await fixture();
+    const [steering, slack] = await Promise.all([f.steering(), f.slack()]);
+    expect(steering.status).toBe(200);
+    expect(slack.status).toBe("accepted");
+    expect(f.snapshot()).toMatchObject({ committedReplyCount: 1, replyCount: 1, resolutionCount: 1, nudgeCount: 1 });
+  });
+
+  it("does not mistake a legacy recovered late message for the person's answer", async () => {
+    const f = await fixture();
+    await f.repo.create({ qitemId: "qitem-slack-inbound-recovered", sourceSession: PERSON, destinationSession: ASKER,
+      inboundReplyTo: f.ask.qitemId, summary: "[Recovered after gap] Founder via Slack: earlier context",
+      body: "Earlier context, not an answer", tags: ["inbound"], nudge: false });
+    expect(f.repo.getHumanAnswer(f.ask.qitemId, PERSON)).toBeNull();
+    expect(await f.steering()).toMatchObject({ status: 200, body: { answer: { outcome: "answered" } } });
+    expect(f.snapshot()).toMatchObject({ committedReplyCount: 2, resolutionCount: 1, nudgeCount: 1 });
+  });
+
+});

packages/daemon/test/slack-reply-to-thread.test.ts

@@ -270,7 +270,7 @@ describe("update --reply-to an earlier item's thread (#96)", () => {
       await humanReply("1.1", "903.1");
       await vi.waitFor(() => expect(repo.getById(decision.qitemId)?.state).toBe("done"));
       await postPendingNotice(decision.qitemId);
-      const inbound = inboundRow("903.1")!;
+      const inbound = repo.getHumanAnswer(decision.qitemId, decision.destinationSession)!;
       expect(inbound).toMatchObject({ destinationSession: "author@rig", replyTo: decision.qitemId });
 
       const answer = await repo.create({ ...request, humanIntent: "update", body: "On it.", replyTo: inbound.qitemId });
@@ -294,7 +294,7 @@ describe("update --reply-to an earlier item's thread (#96)", () => {
           actions: [{ type: "button", block_id: "or-q:q1", action_id: "or-opt:yes", value: "yes", action_ts: "910.1" }],
         },
       }) });
-      const answerId = `qitem-slack-answers-${createHash("sha256").update(decision.qitemId).digest("hex").slice(0, 20)}`;
+      const answerId = `qitem-human-answer-${createHash("sha256").update(decision.qitemId).digest("hex").slice(0, 20)}`;
       await vi.waitFor(() => expect(repo.getById(answerId)).not.toBeNull());
       await vi.waitFor(() => expect(repo.getById(decision.qitemId)?.state).toBe("done"));
       const inbound = repo.getById(answerId)!;
@@ -434,7 +434,7 @@ describe("update --reply-to an earlier item's thread (#96)", () => {
       await humanReply("1.1", "905.1");
       await vi.waitFor(() => expect(repo.getById(decision.qitemId)?.state).toBe("done"));
       await postPendingNotice(decision.qitemId);
-      const inbound = inboundRow("905.1")!;
+      const inbound = repo.getHumanAnswer(decision.qitemId, decision.destinationSession)!;
       expect(inbound.replyTo).toBe(decision.qitemId);
 
       repo.update({ qitemId: inbound.qitemId, actorSession: "author@rig", state: "blocked", blockedOn: "human-founder@kernel", summary: "Which branch?", evidenceRef: "/proof/branches.md", transitionNote: "parked for a question" });

packages/daemon/test/slack-structured-questions.test.ts

@@ -384,8 +384,8 @@ describe("structured human questions (#193)", () => {
       await click("ship", "yes", { actionTs: "2001.1" }); // a real late click
       await click("ship", "yes", { actionTs: "2001.1" }); // and its redelivery
       await vi.waitFor(() => expect(threadAcks()).toHaveLength(3));
-      expect(threadAcks()[2]).toBe(CLICK_REPLIES.closed);
-      expect(threadAcks().filter((t) => t === CLICK_REPLIES.closed)).toHaveLength(1);
+      expect(threadAcks()[2]).toContain("Already answered; the original answer went to author@rig as ");
+      expect(threadAcks().filter((t) => t.startsWith("Already answered"))).toHaveLength(1);
       expect(repliesToSeat()).toHaveLength(1);
     });
 
@@ -442,7 +442,7 @@ describe("structured human questions (#193)", () => {
       expect(threadAcks()).toHaveLength(1);
       await click("ship", "yes", { actionTs: "2200.1" }); // Socket Mode redelivery of that click
       await vi.waitFor(() => expect(threadAcks()).toHaveLength(2));
-      expect(threadAcks()[1]).toBe(CLICK_REPLIES.onRecord);
+      expect(threadAcks()[1]).toContain("Already answered; the original answer went to author@rig as ");
       await click("ship", "yes", { actionTs: "2200.1" }); // and once posted, never again
       expect(threadAcks()).toHaveLength(2);
       expect(repliesToSeat()).toHaveLength(1);

packages/daemon/test/steering-only-asks.test.ts

@@ -350,7 +350,7 @@ describe("an ask to a person is on Steering and the TUI; Slack gets it by their
     expect(resolved(row)).toBe(1);
     expect(onSteering()).not.toContain(row.qitemId);
     // A repeated tap on the same view records, sends and closes nothing more.
-    await expect(t.run(second.act('Answer "Ship today?": No'))).rejects.toThrow();
+    expect(await t.run(second.act('Answer "Ship today?": No'))).toMatchObject({ answer: { message: expect.stringContaining("Already answered") } });
     expect(replies(row)).toHaveLength(1);
     expect(resolved(row)).toBe(1);
     expect(repo.getById(row.qitemId)!.humanAnswers).toEqual({ q1: "b", q2: "y" });
@@ -402,7 +402,7 @@ describe("an ask to a person is on Steering and the TUI; Slack gets it by their
     expect(resolved(row)).toBe(1);
   });
 
-  it("a typed retry in other words than the answer already returned is refused, not closed on; the same words close it", async () => {
+  it("a typed retry closes on the original committed answer, without returning later words", async () => {
     const row = await ask();
     const t = await tui(readers());
     // The first answer went back to the asker, and then its close didn't commit.
@@ -410,14 +410,14 @@ describe("an ask to a person is on Steering and the TUI; Slack gets it by their
     await expect(t.reply(row, "Yes, ship it")).rejects.toThrow();
     expect(replies(row)).toHaveLength(1);
     expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending" });
-    // Other words: nothing is sent, nothing closes, and the asker still holds the first answer.
-    await expect(t.reply(row, "No, hold it")).rejects.toThrow();
+    // Later words are not sent: the committed answer wins, and its interrupted close finishes.
+    expect(await t.reply(row, "No, hold it")).toMatchObject({ answer: { message: expect.stringContaining("Already answered") } });
     expect(replies(row)).toHaveLength(1);
     expect(replies(row)[0]!.body).toContain("Yes, ship it");
-    expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending" });
-    expect(resolved(row)).toBe(0);
-    // The same words: the same reply row, and the ask closes once.
-    expect(await t.reply(row, "Yes, ship it")).toMatchObject({ answer: { outcome: "answered" } });
+    expect(repo.getById(row.qitemId)).toMatchObject({ state: "done" });
+    expect(resolved(row)).toBe(1);
+    // The closed row no longer offers a Reply action in the TUI.
+    await expect(t.reply(row, "Yes, ship it")).rejects.toThrow();
     expect(replies(row)).toHaveLength(1);
     expect(repo.getById(row.qitemId)).toMatchObject({ state: "done" });
     expect(resolved(row)).toBe(1);

packages/tui/src/daemon-client.ts

@@ -404,7 +404,7 @@ export class DaemonClient {
   /** The mission-control write Steering uses to answer a request (POST /api/mission-control/action). */
   missionControlAction(input: { verb: "approve" | "answer"; qitemId: string; actorSession: string; questionId?: string; optionId?: string; decision?: string }) {
     return this.post("/api/mission-control/action", input) as Promise<{ verb: string; qitemId: string; closedQitem: unknown;
-      answer?: { outcome: "recorded" | "answered"; unanswered: string[] } }>;
+      answer?: { outcome: "recorded" | "answered"; unanswered: string[]; message?: string } }>;
   }
   attentionAggregate() {
     return this.get(`/api/queue/attention-aggregate`);

packages/tui/src/main.ts

@@ -445,8 +445,8 @@ async function run(): Promise<void> {
         const result = await client.missionControlAction(input);
         view.dispatch({ type: "notice", message: action.verb === "approve"
           ? `Approved ${action.qitemId} as ${action.actorSession}: the request is closed on its own row.`
-          : result.answer?.outcome === "recorded" ? `Recorded on ${action.qitemId}. Still to answer: ${result.answer.unanswered.join("; ")}`
-          : `Answered ${action.qitemId} as ${action.actorSession}: the answer went back to the asker, and the request is closed.` });
+          : result.answer?.message ?? (result.answer?.outcome === "recorded" ? `Recorded on ${action.qitemId}. Still to answer: ${result.answer.unanswered.join("; ")}`
+          : `Answered ${action.qitemId} as ${action.actorSession}: the answer went back to the asker, and the request is closed.`) });
       }
     } catch (err) {
       const message = err instanceof Error ? err.message : String(err);