From 8bb2e5acd849955064ed4d4343357e6f210ea746 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 12 Sep 2026 21:58:40 -0500 Subject: [PATCH] fix: preserve handoff queues without restarting the former assignee Co-Authored-By: Paperclip --- .../wake-queue/adapters/postgres.test.ts | 21 +++++++ .../modules/wake-queue/adapters/postgres.ts | 3 +- .../modules/wake-queue/application/ports.ts | 2 +- .../wake-queue/application/use-cases.test.ts | 62 ++++++++++++++++++- .../wake-queue/application/use-cases.ts | 31 +++++++++- 5 files changed, 114 insertions(+), 5 deletions(-) diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 8a5a2b1951..94dacfd82a 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -292,6 +292,27 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { // Review test (a): a foreign-company agent id produces the current failed // wake status and the current error text, and creates no run. + it("skips preserved handoff receipts for one drain without changing their durable state", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId }); + const runId = await seedRun({ companyId, agentId, contextSnapshot: { issueId }, status: "succeeded" }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const previous = await seedDeferredWake({ companyId, agentId, issueId }); + const next = await seedDeferredWake({ companyId, agentId, issueId }); + await db.update(agentWakeupRequests).set({ requestedAt: new Date("2026-01-01") }).where(eq(agentWakeupRequests.id, previous)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async (_locked, ports) => { + expect((await ports.transaction.findNextDeferredWake({ companyId, issueId }))?.id).toBe(previous); + expect((await ports.transaction.findNextDeferredWake({ companyId, issueId, excludedWakeIds: [previous] }))?.id).toBe(next); + expect(await ports.transaction.findNextDeferredWake({ companyId, issueId, excludedWakeIds: [previous, next] })).toBeNull(); + return { outcome: { kind: "released" as const }, postCommitEffects: [] }; + }); + const [preserved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, previous)); + expect(preserved.status).toBe("deferred_issue_execution"); + expect(preserved.runId).toBeNull(); + }); + it("fails a deferred wake whose agent belongs to a different company, without creating a run", async () => { const companyId = await seedCompany(); const otherCompanyId = await seedCompany(); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index e22e2caeb9..e52d385791 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -198,7 +198,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable }; }, - async findNextDeferredWake({ companyId, issueId }) { + async findNextDeferredWake({ companyId, issueId, excludedWakeIds }) { while (true) { const row = await tx .select() @@ -207,6 +207,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru and( eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS), + excludedWakeIds?.length ? notInArray(agentWakeupRequests.id, excludedWakeIds) : undefined, sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, interruptQueueId ? eq(agentWakeupRequests.id, interruptQueueId) : undefined, interruptQueueId ? eq(agentWakeupRequests.agentId, run.agentId) : undefined, diff --git a/server/src/modules/wake-queue/application/ports.ts b/server/src/modules/wake-queue/application/ports.ts index c4e4cc0f9e..47fd3bd374 100644 --- a/server/src/modules/wake-queue/application/ports.ts +++ b/server/src/modules/wake-queue/application/ports.ts @@ -110,7 +110,7 @@ export type PromoteDeferredWakeInput = { */ export interface WakeQueueTransaction { findInvokableAgent(input: { companyId: string; agentId: string }): Promise; - findNextDeferredWake(input: { companyId: string; issueId: string }): Promise; + findNextDeferredWake(input: { companyId: string; issueId: string; excludedWakeIds?: string[] }): Promise; getQueuedCommentLiveness(input: { companyId: string; issueId: string; diff --git a/server/src/modules/wake-queue/application/use-cases.test.ts b/server/src/modules/wake-queue/application/use-cases.test.ts index 528acc5391..e5c8fae2ed 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -144,11 +144,71 @@ function createFakeRecovery(): RecoveryEscalationPort { } describe("releaseIssueExecution", () => { + it("preserves the former owner's queue for handoff adoption while draining the new owner's wake", async () => { + const stale = wakeCandidate({ agentId: RUN.agentId, queuedCommentIds: ["saved-user-direction"] }); + const current = wakeCandidate({ id: "wake-new-owner", agentId: "new-agent" }); + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async (input: { companyId: string; issueId: string; excludedWakeIds?: string[] }) => + input.excludedWakeIds?.includes(stale.id) ? current : stale), + getQueuedCommentLiveness: vi.fn(async () => ({ liveNonSelfCommentIds: ["saved-user-direction"], containedSelfAuthoredComment: false })), + }); + const release = createReleaseIssueExecution({ + issueLock: createFakeIssueLock(createFakeHost(), transaction, { ...ISSUE, assigneeAgentId: "new-agent" }), + recovery: createFakeRecovery(), + }); + const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() }); + expect(result.outcome.kind).toBe("promoted"); + expect(transaction.finalizePromotedWake).toHaveBeenCalledWith(expect.objectContaining({ wakeId: current.id })); + expect(transaction.cancelDeferredWake).not.toHaveBeenCalled(); + }); + + it.each(["done", "in_progress"])("does not promote a former assignee's saved instruction after handoff (%s)", async (status) => { + const queuedCommentIds = ["saved-user-direction"]; + const queue = [wakeCandidate({ + agentId: "previous-agent", + reason: "issue_execution_deferred", + queuedCommentIds, + deferredCommentIds: queuedCommentIds, + deferredContextSeed: { wakeReason: "issue_commented", wakeCommentIds: queuedCommentIds }, + })]; + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async () => queue.shift() ?? null), + getQueuedCommentLiveness: vi.fn(async () => ({ liveNonSelfCommentIds: queuedCommentIds, containedSelfAuthoredComment: false })), + }); + const release = createReleaseIssueExecution({ + issueLock: createFakeIssueLock(createFakeHost(), transaction, { ...ISSUE, status }), + recovery: createFakeRecovery(), + }); + await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date(), suppressImmediateRecovery: true }); + expect(transaction.cancelDeferredWake).toHaveBeenCalledWith(expect.objectContaining({ wakeId: "wake-1" })); + expect(transaction.claimDeferredWakeForPromotion).not.toHaveBeenCalled(); + expect(transaction.finalizePromotedWake).not.toHaveBeenCalled(); + expect(transaction.reopenIssue).not.toHaveBeenCalled(); + }); + + it.each([ + { agentId: ISSUE.assigneeAgentId!, wakeReason: "issue_commented", preservesIndependentContinuation: false, authorizedFailedChatRetry: false }, + { agentId: "mentioned-agent", wakeReason: "issue_comment_mentioned", preservesIndependentContinuation: false, authorizedFailedChatRetry: false }, + { agentId: "interaction-agent", wakeReason: "issue_commented", preservesIndependentContinuation: true, authorizedFailedChatRetry: false }, + { agentId: "chat-agent", wakeReason: "issue_commented", preservesIndependentContinuation: false, authorizedFailedChatRetry: true }, + ])("preserves the independently authorized $agentId/$wakeReason wake", async (authority) => { + const queuedCommentIds = ["saved-user-direction"]; + const queue = [wakeCandidate({ ...authority, queuedCommentIds, deferredCommentIds: queuedCommentIds })]; + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async () => queue.shift() ?? null), + getQueuedCommentLiveness: vi.fn(async () => ({ liveNonSelfCommentIds: queuedCommentIds, containedSelfAuthoredComment: false })), + }); + const release = createReleaseIssueExecution({ issueLock: createFakeIssueLock(createFakeHost(), transaction), recovery: createFakeRecovery() }); + expect((await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() })).outcome.kind).toBe("promoted"); + expect(transaction.cancelDeferredWake).not.toHaveBeenCalled(); + }); + it.each([true, false])( "preserves failed-chat retry input without reopening only with adapter proof: %s", async (authorizedFailedChatRetry) => { const queue = [ wakeCandidate({ + agentId: authorizedFailedChatRetry ? AGENT.id : ISSUE.assigneeAgentId!, authorizedFailedChatRetry, queuedCommentIds: ["original-comment"], deferredCommentIds: ["original-comment"], @@ -292,7 +352,7 @@ describe("releaseIssueExecution", () => { ); const transaction = createFakeTransaction({ findNextDeferredWake, findInvokableAgent, getQueuedCommentLiveness }); const host = createFakeHost(); - const issueLock = createFakeIssueLock(host, transaction); + const issueLock = createFakeIssueLock(host, transaction, { ...ISSUE, assigneeAgentId: AGENT.id }); const releaseIssueExecution = createReleaseIssueExecution({ issueLock, recovery: createFakeRecovery() }); const result = await releaseIssueExecution({ companyId: "company-1", runId: "run-1", now: new Date() }); diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index c9c35fee7d..434911bae7 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -143,15 +143,20 @@ async function runReleaseDrain( return runReleaseRecoveryTail(issue, run, ports.host, ports.transaction, input, postCommitEffects); } - // Each `continue` path below leaves the wake row off the + // Each `continue` path either excludes a pending handoff receipt from + // this drain or leaves the wake row off the // `deferred_issue_execution` status, so the next queue read cannot // return that same row again. That invariant is what ends this loop. // The `processedWakeIds` guard below makes a break of the invariant // fail loudly, instead of holding this transaction open forever. const processedWakeIds = new Set(); + const handoffWakeIds: string[] = []; while (true) { - const candidate = await ports.transaction.findNextDeferredWake({ companyId: run.companyId, issueId: issue.id }); + const candidate = await ports.transaction.findNextDeferredWake({ + companyId: run.companyId, issueId: issue.id, + ...(handoffWakeIds.length ? { excludedWakeIds: handoffWakeIds } : {}), + }); if (!candidate) break; if (processedWakeIds.has(candidate.id)) { throw new WakeQueueApplicationError( @@ -162,6 +167,28 @@ async function runReleaseDrain( } processedWakeIds.add(candidate.id); + const ordinaryTaskComment = !candidate.authorizedFailedChatRetry && + !candidate.preservesIndependentContinuation && candidate.queuedCommentIds.length > 0 && + ["issue_commented", "issue_reopened_via_comment"].includes(candidate.wakeReason ?? candidate.reason ?? ""); + if (ordinaryTaskComment && candidate.agentId !== issue.assigneeAgentId) { + if (run.agentId !== issue.assigneeAgentId) { + // The old owner can release before assignment admission adopts these + // exact IDs. Leave its receipt intact, skip it for this drain, and let + // a current-assignee wake behind it proceed. + handoffWakeIds.push(candidate.id); + } else { + // The current owner has finished. An obsolete assignment cannot + // launch another former-owner run or reopen its completed task. + await ports.transaction.cancelDeferredWake({ + companyId: run.companyId, + wakeId: candidate.id, + reason: "Deferred task messages now belong to the current assignee", + now: input.now, + }); + } + continue; + } + let liveness = { liveNonSelfCommentIds: candidate.queuedCommentIds, containedSelfAuthoredComment: false }; if ( !candidate.authorizedFailedChatRetry &&