Retain skipped-lock settlement and failed native cancellation work

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

server/src/__tests__/live-recovery-work.test.ts

@@ -29,6 +29,8 @@ import { createAgentLifecycle, configureAgentLifecycle } from "../services/agent
 import { issueTreeControlService } from "../services/issue-tree-control.js";
 import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
 import { issueService } from "../services/issues.js";
+import { nativeRetryCancellationEligible, pendingFailedNativeRetryCancellationCondition } from "../services/native-runtime/native-cancellation-request.js";
+import { requestNativeQuestionRunCancellation } from "../services/native-runtime/native-question-bridge.js";
 import { readNativeLocalProcessStop } from "../services/native-local-process-stop.js";
 import { remoteTerminationReceipt } from "../services/remote-execution-termination.js";
 import { logger } from "../middleware/logger.js";
@@ -161,6 +163,78 @@ describe("residual recovery production matrix", () => {
     expect((await getRun(run.id)).resultJson).toMatchObject({ cancelledByIssueStatus: "done", cancelledIssueId: f.issue.id });
     await quiet();
   });
+  it.each([false, true])("finishes a question cancellation whose run fails into a pending native retry before consumption (partial Stop fault: %s)", async partialStopFault => {
+    const f = await fixture();
+    let stopReceipt: Record<string, unknown> | undefined;
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      status: "running", runtimeMode: "native", nativeIssueId: f.issue.id, nativePhase: "awaiting_input",
+      contextSnapshot: { issueId: f.issue.id } }).returning();
+    await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id,
+      runId: run.id, phase: "observed", recoveryState: "awaiting_evidence" });
+    if (partialStopFault) {
+      await db.execute(sql`create sequence question_cancel_failure_once`);
+      await db.execute(sql.raw(`create function question_cancel_failure_once() returns trigger language plpgsql as $$
+        begin if new.id = '${run.id}'::uuid and new.status = 'cancelled' and nextval('question_cancel_failure_once') = 1
+          then raise exception 'fixture question cancellation commit unavailable'; end if; return new; end $$`));
+      await db.execute(sql`create trigger question_cancel_failure_once before update on heartbeat_runs for each row execute function question_cancel_failure_once()`);
+      extraStops.push(async () => {
+        await db.execute(sql`drop trigger if exists question_cancel_failure_once on heartbeat_runs`);
+        await db.execute(sql`drop function if exists question_cancel_failure_once()`);
+        await db.execute(sql`drop sequence if exists question_cancel_failure_once`);
+      });
+    }
+    const marker = { version: 1, issueId: f.issue.id, kind: "issue_terminal", issueStatus: "done" };
+    const history = await db.insert(heartbeatRuns).values(Array.from({ length: 25 }, () => ({
+      companyId: f.companyId, agentId: f.agent.id, runtimeMode: "native", nativeIssueId: f.issue.id,
+      status: "failed", finishedAt: new Date(), nativePhase: "terminal_failure",
+      contextSnapshot: { nativeQuestionCancellation: marker }, resultJson: { nativeCancellation: {
+        schema: "paperclip.native-cancellation.v1", runId: randomUUID(), companyId: f.companyId, issueId: f.issue.id,
+        scope: "run", intentId: randomUUID(), intentAuditId: randomUUID(), dispatched: false, dispatchState: "pending",
+      } },
+    }))).returning();
+    await db.insert(nativeRunFinalizations).values(history.map(row => ({ companyId: f.companyId, issueId: f.issue.id,
+      runId: row.id, phase: "terminal_failure", failureCode: "native_retry_cancelled", recoveryState: "blocked" })));
+    draining = true; await start();
+    expect(await db.transaction(tx => requestNativeQuestionRunCancellation(tx, { companyId: f.companyId,
+      issueId: f.issue.id, sourceRunId: run.id, idempotencyKey: `paperclip-runner-question:${run.id}:raced-question`,
+      payload: { questionSet: {}, runtimeRequestId: "raced-question" },
+    }, { kind: "issue_terminal", issueStatus: "done" }))).toBe(run.id);
+    expect((await getRun(run.id)).contextSnapshot?.nativeQuestionCancellation).toMatchObject(marker);
+    await db.transaction(async tx => {
+      await notifyExecutionRunWork(tx, { companyId: f.companyId, runId: run.id, agentId: f.agent.id });
+      await tx.update(heartbeatRuns).set({ status: "failed", nativePhase: "retryable_failure", finishedAt: new Date() })
+        .where(eq(heartbeatRuns.id, run.id));
+      await tx.update(nativeRunFinalizations).set({ phase: "retryable_failure", recoveryState: "resuming_session",
+        nextAttemptAt: new Date(), leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, run.id));
+    });
+    draining = false; await advance(1_000);
+    if (partialStopFault) {
+      await vi.waitFor(() => expect(errors).toHaveLength(1));
+      expect(errors.shift()).toBeInstanceOf(AggregateError);
+      expect((await getRun(run.id)).status).toBe("failed");
+      const saved = await getRun(run.id);
+      stopReceipt = saved.resultJson?.nativeCancellation as Record<string, unknown>;
+      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
+      expect(coordinator).toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled", resultId: null });
+      expect(nativeRetryCancellationEligible({ runId: run.id, companyId: f.companyId, issueId: f.issue.id,
+        resultJson: saved.resultJson }, coordinator)).toBe(true);
+      const [debt] = await db.select({ pending: pendingFailedNativeRetryCancellationCondition() }).from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, run.id));
+      expect(debt.pending).toBe(true);
+    }
+    await advance(30_000);
+    await vi.waitFor(async () => expect((await getRun(run.id)).status).toBe("cancelled"));
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0])
+      .toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null });
+    if (stopReceipt) expect((await getRun(run.id)).resultJson?.nativeCancellation).toMatchObject({
+      intentId: stopReceipt.intentId, intentAuditId: stopReceipt.intentAuditId,
+      acknowledgementAuditId: stopReceipt.acknowledgementAuditId,
+    });
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(26);
+    expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.eventType, "native.process_start_requested"))).toEqual([]);
+    expect((await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.id} <> ${run.id}`)).every(row => row.status === "failed")).toBe(true);
+    await quiet();
+  });
   it("preserves a fresh foreign controller until exact expiry and admits one process retry across two owners", async () => {
     const f = await fixture(), expiry = new Date(Date.now() + 3_000);
     const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "running",
@@ -341,6 +415,30 @@ describe("residual recovery production matrix", () => {
     await start(); await vi.waitFor(async () => expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("idle"));
     await quiet();
   });
+  it("retains only skipped agent lock work until a lock-only transaction releases it", async () => {
+    const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
+    await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "succeeded", finishedAt: old });
+    await db.update(agents).set({ status: "running", updatedAt: old }).where(eq(agents.id, f.agent.id));
+    let release!: () => void, locked!: () => void;
+    const held = new Promise<void>(resolve => { release = resolve; });
+    const ready = new Promise<void>(resolve => { locked = resolve; });
+    const lock = db.transaction(async tx => {
+      await tx.select({ id: agents.id }).from(agents).where(eq(agents.id, f.agent.id)).for("update");
+      locked(); await held;
+    });
+    try {
+      await ready; await start();
+      expect(deadlines.get("stranded-agent-status")!()).toBe(Date.now() + 30_000);
+      expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("running");
+    } finally { release(); await lock; }
+    await advance(29_999);
+    expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("running");
+    await advance(1);
+    await vi.waitFor(async () => expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("idle"));
+    expect(await db.select().from(activityLog).where(and(eq(activityLog.entityId, f.agent.id),
+      eq(activityLog.action, "agent.status_reconciled")))).toHaveLength(1);
+    await quiet();
+  });
   it("moves the silence threshold on actual output and folds same-run completion only at strict snooze expiry", async () => {
     const f = await fixture(), now = new Date(), old = new Date(now.getTime() - 10 * 60_000);
     await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id));

server/src/__tests__/native-cancellation-request.integration.test.ts

@@ -2,19 +2,21 @@ import { randomUUID } from "node:crypto";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { and, eq, inArray, sql } from "drizzle-orm";
 import { agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { rethrowNativeCancellationLockConflict, nativeRetryCancellationCommitCondition, cancellationIntentId, claimCancellationRequest, startupCancellationFence } from "../services/native-runtime/native-cancellation-request.js";
 
 describe("atomic caller cancellation request ownership", () => {
-  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   let db: ReturnType<typeof createDb>;
   let cancelNativeSession: typeof import("../services/native-runtime/native-session-executor.js").cancelNativeSession;
   beforeAll(async () => {
     ({ cancelNativeSession } = await import("../services/native-runtime/native-session-executor.js"));
-    database = await startEmbeddedPostgresTestDatabase("caller-stop-");
-    db = createDb(database.connectionString);
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_CANCELLATION_TEST_DATABASE_URL");
+    if (!external) database = await startEmbeddedPostgresTestDatabase("caller-stop-");
+    db = createDb(external ?? database!.connectionString);
   }, 90_000);
-  afterAll(async () => { await database?.cleanup(); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
   async function fixture() {
     const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID();
     await db.insert(companies).values({ id: companyId, name: "Stop claims", issuePrefix: randomUUID().slice(0, 8) });

server/src/services/heartbeat/agent-status.test.ts

@@ -6,23 +6,29 @@ import {
 } from "@paperclipai/db";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
 import { createHeartbeatAgentStatus, settledAgentWorkCondition } from "./agent-status.js";
 
-const support = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_STATUS_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describePostgres = support.supported ? describe : describe.skip;
 if (!support.supported) console.warn(`Skipping agent status database tests: ${support.reason}`);
 
 describePostgres("heartbeat agent status", () => {
   let db: Db;
-  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   let service: ReturnType<typeof createHeartbeatAgentStatus>;
   const live = new Set<string>();
   const old = () => new Date(Date.now() - 60 * 60_000);
 
   beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("paperclip-agent-status-");
-    db = createDb(database.connectionString);
+    if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-agent-status-");
+    db = createDb(external ?? database!.connectionString);
+    if (external) {
+      db.$client.options.onnotice = () => {};
+      await db.execute(sql`truncate companies cascade`);
+    }
     service = createHeartbeatAgentStatus(db);
   }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
   beforeEach(async () => {
@@ -38,7 +44,7 @@ describePostgres("heartbeat agent status", () => {
     await db.delete(agents);
     await db.delete(companies);
   });
-  afterAll(async () => { await database?.cleanup(); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
 
   async function seed(status = "interrupted") {
     const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), issueId = randomUUID();

server/src/services/heartbeat/agent-status.ts

@@ -134,7 +134,13 @@ export function createHeartbeatAgentStatus(db: Db) {
         const [locked] = await tx.select({ id: agents.id }).from(agents).where(and(
           eq(agents.id, candidate.id), eq(agents.companyId, candidate.companyId),
         )).for("update", { skipLocked: true });
-        if (!locked) return null;
+        if (!locked) {
+          const [pending] = await tx.select({ id: agents.id }).from(agents).where(and(
+            eq(agents.id, candidate.id), eq(agents.companyId, candidate.companyId),
+          ));
+          if (pending) scope.onDeferredDeadline?.(Date.now() + 30_000);
+          return null;
+        }
         const [current] = await tx.select().from(agents).where(and(
           eq(agents.id, candidate.id), eq(agents.companyId, candidate.companyId),
           eq(agents.status, "running"), eq(agents.lifecycleState, "ready"),

server/src/services/heartbeat/live-recovery-work.ts

@@ -1,3 +1,4 @@
+import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js";
 import { settledNativeProcessIdentityCondition } from "../native-local-process-stop.js";
 import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
 import { agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
@@ -69,7 +70,8 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof
               kind === "native" ? and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["running", "failed"]),
                 eq(nativeRunFinalizations.phase, "retryable_failure"), isNull(nativeRunFinalizations.resultId),
                 sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`)
-                : kind === "question" ? and(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
+                : kind === "question" ? and(or(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
+                    pendingFailedNativeRetryCancellationCondition()),
                   sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is not null`)
                   : kind === "active" ? and(eq(heartbeatRuns.status, "running"),
                     sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`)

server/src/services/heartbeat/recovery.ts

@@ -1,3 +1,4 @@
+import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js";
 import { registerLiveRecoveryWork } from "./live-recovery-work.js";
 import { nativeEnvironmentSettlementDisposition, nativeEnvironmentSettlementWork } from "../native-runtime/native-environment-settlement-work.js";
 import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js";
@@ -2127,9 +2128,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
         and(
           scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
           scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
-          inArray(heartbeatRuns.status, [
+          or(inArray(heartbeatRuns.status, [
             ...CANCELLABLE_HEARTBEAT_RUN_STATUSES,
-          ]),
+          ]), pendingFailedNativeRetryCancellationCondition()),
           sql`${heartbeatRuns.contextSnapshot} -> ${NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY} is not null`,
         ),
       );

server/src/services/heartbeat/run-control.ts

@@ -1242,18 +1242,19 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
     // under lock, and the final CAS requires its own unchanged acknowledged
     // retry fence. Re-reading only retryable_failure here would strand replay.
     const pendingNativeRetry =
-      run.runtimeMode === "native" && run.status === "failed" && (
+      run.runtimeMode === "native" && run.status === "failed" && Boolean(run.nativeIssueId) && (
         caseHandoffRetryClaimed || Boolean(options.cancellationRequestId) || await db
-          .select({ runId: nativeRunFinalizations.runId })
+          .select({ phase: nativeRunFinalizations.phase, failureCode: nativeRunFinalizations.failureCode })
           .from(nativeRunFinalizations)
           .where(
             and(
               eq(nativeRunFinalizations.runId, run.id),
               eq(nativeRunFinalizations.companyId, run.companyId),
-              eq(nativeRunFinalizations.phase, "retryable_failure"),
+              eq(nativeRunFinalizations.issueId, run.nativeIssueId!),
             ),
           )
-          .then((rows) => rows.length > 0)
+          .then(([coordinator]) => nativeRetryCancellationEligible({ runId: run!.id, companyId: run!.companyId,
+            issueId: run!.nativeIssueId!, resultJson: run!.resultJson }, coordinator))
       );
     if (
       !pendingNativeRetry &&

server/src/services/native-runtime/native-cancellation-request.ts

@@ -41,6 +41,32 @@ export function nativeRetryCancellationEligible(input: {
       && !!native.acknowledgementAuditId && native.acknowledgementAuditId !== native.intentAuditId));
 }
 
+export function pendingFailedNativeRetryCancellationCondition() {
+  const native = sql`${heartbeatRuns.resultJson}->'nativeCancellation'`;
+  const continuingStop = sql`${nativeRunFinalizations.phase} = 'terminal_failure'
+    and ${nativeRunFinalizations.failureCode} = 'native_retry_cancelled'
+    and ${native}->>'schema' = 'paperclip.native-cancellation.v1'
+    and jsonb_typeof(${native}->'intentId') = 'string' and length(${native}->>'intentId') > 0
+    and ${native}->>'runId' = ${heartbeatRuns.id}::text
+    and ${native}->>'companyId' = ${heartbeatRuns.companyId}::text
+    and ${native}->>'issueId' = ${heartbeatRuns.nativeIssueId}::text
+    and ${native}->>'scope' = 'run'
+    and ${native}->>'dispatchState' in ('pending', 'acknowledged')
+    and jsonb_typeof(${native}->'dispatched') = 'boolean'
+    and jsonb_typeof(${native}->'intentAuditId') = 'string' and length(${native}->>'intentAuditId') > 0
+    and (${native}->>'dispatchState' = 'pending' or (
+      jsonb_typeof(${native}->'acknowledgementAuditId') = 'string'
+      and length(${native}->>'acknowledgementAuditId') > 0
+      and ${native}->>'acknowledgementAuditId' <> ${native}->>'intentAuditId'))`;
+  return sql`${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed'
+    and exists (select 1 from ${nativeRunFinalizations}
+      where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}
+        and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId}
+        and ${nativeRunFinalizations.issueId} = ${heartbeatRuns.nativeIssueId}
+        and (${nativeRunFinalizations.phase} = 'retryable_failure' or (${continuingStop}))
+        and ${nativeRunFinalizations.resultId} is null)`;
+}
+
 /** Final failed -> cancelled CAS: a later board/recovery outcome must win.
  * The row-locking subquery prevents a coordinator-only writer from changing
  * the retry disposition between this check and the status update. */