RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: preserve handoff queues without restarting the former assignee 

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta 2026-09-12 21:58:40 -05:00
parent 8f53d44418
commit 8bb2e5acd8
5 changed files with 114 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
server/src/modules/wake-queue/adapters/postgres.test.ts
View File

@ -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();

3
server/src/modules/wake-queue/adapters/postgres.ts
View File

@ -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,

2
server/src/modules/wake-queue/application/ports.ts
View File

@ -110,7 +110,7 @@ export type PromoteDeferredWakeInput = {
*/
export interface WakeQueueTransaction {
findInvokableAgent(input: { companyId: string; agentId: string }): Promise<InvokableAgentSnapshot | null>;
findNextDeferredWake(input: { companyId: string; issueId: string }): Promise<DeferredWakeCandidate | null>;
findNextDeferredWake(input: { companyId: string; issueId: string; excludedWakeIds?: string[] }): Promise<DeferredWakeCandidate | null>;
getQueuedCommentLiveness(input: {
companyId: string;
issueId: string;

62
server/src/modules/wake-queue/application/use-cases.test.ts
View File

@ -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() });

31
server/src/modules/wake-queue/application/use-cases.ts
View File

@ -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<string>();
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 &&