From ea42a2e41715c70a6822a25683136e48e35ae811 Mon Sep 17 00:00:00 2001 From: Rockiez Date: Thu, 5 Nov 2026 15:54:51 +0800 Subject: [PATCH] fix(recovery): coalesce duplicate legacy no-replay holds (#17055) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Recovery records keep uncertain execution outcomes from being replayed. > - A settled record can still block replay while it waits for a verified decision. > - Repeated scans can create more settled records for the same failed execution. > - Resolving only one duplicate leaves the task behind another hold. > - This pull request retains the settled hold and applies one verified decision to its exact duplicates. > - The task can continue through the existing delivery path without creating another delivery owner for each duplicate. ## Linked Issues or Issue Description Refs #14600. This change addresses duplicate settled holds and their same-source resolution. The visibility and other deferred-wake paths remain outside this change. Related: #10011 handles unchanged active incidents. This patch handles settled no-replay records. #15285 covers missing reconciliation input. #17031 covers a handoff to a new task owner. External-adapter continuation is covered separately by #15982. ## What Changed - Retain a resolved legacy no-replay decision when a later scan finds the same company, task, cause, fingerprint, and source run. - After a verified reconciliation, resolve duplicate records that also match the selected action's kind. Keep the selected action as the delivery owner for this group. Record the merged action IDs in the activity log. - Preserve independent workspace and system recovery records, existing reconciliations, and records for other sources. - Add PostgreSQL regression tests for repeated scans, historical duplicates, delivery count, repeated resolution, scope boundaries, simultaneous resolutions, and rejected reconciliation rollback. - Update the execution semantics and recovery maps. This patch has no schema migration or adapter-specific code. ## Verification Current head: `3c84bc89325d40bfdda5d28f24ed2fb67d48756c`. It merges `master` at `043f2514e67c50320433a0013fb228a75a51325c` without rewriting the existing commits. The only conflict was in `markExecutionReconciliation`. The resolution retains the upstream transaction and all three queue notifications. Source, sibling, and audit writes use the same transaction. The final merged tree passed: ```sh pnpm exec vitest run server/src/__tests__/issue-recovery-actions.test.ts server/src/services/legacy-execution-recovery.test.ts server/src/__tests__/heartbeat-process-recovery.test.ts server/src/__tests__/execution-control-work.test.ts --maxWorkers=1 # 549 passed; zero skipped pnpm --filter @paperclipai/server exec tsc --noEmit # exit 0 node scripts/check-maps.mjs # passed: 35 areas, 191 capability IDs git diff origin/master --check # passed ``` Three new cases add real PostgreSQL and HTTP-route evidence. Two resolution requests target different duplicate holds while both wait on database locks. They leave one continuation owner, one coalescing audit record, and one queued successor. A live-provider rejection leaves all records unchanged. A checkout rejection after reconciliation writes rolls back the source, siblings, and audit row. The fixture occupies the dispatch slot, so no external model provider runs. The new contention case was also run with only `execution-recovery-resolution.ts` replaced by its parent revision at `65f59e339a9fb6e0635bcbd8c36791bd610326a6`. It failed with two continuation owners instead of one. The source was restored byte-for-byte, and all three new cases passed again before the merge. The final four-suite run above validates the merge with the upstream notification changes. The original positive regressions also failed on that parent revision: three scan-and-settle cycles left four records instead of one, and resolving one historical hold left its two duplicates unreconciled. The scope matrix covers company, task, kind, cause, fingerprint, source run, replay state, protected recovery records, and prior reconciliation. Fresh-checkout collection initially failed because plugin SDK build outputs were absent. After building them, PostgreSQL startup exposed native library symlinks missing from an installation with scripts disabled. The package's symlink setup resolved that failure. The standard server typecheck attempt started a Rust vendor prerequisite build and was stopped; direct TypeScript checking passed. These failed or incomplete attempts are not counted as passes. Full repository build, recursive typecheck, and CI were successful on the earlier `08e5c0285` revision ([CI run](https://github.com/paperclipai/paperclip/actions/runs/39541949702)). They are historical evidence, not results for this merge. The current local proof is the four suites and checks listed above. Full upstream CI for the merge is pending. Greptile reviewed `3c84bc89325d40bfdda5d28f24ed2fb67d48756c` with 5/5 confidence and no actionable findings; that source review is separate from test execution. ## Risks - A match that is too broad could clear an independent hold. Tests cover company, task, kind, cause, fingerprint, source run, replay state, workspace recovery, system recovery, and prior reconciliation boundaries. - The existing issue-lock transaction covers the selected action and its duplicates. The patch does not create a new lock or an automatic retry path. - The patch does not infer that a network or provider failure means no action occurred. Existing reconciliation validation still checks the source, owner, stopped processes, leases, and observed action outcome. - Previously reconciled records and their delivery state remain unchanged. This patch prevents new duplicate delivery owners when it resolves unreconciled duplicates. It does not migrate every historical delivery record. ## Model Used OpenAI Codex, based on GPT-6, assisted with the implementation, tests, and review. It used repository tools and code execution. The exact runtime model ID, context window size, and reasoning setting are not exposed in this session. An independent AGY review also ran; its provider model ID was not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with the available version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have linked the existing issue with `Refs #14600` - [x] I have not referenced internal/instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal task IDs or instance details - [x] I have run the focused tests locally and they pass; full upstream CI is pending for the final merge - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented risks above - [ ] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Rockiez <7697772+Rockiez@users.noreply.github.com> Co-authored-by: Devin Foley Co-authored-by: Paperclip --- doc/execution-semantics.md | 7 + .../heartbeat-process-recovery.test.ts | 43 +++ .../__tests__/issue-recovery-actions.test.ts | 331 +++++++++++++++++- .../services/execution-recovery-resolution.ts | 42 ++- .../src/services/legacy-execution-recovery.ts | 6 + 5 files changed, 426 insertions(+), 3 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index aae58c6b2a7..d3e26a36722 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -419,6 +419,13 @@ provider has stopped or change native cancellation acknowledgement. A known execution hold is a waiting condition, not a new execution attempt. Every issue wake must read the current effective reconciliation hold under the issue admission lock before creating a run. Resolved recovery bookkeeping can still carry a no-replay hold; only clearing the effective hold makes admission eligible again. The final dispatch gate remains required for changes after admission. +Repeated legacy recovery scans retain one settled no-replay decision for the same +company, task, cause, fingerprint, and source run. A different source run can +still require its own decision. When the board reconciles that exact execution, +historical duplicate holds share the verified decision and one continuation +delivery owner. Separate workspace or system recovery obligations and records +with an existing reconciliation remain unchanged. + A queued legacy run denied by that final gate is still a wait. Its matching dispatch-gate receipt, task identity, and absence of runtime selection, execution ownership, environment leases, coordinator, or execution events prove that it diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index f55ea858677..d7edf39fe74 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -4952,6 +4952,49 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { ).toHaveLength(1); }); + it("preserves a settled no-replay decision across repeated terminalization while holding a new source run", async () => { + const { companyId, agentId, runId, issueId } = await seedRunFixture({ + adapterType: "process", + agentStatus: "idle", + runStatus: "failed", + }); + const { settleUnrecoverableExecutions } = await import("../services/execution-recovery-resolution.js"); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + await terminalizeLegacyExecution({ db, run: run!, status: "failed" }); + await settleUnrecoverableExecutions(db); + const [settled] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(settled).toMatchObject({ + status: "resolved", + outcome: "blocked", + evidence: { runId, automaticRecovery: { replay: "blocked" } }, + }); + + for (let attempt = 0; attempt < 3; attempt += 1) { + const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + await terminalizeLegacyExecution({ db, run: current!, status: "failed" }); + await settleUnrecoverableExecutions(db); + } + expect(await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId))).toEqual([settled]); + expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ runId }); + + const [newSource] = await db.insert(heartbeatRuns).values({ + companyId, agentId, status: "failed", contextSnapshot: { issueId }, + }).returning(); + await terminalizeLegacyExecution({ db, run: newSource!, status: "failed" }); + const actions = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, issueId)); + expect(actions).toHaveLength(2); + expect(actions.find((action) => action.id === settled!.id)).toEqual(settled); + expect(actions.find((action) => action.id !== settled!.id)).toMatchObject({ + status: "active", + cause: "legacy_execution_requires_reconciliation", + fingerprint: `legacy-execution:${newSource!.id}`, + evidence: { runId: newSource!.id }, + }); + }); + it("does not reset an exhausted incident budget on server restart", async () => { const { agentId, runId, issueId } = await seedRunFixture({ agentStatus: "running", diff --git a/server/src/__tests__/issue-recovery-actions.test.ts b/server/src/__tests__/issue-recovery-actions.test.ts index 85d6bf6afc0..e8e4364fe22 100644 --- a/server/src/__tests__/issue-recovery-actions.test.ts +++ b/server/src/__tests__/issue-recovery-actions.test.ts @@ -39,7 +39,7 @@ import { import { errorHandler } from "../middleware/index.js"; import { issueRoutes } from "../routes/issues.js"; import { buildPaperclipWakePayload, heartbeatService } from "../services/heartbeat.js"; -import { deliverReconciledExecutions } from "../services/execution-recovery-resolution.js"; +import { deliverReconciledExecutions, markExecutionReconciliation } from "../services/execution-recovery-resolution.js"; import { workspaceOperationService, isNativeWorkspaceFinalizationOperationActive } from "../services/workspace-operations.js"; import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; import { recoveryService } from "../services/recovery/service.js"; @@ -2212,6 +2212,335 @@ describeEmbeddedPostgres("issue recovery actions", () => { }; } + async function seedDuplicateReconciliationHolds() { + const fixture = await seedReconciledDelivery(); + const { action, companyId, sourceIssueId, previousRunId, coderId } = fixture; + const evidence = { + runId: previousRunId, + automaticRecovery: { replay: "blocked", actionOutcome: "unknown" }, + }; + await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, sourceIssueId)); + const [canonical] = await db.update(issueRecoveryActions).set({ + outcome: "blocked", evidence, cause: "legacy_execution_requires_reconciliation", + fingerprint: `legacy-execution:${previousRunId}`, + }).where(eq(issueRecoveryActions.id, action.id)).returning(); + const siblings = await db.insert(issueRecoveryActions).values([0, 1].map(() => ({ + companyId, sourceIssueId, kind: "active_run_watchdog", status: "resolved", outcome: "blocked", + ownerType: "board", returnOwnerAgentId: coderId, cause: canonical!.cause, + fingerprint: canonical!.fingerprint, nextAction: "Preserve recorded work without replay.", evidence, + }))).returning(); + const body = { + actionId: action.id, + outcome: "restored", + sourceIssueStatus: "todo", + executionReconciliation: { + runId: previousRunId, + providerStopped: true, + actionOutcome: "not_performed", + outcomeEvidence: "Provider receipts confirm no external action was submitted and the provider is stopped.", + }, + } as const; + return { ...fixture, action: canonical!, siblings, body }; + } + + it("reconciles duplicate source holds in one request and delivers exactly one continuation", async () => { + const { companyId, sourceIssueId, previousRunId, action, siblings, body, heartbeat } = + await seedDuplicateReconciliationHolds(); + const app = createApp(); + await request(app).post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`).send(body).expect(200); + const actions = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, sourceIssueId)); + expect(actions).toHaveLength(3); + const canonical = actions.find((row) => row.id === action.id)!; + expect(canonical).toMatchObject({ status: "resolved", evidence: { + executionReconciliation: { runId: previousRunId }, continuationDelivery: "pending", + } }); + expect(canonical.evidence).not.toHaveProperty("automaticRecovery"); + for (const sibling of siblings) { + const recorded = actions.find((row) => row.id === sibling.id)!; + expect(recorded).toMatchObject({ status: "resolved", evidence: { + executionReconciliation: { runId: previousRunId }, + continuationDelivery: "invalidated", + coalescedIntoRecoveryActionId: action.id, + } }); + expect(recorded.evidence).not.toHaveProperty("automaticRecovery"); + } + await request(app).post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`).send(body).expect(200); + await request(app).post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`) + .send({ ...body, actionId: siblings[0]!.id }).expect(200); + expect(await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, sourceIssueId))).toEqual(actions); + + await deliverReconciledExecutions(db, heartbeat.wakeup); + await deliverReconciledExecutions(db, heartbeat.wakeup); + const wakes = await db.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.companyId, companyId)); + expect(wakes).toHaveLength(1); + expect(wakes[0]).toMatchObject({ + status: "queued", idempotencyKey: `execution-reconciliation:${action.id}`, + }); + const successors = await db.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, previousRunId)); + expect(successors).toHaveLength(1); + expect(successors[0]!.id).toBe(wakes[0]!.runId); + const receipts = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, sourceIssueId)); + expect(receipts.filter((row) => row.evidence.continuationDelivery === "delivered")).toHaveLength(1); + expect(receipts.filter((row) => row.evidence.continuationDelivery === "invalidated")).toHaveLength(2); + expect(receipts.some((row) => row.evidence.continuationDelivery === "pending")).toBe(false); + }); + + it("keeps one continuation owner when duplicate resolutions contend for the issue lock", async () => { + const { companyId, sourceIssueId, previousRunId, action, siblings, body, heartbeat } = + await seedDuplicateReconciliationHolds(); + const app = createApp(); + let release!: () => void; + let ready!: () => void; + const locked = new Promise((resolve) => { ready = resolve; }); + const gate = new Promise((resolve) => { release = resolve; }); + const holder = db.transaction(async (tx) => { + await tx.select({ id: issues.id }).from(issues) + .where(eq(issues.id, sourceIssueId)).for("update"); + ready(); + await gate; + }); + await locked; + const resolutions = Promise.all([action.id, siblings[0]!.id].map((actionId) => + request(app).post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`) + .send({ ...body, actionId }).then((response) => response), + )); + let waiting = 0; + try { + const deadline = Date.now() + 5_000; + while (waiting < 2 && Date.now() < deadline) { + const [state] = await db.execute<{ waiting: number }>(sql` + select count(*)::integer as waiting from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + `); + waiting = state!.waiting; + if (waiting < 2) await new Promise((resolve) => setTimeout(resolve, 20)); + } + } finally { + release(); + await holder; + } + const responses = await resolutions; + expect(waiting, "both resolution requests must wait on actual database locks").toBe(2); + expect(responses.map((response) => response.status)).toEqual([200, 200]); + const records = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, sourceIssueId)); + const owners = records.filter((row) => row.evidence.continuationDelivery === "pending"); + expect(owners).toHaveLength(1); + const owner = owners[0]!; + expect([action.id, siblings[0]!.id]).toContain(owner.id); + const coalesced = records.filter((row) => row.id !== owner.id); + expect(coalesced).toHaveLength(2); + for (const row of coalesced) { + expect(row.evidence).toMatchObject({ + continuationDelivery: "invalidated", coalescedIntoRecoveryActionId: owner.id, + }); + } + expect(records.every((row) => !("automaticRecovery" in row.evidence))).toBe(true); + const audits = await db.select().from(activityLog).where(and( + eq(activityLog.entityId, sourceIssueId), + sql`${activityLog.details}->>'reason' = 'same_source_execution_reconciled'`, + )); + expect(audits).toHaveLength(1); + expect(audits[0]!.details).toMatchObject({ recoveryActionId: owner.id }); + + await deliverReconciledExecutions(db, heartbeat.wakeup); + await deliverReconciledExecutions(db, heartbeat.wakeup); + const wakes = await db.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.companyId, companyId)); + expect(wakes).toHaveLength(1); + expect(wakes[0]).toMatchObject({ + status: "queued", idempotencyKey: `execution-reconciliation:${owner.id}`, + }); + const successors = await db.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, previousRunId)); + expect(successors).toHaveLength(1); + expect(successors[0]!.id).toBe(wakes[0]!.runId); + }); + + it.each(["live provider", "active checkout"] as const)( + "preserves all duplicate holds when a %s rejects reconciliation", + async (guard) => { + const { companyId, sourceIssueId, previousRunId, coderId, body, heartbeat } = + await seedDuplicateReconciliationHolds(); + if (guard === "live provider") { + await db.update(heartbeatRuns).set({ processPid: process.pid }) + .where(eq(heartbeatRuns.id, previousRunId)); + } else { + const [running] = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.agentId, coderId), eq(heartbeatRuns.status, "running"), + )); + await db.update(issues).set({ executionRunId: running!.id }) + .where(eq(issues.id, sourceIssueId)); + } + const before = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, sourceIssueId)).orderBy(issueRecoveryActions.id); + const [issueBefore] = await db.select().from(issues).where(eq(issues.id, sourceIssueId)); + const response = await request(createApp()) + .post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`).send(body).expect(409); + expect(response.body.error).toContain(guard === "live provider" + ? "previous provider is still running" : "locked by another active checkout or run"); + expect(await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, sourceIssueId)).orderBy(issueRecoveryActions.id)).toEqual(before); + expect((await db.select().from(issues).where(eq(issues.id, sourceIssueId)))[0]).toEqual(issueBefore); + expect(await db.select().from(activityLog).where(eq(activityLog.entityId, sourceIssueId))).toEqual([]); + await deliverReconciledExecutions(db, heartbeat.wakeup); + expect(await db.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.companyId, companyId))).toEqual([]); + expect(await db.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.retryOfRunId, previousRunId))).toEqual([]); + }, + ); + + it.each(["run", "issue", "company", "cause", "fingerprint", "kind", "replay"] as const)( + "keeps an unrelated %s recovery hold when reconciling duplicate source holds", + async (scope) => { + const { action, sourceIssueId, previousRunId, companyId, coderId, body } = + await seedDuplicateReconciliationHolds(); + let unrelatedCompanyId = companyId; + let unrelatedIssueId: string = sourceIssueId; + let unrelatedAgentId = coderId; + let unrelatedRunId = previousRunId; + if (scope === "run") { + unrelatedRunId = randomUUID(); + await seedHeartbeatRun({ companyId, agentId: coderId, runId: unrelatedRunId, + issueId: sourceIssueId, status: "failed" }); + } else if (scope === "issue") { + const [otherIssue] = await db.insert(issues).values({ + companyId, title: "Unrelated blocked task", status: "blocked", assigneeAgentId: coderId, + }).returning(); + unrelatedIssueId = otherIssue!.id; + } else if (scope === "company") { + const other = await seedCompany(); + unrelatedCompanyId = other.companyId; + unrelatedAgentId = other.coderId; + } + const [unrelated] = await db.insert(issueRecoveryActions).values({ + companyId: unrelatedCompanyId, sourceIssueId: unrelatedIssueId, + kind: scope === "kind" ? "missing_disposition" : "active_run_watchdog", + status: "resolved", outcome: "blocked", + ownerType: "board", returnOwnerAgentId: unrelatedAgentId, + cause: scope === "cause" ? "uncertain_external_action" : action.cause, + fingerprint: scope === "fingerprint" ? randomUUID() : action.fingerprint, + nextAction: "Preserve this independent hold.", + evidence: { + runId: unrelatedRunId, + automaticRecovery: { + replay: scope === "replay" ? "conversation_continuation" : "blocked", + actionOutcome: "unknown", + }, + }, + }).returning(); + await request(createApp()).post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`) + .send(body).expect(200); + const [after] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, unrelated!.id)); + expect(after).toEqual(unrelated); + }, + ); + + const protectedRecoveryEvidence = [ + { key: "workspaceRestoreFailure", value: "restore_failed" }, + { key: "workspaceRestoreRecovery", value: { schema: "paperclip.workspace-restore-recovery.v1" } }, + { key: "systemRecovery", value: { state: "pending" } }, + ] as const; + + it.each(protectedRecoveryEvidence)( + "does not coalesce duplicate holds from a source with $key evidence", + async ({ key, value }) => { + const { action, siblings, body } = await seedDuplicateReconciliationHolds(); + const [source] = await db.update(issueRecoveryActions).set({ + evidence: { ...action.evidence, [key]: value }, + }).where(eq(issueRecoveryActions.id, action.id)).returning(); + await db.transaction(async (tx) => { + await tx.select().from(issues).where(eq(issues.id, action.sourceIssueId)).for("update"); + await markExecutionReconciliation(tx as unknown as typeof db, source!, body.executionReconciliation, "operator"); + }); + for (const sibling of siblings) { + const [after] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, sibling.id)); + expect(after).toEqual(sibling); + } + }, + ); + + it.each(protectedRecoveryEvidence)( + "preserves sibling $key evidence when reconciling ordinary duplicate holds", + async ({ key, value }) => { + const { sourceIssueId, siblings, body, action } = await seedDuplicateReconciliationHolds(); + const [protectedSibling] = await db.update(issueRecoveryActions).set({ + evidence: { ...siblings[0]!.evidence, [key]: value }, + }).where(eq(issueRecoveryActions.id, siblings[0]!.id)).returning(); + await request(createApp()).post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`) + .send(body).expect(200); + const [after] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, protectedSibling!.id)); + expect(after).toEqual(protectedSibling); + const [ordinarySibling] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, siblings[1]!.id)); + expect(ordinarySibling!.evidence).toMatchObject({ + continuationDelivery: "invalidated", coalescedIntoRecoveryActionId: action.id, + }); + }, + ); + + it("preserves an already reconciled sibling when reconciling ordinary duplicate holds", async () => { + const { sourceIssueId, siblings, body, action } = await seedDuplicateReconciliationHolds(); + const [reconciledSibling] = await db.update(issueRecoveryActions).set({ + evidence: { + ...siblings[0]!.evidence, + executionReconciliation: { ...body.executionReconciliation, actorId: "previous-operator" }, + continuationDelivery: "pending", + }, + }).where(eq(issueRecoveryActions.id, siblings[0]!.id)).returning(); + await request(createApp()).post(`/api/issues/${sourceIssueId}/recovery-actions/resolve`) + .send(body).expect(200); + const [after] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, reconciledSibling!.id)); + expect(after).toEqual(reconciledSibling); + const [ordinarySibling] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, siblings[1]!.id)); + expect(ordinarySibling!.evidence).toMatchObject({ + continuationDelivery: "invalidated", coalescedIntoRecoveryActionId: action.id, + }); + }); + + it("records retained workspace repair without coalescing holds or requesting a continuation", async () => { + const { action, siblings, body } = await seedDuplicateReconciliationHolds(); + const [source] = await db.update(issueRecoveryActions).set({ + evidence: { + ...action.evidence, + workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [] }, + }, + }).where(eq(issueRecoveryActions.id, action.id)).returning(); + await db.transaction(async (tx) => { + await tx.select().from(issues).where(eq(issues.id, action.sourceIssueId)).for("update"); + await markExecutionReconciliation(tx as unknown as typeof db, source!, { + ...body.executionReconciliation, + workspaceRepairEvidence: "Recovered the retained workspace files and verified the saved manifest.", + }, "operator", undefined, { workspaceRepairOnly: true }); + }); + const [repaired] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, source!.id)); + expect(repaired!.evidence).toMatchObject({ + executionReconciliation: { runId: body.executionReconciliation.runId }, + continuationDelivery: "not_requested", + }); + for (const sibling of siblings) { + const [after] = await db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, sibling.id)); + expect(after).toEqual(sibling); + } + const wake = vi.fn(async () => null); + await deliverReconciledExecutions(db, wake); + expect(wake).not.toHaveBeenCalled(); + }); + it("delivers a reconciled execution once across concurrent sweeps without a deferred duplicate", async () => { const { action, heartbeat } = await seedReconciledDelivery(); let entered = 0; diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index 5d9d9e3b815..1d27e125d02 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -224,7 +224,7 @@ export async function markExecutionReconciliation( eq(nativeRunFinalizations.runId, decision.runId), ), ); - await tx + const [source] = await tx .update(issueRecoveryActions) .set({ evidence: { @@ -244,7 +244,45 @@ export async function markExecutionReconciliation( eq(issueRecoveryActions.companyId, action.companyId), eq(issueRecoveryActions.id, action.id), ), - ); + ).returning({ + cause: issueRecoveryActions.cause, + fingerprint: issueRecoveryActions.fingerprint, + kind: issueRecoveryActions.kind, + evidence: issueRecoveryActions.evidence, + }); + if ( + !source || options?.workspaceRepairOnly || + source.cause !== "legacy_execution_requires_reconciliation" || + source.fingerprint !== `legacy-execution:${decision.runId}` || + ["workspaceRestoreFailure", "workspaceRestoreRecovery", "systemRecovery"].some(key => key in source.evidence) + ) return; + const now = new Date(); + const siblings = await tx.update(issueRecoveryActions).set({ + status: "resolved", outcome: "cancelled", resolvedAt: now, updatedAt: now, + wakePolicy: null, monitorPolicy: null, + resolutionNote: "The same source execution was reconciled through another recovery action.", + evidence: sql`(${issueRecoveryActions.evidence} - 'automaticRecovery') || ${JSON.stringify({ + executionReconciliation: { ...decision, actorId, recordedAt: now.toISOString() }, + continuationDelivery: "invalidated", + coalescedIntoRecoveryActionId: action.id, + })}::jsonb`, + }).where(and( + eq(issueRecoveryActions.companyId, action.companyId), + eq(issueRecoveryActions.sourceIssueId, action.sourceIssueId), + not(eq(issueRecoveryActions.id, action.id)), + eq(issueRecoveryActions.kind, source.kind), + eq(issueRecoveryActions.cause, source.cause), + eq(issueRecoveryActions.fingerprint, source.fingerprint), + sql`${issueRecoveryActions.evidence}->>'runId' = ${decision.runId}`, + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, + sql`not (${issueRecoveryActions.evidence} ?| array['executionReconciliation', 'workspaceRestoreFailure', 'workspaceRestoreRecovery', 'systemRecovery'])`, + )).returning({ id: issueRecoveryActions.id }); + if (siblings.length) await persistActivity(tx as unknown as Db, { + companyId: action.companyId, actorType: "user", actorId, + action: "issue.recovery_action_resolved", entityType: "issue", entityId: action.sourceIssueId, + details: { reason: "same_source_execution_reconciled", recoveryActionId: action.id, + coalescedRecoveryActionIds: siblings.map(sibling => sibling.id), runId: decision.runId }, + }); }); } diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index b601258cc87..716ec8bd3a2 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -276,6 +276,12 @@ export async function terminalizeLegacyExecution(input: { sql`${issueRecoveryActions.evidence}->'workspaceRestoreRecovery'->>'schema' = ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}`), sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, ), + and( + eq(issueRecoveryActions.cause, LEGACY_RECOVERY_CAUSE), + eq(issueRecoveryActions.fingerprint, `legacy-execution:${run.id}`), + sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, + sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, + ), ), )).limit(1); if (reconciled) return updated;