fix(runtime): defer native routine recovery collisions (#18916)

Treat the exact routine execution unique conflict as a pending recovery obligation after the complete losing transaction rolls back. Preserve the current routine owner, provider attempt and checkpoint, and continue the startup batch. Cover real constraint failures, row preservation, one reclaim after release and unrelated database failures.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/execution-semantics.md

@@ -1264,6 +1264,11 @@ those locks in place and clears them only when the task is done or cancelled.
 If a scheduled resume finds its execution lock free, restart recovery reclaims
 it, as the local resume claim does, unless another run for the task is queued,
 running, or scheduled. A lock held by another run still blocks the resume.
+Another task can hold the same routine execution slot even when this task's
+lock is free. If that unique claim fails, restart recovery rolls back the whole
+candidate claim and waits through the existing recovery deadline. It preserves
+the current routine owner, checkpoint, and provider attempt, and continues the
+startup batch. Other database failures still fail the recovery pass.
 
 Automatic productivity reviews are retired. Run counts, missing comments, and elapsed task time do not create review tasks or impose continuation holds. Bounded continuation, provider recovery, budget limits, explicit blockers, and normal review/approval stages remain in force. Existing productivity-review tasks, comments, assignments, and dependencies remain unchanged and readable; their historical origins still identify them as recovery work for recursion suppression.
 

server/src/services/native-runtime/native-restart-recovery.ts

@@ -34,6 +34,7 @@ import { REMOTE_CONTROLLER_LEASE_MS } from "./native-runner-recovery-contract.js
 import { readProcessStartedAt } from "../hot-restart.js";
 import { getServerProcessStartedAt } from "../../server-info.js";
 import { redactSensitiveText } from "../../redaction.js";
+import { isUniqueViolation } from "../../db-errors.js";
 import { issueRecoveryActionService } from "../issue-recovery-actions.js";
 import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js";
 import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
@@ -1236,6 +1237,12 @@ export async function claimNativeRestartRecoveries(input: {
         kind: claimKind,
         ...common,
       } satisfies NativeRestartRecoveryClaim;
+    }).catch((error: unknown) => {
+      if (!isUniqueViolation(error, "issues_open_routine_execution_uq")) throw error;
+      terminalRunToReport = null;
+      claimScope = undefined;
+      return { kind: "awaiting_evidence" as const, runId: candidate.runId,
+        reason: "routine_execution_busy", retryAt: new Date(now.getTime() + 30_000) };
     });
     if (terminalRunToReport)
       void reportRunFailure(input.db, terminalRunToReport);

server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts

@@ -152,6 +152,102 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
     expect(next.kind).toBe("bootstrap_incomplete");
   });
 
+  it("preserves a retry when another task owns its routine slot and continues startup recovery", async () => {
+    const candidates = [await seedRun("ROUTINE-COLLISION"), await seedRun("AFTER-ROUTINE-COLLISION")]
+      .sort((a, b) => a.runId.localeCompare(b.runId));
+    const [waiting, unrelated] = candidates;
+    const holder = await seedRun("ROUTINE-SLOT-OWNER", waiting.db);
+    const routineId = randomUUID();
+    const now = new Date();
+    await waiting.db.update(issues).set({ executionRunId: null, originKind: "routine_execution",
+      originId: routineId, originFingerprint: "scheduled" }).where(eq(issues.id, waiting.issueId));
+    await waiting.db.update(issues).set({ originKind: "routine_execution",
+      originId: routineId, originFingerprint: "scheduled" }).where(eq(issues.id, holder.issueId));
+    await waiting.db.update(heartbeatRuns).set({ status: "failed", finishedAt: now,
+      nativePhase: "retryable_failure", runnerProfileJson: { ...waiting.preparedProfile,
+        nativeWorkspaceSync: remotePreparedWorkspace(),
+        sessionCheckpoint: { identity: { companyId, agentId, issueId: waiting.issueId, runId: waiting.runId,
+          sessionId: waiting.native.normalizedSessionId }, providerSessionId: "retained-routine-session" },
+      } }).where(eq(heartbeatRuns.id, waiting.runId));
+    await waiting.db.update(nativeRunFinalizations).set({ controllerBootId: successor.bootId,
+      leaseOwner: "predecessor-routine-controller", leaseExpiresAt: new Date(now.getTime() + 60_000) })
+      .where(eq(nativeRunFinalizations.runId, waiting.runId));
+    const authority = new DurablePrpControlPlane({ stateDirectory: resolve(runtimeRoot, waiting.runId),
+      identity: transportOptions(waiting, runtimeRoot).prpIdentity,
+      expectedRunnerVersion: "routine-recovery-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
+    try {
+      await registerNativeRunnerCheckpoint({ db: waiting.db,
+        binding: { companyId, issueId: waiting.issueId, runId: waiting.runId }, bootId: successor.bootId, authority });
+    } finally { await authority.stop(); }
+    await waiting.db.update(nativeRunFinalizations).set({ phase: "retryable_failure", attempt: 1,
+      leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: new Date(0), failureCode: "provider_transport_interrupted" })
+      .where(eq(nativeRunFinalizations.runId, waiting.runId));
+    const state = async () => ({
+      run: await waiting.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, waiting.runId)),
+      coordinator: await waiting.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, waiting.runId)),
+      checkpoint: await waiting.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, waiting.runId)),
+      task: await waiting.db.select().from(issues).where(eq(issues.id, waiting.issueId)),
+      holder: await waiting.db.select().from(issues).where(eq(issues.id, holder.issueId)),
+    });
+    const before = await state();
+    expect(before.checkpoint.length).toBeGreaterThan(0);
+    const onClaim = vi.fn();
+    const input = { db: waiting.db, controller: successor, restartKind: "hard" as const, now };
+    const dispositions = await claimNativeRestartRecoveries({ ...input, onClaim,
+      runIds: candidates.map(candidate => candidate.runId) });
+    expect(dispositions).toEqual([
+      { kind: "awaiting_evidence", runId: waiting.runId, reason: "routine_execution_busy",
+        retryAt: new Date(now.getTime() + 30_000) },
+      expect.objectContaining({ kind: "bootstrap_incomplete", runId: unrelated.runId }),
+    ]);
+    expect(onClaim).toHaveBeenCalledOnce();
+    expect(onClaim.mock.calls[0][0].runId).toBe(unrelated.runId);
+    expect(await state()).toEqual(before);
+    expect(await waiting.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, waiting.runId),
+      eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(0);
+    expect(await waiting.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, waiting.runId))).toHaveLength(0);
+
+    const heartbeat = heartbeatService(waiting.db);
+    try {
+      await expect(heartbeat.recoverNativeRunsAfterRestart(now, [waiting.runId]))
+        .resolves.toMatchObject({ claims: [], awaitingEvidenceRunIds: [waiting.runId] });
+    } finally { await heartbeat.drainActiveRunExecutions(); }
+    expect(await state()).toEqual({ ...before, run: [{ ...before.run[0],
+      nextEventSeq: before.run[0].nextEventSeq + 1, updatedAt: expect.any(Date) }] });
+
+    await waiting.db.update(issues).set({ executionRunId: null }).where(eq(issues.id, holder.issueId));
+    const recovered = (await Promise.all([1, 2].map(() => claimNativeRestartRecoveries({ ...input,
+      runIds: [waiting.runId] })))).flat();
+    expect(recovered.filter(row => row.kind === "reattach_remote_runner")).toHaveLength(1);
+    const after = await state();
+    expect(after.run[0].runnerProfileJson).toEqual(before.run[0].runnerProfileJson);
+    expect(after.checkpoint).toEqual(before.checkpoint);
+    expect(after.coordinator[0]).toMatchObject({ attempt: 1, controllerGeneration: 1 });
+    expect(after.task[0].executionRunId).toBe(waiting.runId);
+    expect(await waiting.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, waiting.runId),
+      eq(heartbeatRunEvents.eventType, "native.recovery.execution_reclaimed")))).toHaveLength(1);
+  });
+
+  it("does not turn another database constraint failure into a routine wait", async () => {
+    const fixture = await seedRun("UNRELATED-DATABASE-FAILURE");
+    const db = new Proxy(fixture.db, { get(target, key) {
+      if (key === "transaction") return () => target.transaction(async tx => {
+        await tx.execute(sql`create temp table recovery_constraint_control
+          (id integer constraint recovery_unrelated_unique unique) on commit drop`);
+        await tx.execute(sql`insert into recovery_constraint_control values (1), (1)`);
+      });
+      const value = Reflect.get(target, key);
+      return typeof value === "function" ? value.bind(target) : value;
+    } });
+    const onClaim = vi.fn();
+    await expect(claimNativeRestartRecoveries({ db, controller: successor,
+      restartKind: "hard", runIds: [fixture.runId], onClaim })).rejects.toThrow(/recovery_constraint_control/);
+    expect(onClaim).not.toHaveBeenCalled();
+    const [coordinator] = await fixture.db.select().from(nativeRunFinalizations)
+      .where(eq(nativeRunFinalizations.runId, fixture.runId));
+    expect(coordinator).toMatchObject({ phase: "observed", controllerGeneration: 0, leaseOwner: null });
+  });
+
   it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "finalizer-wait", "warm-suspended", "stale-projection", "legacy", "changed", "same-server", "diskless", "diskless-per-turn", "diskless-per-turn-read-retry", "diskless-warm-suspended", "diskless-warm-suspended-read-retry", "diskless-live-inspection"])("preserves remote warm continuity across completed turns: %s", async (scenario) => {
     const hardRestart = !["warm", "same-server", "warm-suspended"].includes(scenario);
     const diskless = scenario.startsWith("diskless");