Merge current master execution recovery into idle ingress

server/src/services/heartbeat/deferred-receipt-work.ts

@@ -6,7 +6,7 @@ import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
 import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
 
-type Entry = { id: string; companyId: string; dueAt: number };
+type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean };
 type NativeIntentOptions = DeferredExecutionScope & {
   onDeferredDeadline(at: number): void;
   onError(error: unknown): void;
@@ -39,6 +39,11 @@ export function registerDeferredReceiptWork(
       },
       decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint),
       matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.requestId || entry.id === scope.requestId),
+      // A busy company can invalidate a paginated scan before the pending
+      // batch drains. Keep unread ready receipts ahead of freshly reloaded
+      // receipts already checked; actual wait/claim deadlines still replace.
+      mergeEntry: (pending, loaded) => pending.immediate && loaded.immediate
+        ? { ...loaded, dueAt: Math.min(pending.dueAt, loaded.dueAt) } : loaded,
       load: async scope => {
         const key = scope ? JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }) : "*";
         if (!enabled) { cursors.delete(key); return []; }
@@ -76,12 +81,15 @@ export function registerDeferredReceiptWork(
         )).orderBy(asc(agentWakeupRequests.id)).limit(100);
         const complete = rows.length < 100;
         if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
-        return { rows: rows.map(row => ({ id: row.id, companyId: row.companyId,
-          dueAt: kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000
+        const now = Date.now();
+        return { rows: rows.map(row => {
+          const deadline = kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000
             : kind === "native" && row.error ? row.updatedAt.getTime() + 30_000
             : kind === "deferred" && row.delayed && !row.interruptedLegacy ? row.updatedAt.getTime() + 30_000
-            : Date.now(),
-        })), complete };
+            : null;
+          const immediate = deadline === null || deadline <= now;
+          return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate };
+        }), complete };
       },
       check: async entry => {
         if (!enabled) return null;
@@ -101,7 +109,7 @@ export function registerDeferredReceiptWork(
               : kind === "deferred" && current.status === "deferred_issue_execution" && !current.interruptedLegacy
                 && (current.payload?.queuedCommentInterrupt || current.payload?.executionWait)
                 ? current.updatedAt.getTime() + 30_000 : null;
-          if (due !== null && due > Date.now()) return { ...entry, dueAt: due };
+          if (due !== null && due > Date.now()) return { ...entry, dueAt: due, immediate: false };
         }
         const errors: unknown[] = [];
         let next: number | null = null;
@@ -117,7 +125,7 @@ export function registerDeferredReceiptWork(
           catch (error) { failed(error); }
         }
         if (errors.length) throw new AggregateError(errors, `Deferred receipt ${entry.id} could not settle`);
-        return next === null ? null : { ...entry, dueAt: next };
+        return next === null ? null : { ...entry, dueAt: next, immediate: false };
       },
     }, canRun);
   }

server/src/services/native-runtime/agent-chat-response-recovery.test.ts

@@ -65,6 +65,26 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () =>
     const input = { companyId, issueId, runId };
     return { ...input, input, agentId, userId, wakeId, prpEvent, reply };
   }
+  it("settles the exact claimed wake when recovery projects a committed native run", async () => {
+    // The provider and result are already committed. Recovery owns terminal
+    // projection because the live executor will not return to completeRun.
+    const f = await fixture(false, { succeed: false });
+    const sourceWakeId = randomUUID();
+    await db.insert(agentWakeupRequests).values({ id: sourceWakeId, companyId: f.companyId,
+      agentId: f.agentId, source: "issue", reason: "issue_assigned", status: "claimed",
+      runId: f.runId, claimedAt: new Date(), payload: { issueId: f.issueId } });
+    await db.update(heartbeatRuns).set({ wakeupRequestId: sourceWakeId }).where(eq(heartbeatRuns.id, f.runId));
+
+    await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
+    const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, sourceWakeId));
+    expect(run).toMatchObject({ status: "succeeded", nativePhase: "committed" });
+    expect(run.finishedAt).not.toBeNull();
+    expect(wake).toMatchObject({ status: "completed", runId: f.runId });
+    expect(wake.finishedAt).not.toBeNull();
+  });
+
   const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId));
   const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0];
 

server/src/services/native-runtime/native-run-finalizer.ts

@@ -3,7 +3,7 @@ import { notifyExternalAgentSettlement } from "../../modules/external-agents/ind
 import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js";
 import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js";
 import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
-import { projectCommittedRun, recoveredExecutionFailureMetadata } from "./native-run-projection.js";
+import { projectCommittedRun, recoveredExecutionFailureMetadata, settleNativeTerminalWake } from "./native-run-projection.js";
 import { notifyNativeResultCommit } from "../native-result-commit-notifications.js";
 import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js";
 import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
@@ -465,6 +465,7 @@ async function recordRetryableFailure(input: {
       })
       .where(eq(heartbeatRuns.id, input.run.id))
       .returning();
+    if (projectsTerminalStatus && updatedRun) await settleNativeTerminalWake(tx as unknown as Db, updatedRun);
     if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null;
     if (supersededByNewerRun) {
       await issueRecoveryActionService(
@@ -1679,6 +1680,7 @@ export async function finalizeNativeRun(input: {
         await notifyExecutionRunWork(tx, { companyId: run.companyId });
         await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
         const [after] = await projectRun(tx);
+        if (after && input.projectRunStatus) await settleNativeTerminalWake(tx, after);
         return after ? { before, after } : null;
       });
       const updatedRun = projection?.after;

server/src/services/native-runtime/native-run-projection.ts

@@ -1,3 +1,5 @@
+import { notifyExternalConversationSettlement } from "../external-conversation-settlement-notifications.js";
+import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js";
 import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js";
@@ -5,7 +7,7 @@ import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-wor
 import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, exists, inArray, isNotNull, isNull, not, or, sql, type SQL } from "drizzle-orm";
-import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
+import { agentWakeupRequests, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
 import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js";
 import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
@@ -26,11 +28,41 @@ export function projectNativeTerminalRunStatus(
 
 
 
+function claimedNativeWakeNeedsSettlement() {
+  return sql`exists (select 1 from ${agentWakeupRequests} where
+    ${agentWakeupRequests.id} = ${heartbeatRuns.wakeupRequestId}
+    and ${agentWakeupRequests.companyId} = ${heartbeatRuns.companyId}
+    and ${agentWakeupRequests.agentId} = ${heartbeatRuns.agentId}
+    and ${agentWakeupRequests.runId} = ${heartbeatRuns.id}
+    and ${agentWakeupRequests.status} = 'claimed')`;
+}
+
+/** Settle only the exact persisted source claim inside terminal projection. */
+export async function settleNativeTerminalWake(tx: Db, run: Pick<typeof heartbeatRuns.$inferSelect,
+  "id" | "companyId" | "agentId" | "wakeupRequestId" | "runtimeMode" | "status" | "finishedAt" | "error">) {
+  if (run.runtimeMode !== "native" || !run.wakeupRequestId || !run.finishedAt
+    || !["succeeded", "failed", "cancelled"].includes(run.status)) return;
+  const claim = and(eq(agentWakeupRequests.id, run.wakeupRequestId),
+    eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
+    eq(agentWakeupRequests.runId, run.id), eq(agentWakeupRequests.status, "claimed"));
+  const [current] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests)
+    .where(claim).for("update").limit(1);
+  if (!current) return;
+  await notifyExternalConversationSettlement(tx, { companyId: run.companyId });
+  await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId });
+  await tx.update(agentWakeupRequests).set({
+    status: run.status === "succeeded" ? "completed" : run.status,
+    finishedAt: run.finishedAt, error: run.error, updatedAt: new Date(),
+  }).where(claim);
+}
+
 export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "failed" | "cancelled" | SQL) {
   return and(
     eq(heartbeatRuns.runtimeMode, "native"),
     or(
       inArray(heartbeatRuns.status, ["queued", "running", "failed"]),
+      and(eq(heartbeatRuns.status, projectedStatus),
+        inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled"]), claimedNativeWakeNeedsSettlement()),
       and(
         eq(heartbeatRuns.status, "succeeded"),
         or(
@@ -51,6 +83,7 @@ export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "f
 
     or(
       sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`,
+      claimedNativeWakeNeedsSettlement(),
       isNull(heartbeatRuns.finishedAt),
       sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
       sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
@@ -164,6 +197,7 @@ export async function projectCommittedRun(input: {
       })
       .where(needsRepair)
       .returning();
+    if (after) await settleNativeTerminalWake(tx, after);
     return after ? { before, after } : null;
   });
 

server/src/services/scoped-deadline-work.ts

@@ -18,6 +18,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     decodeScope(hint: string): Scope | undefined;
     matches(entry: Entry, scope?: Scope): boolean;
     load(scope?: Scope, signal?: AbortSignal): Promise<Entry[] | { rows: Entry[] | null; complete: boolean }>;
+    // Owners may retain pending queue priority while refreshing eligibility.
+    mergeEntry?(pending: Entry, loaded: Entry): Entry;
     check(entry: Entry, signal?: AbortSignal): Promise<Entry | null>;
   },
   canRun: () => boolean,
@@ -90,7 +92,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
           const loaded = await options.load(scope, signal);
           const result = Array.isArray(loaded) ? { rows: loaded, complete: true } : loaded;
           for (const entry of result.rows ?? []) {
-            entries.set(entry.id, entry);
+            const pending = entries.get(entry.id);
+            entries.set(entry.id, pending && options.mergeEntry ? options.mergeEntry(pending, entry) : entry);
             for (const [other, progress] of scans) if (options.matches(entry, options.decodeScope(other))) progress.seen.add(entry.id);
           }
           if (!result.complete) { dirty.add(hint); continue; }