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,