diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e316b51891..2ecf5fe725 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -837,6 +837,14 @@ Every continuation carries the triggering request, ordered user direction, inter ### Interrupted conversation continuation +Before provider dispatch, chat-control admission retries transient database lock +contention with up to 50 waits of 100 ms. Each attempt starts a new transaction +and rechecks the current run and committed conversation-close evidence. No lock +is held between attempts, and no provider call is retried. Queue claims remain +nonblocking. Persistent contention retains the bounded admission failure, with +an explicit database-lock error; missing or invalid source evidence still stops +the run without retrying the admission check. + An interrupted conversation does not permanently block its task. For local conversational adapters, Paperclip starts a new bounded turn with the existing session when compatible, or the full task conversation when the session is unavailable. The prompt says: “Your previous run was interrupted. Continue from where you left off.” The agent decides what remains from the history and latest user request. Paperclip never automatically replays recorded tool calls. Unknown past action outcomes are not a task-wide execution gate, and no action-reconciliation questionnaire is required. Shutdown, process loss, and provider failure use the existing durable failure retry counter and delay. Ordinary failure recovery permits at most two automatic retries in a failure chain. Accepted-interaction infrastructure recovery retains its existing bounded policy. Repeated scheduler visits reuse the same successor; restarting the server does not reset the counter. After exhaustion, automatic attempts stop. A new explicit user message can start a fresh run and failure budget. Productive max-turn continuation and confirmed workspace waits keep their separate existing semantics. diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 55b16971cd..ff517559ba 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -11335,6 +11335,96 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }, ); + it.each(["issue", "wake", "run", "native", "close", "cancel"] as const)( + "rechecks admission after transient database contention: %s", + async (mode) => { + const source = await seedCommittedChatControlStop(); + await db.update(chatPublications).set({ state: "pending" }) + .where(eq(chatPublications.id, source.publicationId)); + const child = await seedChatAutomaticChild(source); + // Board comments use the automation transport but are fresh user work. + if (!["close", "cancel"].includes(mode)) { + await db.update(agentWakeupRequests).set({ + requestedByActorType: "user", requestedByActorId: "responsible-user", + reason: "issue_commented", + }).where(eq(agentWakeupRequests.id, child.wakeupRequestId)); + await db.update(heartbeatRuns).set({ retryOfRunId: null }) + .where(eq(heartbeatRuns.id, child.runId)); + } + if (mode === "native") { + await db.update(agents).set({ + adapterType: "paperclip_runner", + adapterConfig: { provider: "codex", model: "gpt-5.6-luna" }, + }).where(eq(agents.id, source.agentId)); + } + const factory = vi.fn(() => { throw new NativeRunnerOwnershipUnverifiedError(); }); + let release!: () => void; + let locked: Promise | undefined; + let timer: ReturnType | undefined; + const heartbeat = heartbeatService(db, { + nativeSessionBackendFactory: factory, + beforeChatControlRecoveryCheck: async ({ stage }) => { + if (stage !== "dispatch") return; + let ready!: () => void; + const acquired = new Promise((resolve) => { ready = resolve; }); + const held = new Promise((resolve) => { release = resolve; }); + locked = db.transaction(async (tx) => { + if (mode === "wake") { + await tx.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, child.wakeupRequestId)).for("update"); + } else if (mode === "run" || mode === "cancel") { + await tx.select().from(heartbeatRuns) + .where(eq(heartbeatRuns.id, child.runId)).for("update"); + } else if (mode === "close") { + await tx.select().from(chatConversations) + .where(eq(chatConversations.id, source.conversationId)).for("update"); + } else { + await tx.select().from(issues) + .where(eq(issues.id, source.issueId)).for("update"); + } + ready(); + await held; + expect(mockAdapterExecute).not.toHaveBeenCalled(); + expect(factory).not.toHaveBeenCalled(); + if (mode === "close") { + await tx.update(chatPublications).set({ state: "published" }) + .where(eq(chatPublications.id, source.publicationId)); + } else if (mode === "cancel") { + await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, child.runId)); + } + }); + await acquired; + timer = setTimeout(release, 250); + }, + }); + try { + await heartbeat.resumeQueuedRuns(); + await heartbeat.drainActiveRunExecutions(); + } finally { + if (timer) clearTimeout(timer); + release?.(); + await locked; + await heartbeat.drainActiveRunExecutions(); + } + expect(locked).toBeDefined(); + const settled = await heartbeat.getRun(child.runId); + expect(settled?.errorCode).not.toBe(CHAT_CONTROL_RECOVERY_UNRESOLVED_CODE); + if (mode === "native") { + expect(factory).toHaveBeenCalledTimes(1); + expect(readChatControlRecoveryAdmission(settled!)).toBe("admitted"); + } else if (mode === "close" || mode === "cancel") { + expect(mockAdapterExecute).not.toHaveBeenCalled(); + expect(settled?.status).toBe("cancelled"); + if (mode === "close") expect(settled?.errorCode).toBe(CHAT_CONTROL_RECOVERY_STOP_CODE); + } else { + expect(mockAdapterExecute).toHaveBeenCalledTimes(1); + expect(settled?.status).toBe("succeeded"); + expect(readChatControlRecoveryAdmission(settled!)).toBe("admitted"); + } + }, + ); + it("defers unresolved automatic ancestry at claim and records a distinct nonretrying failure after claim", async () => { const source = await seedCommittedChatControlStop(); await db diff --git a/server/src/services/chat-control-admission-retry.test.ts b/server/src/services/chat-control-admission-retry.test.ts new file mode 100644 index 0000000000..b9ef4860be --- /dev/null +++ b/server/src/services/chat-control-admission-retry.test.ts @@ -0,0 +1,33 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { retryChatControlAdmission } from "./chat-control-admission-retry.js"; + +afterEach(() => vi.useRealTimers()); + +it("retries a rolled-back lock conflict and returns the fresh admission result", async () => { + vi.useFakeTimers(); + const attempt = vi.fn() + .mockRejectedValueOnce(new Error("query failed", { cause: { code: "55P03" } })) + .mockResolvedValueOnce(null); + const result = retryChatControlAdmission(attempt); + await vi.advanceTimersByTimeAsync(100); + await expect(result).resolves.toBeNull(); + expect(attempt).toHaveBeenCalledTimes(2); +}); + +it("does not retry unrelated database failures", async () => { + const error = new Error("constraint violation", { cause: { code: "23505" } }); + const attempt = vi.fn().mockRejectedValue(error); + await expect(retryChatControlAdmission(attempt)).rejects.toBe(error); + expect(attempt).toHaveBeenCalledTimes(1); +}); + +it("stops persistent contention after fifty delays", async () => { + vi.useFakeTimers(); + const error = new Error("query failed", { cause: { code: "55P03" } }); + const attempt = vi.fn().mockRejectedValue(error); + const rejected = expect(retryChatControlAdmission(attempt)).rejects.toBe(error); + await vi.advanceTimersByTimeAsync(5_000); + await rejected; + expect(attempt).toHaveBeenCalledTimes(51); + expect(vi.getTimerCount()).toBe(0); +}); diff --git a/server/src/services/chat-control-admission-retry.ts b/server/src/services/chat-control-admission-retry.ts new file mode 100644 index 0000000000..e3e618db99 --- /dev/null +++ b/server/src/services/chat-control-admission-retry.ts @@ -0,0 +1,15 @@ +import { isExternalChatWaitAuthorizationContention } from "./native-runtime/chat-attachment-reuse.js"; + +/** Retry only a rolled-back admission transaction, never provider execution. */ +export async function retryChatControlAdmission(attempt: () => Promise): Promise { + for (let retry = 0; ; retry += 1) { + try { + return await attempt(); + } catch (error) { + if (retry >= 50 || !isExternalChatWaitAuthorizationContention(error)) throw error; + } + // The previous transaction has released all locks. The next attempt must + // read current run ownership and close evidence again before admission. + await new Promise((resolve) => setTimeout(resolve, 100)); + } +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 586b1bea03..2825c56d71 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -547,6 +547,7 @@ import { extractSkillMentionIds, isUuidLike } from "@paperclipai/shared"; import { evaluateCodexCredentialReadiness } from "@paperclipai/adapter-codex-local/server"; import { environmentService } from "./environments.js"; import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js"; +import { retryChatControlAdmission } from "./chat-control-admission-retry.js"; import { environmentRuntimeService, type ProviderResourceDisposition, @@ -799,7 +800,7 @@ function nonRetryablePreflightFailureCode(error: unknown): string | null { class ChatControlRecoveryUnresolvedError extends Error { constructor() { super( - "Automatic continuation source could not be verified before provider admission. Review the task and send a fresh request; this attempt will not automatically retry.", + "Run admission could not acquire its database locks after bounded retries. No provider work started. Review database contention and send a fresh request; this attempt will not automatically retry.", ); } } @@ -16401,9 +16402,11 @@ export function heartbeatService( } let terminal: typeof heartbeatRuns.$inferSelect | null = null; try { - const result = await db.transaction(async (tx) => { + const attempt = () => db.transaction(async (tx) => { + terminal = null; // Same queue-edit lock order, then the close committer's conversation - // row. NOWAIT makes contention a scoped deferral, never authority. + // row. NOWAIT releases partial locks on contention. Claim defers to the + // queue; dispatch retries this transaction before considering failure. const [issue] = await tx .select({ id: issues.id }) .from(issues) @@ -16567,6 +16570,9 @@ export function heartbeatService( ); return null; }); + const result = stage === "dispatch" + ? await retryChatControlAdmission(attempt) + : await attempt(); if (terminal) { const settled = terminal as typeof heartbeatRuns.$inferSelect; publishLiveEvent({