From f044e17ed5e88322c50732d3308f002425906544 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 22 Sep 2026 09:58:20 -0500 Subject: [PATCH] fix: fence legacy repair accounting and preserve source identity --- ...at-accepted-plan-workspace-refresh.test.ts | 15 ++++- .../heartbeat-comment-wake-batching.test.ts | 56 +++++++------------ ...artbeat-responsible-user-invariant.test.ts | 20 +++++-- .../legacy-continuation-authority.test.ts | 30 +++++++++- .../legacy-finalization-regression.test.ts | 5 ++ .../native-session-resumption.test.ts | 5 ++ server/src/services/recovery/service.ts | 15 ++++- tests/runner-e2e/lifecycle-live-cases.ts | 7 ++- 8 files changed, 104 insertions(+), 49 deletions(-) diff --git a/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts b/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts index 490c070204a..211c9e662cd 100644 --- a/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts +++ b/server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts @@ -112,7 +112,6 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => { }, 20_000); afterEach(async () => { - adapterExecute.mockClear(); // Await every in-flight background heartbeat run to quiescence before the // deletes below. A wakeup claims a run and dispatches its execution // fire-and-forget, and that run can dispatch a follow-up wakeup, so a run or @@ -121,6 +120,7 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => { // wakeup that is still before run registration, which a plain run table // status poll cannot see. await drainHeartbeatRunsToQuiescence(db, heartbeatService(db)); + adapterExecute.mockClear(); while (tempRoots.length > 0) { const root = tempRoots.pop(); if (root) await rm(root, { recursive: true, force: true }).catch(() => undefined); @@ -535,7 +535,14 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => { }); const heartbeat = heartbeatService(db); - adapterExecute.mockImplementationOnce(async () => ({ + adapterExecute.mockImplementationOnce(async () => { + // Planning is awaiting a real confirmation, not a prose-only promise. + await db.insert(issueThreadInteractions).values({ + companyId, issueId: sourceIssueId, kind: "request_confirmation", status: "pending", + requestedResolverPolicy: "anyone", effectiveResolverPolicy: "anyone", + payload: { version: 1, prompt: "Review the source plan" }, + }); + return { exitCode: 0, signal: null, timedOut: false, @@ -544,7 +551,7 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => { summary: "Realized the planning source workspace.", provider: "test", model: "test-model", - })); + }; }); const sourceRun = await heartbeat.wakeup(agentId, { source: "automation", @@ -582,6 +589,8 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => { await runGit(repoRoot, ["push", "origin", "HEAD:master"]); await runGit(repoRoot, ["fetch", "origin", "master"]); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + await db.delete(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, sourceIssueId)); const acceptedPlanRevisionId = await seedAcceptedPlanAcceptance({ companyId, issueId: sourceIssueId, diff --git a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts index 129249352e4..8ee01916e17 100644 --- a/server/src/__tests__/heartbeat-comment-wake-batching.test.ts +++ b/server/src/__tests__/heartbeat-comment-wake-batching.test.ts @@ -68,7 +68,7 @@ async function closeDbClient(db: ReturnType | undefined) { await db?.$client?.end?.({ timeout: 0 }); } -async function createControlledGatewayServer() { +async function createControlledGatewayServer(beforeComplete?: (turn: number) => Promise) { const server = createServer(); const wss = new WebSocketServer({ server }); const agentPayloads: Array> = []; @@ -151,6 +151,7 @@ async function createControlledGatewayServer() { if (waitCount === 1) { await firstWaitGate; } + await beforeComplete?.(waitCount); socket.send( JSON.stringify({ type: "res", @@ -697,7 +698,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { }, 120_000); it("cancels an empty deferred comment wake instead of promoting deleted input", async () => { - const gateway = await createControlledGatewayServer(); + const gateway = await createControlledGatewayServer(async turn => { + if (turn === 1) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + }); const companyId = randomUUID(); const agentId = randomUUID(); const issueId = randomUUID(); @@ -1644,7 +1647,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { }, 120_000); it("promotes an interaction continuation with its full authoritative source comment after removing a coalesced self-comment", async () => { - const gateway = await createControlledGatewayServer(); + const gateway = await createControlledGatewayServer(async turn => { + if (turn === 2) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + }); const companyId = randomUUID(); const agentId = randomUUID(); const issueId = randomUUID(); @@ -2929,7 +2934,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { }, 120_000); it("promotes an interaction continuation after removing a coalesced self-authored comment", async () => { - const gateway = await createControlledGatewayServer(); + const gateway = await createControlledGatewayServer(async turn => { + if (turn === 2) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + }); const companyId = randomUUID(); const agentId = randomUUID(); const issueId = randomUUID(); @@ -3616,7 +3623,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { } }, 120_000); it("treats the automatic run summary as fallback-only when the run already posted a comment", async () => { - const gateway = await createControlledGatewayServer(); + const gateway = await createControlledGatewayServer(async turn => { + if (turn === 2) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + }); const companyId = randomUUID(); const agentId = randomUUID(); const issueId = randomUUID(); @@ -3714,30 +3723,10 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { expect(sourceRun?.issueCommentStatus).toBe("satisfied"); expect(sourceRun?.issueCommentSatisfiedByCommentId).not.toBeNull(); - await waitFor(async () => { - const comments = await db - .select() - .from(issueComments) - .where(eq(issueComments.issueId, issueId)); - const wakeups = await db - .select() - .from(agentWakeupRequests) - .where( - and( - eq(agentWakeupRequests.companyId, companyId), - eq(agentWakeupRequests.agentId, agentId), - ), - ); - - const hasHandoffComment = comments.some( - (comment) => - comment.body === SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY, - ); - const hasHandoffWake = wakeups.some( - (wakeup) => wakeup.reason === "finish_successful_run_handoff", - ); - return hasHandoffComment && hasHandoffWake; - }); + await waitFor(async () => (await db.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.companyId, companyId))) + .some(wake => wake.reason === "issue_disposition_repair")); + await heartbeat.drainActiveRunExecutions(); const comments = await db .select() @@ -3751,12 +3740,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { comment.body === "Manual completion comment from the run.", ), ).toBe(true); - expect( - comments.some( - (comment) => - comment.body === SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY, - ), - ).toBe(true); + expect(comments.some(comment => comment.body === SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY)).toBe(false); expect( comments.every((comment) => !comment.body.startsWith("## Run summary")), ).toBe(true); @@ -3776,7 +3760,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => { ).toBe(false); expect( wakeups.some( - (wakeup) => wakeup.reason === "finish_successful_run_handoff", + (wakeup) => wakeup.reason === "issue_disposition_repair", ), ).toBe(true); } finally { diff --git a/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts b/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts index 0fd788d2589..18d642732a6 100644 --- a/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts +++ b/server/src/__tests__/heartbeat-responsible-user-invariant.test.ts @@ -87,6 +87,14 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => { tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-responsible-user-"); db = createDb(tempDb.connectionString); heartbeat = heartbeatService(db); + const baseExecute = mockAdapterExecute.getMockImplementation()!; + mockAdapterExecute.mockImplementation(async (...args: unknown[]) => { + const context = (args[0] as { context?: Record } | undefined)?.context; + if (context?.wakeReason === "issue_disposition_repair" && typeof context.issueId === "string") { + await db.update(issues).set({ status: "done" }).where(eq(issues.id, context.issueId)); + } + return baseExecute(); + }); }, 20_000); afterEach(async () => { @@ -301,6 +309,7 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => { const sourceRunIds: string[] = []; for (let attempt = 0; attempt < 3; attempt += 1) { + await db.update(issues).set({ status: "todo" }).where(eq(issues.id, issueId)); const wakeReason = "issue_blockers_resolved"; const run = await heartbeat.wakeup(agentId, { source: "automation", @@ -321,16 +330,16 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => { await drainHeartbeatRunsToQuiescence(db, heartbeat); } // The deliberately disposition-free adapter response schedules one bounded - // handoff per source run. Those automatic continuations retain its identity. + // repair per source run; the fixture records done on repair. Each retains its identity. const runs = await db.select().from(heartbeatRuns); const handoffs = runs.filter((run) => !sourceRunIds.includes(run.id)); expect(handoffs).toHaveLength(3); expect( - handoffs.map((run) => run.contextSnapshot?.parentRunId).sort(), + handoffs.map((run) => run.contextSnapshot?.retryOfRunId).sort(), ).toEqual(sourceRunIds.sort()); for (const handoff of handoffs) { expect(handoff.contextSnapshot?.wakeReason).toBe( - "finish_successful_run_handoff", + "issue_disposition_repair", ); expect(handoff.responsibleUserId).toBe(issueResponsibleUserId); expect(handoff.status).toBe("succeeded"); @@ -379,7 +388,10 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => { expect(completed?.responsibleUserId).toBe(commenterUserId); const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); expect(issue?.responsibleUserId).toBe(issueResponsibleUserId); - expect(mockAdapterExecute).toHaveBeenCalledTimes(1); + await drainHeartbeatRunsToQuiescence(db, heartbeat); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + expect(runs.every(row => row.responsibleUserId === commenterUserId)).toBe(true); + expect(mockAdapterExecute).toHaveBeenCalledTimes(runs.length); }, ); diff --git a/server/src/__tests__/legacy-continuation-authority.test.ts b/server/src/__tests__/legacy-continuation-authority.test.ts index 9b2369baadd..cc1a6d54f58 100644 --- a/server/src/__tests__/legacy-continuation-authority.test.ts +++ b/server/src/__tests__/legacy-continuation-authority.test.ts @@ -16,13 +16,13 @@ describe("legacy continuation persisted authority", () => { await db.insert(agents).values({ id: agentId, companyId, name: "Worker", role: "engineer", status: "idle", adapterType: "codex_local", runtimeConfig: { heartbeat: { wakeOnDemand: true } } }); await db.insert(issues).values({ id: issueId, companyId, title: "Implement export", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: "fixture-owner" }); await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "succeeded", runtimeMode: "legacy", continuationAttempt, contextSnapshot: { issueId, ...context }, livenessState: "blocked", resultJson: { summary: "All done. Need approval. I will continue." } }); - const createRecovery = () => recoveryService(db, { + const createRecovery = (afterEnqueue?: (run: typeof heartbeatRuns.$inferSelect) => Promise) => recoveryService(db, { enqueueWakeup: async (targetAgentId, opts) => db.transaction(async tx => { const [wake] = await tx.insert(agentWakeupRequests).values({ companyId, agentId: targetAgentId, source: "automation", reason: opts?.reason, payload: opts?.payload, idempotencyKey: opts?.idempotencyKey, status: "queued" }).returning(); const [run] = await tx.insert(heartbeatRuns).values({ companyId, agentId: targetAgentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", wakeupRequestId: wake.id, contextSnapshot: opts?.contextSnapshot }).returning(); await tx.update(agentWakeupRequests).set({ runId: run.id }).where(eq(agentWakeupRequests.id, wake.id)); return run; - }), + }).then(async run => { await afterEnqueue?.(run); return run; }), }); const runs = () => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); const actions = () => db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, companyId)); @@ -43,6 +43,17 @@ describe("legacy continuation persisted authority", () => { const repair = (await f.runs()).find(r => r.id !== f.runId)!; expect(repair.contextSnapshot).toMatchObject({ legacyDispositionEpisode: { id: f.runId, attempt: 1, maxAttempts: 2 } }); }); + it("does not roll back the ledger when a fast repair finishes before enqueue returns", async () => { + const f = await fixture(); + await f.createRecovery(async first => { + await f.finish(first); + await f.createRecovery().reconcileLegacyContinuation(first.id); + expect((await f.actions())[0].attemptCount).toBe(2); + }).reconcileLegacyContinuation(f.runId); + const second = (await f.runs()).find(r => r.status === "scheduled_retry")!; + expect((await f.actions())[0]).toMatchObject({ attemptCount: 2, wakePolicy: { scheduledRunId: second.id, attempt: 2 } }); + expect(await f.runs()).toHaveLength(3); + }); it("keeps the same bounded episode across restart and commentary, then escalates only after agent attempts", async () => { const f = await fixture(); await f.createRecovery().reconcileLegacyContinuation(f.runId); @@ -59,6 +70,21 @@ describe("legacy continuation persisted authority", () => { await f.createRecovery().reconcileLegacyContinuation(second.id); expect(await f.runs()).toHaveLength(3); }); + it("persists the source identity while the second repair waits to dispatch", async () => { + const f = await fixture(); + await f.createRecovery().reconcileLegacyContinuation(f.runId); + const first = (await f.runs()).find(r => r.id !== f.runId)!; + await db.update(heartbeatRuns).set({ responsibleUserId: "initiating-operator" }).where(eq(heartbeatRuns.id, first.id)); + await f.finish(first); + await f.createRecovery().reconcileLegacyContinuation(first.id); + expect((await f.runs()).find(r => r.status === "scheduled_retry")?.responsibleUserId).toBe("initiating-operator"); + }); + it.each(["retry_queued", "retry_exhausted"])("does not stack a new repair budget on the %s comment policy", async issueCommentStatus => { + const f = await fixture(); + await db.update(heartbeatRuns).set({ issueCommentStatus }).where(eq(heartbeatRuns.id, f.runId)); + expect(await f.createRecovery().reconcileLegacyContinuation(f.runId)).toBe("skipped"); + expect(await f.runs()).toHaveLength(1); + }); it("does not grant a new budget to exhausted pre-upgrade continuation", async () => { const f = await fixture({}, 2); expect(await f.createRecovery().reconcileLegacyContinuation(f.runId)).toBe("escalated"); diff --git a/server/src/__tests__/legacy-finalization-regression.test.ts b/server/src/__tests__/legacy-finalization-regression.test.ts index c400d364ccb..f53692b9fbe 100644 --- a/server/src/__tests__/legacy-finalization-regression.test.ts +++ b/server/src/__tests__/legacy-finalization-regression.test.ts @@ -104,6 +104,11 @@ describe("P6-32 legacy finalization regression", () => { }); it("executes a flag-off heartbeat through the legacy adapter with byte-stable reads and zero native rows", async () => { + const execute = adapterExecute.getMockImplementation()!; + adapterExecute.mockImplementationOnce(async () => { + await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId)); + return execute(); + }); const heartbeat = heartbeatService(db); const queued = await heartbeat.wakeup(agentId, { source: "automation", diff --git a/server/src/__tests__/native-session-resumption.test.ts b/server/src/__tests__/native-session-resumption.test.ts index 169c1acb5e1..1283b29a463 100644 --- a/server/src/__tests__/native-session-resumption.test.ts +++ b/server/src/__tests__/native-session-resumption.test.ts @@ -1392,6 +1392,11 @@ describe.each(["unchanged", "newer_active", "stale_idle"] as const)( assigneeAgentId: agentId, workMode: "standard", }); + const directExecute = legacyAdapterExecute.getMockImplementation()!; + legacyAdapterExecute.mockImplementationOnce(async () => { + await db.update(issues).set({ status: "done" }).where(eq(issues.id, freshIssueId)); + return directExecute(); + }); const fresh = await heartbeat.wakeup(agentId, { source: "automation", triggerDetail: "system", diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index b64e87be663..0c7b03d92d4 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -3093,6 +3093,11 @@ export function recoveryService( }) { const agentId = input.issue.assigneeAgentId; if (!agentId) return null; + // Preserve the initiating identity on the durable row, including while a + // delayed repair is waiting to dispatch. Never substitute the issue owner. + const sourceRun = input.latestRun?.id ? await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.latestRun.id), eq(heartbeatRuns.companyId, input.issue.companyId))) + .limit(1).then(rows => rows[0]) : null; const timing = dispositionRepairDelayMs(input.attempt, input.fingerprint); const now = new Date(); const retryAt = new Date(now.getTime() + timing.delayMs); @@ -3208,6 +3213,7 @@ export function recoveryService( scheduledRetryAt: retryAt, scheduledRetryAttempt: input.attempt, scheduledRetryReason: ISSUE_DISPOSITION_REPAIR_RETRY_REASON, + responsibleUserId: sourceRun?.responsibleUserId ?? null, contextSnapshot: context, updatedAt: now, }) @@ -3254,6 +3260,12 @@ export function recoveryService( and( eq(issueRecoveryActions.id, input.action.id), eq(issueRecoveryActions.companyId, input.issue.companyId), + // A fast successor can reserve the next slot before enqueue returns. + // An older scheduler must not rewind its ledger or resurrect a wait. + eq(issueRecoveryActions.status, "active"), + eq(issueRecoveryActions.ownerType, "agent"), + eq(issueRecoveryActions.fingerprint, input.fingerprint), + sql`${issueRecoveryActions.attemptCount} <= ${input.attempt}`, ), ); @@ -3658,7 +3670,8 @@ export function recoveryService( paused: pause, budgetBlocked: budget, pendingWait: state.hasDurableWaitingPath || durableWait || parseIssueExecutionState(issue.executionState)?.status === "pending" || Boolean(issue.monitorNextCheckAt), activeExecution: state.hasActiveExecutionPath || latest?.id !== (dispatchRunId ?? run.id), - ownedLifecycle: Boolean(readNonEmptyString(context.goalControlRequestId)) || context.resumeSessionGoalHeartbeat === true || + ownedLifecycle: run.issueCommentStatus === "retry_queued" || run.issueCommentStatus === "retry_exhausted" || + Boolean(readNonEmptyString(context.goalControlRequestId)) || context.resumeSessionGoalHeartbeat === true || isPluginManagedIssueLifecycle(issue) || routine.length > 0 || workspaceChildren.length > 0 || Boolean(goal[0]?.status && goal[0].status !== "complete") || Boolean(active && !ownsRepair), conversation: Boolean(issue.conversationAgentId) || isWaitingConversation(issue), diff --git a/tests/runner-e2e/lifecycle-live-cases.ts b/tests/runner-e2e/lifecycle-live-cases.ts index e5b90108b62..bdf4270629b 100644 --- a/tests/runner-e2e/lifecycle-live-cases.ts +++ b/tests/runner-e2e/lifecycle-live-cases.ts @@ -60,7 +60,7 @@ export const lifecycleLiveCases = lifecycleNarrativePairs.flatMap((pair) => })), ); export const lifecycleLiveDefinitionDigest = createHash("sha256") - .update(JSON.stringify({ version: 4, cases: lifecycleLiveCases })) + .update(JSON.stringify({ version: 5, cases: lifecycleLiveCases })) .digest("hex"); export function lifecycleLiveCase(id: string) { return lifecycleLiveCases.find((c) => c.id === id); @@ -107,8 +107,9 @@ export const lifecycleLiveTasks: readonly RunnerTaskFixture[] = buildVisibleMarker: nonce => `REPAIRED_${nonce}`, buildPrompt: nonce => [ "This is a bounded two-turn disposition repair exercise. Use the public Paperclip API with normal authorization and run attribution headers.", - `On the initial assignment turn only: post exactly one task comment whose complete body is ${JSON.stringify(probe.narrative)}, then end the provider turn with the same text. Leave task status in_progress. Do not record completion, a blocker, an approval, a question, a monitor, children, or a continuation. The quoted words do not describe actual authorization or missing prerequisites.`, - `When Paperclip wakes you for task disposition repair: PATCH the current task to status done, with the exact comment REPAIRED_${nonce}. End the provider turn. Do not repeat the initial comment or create further work.`, + "Before any mutation, read PAPERCLIP_WAKE_REASON from the process environment. Branch only on that exact value: issue_assigned is INITIAL; issue_disposition_repair is REPAIR. Never infer the phase from the task title, these instructions, or quoted words. Execute only the selected branch, then end this turn.", + `INITIAL branch (PAPERCLIP_WAKE_REASON is issue_assigned): post exactly one task comment whose complete body is ${JSON.stringify(probe.narrative)}, then end the provider turn with the same text. Leave task status in_progress. Do not record completion, a blocker, an approval, a question, a monitor, children, or a continuation. The quoted words do not describe actual authorization or missing prerequisites.`, + `REPAIR branch (PAPERCLIP_WAKE_REASON is issue_disposition_repair): PATCH the current task to status done, with the exact comment REPAIRED_${nonce}. End the provider turn. Do not repeat the initial comment or create further work.`, ].join("\n"), buildMatchers: (nonce, execution) => [ { kind: "message_exact", expected: `REPAIRED_${nonce}` },