fix: match answered questions to originating responsibility run

openbot/server/src/approvals/question-resume.ts

@@ -68,27 +68,55 @@ export async function resumePersonQuestion(input: {
       throw new ApprovalRefusedError(
         "That answer does not belong to this responsibility's conversation.",
       );
-    const [latest] = await input.responsibilities.listRuns(
+    const runs = await input.responsibilities.listRuns(
       ownerUserId,
       responsibility.id,
     );
+    const [latest] = runs;
     /*
-     * The run that asked is waiting for exactly this: resume it, with the answer as its question's
-     * result. It runs again as the responsibility's current run, so it may record progress and
-     * complete, and its reply goes wherever a responsibility's replies go.
+     * Concurrent events can create several runs for one responsibility. The newest run may
+     * be waiting for a different question: resuming it with this older question's answer
+     * consumes the wrong interrupt and leaves the original question unanswered.
+     *
+     * A headless suspension carries the AG-UI runId in its continuation; use it to find
+     * the run that actually asked this question, even when it is not the newest one.
+     * Older waiting records without a continuation remain supported only when there is
+     * exactly one run to resume, so ambiguous legacy records cannot cross-answer.
      */
-    if (
+    const waitingForQuestion = runs.find((run) => {
+      if (
+        run.status !== "waiting" ||
+        run.waiting?.kind !== PERSON_QUESTION_WAITING
+      )
+        return false;
+      const continuation = run.waiting.continuation;
+      return (
+        continuation !== null &&
+        typeof continuation === "object" &&
+        "runId" in continuation &&
+        continuation.runId === question.runId
+      );
+    }) ?? (
+      runs.length === 1 &&
       latest?.status === "waiting" &&
-      latest.waiting?.kind === PERSON_QUESTION_WAITING
-    ) {
-      await input.responsibilities.resumeWaiting(ownerUserId, latest.id, {
-        content: `The person answered: ${input.response}`,
-      });
-      return { replyText: "", responsibilityId: responsibility.id };
+      latest.waiting?.kind === PERSON_QUESTION_WAITING &&
+      !latest.waiting.continuation
+        ? latest
+        : undefined
+    );
+    if (waitingForQuestion) {
+      const resumed = await input.responsibilities.resumeWaiting(
+        ownerUserId,
+        waitingForQuestion.id,
+        { content: `The person answered: ${input.response}` },
+      );
+      if (resumed)
+        return { replyText: "", responsibilityId: responsibility.id };
     }
+    // Never bind a detached answer to some *other* suspended run's progress tools.
     const run = input.responsibilityTurn({
       responsibility,
-      responsibilityRunId: latest?.id,
+      responsibilityRunId: latest?.status === "waiting" ? undefined : latest?.id,
     });
     const outcome = await run({
       ownerUserId,

openbot/server/tests/approvals-question-resume-correlation.test.ts

@@ -0,0 +1,123 @@
+import { describe, expect, test } from "bun:test";
+import type { TurnRunner } from "@openbot/toolkit/run-turn";
+import { resumePersonQuestion } from "../src/approvals/question-resume";
+import type { PersonQuestion } from "../src/approvals/questions";
+import type {
+  Responsibility,
+  ResponsibilityRun,
+  ResponsibilityStore,
+} from "../src/responsibilities/types";
+
+const ownerUserId = "question-answer-owner";
+const responsibility = {
+  id: "goal-1",
+  ownerUserId,
+  agentId: "bot-1",
+  threadId: "thread-1",
+  status: "active",
+} as Responsibility;
+const question = (runId: string): PersonQuestion => ({
+  actorId: ownerUserId,
+  botId: responsibility.agentId,
+  threadId: responsibility.threadId,
+  runId,
+  question: "Which deployment should I choose?",
+  mode: "completed_question",
+  initiator: { kind: "responsibility", id: responsibility.id },
+});
+const waitingRun = (id: string, runId?: string) =>
+  ({
+    id,
+    status: "waiting",
+    waiting: {
+      kind: "person_question",
+      ...(runId ? { continuation: { runId } } : {}),
+    },
+  }) as ResponsibilityRun;
+
+function fixture(
+  runs: ResponsibilityRun[],
+  resumptionAccepted = true,
+) {
+  const resumed: { runId: string; response: unknown }[] = [];
+  const continued: (string | undefined)[] = [];
+  const store = {
+    get: async () => responsibility,
+    listRuns: async () => runs,
+    resumeWaiting: async (
+      _owner: string,
+      runId: string,
+      response: unknown,
+    ) => {
+      resumed.push({ runId, response });
+      return resumptionAccepted;
+    },
+  } as Pick<ResponsibilityStore, "get" | "listRuns" | "resumeWaiting">;
+  const plain: TurnRunner = async () => {
+    throw new Error("A responsibility answer must not use the plain runner");
+  };
+  const respond = (id: string) =>
+    resumePersonQuestion({
+      ownerUserId,
+      question: question(id),
+      response: "Staging",
+      messageId: "person-response:1",
+      runTurn: plain,
+      responsibilities: store,
+      responsibilityTurn: ({ responsibilityRunId }) => {
+        continued.push(responsibilityRunId);
+        return async () => ({ replyText: "Recorded staging." });
+      },
+    });
+  return { respond, resumed, continued };
+}
+
+describe("person question run correlation", () => {
+  test("resumes the older originating run, not the newest run waiting on another question", async () => {
+    const f = fixture([
+      waitingRun("newer-task", "agent-run-B"),
+      waitingRun("original-task", "agent-run-A"),
+    ]);
+    expect(await f.respond("agent-run-A")).toEqual({
+      replyText: "",
+      responsibilityId: responsibility.id,
+    });
+    expect(f.resumed).toEqual([{
+      runId: "original-task",
+      response: { content: "The person answered: Staging" },
+    }]);
+    expect(f.continued).toEqual([]);
+  });
+
+  test("does not resume a different run when the answered run is no longer waiting", async () => {
+    const f = fixture([waitingRun("newer-task", "agent-run-B")]);
+    expect(await f.respond("agent-run-A")).toEqual({
+      replyText: "Recorded staging.",
+      responsibilityId: responsibility.id,
+    });
+    expect(f.resumed).toEqual([]);
+    // The detached turn must not gain another waiting run's progress authority.
+    expect(f.continued).toEqual([undefined]);
+  });
+
+  test("supports a legacy single waiting run without continuation metadata", async () => {
+    const f = fixture([waitingRun("only-task")]);
+    await f.respond("older-run");
+    expect(f.resumed.map((entry) => entry.runId)).toEqual(["only-task"]);
+    expect(f.continued).toEqual([]);
+  });
+
+  test("does not guess between ambiguous legacy runs", async () => {
+    const f = fixture([waitingRun("second-task"), waitingRun("first-task")]);
+    await f.respond("older-run");
+    expect(f.resumed).toEqual([]);
+    expect(f.continued).toEqual([undefined]);
+  });
+
+  test("falls back to a detached answer if the matched run has already been resumed", async () => {
+    const f = fixture([waitingRun("original-task", "agent-run-A")], false);
+    await f.respond("agent-run-A");
+    expect(f.resumed.map((entry) => entry.runId)).toEqual(["original-task"]);
+    expect(f.continued).toEqual([undefined]);
+  });
+});