diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index d584a990654..a44b0a8bf71 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -145,7 +145,8 @@ describe("report-only ownership reconciliation", () => { expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", - "native_workspace_finalize_attempt", "wake_request_claimed", "legacy_controller_lease"]); + "native_workspace_finalize_attempt", "wake_request_claimed", "legacy_controller_lease", + "issue_execution_lock", "issue_checkout_lock"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); @@ -292,6 +293,39 @@ describe("report-only ownership reconciliation", () => { .toEqual([wakes.pending.id, wakes.unprojected.id].sort()); }); + // Non-wake recovery (Phase 3): a task lock is owned by the run it names. A + // restart between the run's terminal write and the lock release left the + // task locked to a dead run until the next periodic sweep. The reconciler + // now settles it on the proof (the run's terminal row), never by age. + it("settles a task lock whose run ended without releasing it, and keeps live or resuming ones", async () => { + const ids = { dead: randomUUID(), live: randomUUID(), resuming: randomUUID() }; + const runs = { dead: randomUUID(), live: randomUUID(), resuming: randomUUID() }; + await db.insert(heartbeatRuns).values([ + { id: runs.dead, companyId, agentId, status: "failed", invocationSource: "manual", finishedAt: minutesAgo(1) }, + { id: runs.live, companyId, agentId, status: "running", invocationSource: "manual" }, + { id: runs.resuming, companyId, agentId, status: "failed", invocationSource: "manual", finishedAt: minutesAgo(1), + runtimeMode: "native", nativeIssueId: ids.resuming }, + ]); + await db.insert(issues).values(Object.entries(ids).map(([name, id]) => ({ id, companyId, title: name, status: "in_progress", + assigneeAgentId: agentId, executionRunId: runs[name as keyof typeof runs], checkoutRunId: runs[name as keyof typeof runs], + executionLockedAt: minutesAgo(1) }))); + // A same-run resume still needs the failed run's lock. + await db.insert(nativeRunFinalizations).values({ runId: runs.resuming, companyId, issueId: ids.resuming, phase: "retryable_failure" }); + const locks = async () => Object.fromEntries((await db.select({ id: issues.id, executionRunId: issues.executionRunId, + checkoutRunId: issues.checkoutRunId }).from(issues)).map(row => + [Object.keys(ids).find(name => ids[name as keyof typeof ids] === row.id), row.executionRunId === null && row.checkoutRunId === null])); + const allHeld = { dead: false, live: false, resuming: false }; + + // A read-only pass never settles. + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(await locks()).toEqual(allHeld); + + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(await locks()).toEqual({ dead: true, live: false, resuming: false }); + expect(await db.select({ action: activityLog.action }).from(activityLog).where(eq(activityLog.entityId, ids.dead))) + .toContainEqual({ action: "issue.stale_lock_cleared" }); + }); + it("settles a finalize attempt whose process died mid-copyback, and keeps live or independent ones", async () => { // Crash points workspace.finalize.after_start_before_copyback and // .after_copyback_before_done: the run finished through a later attempt, diff --git a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts index 8efca7b2f77..535c1c9de6a 100644 --- a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts +++ b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts @@ -9,6 +9,7 @@ import { costEvents, activityLog, agents, + agentWakeupRequests, companies, completionContracts, createDb, @@ -65,6 +66,7 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { await db.delete(heartbeatRuns); await db.delete(completionContracts); await db.delete(issues); + await db.delete(agentWakeupRequests); await db.delete(agents); await db.delete(companies); }); @@ -147,6 +149,43 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.executionRunId).toBe(runningRunId); }); + // Section E: clearing a stale lock is a hold-release event. A message saved + // behind the lock is made due at once instead of waiting out its backoff. + it("promotes a message saved behind a stale lock when the sweep clears it", async () => { + const { companyId, agentId, failedRunId } = await seed(); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Saved behind a stale lock", status: "in_progress", + priority: "high", assigneeAgentId: agentId, checkoutRunId: failedRunId, executionRunId: failedRunId }); + const [saved] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", + status: "deferred_issue_execution", requestedByActorType: "user", updatedAt: new Date(), + payload: { issueId, executionWait: { reason: "company.inactive", trigger: "pause_lifted" } } }).returning(); + const result = await recoveryService(db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks({ companyId, issueId }); + expect(result.issueIds).toEqual([issueId]); + const [after] = await db.select({ updatedAt: agentWakeupRequests.updatedAt }).from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, saved.id)); + expect(after.updatedAt.getTime()).toBeLessThanOrEqual(Date.now() - 29_000); + }); + + it("does not promote a saved message when the guarded clear keeps the lock", async () => { + const { companyId, agentId, failedRunId } = await seed(); + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Lock taken back", status: "in_progress", + priority: "high", assigneeAgentId: agentId, checkoutRunId: failedRunId, executionRunId: failedRunId }); + const savedAt = new Date(); + const [saved] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", + status: "deferred_issue_execution", requestedByActorType: "user", updatedAt: savedAt, + payload: { issueId, executionWait: { reason: "company.inactive", trigger: "pause_lifted" } } }).returning(); + // A resume returns the run to running after the sweep read it. + const beforeStaleLockClear = vi.fn(async () => { + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, failedRunId)); + }); + const result = await recoveryService(db, { enqueueWakeup: vi.fn(), beforeStaleLockClear }).sweepStaleIssueLocks({ companyId, issueId }); + expect(result.issueIds).toEqual([]); + const [after] = await db.select({ updatedAt: agentWakeupRequests.updatedAt }).from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, saved.id)); + expect(after.updatedAt.getTime()).toBe(savedAt.getTime()); + }); + it("clears lock columns when checkoutRunId points at a terminal heartbeat run", async () => { const { companyId, agentId, failedRunId } = await seed(); const issueId = randomUUID(); diff --git a/server/src/services/recovery/issue-lock-settlement.ts b/server/src/services/recovery/issue-lock-settlement.ts new file mode 100644 index 00000000000..8ffe6a88c0d --- /dev/null +++ b/server/src/services/recovery/issue-lock-settlement.ts @@ -0,0 +1,28 @@ +import type { Db } from "@paperclipai/db"; +import type { OwnerProof } from "../owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; + +/** + * Settle of the `issue_execution_lock` and `issue_checkout_lock` ownership + * kinds (plan Phase 2, taken over by Phase 3 for non-wake recovery). + * + * A task's lock is owned by the run it names. That run's terminal row, or a + * missing row, is the proof the reconciler acts on; age never is. Settling + * runs the stale-lock sweep for exactly this task, so the one compare-and-swap + * clear decides: it re-reads both lock columns under the task row lock, keeps + * a lock that a same-run resume, an explicit-retry claim, a manual admission + * hold or a newly live run still needs, and releases messages waiting behind + * the lock. The sweep loads lazily so the registry stays light. + */ +export async function settleIssueLock(db: Db, record: InProgressRecord, + _context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "run" || !record.companyId || !record.issueId) return { settled: false, reason: "owner_not_a_run" }; + const { recoveryService } = await import("./service.js"); + const result = await recoveryService(db, { + // Clearing a stale lock never starts work itself; the release drain does. + enqueueWakeup: async () => null, + }).sweepStaleIssueLocks({ companyId: record.companyId, issueId: record.issueId }); + return result.issueIds.includes(record.issueId) + ? { settled: true, reason: "stale_lock_cleared" } + : { settled: false, reason: "lock_still_needed" }; +} diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 848c0992fdf..e9c93464cd6 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -34,6 +34,7 @@ import { import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; import { settleAbandonedWorkspaceFinalizeAttempt } from "./workspace-finalize-attempt-settlement.js"; import { nativeControllerOwnerStamp, runProcessOwnerStamp } from "../native-runtime/native-owner-stamps.js"; +import { settleIssueLock } from "./issue-lock-settlement.js"; import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js"; import { nativeLeaseHolderStamp, @@ -892,7 +893,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ [2]), { id: "issue_execution_lock", title: "Issue execution lock", obligation: "ownership", storage: "database", - settle: null, invariantChecks: [2, 3, 7], + settle: settleIssueLock, invariantChecks: [2, 3, 7], listInProgress: async (db, limit, offset) => { const rows = await db.select({ id: issues.id, companyId: issues.companyId, runId: issues.executionRunId, updatedAt: issues.updatedAt }).from(issues).where(isNotNull(issues.executionRunId)).orderBy(asc(issues.id)).limit(limit + 1).offset(offset); @@ -905,7 +906,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ }, { id: "issue_checkout_lock", title: "Issue checkout lock", obligation: "ownership", storage: "database", - settle: null, invariantChecks: [2, 7], + settle: settleIssueLock, invariantChecks: [2, 7], listInProgress: async (db, limit, offset) => { const rows = await db.select({ id: issues.id, companyId: issues.companyId, runId: issues.checkoutRunId, updatedAt: issues.updatedAt }).from(issues).where(isNotNull(issues.checkoutRunId)).orderBy(asc(issues.id)).limit(limit + 1).offset(offset); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 7f27b08a211..75e7657415c 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,4 +1,5 @@ import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js"; +import { promoteDeferredWorkIntents } from "../work-intent-outbox.js"; import { isContextCompactionRun } from "../context-compaction.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; @@ -6544,7 +6545,7 @@ export function recoveryService( } await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id }); await notifyNativeStatusReassessment(tx, { companyId: issue.companyId, issueId: issue.id }); - return tx + const cleared = await tx .update(issues) .set({ checkoutRunId: null, @@ -6570,6 +6571,12 @@ export function recoveryService( ) .returning({ id: issues.id }) .then((rows) => rows[0] ?? null); + // Releasing the lock is a hold-release event: messages saved behind it + // are promoted through the one outbox path (section E), only when the + // guarded clear actually released it. The promotion registers its own + // work signals before its write. + if (cleared) await promoteDeferredWorkIntents(tx, { companyId: issue.companyId, trigger: "lock_released", issueId: issue.id }); + return cleared; }); if (!updated) continue;