From 32af2398d64fb766017dc5fffb03dc1ffede026a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 22 Sep 2026 04:44:10 +0000 Subject: [PATCH] fix(adapters): retry delegated bot returns when wake message is gone When a delegated bot finishes but its wake message was deleted (e.g. thread clear), returnBotMessageOutcome used to mark botOutcomeReturnedAt and drop the coordinator reply silently. Retry instead, and block clearThread while completed bot_message runs still owe a return. Co-authored-by: cynarAI --- apps/api/src/router.ts | 22 ++++++++---- packages/adapters/src/bot-messages.test.ts | 16 +++++++++ packages/adapters/src/bot-messages.ts | 10 ++++-- packages/db/src/events.test.ts | 41 ++++++++++++++++++++++ packages/db/src/events.ts | 14 ++++++++ packages/db/src/scope.ts | 8 +++++ 6 files changed, 102 insertions(+), 9 deletions(-) diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index 2438c12..1439cae 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -99,6 +99,7 @@ import type { PrismaClient, ThreadEvents } from "@rakazo/db"; import { appendEventInTransaction, BotSectionNameConflictError, + ClearThreadPendingDelegatedReturnsError, CannotDeleteDefaultSpaceError, CannotDeleteLastSpaceError, CannotDeleteSpaceAsNonOwnerError, @@ -1479,12 +1480,21 @@ export function createRouter(deps: RouterDeps) { const target = await resolveThreadTarget(deps.prisma, context.actor, input); const contextBotId = target.kind === "bot" ? target.botId : target.memberBotIds[0]; if (!contextBotId) throw new IsolationError(); - const { cancelledRunIds, historyCompactionGeneration } = await deps.events.clearThread({ - spaceId: context.actor.spaceId, - threadId: target.threadId, - botId: contextBotId, - ...(target.kind === "group" ? { groupId: target.groupId } : {}), - }); + let cancelledRunIds: string[]; + let historyCompactionGeneration: number; + try { + ({ cancelledRunIds, historyCompactionGeneration } = await deps.events.clearThread({ + spaceId: context.actor.spaceId, + threadId: target.threadId, + botId: contextBotId, + ...(target.kind === "group" ? { groupId: target.groupId } : {}), + })); + } catch (error) { + if (error instanceof ClearThreadPendingDelegatedReturnsError) { + throw new ORPCError("CONFLICT", { message: error.message }); + } + throw error; + } const [configuredMemory] = await Promise.all([ target.kind === "bot" ? deps.memoryProviders diff --git a/packages/adapters/src/bot-messages.test.ts b/packages/adapters/src/bot-messages.test.ts index f980dad..7a4bae1 100644 --- a/packages/adapters/src/bot-messages.test.ts +++ b/packages/adapters/src/bot-messages.test.ts @@ -611,6 +611,22 @@ describe("automatic outcome return", () => { expect(harness.enqueue).toHaveBeenCalledOnce(); }); + it("retries when the wake message is missing instead of dropping the outcome", async () => { + const harness = deps({ hopBlocks: [] }); + vi.mocked(harness.deps.prisma.message.findUnique).mockResolvedValue(null as never); + + const returned = await returnBotMessageOutcome( + harness.deps, + { ...run, sourceMessageId: "message-source" }, + sender, + "The answer is 42.", + ); + + expect(returned).toBe(false); + expect(harness.deps.prisma.run.updateMany).not.toHaveBeenCalled(); + expect(harness.enqueue).not.toHaveBeenCalled(); + }); + it("skips the automatic return when a result was already sent", async () => { const harness = deps({ hopBlocks: [ diff --git a/packages/adapters/src/bot-messages.ts b/packages/adapters/src/bot-messages.ts index 9d947d8..4736a86 100644 --- a/packages/adapters/src/bot-messages.ts +++ b/packages/adapters/src/bot-messages.ts @@ -331,9 +331,13 @@ export async function returnBotMessageOutcome( ) { const source = await loadBotMessageContext(deps.prisma, run.sourceMessageId); if (!source) { - await markBotOutcomeReturned(deps.prisma, run.id); - // Handled: nothing to deliver. Return true so callers do not release a reservation. - return true; + if (!run.sourceMessageId) { + await markBotOutcomeReturned(deps.prisma, run.id); + return true; + } + // Wake message missing (e.g. thread cleared before return). Keep botOutcomeReturnedAt + // unset so the reconciler retries once the coordinator can be resolved again. + return false; } const sourceIntent = source.intent ?? "request"; if (sourceIntent !== "request" && sourceIntent !== "question") { diff --git a/packages/db/src/events.test.ts b/packages/db/src/events.test.ts index 5339407..4b93d7e 100644 --- a/packages/db/src/events.test.ts +++ b/packages/db/src/events.test.ts @@ -1678,6 +1678,45 @@ describe("claimSteering", () => { }); describe("clearThread", () => { + it("refuses to clear while a delegated bot outcome is still pending", async () => { + const fanout = new TestFanout(); + const tx = { + thread: { + update: vi + .fn() + .mockResolvedValueOnce({ nextMessageSeq: 4, historyCompactionGeneration: 0 }), + }, + run: { + findMany: vi.fn().mockResolvedValue([]), + count: vi.fn().mockResolvedValue(1), + }, + attempt: { updateMany: vi.fn() }, + task: { updateMany: vi.fn() }, + computerExecutionLease: { updateMany: vi.fn() }, + computer: { updateMany: vi.fn() }, + message: { deleteMany: vi.fn() }, + event: { deleteMany: vi.fn(), create: vi.fn() }, + bot: { update: vi.fn() }, + }; + const prisma = { + $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), + } as unknown as PrismaClient; + + const { ClearThreadPendingDelegatedReturnsError } = await import("./scope.js"); + + await expect( + clearThread(prisma, { spaceId: "workspace-1", threadId: "thread-1", botId: "bot-1" }, fanout), + ).rejects.toBeInstanceOf(ClearThreadPendingDelegatedReturnsError); + expect(tx.run.count).toHaveBeenCalledWith({ + where: expect.objectContaining({ + trigger: "bot_message", + status: { in: ["completed", "failed"] }, + botOutcomeReturnedAt: null, + }), + }); + expect(tx.message.deleteMany).not.toHaveBeenCalled(); + }); + it("releases the computer execution lease without dropping the computer", async () => { const fanout = new TestFanout(); const publish = vi.spyOn(fanout, "publish"); @@ -1690,6 +1729,7 @@ describe("clearThread", () => { }, run: { findMany: vi.fn().mockResolvedValue([{ id: "run-1", taskId: "task-1" }]), + count: vi.fn().mockResolvedValue(0), updateMany: vi.fn(), }, attempt: { updateMany: vi.fn() }, @@ -1754,6 +1794,7 @@ describe("clearThread", () => { { id: "group-run-1", taskId: "task-1" }, { id: "group-run-2", taskId: "task-2" }, ]), + count: vi.fn().mockResolvedValue(0), updateMany: vi.fn(), }, attempt: { updateMany: vi.fn() }, diff --git a/packages/db/src/events.ts b/packages/db/src/events.ts index 3a3de1f..120219d 100644 --- a/packages/db/src/events.ts +++ b/packages/db/src/events.ts @@ -21,6 +21,7 @@ import { createThreadMessageInTransaction, RunHistoryWriteError, } from "./messages.js"; +import { ClearThreadPendingDelegatedReturnsError } from "./scope.js"; import { withTransactionRetry } from "./transaction-retry.js"; const EVENT_BATCH_SIZE = 200; @@ -271,6 +272,19 @@ export async function clearThread( const now = new Date(); const runIds = activeRuns.map((run) => run.id); await cancelRunsInTransaction(tx, activeRuns, now); + const pendingDelegatedReturns = await tx.run.count({ + where: { + spaceId: input.spaceId, + threadId: input.threadId, + ...(input.groupId ? {} : { botId: input.botId }), + trigger: "bot_message", + status: { in: ["completed", "failed"] }, + botOutcomeReturnedAt: null, + }, + }); + if (pendingDelegatedReturns > 0) { + throw new ClearThreadPendingDelegatedReturnsError(); + } // Expire as tombstones so a still-open provider screen claim cannot reset fencing to 1. await expireComputerExecutionLeases(tx, { runId: { in: runIds } }); await tx.computer.updateMany({ diff --git a/packages/db/src/scope.ts b/packages/db/src/scope.ts index cc2ec87..c99b0a2 100644 --- a/packages/db/src/scope.ts +++ b/packages/db/src/scope.ts @@ -16,6 +16,14 @@ export class BotSectionNameConflictError extends Error { } } +/** Thread clear blocked while a delegated bot run still owes its coordinator a return. */ +export class ClearThreadPendingDelegatedReturnsError extends Error { + constructor(message = "Delegated bot work is still finishing") { + super(message); + this.name = "ClearThreadPendingDelegatedReturnsError"; + } +} + export async function requireMembership( prisma: PrismaClient, userId: string,