mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
feat(recovery): settle stale issue execution and checkout locks on the shared ownership path (#20006)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A task is locked to the run that executes it, and the lock must be released when that run ends > - A restart between the run's terminal write and the lock release left the task locked to a dead run until a periodic sweep noticed > - The durable ownership plan moves each in-progress record onto the shared reconciler, which settles a record in its first pass after its owner is proven gone > - This pull request registers the task execution and checkout locks with the reconciler, and makes the lock release promote any messages saved behind it > - The benefit is that a task stuck behind a dead run is freed in the first reconciler pass, and its waiting messages go out at once ## Linked Issues or Issue Description Refs #19831 (plan, Phase 2 sections A and C), #19838 (inventory: "Issue execution lock" and "Issue checkout lock"), #19956 (work-intent outbox), #20004 (lost-process reapers). **What happened?** The task locks were report-only kinds in the ownership registry. A lock whose run had already ended was released only by `sweepStaleIssueLocks` on its own schedule. Messages saved behind the lock then still waited out their 30-second refusal backoff. **Expected behavior** A lock whose owner run is proven ended (a terminal or missing run row) is settled by the reconciler on that proof, never by age. A lock that a same-run resume, an explicit-retry claim, a manual admission hold or a newly live run still needs is kept. ## What Changed - New `server/src/services/recovery/issue-lock-settlement.ts`: `settleIssueLock` runs the stale-lock sweep for exactly that task, so its existing compare-and-swap clear and all of its guards decide. It loads the sweep lazily. - `recovery/ownership-registry.ts`: `issue_execution_lock` and `issue_checkout_lock` register `settleIssueLock`. - `recovery/service.ts`: clearing a stale lock now calls `promoteDeferredWorkIntents(trigger: "lock_released")` before its write, in the same transaction. - The periodic sweep stays, because its pre-pass also hands lost-process runs to the single reaper (#20004). The reconciler is now the faster backstop for the lock itself. ## Verification Each focused file ran alone, in one process, on embedded Postgres: - `ownership-reconciliation-report.test.ts` (25 pass). A new test covers dead, live and resuming locks: a read-only pass never settles; the dead run's lock is cleared with an `issue.stale_lock_cleared` row; the live and same-run-resume locks are kept. It fails against `master`'s registry and passes here. The test that lists settling kinds now includes both lock kinds. - `recovery-stale-issue-lock-sweep.test.ts` (49 pass). A new test checks that a message saved behind a stale lock is made due when the lock is cleared. It fails with `master`'s `service.ts` and passes here. - The server package typecheck passes. ## Risks - The reconciler can now clear a lock that only the periodic sweep cleared before. It goes through the same compare-and-swap and guards, so the outcome is the same, only sooner. - More passes call the sweep for a single task. Each call is scoped to one issue. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with 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 either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [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 any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
7c508d3e42
commit
ada44a669a
5 files changed
+113
-4
No files matched your search
@@ -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,
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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<SettleOutcome> {
|
||||
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" };
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in new issue
Block a user