diff --git a/openbot/server/src/approvals/question-resume.ts b/openbot/server/src/approvals/question-resume.ts index 96be38e5..f88943a2 100644 --- a/openbot/server/src/approvals/question-resume.ts +++ b/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, diff --git a/openbot/server/tests/approvals-question-resume-correlation.test.ts b/openbot/server/tests/approvals-question-resume-correlation.test.ts new file mode 100644 index 00000000..6bb644c1 --- /dev/null +++ b/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; + 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]); + }); +});