diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7ebfdee4059..fffdb1dad48 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/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index 6086cc3e3ea..a265606183e 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -55,6 +55,7 @@ import { isClaudeTransientUpstreamError, isClaudeUnknownSessionError, isClaudePoisonedPreviousMessageIdError, + isClaudeImageProcessingError, } from "./parse.js"; import { prepareClaudeConfigSeed, resolveSharedClaudeConfigDir } from "./claude-config.js"; import { resolveClaudeDesiredSkillNames } from "./skills.js"; @@ -983,6 +984,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise { @@ -185,6 +186,50 @@ describe("isClaudeUnknownSessionError", () => { }); }); +describe("isClaudeImageProcessingError", () => { + it("detects the 'Could not process image' 400 error in the result field", () => { + expect( + isClaudeImageProcessingError({ + subtype: "success", + is_error: true, + result: "API Error: 400 Could not process image: image source URL has expired", + }), + ).toBe(true); + }); + + it("detects the error in the errors array", () => { + expect( + isClaudeImageProcessingError({ + is_error: true, + result: "", + errors: [{ message: "400 Could not process image" }], + }), + ).toBe(true); + }); + + it("is case-insensitive", () => { + expect( + isClaudeImageProcessingError({ + is_error: true, + result: "could not process image attached to message", + }), + ).toBe(true); + }); + + it("returns false for unrelated errors", () => { + expect( + isClaudeImageProcessingError({ + is_error: true, + result: "No conversation found with session id abc-123", + }), + ).toBe(false); + }); + + it("returns false for empty parsed result", () => { + expect(isClaudeImageProcessingError({})).toBe(false); + }); +}); + describe("extractClaudeRetryNotBefore", () => { it("parses the 'resets 4pm' hint in its explicit timezone", () => { const now = new Date("2026-04-22T15:15:00.000Z"); diff --git a/packages/adapters/claude-local/src/server/parse.ts b/packages/adapters/claude-local/src/server/parse.ts index 36d13fd60ed..fbf66f792b6 100644 --- a/packages/adapters/claude-local/src/server/parse.ts +++ b/packages/adapters/claude-local/src/server/parse.ts @@ -209,6 +209,17 @@ export function isClaudePoisonedPreviousMessageIdError(parsed: Record): boolean { + const resultText = asString(parsed.result, "").trim(); + const allMessages = [resultText, ...extractClaudeErrorMessages(parsed)] + .map((msg) => msg.trim()) + .filter(Boolean); + + return allMessages.some((msg) => + /could not process image/i.test(msg), + ); +} + function buildClaudeTransientHaystack(input: { parsed?: Record | null; stdout?: string | null; @@ -388,7 +399,7 @@ export function isClaudeTransientUpstreamError(input: { }): boolean { const parsed = input.parsed ?? null; // Deterministic failures are handled by their own classifiers. - if (parsed && (isClaudeMaxTurnsResult(parsed) || isClaudeUnknownSessionError(parsed) || isClaudePoisonedPreviousMessageIdError(parsed))) { + if (parsed && (isClaudeMaxTurnsResult(parsed) || isClaudeUnknownSessionError(parsed) || isClaudePoisonedPreviousMessageIdError(parsed) || isClaudeImageProcessingError(parsed))) { return false; } const loginMeta = detectClaudeLoginRequired({ diff --git a/packages/adapters/codex-local/src/server/codex-home.test.ts b/packages/adapters/codex-local/src/server/codex-home.test.ts index 864837510b3..840db456916 100644 --- a/packages/adapters/codex-local/src/server/codex-home.test.ts +++ b/packages/adapters/codex-local/src/server/codex-home.test.ts @@ -2,7 +2,7 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; -import { prepareManagedCodexHome } from "./codex-home.js"; +import { ensureSymlink, prepareManagedCodexHome } from "./codex-home.js"; describe("codex managed home", () => { afterEach(() => { @@ -54,4 +54,95 @@ describe("codex managed home", () => { await fs.rm(root, { recursive: true, force: true }); } }); + + // Regression for #5028: older Paperclip versions copied auth.json into the + // managed home instead of symlinking. After upgrading to the symlink-based + // logic, the stale regular file at the target stayed in place and every + // subsequent codex_local run failed with refresh_token_reused as soon as the + // source token rotated. `ensureSymlink` now heals the upgrade path by + // unlinking the stale copy and creating a symlink to the live source. + it("replaces a stale regular-file auth.json with a symlink to the live source (#5028)", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-codex-home-")); + try { + const sharedCodexHome = path.join(root, "shared-codex-home"); + const paperclipHome = path.join(root, "paperclip-home"); + const managedCodexHome = path.join( + paperclipHome, + "instances", + "default", + "companies", + "company-1", + "codex-home", + ); + const sharedAuth = path.join(sharedCodexHome, "auth.json"); + const managedAuth = path.join(managedCodexHome, "auth.json"); + + await fs.mkdir(sharedCodexHome, { recursive: true }); + // The live source has rotated since the stale copy was written. + await fs.writeFile(sharedAuth, '{"token":"fresh"}', "utf8"); + + // Simulate a stale copy left by a previous Paperclip version. + await fs.mkdir(managedCodexHome, { recursive: true }); + await fs.writeFile(managedAuth, '{"token":"stale-from-copy"}', "utf8"); + + await prepareManagedCodexHome( + { + CODEX_HOME: sharedCodexHome, + PAPERCLIP_HOME: paperclipHome, + PAPERCLIP_INSTANCE_ID: "default", + }, + async () => {}, + "company-1", + ); + + expect((await fs.lstat(managedAuth)).isSymbolicLink()).toBe(true); + expect(await fs.readFile(managedAuth, "utf8")).toBe('{"token":"fresh"}'); + } finally { + await fs.rm(root, { recursive: true, force: true }); + } + }); + + // Direct unit coverage for the new ensureSymlink branch (#5028). The + // regression test above goes through prepareManagedCodexHome, whose + // pre-existing apikey-mode cleanup `fs.rm`s the stale auth.json before + // ensureSymlink runs — so the heal branch never executes there. Call + // ensureSymlink directly to prove the unlink-and-recreate path itself. + it("ensureSymlink: unlinks a stale regular file and recreates the symlink", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-ensure-symlink-")); + try { + const source = path.join(root, "live-source.json"); + const target = path.join(root, "stale-target.json"); + await fs.writeFile(source, '{"token":"fresh"}', "utf8"); + await fs.writeFile(target, '{"token":"stale-from-copy"}', "utf8"); + + await ensureSymlink(target, source); + + expect((await fs.lstat(target)).isSymbolicLink()).toBe(true); + expect(await fs.readFile(target, "utf8")).toBe('{"token":"fresh"}'); + } finally { + await fs.rm(root, { recursive: true, force: true }); + } + }); + + // The isDirectory() guard added with the heal branch must keep an unexpected + // directory in place rather than throwing EISDIR. We treat a directory at + // this path as operator-owned, not a stale Paperclip copy. + it("ensureSymlink: leaves an unexpected directory in place instead of throwing", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-ensure-symlink-dir-")); + try { + const source = path.join(root, "live-source.json"); + const target = path.join(root, "unexpected-dir"); + await fs.writeFile(source, '{"token":"fresh"}', "utf8"); + await fs.mkdir(target); + await fs.writeFile(path.join(target, "sentinel"), "keep-me", "utf8"); + + await expect(ensureSymlink(target, source)).resolves.toBeUndefined(); + + expect((await fs.lstat(target)).isDirectory()).toBe(true); + expect(await fs.readFile(path.join(target, "sentinel"), "utf8")).toBe("keep-me"); + } finally { + await fs.rm(root, { recursive: true, force: true }); + } + }); + }); diff --git a/packages/adapters/codex-local/src/server/codex-home.ts b/packages/adapters/codex-local/src/server/codex-home.ts index ee87e6cbf5e..bd9d4c02e30 100644 --- a/packages/adapters/codex-local/src/server/codex-home.ts +++ b/packages/adapters/codex-local/src/server/codex-home.ts @@ -65,7 +65,7 @@ async function createExpectedSymlink(target: string, source: string): Promise { +export async function ensureSymlink(target: string, source: string): Promise { const existing = await fs.lstat(target).catch(() => null); if (!existing) { await ensureParentDir(target); @@ -74,6 +74,19 @@ async function ensureSymlink(target: string, source: string): Promise { } if (!existing.isSymbolicLink()) { + // A previous Paperclip version copied this file into the managed home + // instead of symlinking it. Codex refresh tokens rotate and are + // single-use, so a stale copy fails with refresh_token_reused on the next + // run (#5028). Replace the regular file with a symlink so the CLI follows + // the live source. Safe to delete: target is always under the + // Paperclip-managed company home, never the user's real ~/.codex. + // Directories are left alone — `fs.unlink` would throw EISDIR on Unix + // (and behave inconsistently on Windows). A directory at this path is not + // a Paperclip-written stale copy and warrants operator inspection rather + // than silent removal. + if (existing.isDirectory()) return; + await fs.unlink(target); + await createExpectedSymlink(target, source); return; } diff --git a/packages/adapters/codex-local/src/server/parse.test.ts b/packages/adapters/codex-local/src/server/parse.test.ts index 93b76b1865e..bfd15e24ae8 100644 --- a/packages/adapters/codex-local/src/server/parse.test.ts +++ b/packages/adapters/codex-local/src/server/parse.test.ts @@ -80,6 +80,7 @@ describe("isCodexUnknownSessionError", () => { it("still detects existing stale-session wordings", () => { expect(isCodexUnknownSessionError("unknown thread id", "")).toBe(true); expect(isCodexUnknownSessionError("", "state db missing rollout path for thread abc")).toBe(true); + expect(isCodexUnknownSessionError("", "state db returned stale rollout path for thread abc")).toBe(true); }); it("does not classify unrelated Codex failures as stale sessions", () => { diff --git a/packages/adapters/codex-local/src/server/parse.ts b/packages/adapters/codex-local/src/server/parse.ts index 679a3f8f4df..6dd3d5fb3f2 100644 --- a/packages/adapters/codex-local/src/server/parse.ts +++ b/packages/adapters/codex-local/src/server/parse.ts @@ -78,7 +78,7 @@ export function isCodexUnknownSessionError(stdout: string, stderr: string): bool .map((line) => line.trim()) .filter(Boolean) .join("\n"); - return /unknown (session|thread)|session .* not found|thread .* not found|conversation .* not found|missing rollout path for thread|state db missing rollout path|no rollout found for thread id/i.test( + return /unknown (session|thread)|session .* not found|thread .* not found|conversation .* not found|missing rollout path for thread|state db missing rollout path|state db returned stale rollout path|no rollout found for thread id/i.test( haystack, ); } diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 6e47ecc3c68..3954ccb02f8 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 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__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index 6569815cb63..9b0d7a8342f 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 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 38caf44a518..a81e50b579e 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 7b2da0653c6..b988dcb32ec 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 d422f4d9281..429205a1b4f 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 c9f4b4fdf90..1a5fb8c5d6a 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,