fix(server): recover captured instruction saves from committed deadlines

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

server/src/services/agent-deletion.ts

@@ -1,3 +1,4 @@
+import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
 import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
 import { and, eq, or, sql } from "drizzle-orm";
 import { agents, agentApiKeys, agentRuntimeState, agentTaskSessions, agentWakeupRequests, activityLog,
@@ -44,6 +45,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str
   await tx.delete(issueExecutionDecisions).where(eq(issueExecutionDecisions.actorAgentId, id));
   await tx.delete(issueComments).where(eq(issueComments.authorAgentId, id));
   await notifyScheduledRetryCancellation(tx, { companyId: companyId });
+  await notifyInstructionCommit(tx, { companyId: companyId });
   await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.agentId, id));
   await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, id));
   await tx.delete(agentApiKeys).where(eq(agentApiKeys.agentId, id));

server/src/services/agent-instruction-working-copies.ts

@@ -1,10 +1,14 @@
+import { notifyInstructionCommit, type InstructionCommitScope } from "./instruction-commit-notifications.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
 import { agentDirectoryWorkingCopyService, isAgentDirectoryCopy } from "./agent-directory-working-copies.js";
 import { createHash } from "node:crypto";
 import { execFile as execFileCallback } from "node:child_process";
 import { promisify } from "node:util";
 import fs from "node:fs/promises";
 import path from "node:path";
-import { and, asc, desc, eq, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm";
+import { and, asc, desc, eq, gt, inArray, lte, or, isNull, isNotNull, sql } from "drizzle-orm";
 import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db";
 import {
   prepareAdapterExecutionTargetRuntime,
@@ -73,11 +77,22 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
     const [row] = await db.select().from(copies).where(scope(companyId, runId));
     return row ?? null;
   }
+  async function tryPatch(row: Copy, values: Partial<typeof copies.$inferInsert>, requireDue = false) {
+    const signalsCommit = row.state === "pending_commit" || values.state === "pending_commit";
+    const write = async (tx: Db) => {
+      if (signalsCommit)
+        await notifyInstructionCommit(tx, { companyId: row.companyId, runId: row.runId });
+      const [updated] = await tx.update(copies).set({ ...values, updatedAt: new Date() }).where(and(
+        scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash),
+        sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`,
+        requireDue ? or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())) : undefined,
+      )).returning();
+      return updated;
+    };
+    return signalsCommit ? db.transaction(tx => write(tx as unknown as Db)) : write(db);
+  }
   async function patch(row: Copy, values: Partial<typeof copies.$inferInsert>) {
-    const [updated] = await db.update(copies).set({ ...values, updatedAt: new Date() }).where(and(
-      scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash),
-      sql`${copies.receipt}->>'retainedByRunId' IS NOT DISTINCT FROM ${typeof row.receipt?.retainedByRunId === "string" ? row.receipt.retainedByRunId : null}`,
-    )).returning();
+    const updated = await tryPatch(row, values);
     // A late cleanup must not overwrite an explicit resolution or a newer
     // baseline. Canonical content has its own independent head CAS.
     return updated ?? (await get(row.companyId, row.runId))!;
@@ -300,16 +315,13 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
   }
 
   /** Restarts can retry persisted bytes without touching a live filesystem or launching a model. */
-  async function recoverCaptured() {
-    const pending = await db.select().from(copies).where(and(
+  async function recoverCaptured(options: { skipCommit?: boolean } = {}) {
+    const pending = options.skipCommit ? [] : await db.select().from(copies).where(and(
       eq(copies.state, "pending_commit"),
       or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())),
       lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1),
     )).limit(20);
-    for (const row of pending) {
-      const result = await commitCandidate(await patch(row, { attempts: row.attempts + 1 }));
-      if (isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result);
-    }
+    for (const row of pending) await recoverCommit(commitEntry(row)!, () => true);
     // A crash between a durable save and release must not retain run snapshots
     // forever. Only discard copies with recorded stop proof and a terminal receipt.
     const cleanup = await db.select().from(copies).where(and(
@@ -374,6 +386,53 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
     return pending.length;
   }
 
+  type CommitEntry = { id: string; companyId: string; dueAt: number };
+  function commitEntry(row: Pick<Copy, "runId" | "companyId" | "state" | "attempts" | "nextAttemptAt"> | null): CommitEntry | null {
+    return row?.state === "pending_commit" && row.attempts < MAX_COLLECTION_ATTEMPTS
+      ? { id: row.runId, companyId: row.companyId, dueAt: row.nextAttemptAt?.getTime() ?? 0 } : null;
+  }
+  async function loadCommits(scope: InstructionCommitScope | undefined, admitted: () => boolean) {
+    const entries: CommitEntry[] = [];
+    let cursor: string | undefined;
+    while (admitted()) {
+      const rows = await db.select({ runId: copies.runId, companyId: copies.companyId, state: copies.state,
+        attempts: copies.attempts, nextAttemptAt: copies.nextAttemptAt }).from(copies).where(and(
+        eq(copies.state, "pending_commit"), lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1),
+        scope ? eq(copies.companyId, scope.companyId) : undefined,
+        scope?.runId ? eq(copies.runId, scope.runId) : undefined,
+        cursor ? gt(copies.runId, cursor) : undefined,
+      )).orderBy(asc(copies.runId)).limit(100);
+      for (const row of rows) entries.push(commitEntry(row)!);
+      if (rows.length < 100) return entries;
+      cursor = rows[rows.length - 1].runId;
+    }
+    throw new Error("Instruction commit reconstruction stopped");
+  }
+  async function recoverCommit(entry: CommitEntry, admitted: () => boolean) {
+    const row = await get(entry.companyId, entry.id);
+    const current = commitEntry(row);
+    if (!row || !current || current.dueAt > Date.now()) return current;
+    if (!admitted()) throw new Error("Instruction commit recovery stopped");
+    const claimed = await tryPatch(row, { attempts: row.attempts + 1 }, true);
+    if (claimed) {
+      const result = await commitCandidate(claimed);
+      if (isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result);
+    }
+    return commitEntry(await get(entry.companyId, entry.id));
+  }
+  function registerCapturedCommitWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+    canRun: () => boolean, canRecover: () => Promise<boolean>) {
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.instructionCommit,
+      batchSize: 20,
+      decodeScope: (hint: string): InstructionCommitScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: CommitEntry, scope?: InstructionCommitScope) => !scope ||
+        entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId),
+      load: async (scope, signal) => await canRecover() ? loadCommits(scope, () => canRun() && !signal?.aborted) : [],
+      check: async (entry, signal) => await canRecover() ? recoverCommit(entry, () => canRun() && !signal?.aborted) : null,
+    }, canRun);
+  }
+
   async function list(companyId: string, agentId: string, actor: AuthorizationActor) {
     await revisions.readCurrent({ companyId, agentId }, actor);
     const rows = await db.select().from(copies).where(and(eq(copies.companyId, companyId), eq(copies.agentId, agentId), inArray(copies.state, ["conflict", "pending_collection", "pending_commit", "unavailable"]))).orderBy(desc(copies.createdAt)).limit(50);
@@ -422,5 +481,5 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
     }
     return row;
   }
-  return { prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
+  return { registerCapturedCommitWork, prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
 }

server/src/services/company-deletion.ts

@@ -1,3 +1,4 @@
+import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
 import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
 import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-work-notifications.js";
 import { chatCompanyDeletion } from "./chat-channels/company-deletion.js";
@@ -96,6 +97,7 @@ export async function deleteCompany(db: Db, id: string) {
         await tx.delete(activityLog).where(eq(activityLog.companyId, id));
         await tx.delete(runIdentityContexts).where(eq(runIdentityContexts.companyId, id));
         await notifyScheduledRetryCancellation(tx, { companyId: id });
+        await notifyInstructionCommit(tx, { companyId: id });
         await tx.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, id));
         await tx.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, id));
         await tx.delete(agentApiKeys).where(eq(agentApiKeys.companyId, id));

server/src/services/delivery-work-notifications.ts

@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  instructionCommit: "instruction-commit",
   scheduledRetryCancellation: "scheduled-retry-cancellation",
   executionFinalization: "execution-finalization",
   executionReplacement: "execution-replacement",

server/src/services/heartbeat.ts

@@ -4062,6 +4062,14 @@ export function heartbeatService(
     triggerIssueMonitor,
 
     reportRunActivity: clearDetachedRunWarning,
+    registerCapturedInstructionCommitWork: (
+      coordinator: Parameters<typeof instructionCopies.registerCapturedCommitWork>[0], canRun: () => boolean,
+    ) => instructionCopies.registerCapturedCommitWork(coordinator,
+      () => canRun() && readTaskDrain(new Date()) === null,
+      async () => {
+        const state = await getSchedulingSuppression(true);
+        return !state.suppressed || state.reason === "task_drain";
+      }),
     registerScheduledRetryCancellationWork: (
       coordinator: Parameters<typeof registerScheduledRetryCancellationWork>[0], canRun: () => boolean,
     ) => registerScheduledRetryCancellationWork(coordinator,

server/src/services/heartbeat/recovery.ts

@@ -1915,7 +1915,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
 
   let settledProcessesDiscovered = false;
 
-  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean }) {
+  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean }) {
     await preserveCancelledNativeWorkspaces(db);
     const staleThresholdMs = opts?.staleThresholdMs ?? 0;
     const now = new Date();
@@ -1925,7 +1925,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     await instructionCopies.recoverStopped().catch(error => {
       logger.warn({ err: error }, "failed to recover stopped instruction copies");
     });
-    await instructionCopies.recoverCaptured().catch(error => {
+    await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit }).catch(error => {
       logger.warn({ err: error }, "failed to retry captured instruction revisions");
     });
 

server/src/services/instance-settings.ts

@@ -1,3 +1,4 @@
+import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
 import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
 import { notifyVoiceChange } from "./voice/voice-recovery-notifications.js";
 import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
@@ -617,6 +618,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
         if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
+        if (patch.enableWorktreeRunExecution !== undefined) await notifyInstructionCommit(tx);
         if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx);
         await notifyToolConnectionHealth(tx, "*");
         return tx

server/src/services/instruction-commit-notifications.ts

@@ -0,0 +1,7 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type InstructionCommitScope = { companyId: string; runId?: string };
+
+export async function notifyInstructionCommit(transaction: object, scope?: InstructionCommitScope) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.instructionCommit, scope ? JSON.stringify(scope) : "*");
+}