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 ced38bced04..a74ecb1a1c6 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__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts new file mode 100644 index 00000000000..de6baf9016b --- /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 105f23ecaf7..9180849c17c 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -707,6 +707,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) { @@ -773,6 +779,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 3fa574b7783..9728466d145 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -6593,6 +6593,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); @@ -8158,16 +8162,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)), + ), + ); } while (true) { @@ -9719,6 +9733,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 fb47582fe04..e0839371d59 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -3382,8 +3382,58 @@ 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 }) + .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; + + 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), + ), + ) + .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") { @@ -4610,6 +4660,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 ?? []; @@ -4739,6 +4790,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, @@ -4817,6 +4869,7 @@ export function issueService(db: Db) { release: async (id: string, actorAgentId?: string, actorRunId?: string | null) => { await clearExecutionRunIfTerminal(id); + await clearCheckoutRunIfTerminal(id); const existing = await db .select() .from(issues) diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index f94d5b52c0a..341b444f125 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -3039,6 +3039,112 @@ 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 TERMINAL_STATUSES = ["succeeded", "failed", "cancelled", "timed_out"] as const; + 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)`, + ); + + for (const issue of candidates) { + const referencedRunIds = [issue.checkoutRunId, issue.executionRunId].filter( + (id): id is string => !!id, + ); + if (referencedRunIds.length === 0) continue; + + const runRows = 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_STATUSES as readonly string[]).includes(status); + }; + + 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, @@ -3046,6 +3152,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) recordWatchdogDecision, scanSilentActiveRuns, reconcileStrandedAssignedIssues, + sweepStaleIssueLocks, buildIssueGraphLivenessAutoRecoveryPreview, reconcileIssueGraphLiveness, readRecoveryTimerIntervalMs,