Merge remote-tracking branch 'origin/master' into PAP-10341-paperclip-needs-good-permissions-around-mcp-servers-consider-natoma-and-make-a-plan

This commit is contained in:
Dotta committed 2026-06-10 22:09:03 +00:00
commit 485d2ddd56
16 files changed
+898 -58

No files matched your search

+12 -1
View File
@@ -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.
@@ -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<AdapterExec
? "unknown"
: isClaudePoisonedPreviousMessageIdError(initial.parsed)
? "poisoned"
: isClaudeImageProcessingError(initial.parsed)
? "image"
: null
: null;
@@ -990,6 +993,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
const reason =
sessionErrorKind === "poisoned"
? "returned a poisoned message-id"
: sessionErrorKind === "image"
? "contains an unprocessable image"
: "is unavailable";
await onLog(
"stdout",
@@ -4,6 +4,7 @@ import {
isClaudeTransientUpstreamError,
isClaudePoisonedPreviousMessageIdError,
isClaudeUnknownSessionError,
isClaudeImageProcessingError,
} from "./parse.js";
describe("isClaudeTransientUpstreamError", () => {
@@ -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");
@@ -209,6 +209,17 @@ export function isClaudePoisonedPreviousMessageIdError(parsed: Record<string, un
);
}
export function isClaudeImageProcessingError(parsed: Record<string, unknown>): 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<string, unknown> | 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({
@@ -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 });
}
});
});
@@ -65,7 +65,7 @@ async function createExpectedSymlink(target: string, source: string): Promise<vo
}
}
async function ensureSymlink(target: string, source: string): Promise<void> {
export async function ensureSymlink(target: string, source: string): Promise<void> {
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<void> {
}
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;
}
@@ -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", () => {
@@ -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,
);
}
@@ -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()
@@ -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,
});
});
});
+165
View File
@@ -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", () => {
@@ -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<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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);
});
});
+12
View File
@@ -754,6 +754,12 @@ export async function startServer(): Promise<StartedServer> {
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<StartedServer> {
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) {
+19 -2
View File
@@ -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,
+118 -48
View File
@@ -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<boolean> {
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) => {
+111 -1
View File
@@ -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<string, string>();
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,