fix: wake native response recovery from committed producers

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

server/src/services/native-runtime/native-interaction-bridge.ts

@@ -1,4 +1,4 @@
-import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js";
+import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import type { Db } from "@paperclipai/db";
 import type { NativeInteractionResponseEnvelope } from "../../vendor/paperclip-runner/index.js";
@@ -633,7 +633,7 @@ export async function routePersistedNativeResultAttention(input: {
         const delegation = effects.find((effect) => effect.kind === "create_delegated_issue");
         const finalizationError = effects.find((effect) => effect.kind === "record_finalization_error");
         await input.db.transaction(async tx => {
-          await notifyNativePendingEffects(tx, { companyId: binding.companyId, runId: input.runId });
+          await notifyNativeFinalizationWork(tx, { companyId: binding.companyId, runId: input.runId });
           await tx.update(nativeRunFinalizations).set({
             phase: finalizationError ? "retryable_failure" : "committed",
             assessmentId: existingAssessment.id,

server/src/services/native-runtime/native-pending-effect-work.ts

@@ -2,7 +2,7 @@ import { and, asc, eq, exists, gt, inArray, isNotNull, ne } from "drizzle-orm";
 import { heartbeatRuns, issues, nativeRunFinalizations, statusDecisionEffects, type Db } from "@paperclipai/db";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
-import type { NativePendingEffectScope } from "../native-pending-effect-notifications.js";
+import type { NativeFinalizationWorkScope } from "../native-finalization-notifications.js";
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
 import { dispatchPendingNativeStatusEffects, nativePendingEffectDueAt } from "./status-decision-committer.js";
 
@@ -15,7 +15,7 @@ export function pendingNativeStatusEffectsCondition(db: Db) {
   )));
 }
 
-function findPendingEffects(db: Db, scope?: NativePendingEffectScope, after?: string) {
+function findPendingEffects(db: Db, scope?: NativeFinalizationWorkScope, after?: string) {
   return db.select({
     id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId,
     issueId: nativeRunFinalizations.issueId, decisionId: nativeRunFinalizations.decisionId,
@@ -36,7 +36,7 @@ function findPendingEffects(db: Db, scope?: NativePendingEffectScope, after?: st
 
 export function registerNativePendingEffectWork(db: Db, coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
   type Entry = { id: string; companyId: string; dueAt: number };
-  async function load(scope?: NativePendingEffectScope, signal?: AbortSignal): Promise<Entry[]> {
+  async function load(scope?: NativeFinalizationWorkScope, signal?: AbortSignal): Promise<Entry[]> {
     const entries: Entry[] = [];
     let after: string | undefined;
     while (canRun() && !signal?.aborted) {
@@ -49,8 +49,8 @@ export function registerNativePendingEffectWork(db: Db, coordinator: ReturnType<
   }
   return registerScopedDeadlineWork(coordinator, {
     queue: DELIVERY_QUEUES.nativePendingEffects, batchSize: 25,
-    decodeScope: (hint: string): NativePendingEffectScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
-    matches: (entry: Entry, scope?: NativePendingEffectScope) => !scope || entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId),
+    decodeScope: (hint: string): NativeFinalizationWorkScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+    matches: (entry: Entry, scope?: NativeFinalizationWorkScope) => !scope || entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId),
     load,
     check: async (entry: Entry, signal?: AbortSignal) => {
       if (!canRun() || signal?.aborted) throw new Error("Native pending effect recovery stopped");

server/src/services/native-runtime/native-response-presentation-work.ts

@@ -2,17 +2,17 @@ import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from
 import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, type Db } from "@paperclipai/db";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
-import type { ChatRunMilestoneScope } from "../chat-run-milestone-notifications.js";
+import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js";
 import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wakeup.js";
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
 import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js";
 
-type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number };
 
 export function nativeResponsePresentationWork(db: Db) {
-  function candidates(scope?: ChatRunMilestoneScope, after?: string, runId?: string) {
+  function candidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) {
     return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
-      issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
+      issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
     }).from(nativeRunFinalizations)
       .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
         eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
@@ -46,15 +46,17 @@ export function nativeResponsePresentationWork(db: Db) {
           inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
         ))),
         scope ? and(eq(heartbeatRuns.companyId, scope.companyId), scope.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined) : undefined,
+        scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
+        scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined,
         after ? gt(heartbeatRuns.id, after) : undefined, runId ? eq(heartbeatRuns.id, runId) : undefined,
       )).orderBy(asc(heartbeatRuns.id)).limit(100);
   }
-  async function load(scope?: ChatRunMilestoneScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
+  async function load(scope?: NativeResponsePresentationScope, signal?: AbortSignal, canRun = () => true): Promise<Entry[]> {
     const entries: Entry[] = [];
     let after: string | undefined;
     while (canRun() && !signal?.aborted) {
       const rows = await candidates(scope, after);
-      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
+      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
       if (rows.length < 100) return entries;
       after = rows[rows.length - 1].id;
     }
@@ -76,8 +78,8 @@ export function nativeResponsePresentationWork(db: Db) {
   function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
     const worker = registerScopedDeadlineWork(coordinator, {
       queue: DELIVERY_QUEUES.nativeResponsePresentation,
-      decodeScope: (hint: string): ChatRunMilestoneScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
-      matches: (entry: Entry, scope?: ChatRunMilestoneScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId),
+      decodeScope: (hint: string): NativeResponsePresentationScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: Entry, scope?: NativeResponsePresentationScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId) && (!scope.agentId || entry.agentId === scope.agentId),
       load: (scope, signal) => load(scope, signal, canRun),
       check: (entry, signal) => {
         if (!canRun()) throw new Error("Native response presentation recovery paused");

server/src/services/native-runtime/native-run-coordinator-store.ts

@@ -1,4 +1,5 @@
-import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js";
+import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js";
+import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { isSafeNativeChatProgressEventType } from "../safe-native-chat-progress.js";
 import { notifyChatNativeProgress } from "../chat-native-progress-notifications.js";
@@ -297,6 +298,9 @@ export class NativeRunCoordinatorStore {
       if (isSafeNativeChatProgressEventType(event.eventType)) {
         await notifyChatNativeProgress(tx, { companyId: this.#binding.companyId, issueId: this.#binding.issueId });
       }
+      if (isNativeResponseEvidence(event.eventType, canonicalPayload)) {
+        await notifyNativeResponsePresentation(tx, this.#binding);
+      }
       const cursor = run.nextEventSeq;
       const [inserted] = await tx
         .insert(heartbeatRunEvents)
@@ -523,7 +527,7 @@ export class NativeRunCoordinatorStore {
       ) {
         throw new Error("native_finalization_lease_conflict");
       }
-      if (row.decisionId) await notifyNativePendingEffects(tx, this.#binding);
+      if (row.decisionId) await notifyNativeFinalizationWork(tx, this.#binding);
       if (
         row.leaseOwner === input.leaseOwner &&
         row.leaseExpiresAt &&
@@ -566,7 +570,7 @@ export class NativeRunCoordinatorStore {
       throw new Error("native_finalization_retry_invalid");
     }
     const [updated] = await this.#db.transaction(async tx => {
-      await notifyNativePendingEffects(tx, this.#binding);
+      await notifyNativeFinalizationWork(tx, this.#binding);
       if (input.failureCode === "native_workspace_sync_out_unsafe_archive") {
         await notifyUnsafeWorkspaceExportRecovery(tx, this.#binding);
       }

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

@@ -1,4 +1,4 @@
-import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js";
+import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
 import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
@@ -214,7 +214,7 @@ async function claimCoordinator(input: {
       coordinator.leaseOwner !== leaseOwner
     )
       throw new Error("native_finalization_lease_busy");
-    if (coordinator.decisionId) await notifyNativePendingEffects(tx, coordinator);
+    if (coordinator.decisionId) await notifyNativeFinalizationWork(tx, coordinator);
     const [updated] = await tx
       .update(nativeRunFinalizations)
       .set({
@@ -382,7 +382,7 @@ async function recordRetryableFailure(input: {
             ? "native_workspace_sync_out_retry_exhausted"
             : "native_finalization_retry_exhausted"
           : input.failureCode;
-    await notifyNativePendingEffects(tx, { companyId: input.run.companyId, runId: input.run.id });
+    await notifyNativeFinalizationWork(tx, { companyId: input.run.companyId, runId: input.run.id });
     if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
     if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") {
       await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });

server/src/services/native-runtime/native-session-executor.ts

@@ -1,4 +1,4 @@
-import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js";
+import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
 import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
@@ -3814,7 +3814,7 @@ export async function reconcileRetainedNativeSessionCleanup(
         renameSync(owned.root, archive);
       }
       renameSync(copy, owned.root);
-      await notifyNativePendingEffects(tx, { companyId: owned.run.companyId, runId: owned.run.id });
+      await notifyNativeFinalizationWork(tx, { companyId: owned.run.companyId, runId: owned.run.id });
       await tx
         .update(nativeRunFinalizations)
         .set({
@@ -3872,7 +3872,7 @@ export async function reconcileRetainedNativeSessionCleanup(
         .limit(1)
         .then((rows) => rows[0]);
       if (!current || current.leaseOwner !== leaseOwner) return;
-      await notifyNativePendingEffects(tx, { companyId: owned.run.companyId, runId: owned.run.id });
+      await notifyNativeFinalizationWork(tx, { companyId: owned.run.companyId, runId: owned.run.id });
       await tx
         .update(nativeRunFinalizations)
         .set({
@@ -9477,7 +9477,7 @@ async function executePaperclipNativeSessionWithinScope(
           )
           .for("update");
         await notifyExecutionRunWork(tx);
-        await notifyNativePendingEffects(tx, input.execution.binding);
+        await notifyNativeFinalizationWork(tx, input.execution.binding);
         const updated = await tx
           .update(nativeRunFinalizations)
           .set({
@@ -9813,7 +9813,7 @@ async function executePaperclipNativeSessionWithinScope(
   }
   const releaseNow = new Date();
   const released = await timeController("coordinator_release", () => input.db.transaction(async tx => {
-    await notifyNativePendingEffects(tx, input.execution.binding);
+    await notifyNativeFinalizationWork(tx, input.execution.binding);
     return tx.update(nativeRunFinalizations)
       .set({
         leaseOwner: null,

server/src/services/native-runtime/native-workspace-export-recovery.ts

@@ -1,4 +1,4 @@
-import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js";
+import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
 import { alias } from "drizzle-orm/pg-core";
 import type { UnsafeWorkspaceExportScope } from "../unsafe-workspace-export-notifications.js";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
@@ -96,7 +96,7 @@ async function recoverCandidate(db: Db, candidate: { runId: string; companyId: s
           phase: "workspace_finalize", status: "succeeded", exitCode: 0, finishedAt: now,
           metadata: { owningService: "native_workspace_finalizer", workspaceSync: { omitted: true, reason: "restore_unsafe_archive", legacy: true } },
         });
-        await notifyNativePendingEffects(tx, { companyId: run.companyId, runId: run.id });
+        await notifyNativeFinalizationWork(tx, { companyId: run.companyId, runId: run.id });
         await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null,
           failureDetail: { workspaceFinalizeAttempt: 0, legacyUnsafeExportOmitted: { resultId: result.id, recoveredAt: now.toISOString() } }, updatedAt: now,
         }).where(eq(nativeRunFinalizations.runId, run.id));

server/src/services/native-runtime/native-workspace-export-retry.ts

@@ -1,4 +1,4 @@
-import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js";
+import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
 import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js";
 import { randomUUID } from "node:crypto";
 import type { Environment, EnvironmentLease } from "@paperclipai/shared";
@@ -135,7 +135,7 @@ export async function retryNativeWorkspaceExport(input: {
         const now = new Date();
         const retry = { ...receipt, requestId: randomUUID(), actorId: input.actorId, repairNote: input.repairNote, requestedAt: now.toISOString(),
           stoppedProvider: initial.lease.metadata?.remoteExecutionTermination };
-        await notifyNativePendingEffects(tx, { companyId: current.run.companyId, runId: current.run.id });
+        await notifyNativeFinalizationWork(tx, { companyId: current.run.companyId, runId: current.run.id });
         await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null,
           failureDetail: { ...current.coordinator.failureDetail, workspaceFinalizeAttempt: 0, workspaceExportRetry: retry }, updatedAt: now }).where(eq(nativeRunFinalizations.runId, current.run.id));
         await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null,

server/src/services/native-runtime/paperclip-control-plane-port.test.ts

@@ -1,3 +1,5 @@
+import { nativeResponsePresentationWork } from "./native-response-presentation-work.js";
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { asc, eq, sql } from "drizzle-orm";
 import { generateKeyPairSync, randomUUID } from "node:crypto";
@@ -61,6 +63,8 @@ import {
   subscribeCompanyLiveEvents,
 } from "../live-events.js";
 
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_CONTROL_PLANE_TEST_DATABASE_URL");
+
 describe("PaperclipControlPlanePort conformance", () => {
   let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
   let db: ReturnType<typeof createDb>;
@@ -83,8 +87,11 @@ describe("PaperclipControlPlanePort conformance", () => {
   const governanceRunnerId = "00000000-0000-4000-8000-000000000021";
 
   beforeAll(async () => {
-    temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-port-");
-    db = createDb(temporary.connectionString);
+    if (external) db = createDb(external);
+    else {
+      temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-port-");
+      db = createDb(temporary.connectionString);
+    }
     const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity;
     await db.insert(companies).values({ id: identity.companyId, name: "Phase 6", issuePrefix: "P6C" });
     await db.insert(agents).values({
@@ -609,7 +616,7 @@ describe("PaperclipControlPlanePort conformance", () => {
       completionContractSha256: localContractSha,
       contextSnapshot: { issueId },
     });
-    const observerDb = createDb(temporary!.connectionString);
+    const observerDb = createDb(external ?? temporary!.connectionString);
     const observed: Array<Record<string, unknown>> = [];
     const committedCallbacks: string[] = [];
     let visibilityCheck: Promise<Array<{ eventType: string }>> | null = null;
@@ -1054,7 +1061,10 @@ describe("PaperclipControlPlanePort conformance", () => {
       .where(eq(heartbeatRuns.id, taskRunId));
     expect(await repair()).toBe(false);
     await db.update(heartbeatRuns).set({ contextSnapshot: {} }).where(eq(heartbeatRuns.id, taskRunId));
-    expect(await repair()).toBe(true);
+    const work = nativeResponsePresentationWork(db);
+    const [entry] = await work.load({ companyId: identity.companyId, runId: taskRunId });
+    expect(entry?.id).toBe(taskRunId);
+    expect(await work.check(entry)).toBeNull();
     expect(await repair()).toBe(false);
     expect(await db.select({ body: issueComments.body }).from(issueComments)
       .where(eq(issueComments.createdByRunId, taskRunId)).orderBy(asc(issueComments.createdAt)))

server/src/services/native-runtime/status-decision-committer.ts

@@ -1,4 +1,4 @@
-import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js";
+import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js";
 import { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js";
@@ -151,7 +151,7 @@ export async function dispatchPendingNativeStatusEffects(input: {
         targetId: effect.targetId,
       });
     }
-    await notifyNativePendingEffects(tx, input);
+    await notifyNativeFinalizationWork(tx, input);
     await notifyChatRunMilestones(tx, input);
     await tx
       .update(statusDecisionEffects)
@@ -1238,7 +1238,7 @@ async function materializeDecisionEffect(input: {
       )
       .for("update");
     if (pending.length === 0) throw new Error("native_pending_effect_missing");
-    await notifyNativePendingEffects(input.tx, { companyId: input.companyId });
+    await notifyNativeFinalizationWork(input.tx, { companyId: input.companyId });
     await input.tx
       .update(statusDecisionEffects)
       .set({
@@ -1888,7 +1888,7 @@ export async function commitNativeStatusDecision(input: {
         effect => effect.kind === "record_finalization_error",
       );
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.companyId);
-      await notifyNativePendingEffects(tx, input);
+      await notifyNativeFinalizationWork(tx, input);
       await tx
         .update(nativeRunFinalizations)
         .set({