diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7ebfdee4..fffdb1da 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1,7 +1,7 @@ # Execution Semantics Status: Current implementation guide -Date: 2026-06-08 +Date: 2026-06-10 Audience: Product and engineering This document explains how Paperclip interprets issue assignment, issue status, execution runs, wakeups, parent/sub-issue structure, and blocker relationships. @@ -121,6 +121,17 @@ These are related but not identical: Paperclip already clears stale execution locks and can adopt some stale checkout locks when the original run is gone. +The active-lock lifecycle is part of the checkout contract: + +- a run owns `checkoutRunId` only while that run is non-terminal +- when a run reaches `succeeded`, `failed`, `cancelled`, or `timed_out`, finalization must compare-and-clear lock columns that still point at that run +- finalization must not clear a lock already reacquired by a successor run +- process-loss retry handoff must not leave `checkoutRunId` pinned to the failed run when `executionRunId` moves to the retry run +- checkout and checkout-owner checks may self-heal lock columns that point at terminal or missing runs before evaluating conflicts +- the recovery sweeper may clear rows whose checkout and execution locks all point at terminal or missing runs + +Stale-lock recovery is crash recovery, not a retry loop. Paperclip must not clear or adopt locks held by non-terminal runs. After stale cleanup, a checkout `409` should mean a real live owner, status/assignee mismatch, unresolved blocker, or active gate still prevents checkout. Agents must treat that `409` as an ownership conflict and stop rather than retrying the same checkout. + ## 6. Parent/Sub-Issue vs Blockers Paperclip uses two different relationships for different jobs. diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 6e47ecc3..3954ccb0 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1023,7 +1023,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }) ); expect(issue?.executionRunId).toBe(retryRun?.id ?? null); - expect(issue?.checkoutRunId).toBe(runId); + // Terminal run cleanup releases the checkout lock so future checkout 409s only mean a live owner exists. + expect(issue?.checkoutRunId).toBeNull(); }); it("releases active environment leases when an orphaned run is reaped", async () => { @@ -1270,7 +1271,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { const issue = await db.select().from(issues).where(eq(issues.id, issueId)).then((rows) => rows[0] ?? null); expect(issue?.status).toBe("in_progress"); expect(issue?.executionRunId).toBeNull(); - expect(issue?.checkoutRunId).toBe(runId); + // Terminal run cleanup releases the checkout lock even when paused-tree recovery is suppressed. + expect(issue?.checkoutRunId).toBeNull(); const recoveryIssues = await db .select() diff --git a/server/src/__tests__/issue-stale-execution-lock-routes.test.ts b/server/src/__tests__/issue-stale-execution-lock-routes.test.ts index ced38bce..a74ecb1a 100644 --- a/server/src/__tests__/issue-stale-execution-lock-routes.test.ts +++ b/server/src/__tests__/issue-stale-execution-lock-routes.test.ts @@ -283,4 +283,65 @@ describeEmbeddedPostgres("stale issue execution lock routes", () => { }, }); }); + + it("self-heals a stale checkoutRunId via clearCheckoutRunIfTerminal on checkout (Fix B path)", async () => { + // Reproduces the recurrence pattern: prior owning run died, executionRunId + // was cleared by releaseIssueExecutionAndPromote, but checkoutRunId stayed + // pinned to the dead run. The new agent's POST /checkout would 409 forever + // without the clearCheckoutRunIfTerminal helper in svc.checkout. + const { companyId, agentId, failedRunId, currentRunId } = await seedCompanyAgentAndRuns(); + const issueId = randomUUID(); + const otherAgentId = randomUUID(); + await db.insert(agents).values({ + id: otherAgentId, + companyId, + name: "OtherAgent", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Stale checkout lock after reassignment", + // Status off in_progress + checkoutRunId still set — adoptStaleCheckoutRun + // cannot recover from this; only clearCheckoutRunIfTerminal can. + status: "todo", + priority: "high", + assigneeAgentId: otherAgentId, + checkoutRunId: failedRunId, + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + }); + + const res = await request(createApp(agentActor(companyId, otherAgentId, currentRunId))) + .post(`/api/issues/${issueId}/checkout`) + .send({ + agentId: otherAgentId, + expectedStatuses: ["todo", "backlog", "blocked", "in_review"], + }); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + + const row = await db + .select({ + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ + status: "in_progress", + assigneeAgentId: otherAgentId, + checkoutRunId: currentRunId, + executionRunId: currentRunId, + }); + }); }); diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index 6569815c..9b0d7a83 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -3869,6 +3869,171 @@ describeEmbeddedPostgres("issueService.clearExecutionRunIfTerminal", () => { .then((rows) => rows[0]); expect(row).toEqual({ executionRunId: null, executionLockedAt: null }); }); + + it("does not clear checkout locks when a different execution run is live", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const failedRunId = randomUUID(); + const runningRunId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values([ + { + id: failedRunId, + companyId, + agentId, + status: "failed", + invocationSource: "manual", + finishedAt: new Date("2026-06-10T10:05:00.000Z"), + }, + { + id: runningRunId, + companyId, + agentId, + status: "running", + invocationSource: "manual", + startedAt: new Date("2026-06-10T10:06:00.000Z"), + }, + ]); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Mixed execution lock", + status: "in_progress", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: failedRunId, + executionRunId: runningRunId, + executionAgentNameKey: "codexcoder", + executionLockedAt: new Date("2026-06-10T10:06:00.000Z"), + }); + + await expect(svc.clearCheckoutRunIfTerminal(issueId)).resolves.toBe(false); + + const row = await db + .select({ + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + executionAgentNameKey: issues.executionAgentNameKey, + executionLockedAt: issues.executionLockedAt, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row?.checkoutRunId).toBe(failedRunId); + expect(row?.executionRunId).toBe(runningRunId); + expect(row?.executionAgentNameKey).toBe("codexcoder"); + expect(row?.executionLockedAt).toBeInstanceOf(Date); + }); + + it("does not let stale release clobber a successor checkout lock", async () => { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + const failedRunId = randomUUID(); + const releasingRunId = randomUUID(); + const successorRunId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "CodexCoder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values([ + { + id: failedRunId, + companyId, + agentId, + status: "failed", + invocationSource: "manual", + finishedAt: new Date("2026-06-10T10:05:00.000Z"), + }, + { + id: releasingRunId, + companyId, + agentId, + status: "running", + invocationSource: "manual", + startedAt: new Date("2026-06-10T10:06:00.000Z"), + }, + { + id: successorRunId, + companyId, + agentId, + status: "running", + invocationSource: "manual", + startedAt: new Date("2026-06-10T10:07:00.000Z"), + }, + ]); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Race stale release", + status: "in_progress", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: failedRunId, + executionRunId: failedRunId, + executionAgentNameKey: "codexcoder", + executionLockedAt: new Date("2026-06-10T10:00:00.000Z"), + }); + + const [releaseResult, checkoutResult] = await Promise.allSettled([ + svc.release(issueId, agentId, releasingRunId), + svc.checkout(issueId, agentId, ["todo", "in_progress"], successorRunId), + ]); + + expect(checkoutResult.status).toBe("fulfilled"); + if (releaseResult.status === "rejected") { + expect(releaseResult.reason).toMatchObject({ status: 409 }); + } + + const row = await db + .select({ + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toMatchObject({ + status: "in_progress", + assigneeAgentId: agentId, + checkoutRunId: successorRunId, + executionRunId: successorRunId, + }); + }); }); describeEmbeddedPostgres("accepted plan decomposition", () => { diff --git a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts new file mode 100644 index 00000000..de6baf90 --- /dev/null +++ b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts @@ -0,0 +1,226 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, + agents, + companies, + createDb, + heartbeatRuns, + issueComments, + issueRelations, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() })); +vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient })); + +import { heartbeatService } from "../services/heartbeat.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres stale-lock sweeper tests on this host: ${ + embeddedPostgresSupport.reason ?? "unsupported environment" + }`, + ); +} + +describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-stale-lock-sweep-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + await db.delete(issueComments); + await db.delete(issueRelations); + await db.delete(activityLog); + await db.delete(issues); + await db.delete(heartbeatRuns); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function seed() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const failedRunId = randomUUID(); + const runningRunId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Coder", + role: "engineer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values([ + { + id: failedRunId, + companyId, + agentId, + status: "failed", + invocationSource: "manual", + finishedAt: new Date(), + }, + { + id: runningRunId, + companyId, + agentId, + status: "running", + invocationSource: "manual", + startedAt: new Date(), + }, + ]); + + return { companyId, agentId, failedRunId, runningRunId }; + } + + it("clears lock columns when checkoutRunId points at a terminal heartbeat run", async () => { + const { companyId, agentId, failedRunId } = await seed(); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Stale lock — terminal checkoutRunId", + // Status off in_progress + checkoutRunId still set → exactly the recurrence shape. + status: "todo", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: failedRunId, + executionRunId: null, + }); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.sweepStaleIssueLocks(); + + expect(result.cleared).toBe(1); + expect(result.issueIds).toEqual([issueId]); + + const row = await db + .select({ + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + executionLockedAt: issues.executionLockedAt, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ checkoutRunId: null, executionRunId: null, executionLockedAt: null }); + + const audit = await db + .select({ action: activityLog.action, details: activityLog.details }) + .from(activityLog) + .where(eq(activityLog.action, "issue.stale_lock_cleared")) + .then((rows) => rows[0]); + expect(audit?.action).toBe("issue.stale_lock_cleared"); + expect((audit?.details as { clearedCheckoutRunId?: string } | null)?.clearedCheckoutRunId).toBe( + failedRunId, + ); + }); + + it("does not clear locks while the referenced run is still running", async () => { + const { companyId, agentId, runningRunId } = await seed(); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Live lock — must be preserved", + status: "in_progress", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: runningRunId, + executionRunId: runningRunId, + executionLockedAt: new Date(), + }); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.sweepStaleIssueLocks(); + + expect(result.cleared).toBe(0); + const row = await db + .select({ + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ checkoutRunId: runningRunId, executionRunId: runningRunId }); + }); + + it("does not clear when checkoutRunId is terminal but executionRunId is still running", async () => { + const { companyId, agentId, failedRunId, runningRunId } = await seed(); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Mixed lock — preserve", + status: "in_progress", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: failedRunId, + executionRunId: runningRunId, + executionLockedAt: new Date(), + }); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.sweepStaleIssueLocks(); + + expect(result.cleared).toBe(0); + const row = await db + .select({ + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0]); + expect(row).toEqual({ checkoutRunId: failedRunId, executionRunId: runningRunId }); + }); + + it("is idempotent — second pass finds nothing to clear", async () => { + const { companyId, agentId, failedRunId } = await seed(); + const issueId = randomUUID(); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Idempotency", + status: "todo", + priority: "high", + assigneeAgentId: agentId, + checkoutRunId: failedRunId, + executionRunId: null, + }); + + const heartbeat = heartbeatService(db); + const first = await heartbeat.sweepStaleIssueLocks(); + const second = await heartbeat.sweepStaleIssueLocks(); + expect(first.cleared).toBe(1); + expect(second.cleared).toBe(0); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index 38caf44a..a81e50b5 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -754,6 +754,12 @@ export async function startServer(): Promise { logger.warn({ ...scanned }, "startup active-run output watchdog created review work"); } }) + .then(async () => { + const swept = await heartbeat.sweepStaleIssueLocks(); + if (swept.cleared > 0) { + logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks"); + } + }) .then(async () => { const reviewed = await heartbeat.reconcileProductivityReviews(); if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { @@ -820,6 +826,12 @@ export async function startServer(): Promise { logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); } }) + .then(async () => { + const swept = await heartbeat.sweepStaleIssueLocks(); + if (swept.cleared > 0) { + logger.warn({ ...swept }, "periodic stale-lock sweeper cleared issue locks"); + } + }) .then(async () => { const reviewed = await heartbeat.reconcileProductivityReviews(); if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 7b2da065..b988dcb3 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -5539,6 +5539,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) await tx .update(issues) .set({ + checkoutRunId: null, executionRunId: retryRun.id, executionAgentNameKey: normalizeAgentNameKey(agent.name), executionLockedAt: now, @@ -7475,6 +7476,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return recovery.reconcileStrandedAssignedIssues(); } + async function sweepStaleIssueLocks() { + return recovery.sweepStaleIssueLocks(); + } + function issueIdFromRunContext(contextSnapshot: unknown) { const context = parseObject(contextSnapshot); return readNonEmptyString(context.issueId) ?? readNonEmptyString(context.taskId); @@ -9377,16 +9382,26 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) if (!issue) return null; if (issue.executionRunId && issue.executionRunId !== run.id) return null; - if (issue.executionRunId === run.id) { + // Clear lock columns that point at the terminating run. checkoutRunId is + // cleared here in addition to executionRunId so the issue self-heals even + // if its assignee or status changed between checkout and termination — + // adoptStaleCheckoutRun's narrow WHERE clause cannot cover those paths. + if (issue.executionRunId === run.id || issue.checkoutRunId === run.id) { await tx .update(issues) .set({ + checkoutRunId: null, executionRunId: null, executionAgentNameKey: null, executionLockedAt: null, updatedAt: new Date(), }) - .where(eq(issues.id, issue.id)); + .where( + and( + eq(issues.id, issue.id), + or(eq(issues.executionRunId, run.id), eq(issues.checkoutRunId, run.id)), + ), + ); } if ( @@ -11097,6 +11112,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) reconcileStrandedAssignedIssues, + sweepStaleIssueLocks, + buildIssueGraphLivenessAutoRecoveryPreview, reconcileIssueGraphLiveness, diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index d422f4d9..429205a1 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -380,7 +380,7 @@ function sameRunLock(checkoutRunId: string | null, actorRunId: string | null) { return checkoutRunId == null; } -const TERMINAL_HEARTBEAT_RUN_STATUSES = new Set(["succeeded", "failed", "cancelled", "timed_out"]); +export const TERMINAL_HEARTBEAT_RUN_STATUSES = new Set(["succeeded", "failed", "cancelled", "timed_out"]); const ISSUE_LIST_DESCRIPTION_MAX_CHARS = 1200; const ISSUE_LIST_DESCRIPTION_MAX_BYTES = ISSUE_LIST_DESCRIPTION_MAX_CHARS * 4; @@ -3634,8 +3634,8 @@ export function issueService(db: Db) { ); } - async function isTerminalOrMissingHeartbeatRun(runId: string) { - const run = await db + async function isTerminalOrMissingHeartbeatRun(runId: string, dbOrTx: DbReader = db) { + const run = await dbOrTx .select({ status: heartbeatRuns.status }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)) @@ -3760,8 +3760,73 @@ export function issueService(db: Db) { }); } + // Symmetric to clearExecutionRunIfTerminal. Clears checkoutRunId (and the + // bundled execution lock cols) when the row's checkoutRunId points at a + // heartbeat run that is terminal or no longer exists. No assignee/status + // precondition: a terminal run holds no real claim regardless of who is + // assigned or what status the issue is currently in. + async function clearCheckoutRunIfTerminal(issueId: string): Promise { + return db.transaction(async (tx) => { + await tx.execute( + sql`select ${issues.id} from ${issues} where ${issues.id} = ${issueId} for update`, + ); + const issue = await tx + .select({ checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + if (!issue?.checkoutRunId) return false; + + await tx.execute( + sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${issue.checkoutRunId} for update`, + ); + const run = await tx + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, issue.checkoutRunId)) + .then((rows) => rows[0] ?? null); + if (run && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(run.status)) return false; + + if (issue.executionRunId && issue.executionRunId !== issue.checkoutRunId) { + await tx.execute( + sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${issue.executionRunId} for update`, + ); + const executionRun = await tx + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, issue.executionRunId)) + .then((rows) => rows[0] ?? null); + if (executionRun && !TERMINAL_HEARTBEAT_RUN_STATUSES.has(executionRun.status)) return false; + } + + const updated = await tx + .update(issues) + .set({ + checkoutRunId: null, + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }) + .where( + and( + eq(issues.id, issueId), + eq(issues.checkoutRunId, issue.checkoutRunId), + issue.executionRunId + ? eq(issues.executionRunId, issue.executionRunId) + : isNull(issues.executionRunId), + ), + ) + .returning({ id: issues.id }) + .then((rows) => rows[0] ?? null); + + return Boolean(updated); + }); + } + return { clearExecutionRunIfTerminal, + clearCheckoutRunIfTerminal, list: async (companyId: string, filters?: IssueFilters) => { if (filters?.attention === "blocked") { @@ -5311,6 +5376,7 @@ export function issueService(db: Db) { } await clearExecutionRunIfTerminal(id); + await clearCheckoutRunIfTerminal(id); const dependencyReadiness = await listIssueDependencyReadinessMap(db, issueCompany.companyId, [id]); const unresolvedBlockerIssueIds = dependencyReadiness.get(id)?.unresolvedBlockerIssueIds ?? []; @@ -5440,6 +5506,7 @@ export function issueService(db: Db) { assertCheckoutOwner: async (id: string, actorAgentId: string, actorRunId: string | null) => { await clearExecutionRunIfTerminal(id); + await clearCheckoutRunIfTerminal(id); const current = await db .select({ id: issues.id, @@ -5516,54 +5583,57 @@ export function issueService(db: Db) { }); }, - release: async (id: string, actorAgentId?: string, actorRunId?: string | null) => { - await clearExecutionRunIfTerminal(id); - const existing = await db - .select() - .from(issues) - .where(eq(issues.id, id)) - .then((rows) => rows[0] ?? null); + release: async (id: string, actorAgentId?: string, actorRunId?: string | null) => + db.transaction(async (tx) => { + await tx.execute( + sql`select ${issues.id} from ${issues} where ${issues.id} = ${id} for update`, + ); + const existing = await tx + .select() + .from(issues) + .where(eq(issues.id, id)) + .then((rows) => rows[0] ?? null); - if (!existing) return null; - if (actorAgentId && existing.assigneeAgentId && existing.assigneeAgentId !== actorAgentId) { - throw conflict("Only assignee can release issue"); - } - if ( - actorAgentId && - existing.status === "in_progress" && - existing.assigneeAgentId === actorAgentId && - existing.checkoutRunId && - !sameRunLock(existing.checkoutRunId, actorRunId ?? null) - ) { - const stale = await isTerminalOrMissingHeartbeatRun(existing.checkoutRunId); - if (!stale) { - throw conflict("Only checkout run can release issue", { - issueId: existing.id, - assigneeAgentId: existing.assigneeAgentId, - checkoutRunId: existing.checkoutRunId, - actorRunId: actorRunId ?? null, - }); + if (!existing) return null; + if (actorAgentId && existing.assigneeAgentId && existing.assigneeAgentId !== actorAgentId) { + throw conflict("Only assignee can release issue"); + } + if ( + actorAgentId && + existing.status === "in_progress" && + existing.assigneeAgentId === actorAgentId && + existing.checkoutRunId && + !sameRunLock(existing.checkoutRunId, actorRunId ?? null) + ) { + const stale = await isTerminalOrMissingHeartbeatRun(existing.checkoutRunId, tx); + if (!stale) { + throw conflict("Only checkout run can release issue", { + issueId: existing.id, + assigneeAgentId: existing.assigneeAgentId, + checkoutRunId: existing.checkoutRunId, + actorRunId: actorRunId ?? null, + }); + } } - } - const updated = await db - .update(issues) - .set({ - status: "todo", - assigneeAgentId: null, - checkoutRunId: null, - executionRunId: null, - executionAgentNameKey: null, - executionLockedAt: null, - updatedAt: new Date(), - }) - .where(eq(issues.id, id)) - .returning() - .then((rows) => rows[0] ?? null); - if (!updated) return null; - const [enriched] = await withIssueLabels(db, [updated]); - return enriched; - }, + const updated = await tx + .update(issues) + .set({ + status: "todo", + assigneeAgentId: null, + checkoutRunId: null, + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }) + .where(eq(issues.id, id)) + .returning() + .then((rows) => rows[0] ?? null); + if (!updated) return null; + const [enriched] = await withIssueLabels(tx, [updated]); + return enriched; + }), adminForceRelease: async (id: string, options: { clearAssignee?: boolean } = {}) => db.transaction(async (tx) => { diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index c9f4b4fd..1a5fb8c5 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -35,7 +35,7 @@ import { budgetService } from "../budgets.js"; import { instanceSettingsService } from "../instance-settings.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueTreeControlService } from "../issue-tree-control.js"; -import { issueService } from "../issues.js"; +import { TERMINAL_HEARTBEAT_RUN_STATUSES, issueService } from "../issues.js"; import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; import { getRunLogStore } from "../run-log-store.js"; import { @@ -3604,6 +3604,115 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return Math.max(1, Math.floor(asNumber(raw, fallback))); } + // Backstop sweeper: clears stale lock columns on issues whose checkoutRunId + // or executionRunId points at a heartbeat_runs row that is either missing or + // in a terminal status. Provides self-heal for stale locks that fell outside + // releaseIssueExecutionAndPromote / clearCheckoutRunIfTerminal / adoption. + // Idempotent and safe: clears at most one row's worth of lock columns per + // candidate, and only when the referenced run row is unambiguously terminal. + async function sweepStaleIssueLocks() { + const result = { + cleared: 0, + issueIds: [] as string[], + }; + + const candidates = await db + .select({ + id: issues.id, + companyId: issues.companyId, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + }) + .from(issues) + .where( + sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`, + ); + + const referencedRunIds = [ + ...new Set( + candidates + .flatMap((issue) => [issue.checkoutRunId, issue.executionRunId]) + .filter((id): id is string => !!id), + ), + ]; + const runRows = + referencedRunIds.length > 0 + ? await db + .select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, referencedRunIds)) + : []; + const runStatusById = new Map(); + for (const row of runRows) runStatusById.set(row.id, row.status); + + const isCleanable = (runId: string | null) => { + if (!runId) return true; + const status = runStatusById.get(runId); + if (!status) return true; // missing run row → no real claim + return TERMINAL_HEARTBEAT_RUN_STATUSES.has(status); + }; + + for (const issue of candidates) { + if (!isCleanable(issue.checkoutRunId) || !isCleanable(issue.executionRunId)) { + continue; + } + + const updated = await db + .update(issues) + .set({ + checkoutRunId: null, + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }) + .where( + and( + eq(issues.id, issue.id), + issue.checkoutRunId + ? eq(issues.checkoutRunId, issue.checkoutRunId) + : isNull(issues.checkoutRunId), + issue.executionRunId + ? eq(issues.executionRunId, issue.executionRunId) + : isNull(issues.executionRunId), + ), + ) + .returning({ id: issues.id }) + .then((rows) => rows[0] ?? null); + + if (!updated) continue; + + result.cleared += 1; + result.issueIds.push(updated.id); + + await logActivity(db, { + companyId: issue.companyId, + actorType: "system", + actorId: "system", + agentId: null, + runId: null, + action: "issue.stale_lock_cleared", + entityType: "issue", + entityId: updated.id, + details: { + source: "recovery.sweep_stale_issue_locks", + clearedCheckoutRunId: issue.checkoutRunId, + clearedExecutionRunId: issue.executionRunId, + referencedRunStatuses: Object.fromEntries(runStatusById), + }, + }); + } + + if (result.cleared > 0) { + logger.warn( + { cleared: result.cleared, issueIds: result.issueIds }, + "swept stale issue lock columns", + ); + } + + return result; + } + return { buildRunOutputSilence, escalateStrandedRecoveryIssueInPlace, @@ -3611,6 +3720,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) recordWatchdogDecision, scanSilentActiveRuns, reconcileStrandedAssignedIssues, + sweepStaleIssueLocks, buildIssueGraphLivenessAutoRecoveryPreview, reconcileIssueGraphLiveness, readRecoveryTimerIntervalMs,