fix(server): schedule stopped instruction collection (#17674)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Runs can stop while their private instruction edits still need collection. > - Existing copy rows and stop evidence describe that unfinished work. > - The orphan reaper still queries for it on every maintenance pass. > - This PR activates collection from committed copy and run transitions. > - It uses saved retry deadlines when stop proof or collection remains pending. > - Active copies and completed history no longer cause collection polling. ## Linked Issues or Issue Description Refs #17671 and #17670. Both prerequisites are merged into master. Captured saves and post-save release remain separate reasons with their own workers. **What needs to be improved?** `recoverStopped` queries instruction copies on every orphan-reaper pass, even when no terminal run owes collection. A run can become terminal without changing its copy row, so copy-only notifications would miss real work. **Why is this improvement important?** Repeated empty queries prevent idle tenant instances from becoming quiet. Removing them must preserve edits, stop proof, current ownership, and recovery after restart. **Proposed solution** Use the existing transaction signals, scoped deadline worker, and scheduler. Reconstruct eligible copies at startup, signal supported copy and terminal-run writers, and retain saved retry deadlines for unresolved work. Preserve the existing collection and directory owners. Related PR search found the warm agent-file ownership work (#14695), verified stop boundary (#14311), and task-workspace cleanup (#16464). This PR does not replace those contracts. ROADMAP.md has no duplicate migration. ## What Changed - Add scoped stopped-instruction collection with keyset reconstruction and fair batches of 20. - Signal before copy preparation, mutation and adoption writes, and before terminal run writes in their existing transactions. - Cover heartbeat, native finalization/restart/session recovery, execution-control, queue and dispatch producers. - Signal settings changes and company/agent deletion. Preserve drain, standby, restart and shutdown ownership. - Recheck terminal-run eligibility in the due-time claim. Keep terminal status separate from process-stop proof. - Retain existing stop checks, attempt limits, saved candidate bytes and remote-resource fencing. - Skip only this lane in periodic orphan recovery. Preserve manual/startup recovery. - Add concurrent, replay-safe partial index 0408 and retain the five newest snapshots. Add no table. - Record the owner and producer audit in `doc/plans/2026-10-25-stopped-instruction-collection-sleep-audit.md`. ## Verification - Source `cb2818a8f46ef961f744977a258f0e9042e31a42`: all 1469 cases across 14 complete files pass without skips in 141.69 seconds. This includes all 968 native-session cases, filesystem saves, startup, terminal-run producers, dispatch, queue, execution-control, finalization and lifecycle. - Eleven separate fresh PostgreSQL 17 databases migrated through 0408. The embedded-only lifecycle fixture temporarily used its own external database on this IPC-exhausted host and was restored byte for byte. Assertions and timing limits stayed unchanged. This qualifies external-database behavior; it does not claim an unmodified embedded CI run. - Database build, server typecheck/build, module boundaries and exact-current migration order pass. The original concurrent index SQL is unchanged. All prior executable migration history, journal entries and retained snapshot bytes are preserved; the five newest snapshots are 0404 through 0408. - A populated 0407-to0408 upgrade preserves all 406 prior migration receipts and 38 companies. It appends exactly one receipt. Replay adds none. The collection index is valid and ready on fresh and upgraded databases. - The normal merge of actual parent `267842fc3c6e6542deed8a1bcc2c78bec8df7862` preserves both complete patches, all independent sidebar changes from #18117, and the full accepted server/package/test trees. The complete sidebar test file passes all 36 cases on the combined tree. React act warnings remain. - Prior failures and fixture qualifications remain in the audit. Published head `d176d731f083c7edd2ede688cb901d1e66d8f1d8` has Greptile 5/5, all 57 checks passing or skipped, and no unresolved review threads. No Ponytail review is present. These local tests do not establish deployed host/database suspension or Cloud alarm delivery. ## Risks - Process-local signals cover instrumented writers. Independent SQL writers and additional replicas still need integration. - Terminal-run signals can cause one scoped inspection even if no instruction copy exists. They do not create a recurring scan. - Reconstruction and memory scale with actual pending candidates; active candidates are filtered by the run join. - Admitted file/provider operations can outlive a later drain request. Shutdown joins them. Uncertain ownership remains pending rather than being discarded. - The partial index adds storage and write cost. Its concurrent migration uses the existing interrupted-index repair path. - Cloud alarms, sleep eligibility and provider suspension are unchanged. Worker quiet is not a claim of deployed host/database sleep or measured savings. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, shell tools, and PostgreSQL/filesystem tests. The runtime does not expose a more precise model ID or context-window size for this turn. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-11-11 05:11:49 +01:00 · 2026-10-28 22:13:57 -05:00
1 parent 267842fc3c
commit 8252974141
@@ -433,25 +433,15 @@ describe("persistent agent directories", () => {
      const first = await run();
      expect(await copies.hasChanges({ companyId, runId: first.runId })).toBe(false);
      const secondId = await nextRun();
-      const update = db.update.bind(db);
+      const transaction = db.transaction.bind(db);
      let injected = false;
-      const spy = vi.spyOn(db, "update").mockImplementation(table => {
-        const builder = update(table);
-        if (table !== agentInstructionWorkingCopies) return builder;
-        const set = builder.set.bind(builder);
-        builder.set = (values => {
-          const query = set(values);
-          if (!("processStoppedAt" in values) || !values.processStoppedAt || injected) return query;
-          const returning = query.returning.bind(query);
-          query.returning = (async () => {
-            injected = true;
-            expect(await copies.adopt({ companyId, agentId, runId: secondId, previousRunId: first.runId, cwd: home })).not.toBeNull();
-            return returning();
-          }) as typeof query.returning;
-          return query;
-        }) as typeof builder.set;
-        return builder;
-      });
+      const spy = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => {
+        if (!injected) {
+          injected = true;
+          expect(await copies.adopt({ companyId, agentId, runId: secondId, previousRunId: first.runId, cwd: home })).not.toBeNull();
+        }
+        return transaction(callback, options);
+      }) as typeof db.transaction);
      try {
        expect(await copies.collectStopped({ companyId, runId: first.runId })).toMatchObject({
          receipt: { retainedByRunId: secondId }, processStoppedAt: null, attempts: 0,
@@ -1,3 +1,4 @@
+import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
 import { heartbeatService } from "../services/heartbeat.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
@@ -79,6 +80,216 @@ describe("registered run instruction copies", () => {
    await fs.writeFile(path.join(root, entryFile), initial);
  });

+  describe("stopped instruction collection worker", () => {
+    afterEach(async () => {
+      vi.useRealTimers(); vi.restoreAllMocks();
+      await db.update(agentInstructionWorkingCopies).set({ state: "saved", nextAttemptAt: null })
+        .where(eq(agentInstructionWorkingCopies.companyId, companyId));
+    });
+    const clock = () => vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const coordinator = (canRun = () => true, canReconcile = () => true) =>
+      createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile, onError: () => {} });
+    async function prepared(stopped = true) {
+      const copy = await run();
+      await fs.writeFile(path.join(copy.localRoot, entryFile), "collected after stop");
+      await db.update(heartbeatRuns).set({ status: "running", runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, copy.runId));
+      if (stopped) await db.update(agentInstructionWorkingCopies).set({ processStoppedAt: new Date() })
+        .where(eq(agentInstructionWorkingCopies.runId, copy.runId));
+      return copy;
+    }
+    async function terminal(runId: string, owner = db) {
+      const [run] = await owner.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
+      expect(await terminalizeLegacyExecution({ db: owner, run, status: "failed", patch: { finishedAt: new Date() } }))
+        .toMatchObject({ id: runId, status: "failed" });
+    }
+    async function saved(runId: string) {
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(100);
+        expect((await copies.get(companyId, runId))?.state).toBe("saved");
+      }, { timeout: 5000 });
+      expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("collected after stop");
+    }
+    it("leaves active copies and completed or exhausted history quiet", async () => {
+      const active = await prepared(false), finished = await prepared(), exhausted = await prepared();
+      await terminal(finished.runId); await terminal(exhausted.runId);
+      await db.update(agentInstructionWorkingCopies).set({ state: "saved" }).where(eq(agentInstructionWorkingCopies.runId, finished.runId));
+      await db.update(agentInstructionWorkingCopies).set({ attempts: 3 }).where(eq(agentInstructionWorkingCopies.runId, exhausted.runId));
+      clock(); const worker = coordinator();
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull();
+        expect((await copies.get(companyId, active.runId))?.state).toBe("prepared");
+        expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+      } finally { await worker.stop(); }
+    });
+    it("reactivates when the real terminalization producer changes only the run", async () => {
+      const copy = await prepared(); clock(); const worker = coordinator();
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        expect(worker.nextWakeAt()).toBeNull();
+        await terminal(copy.runId); await saved(copy.runId);
+        await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0);
+        const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull();
+      } finally { await worker.stop(); }
+    });
+    it.each(["restart", "drain", "standby"] as const)("retains terminal collection work across %s", async mode => {
+      const copy = await prepared(); if (mode === "restart") await terminal(copy.runId);
+      let admitted = mode === "restart";
+      clock(); const worker = coordinator(() => admitted, () => mode !== "standby");
+      try {
+        await agentInstructionWorkingCopyService(db).registerStoppedCollectionWork(worker, () => admitted, async () => true).ready;
+        if (mode !== "restart") {
+          await terminal(copy.runId); await vi.advanceTimersByTimeAsync(1000);
+          if (mode === "drain") await expect.poll(() => idleWorkSnapshot().active).toBe(0);
+          const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000);
+          expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+          expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+          admitted = true; await vi.advanceTimersByTimeAsync(30_000);
+        }
+        await saved(copy.runId);
+      } finally { await worker.stop(); }
+    });
+    it("keeps terminalization inside the outer transaction until commit", async () => {
+      const copy = await prepared(); clock(); const worker = coordinator();
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        await expect(db.transaction(async tx => {
+          await terminal(copy.runId, tx as unknown as typeof db);
+          await vi.advanceTimersByTimeAsync(1000);
+          expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+          throw new Error("rollback terminal run");
+        })).rejects.toThrow("rollback terminal run");
+        await vi.advanceTimersByTimeAsync(1000);
+        expect((await copies.get(companyId, copy.runId))?.state).toBe("prepared");
+        await db.transaction(async tx => {
+          await terminal(copy.runId, tx as unknown as typeof db);
+          await vi.advanceTimersByTimeAsync(1000);
+          expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+        });
+        await saved(copy.runId);
+      } finally { await worker.stop(); }
+    });
+    it("keeps the saved future deadline instead of collecting on notification", async () => {
+      const copy = await prepared(); clock(); const due = new Date(Date.now() + 90_000);
+      await db.update(agentInstructionWorkingCopies).set({ nextAttemptAt: due }).where(eq(agentInstructionWorkingCopies.runId, copy.runId));
+      const worker = coordinator();
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        await terminal(copy.runId); await vi.advanceTimersByTimeAsync(1000);
+        await expect.poll(() => idleWorkSnapshot().active).toBe(0);
+        expect(worker.nextWakeAt()).toBe(due.getTime());
+        const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000);
+        expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+        expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+        await vi.advanceTimersByTimeAsync(30_000); await saved(copy.runId);
+      } finally { await worker.stop(); }
+    });
+    it("never substitutes terminal status for stop proof and recovers later recorded evidence", async () => {
+      const copy = await prepared(false); await terminal(copy.runId);
+      await db.update(heartbeatRuns).set({ runtimeMode: "native" }).where(eq(heartbeatRuns.id, copy.runId));
+      clock(); const worker = coordinator();
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        expect(await copies.get(companyId, copy.runId)).toMatchObject({ state: "pending_collection", processStoppedAt: null });
+        expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+        await appendHeartbeatRunEvent(db, { ...target(), runId: copy.runId, eventType: "native.local_process_stopped", stream: "system" });
+        await vi.advanceTimersByTimeAsync(30_000); await saved(copy.runId);
+      } finally { await worker.stop(); }
+    });
+    it("reconstructs more than one page without a recurring idle scan", async () => {
+      const copy = await run();
+      const ids = Array.from({ length: 103 }, () => randomUUID());
+      await db.insert(heartbeatRuns).values(ids.map(id => ({ id, companyId, agentId, status: "failed",
+        runtimeMode: "legacy", invocationSource: "on_demand", responsibleUserId: userId })));
+      await db.insert(agentInstructionWorkingCopies).values(ids.map(runId => ({ ...copy, runId,
+        processStoppedAt: new Date(), nextAttemptAt: null,
+      })));
+      clock(); const worker = coordinator();
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        const rows = () => db.select({ state: agentInstructionWorkingCopies.state }).from(agentInstructionWorkingCopies)
+          .where(eq(agentInstructionWorkingCopies.companyId, companyId));
+        expect((await rows()).filter(row => row.state === "unchanged").length).toBeLessThanOrEqual(20);
+        await vi.waitFor(async () => {
+          await vi.advanceTimersByTimeAsync(1000);
+          expect((await rows()).filter(row => row.state === "unchanged")).toHaveLength(103);
+        }, { timeout: 15_000 });
+        expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+        await vi.advanceTimersByTimeAsync(1000);
+        await expect.poll(() => idleWorkSnapshot().active).toBe(0);
+        const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
+        expect(reads).not.toHaveBeenCalled(); expect(worker.nextWakeAt()).toBeNull();
+      } finally { await worker.stop(); }
+    }, 20_000);
+    it("retries a failed collection claim without losing the preserved bytes", async () => {
+      const copy = await prepared(); await terminal(copy.runId); clock(); const worker = coordinator();
+      const transactions = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("collection claim unavailable"));
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        expect((await copies.get(companyId, copy.runId))?.state).toBe("prepared");
+        expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+        transactions.mockRestore();
+        const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(20_000);
+        expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+        await vi.advanceTimersByTimeAsync(10_000); await saved(copy.runId);
+        await vi.advanceTimersByTimeAsync(1000);
+        await expect.poll(() => idleWorkSnapshot().active).toBe(0);
+        expect(worker.nextWakeAt()).toBeNull();
+      } finally { transactions.mockRestore(); await worker.stop(); }
+    });
+    it("joins an in-flight collection before shutdown and restarts quiet", async () => {
+      const copy = await prepared(); await terminal(copy.runId); clock(); const worker = coordinator();
+      const rename = fs.rename.bind(fs);
+      let release!: () => void, entered!: () => void;
+      const held = new Promise<void>(resolve => { release = resolve; });
+      const reached = new Promise<void>(resolve => { entered = resolve; });
+      const renameSpy = vi.spyOn(fs, "rename").mockImplementation(async (from, to) => {
+        if (to === path.join(root, entryFile)) { entered(); await held; }
+        return rename(from, to);
+      });
+      let stopped: Promise<void> | undefined;
+      try {
+        const registration = copies.registerStoppedCollectionWork(worker, () => true, async () => true);
+        await reached;
+        let complete = false;
+        stopped = worker.stop().then(() => { complete = true; });
+        await new Promise<void>(resolve => setImmediate(resolve));
+        expect(complete).toBe(false);
+        release(); await stopped; await registration.ready;
+        expect((await copies.get(companyId, copy.runId))?.state).toBe("saved");
+        expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe("collected after stop");
+        const restarted = coordinator();
+        try {
+          await copies.registerStoppedCollectionWork(restarted, () => true, async () => true).ready;
+          expect(restarted.nextWakeAt()).toBeNull();
+        } finally { await restarted.stop(); }
+      } finally { release(); renameSpy.mockRestore(); await stopped; await worker.stop(); }
+    });
+    it("checks current run eligibility when a queued callback meets a resumed run", async () => {
+      const copy = await prepared(); await terminal(copy.runId); clock(); const worker = coordinator();
+      const transaction = db.transaction.bind(db); let resumed = false;
+      const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => {
+        if (!resumed) {
+          resumed = true;
+          await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, copy.runId));
+        }
+        return transaction(callback, options);
+      }) as typeof db.transaction);
+      try {
+        await copies.registerStoppedCollectionWork(worker, () => true, async () => true).ready;
+        transactions.mockRestore();
+        expect(resumed).toBe(true);
+        expect((await copies.get(companyId, copy.runId))?.state).toBe("prepared");
+        expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
+        await vi.advanceTimersByTimeAsync(1000); await expect.poll(() => idleWorkSnapshot().active).toBe(0);
+        expect(worker.nextWakeAt()).toBeNull();
+        await terminal(copy.runId); await saved(copy.runId);
+      } finally { transactions.mockRestore(); await worker.stop(); }
+    });
+  });
+
  describe("captured instruction commit worker", () => {
    afterEach(async () => {
      vi.useRealTimers(); vi.restoreAllMocks();
@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { eq } from "drizzle-orm";
@@ -28,6 +29,8 @@ import {
 import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js";
 import { assertRetainedNativeSourceArchiveSettled } from "../services/native-runtime/native-session-executor.js";

+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_FINALIZATION_TEST_DATABASE_URL");
+
 describe("P6-16/P6-25/P6-28 native finalization recovery", () => {
  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
  let db: ReturnType<typeof createDb>;
@@ -41,8 +44,8 @@ describe("P6-16/P6-25/P6-28 native finalization recovery", () => {
  const staleRunId = "72000000-0000-4000-8000-000000000015";

  beforeAll(async () => {
-    temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-recovery-");
-    db = createDb(temporary.connectionString);
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-recovery-");
+    db = createDb(external ?? temporary.connectionString);
    await db.insert(companies).values({ id: companyId, name: "Native recovery", issuePrefix: "NRC" });
    await db.insert(agents).values({
      id: agentId,
@@ -210,7 +213,8 @@ describe("P6-16/P6-25/P6-28 native finalization recovery", () => {
  }, 30_000);

  afterAll(async () => {
-    await temporary.cleanup();
+    await db?.$client.end({ timeout: 1 });
+    await temporary?.cleanup();
  });

  it("fails closed into bounded named recovery without consulting the live flag or falling back", async () => {
@@ -355,10 +359,10 @@ describe("retained native cleanup discovery", () => {
  ];

  beforeAll(async () => {
-    temporary = await startEmbeddedPostgresTestDatabase(
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase(
      "paperclip-native-cleanup-sweep-",
    );
-    db = createDb(temporary.connectionString);
+    db = createDb(external ?? temporary.connectionString);
    await db
      .insert(companies)
      .values({ id: companyId, name: "Cleanup sweep", issuePrefix: "NCS" });
@@ -382,7 +386,8 @@ describe("retained native cleanup discovery", () => {
    });
  });
  afterAll(async () => {
-    await temporary.cleanup();
+    await db?.$client.end({ timeout: 1 });
+    await temporary?.cleanup();
  });

  async function candidate(
@@ -115,6 +115,7 @@ const {
    reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })),
    reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })),
    registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
+    registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
    registerInstructionDirectoryCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
    registerCapturedInstructionCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
    promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })),
@@ -552,6 +553,13 @@ describe("startServer feedback export wiring", () => {
    process.env.BETTER_AUTH_SECRET = "test-secret";
  });

+  it.each([false, true])("registers stopped instruction collection only when maintenance is enabled (%s)", async enabled => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+    await startServer(); await settleStartupHousekeeping();
+    expect(heartbeatServiceMock.registerStoppedInstructionCollectionWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
+    if (enabled) expect(heartbeatServiceMock.registerStoppedInstructionCollectionWork.mock.calls[0][1]()).toBe(true);
+  });
+
  it.each([false, true])("registers directory cleanup recovery only when maintenance is enabled (%s)", async enabled => {
    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
    await startServer(); await settleStartupHousekeeping();
@@ -957,7 +965,7 @@ describe("startServer feedback export wiring", () => {
      intervalCallback?.();
      await new Promise<void>((resolve) => setImmediate(resolve));
      expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
-      expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true });
+      expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true });
      expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
      expect(retiredDetector).not.toHaveBeenCalled();
    } finally {
@@ -1824,6 +1824,8 @@ async function startServerWithDatabaseTeardown(
      () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
    if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork,
      () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
+    if (config.heartbeatSchedulerEnabled) await heartbeat.registerStoppedInstructionCollectionWork(deliveryWork,
+      () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;

    const runRetentionSweep = async () => {
      const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active"));
@@ -1952,7 +1954,7 @@ async function startServerWithDatabaseTeardown(
          // Periodically reap orphaned runs (5-min staleness threshold) and make sure
          // persisted queued work is still being driven forward.
          runPeriodicRecovery(() => heartbeat
-            .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true })
+            .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true })
            .then(() => heartbeat.promoteDueScheduledRetries())
            .then(async (promotion) => {
              await heartbeat.resumeQueuedRuns();
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js";
 import { and, asc, desc, eq, gt, gte, inArray, ne, notInArray, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
@@ -337,6 +338,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
    };

    const transactionResult = await db.transaction(async (tx) => {
+      await notifyInstructionCollection(tx, { companyId, runId: input.run.id });
      const [updatedRun] = await tx
        .update(heartbeatRuns)
        .set({
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js";
 import { notifyScheduledRetryCancellation } from "../../../services/scheduled-retry-cancellation-notifications.js";
 import { hasConversationContinuationPolicy } from "../../../services/conversation-continuation.js";
 import { getExecutionBlocker } from "../../../services/execution-blocker.js";
@@ -667,6 +668,7 @@ export function createPostgresRunDispatchAdapter(
    | { applied: true; run: HeartbeatRun; postCommitEffects: PostCommitEffect[] }
    | { applied: false }
  > {
+    await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
    await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId });
    await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: issueId ?? undefined });
    const [row] = await tx
@@ -710,6 +712,7 @@ export function createPostgresRunDispatchAdapter(
    tx: Db,
    input: CancelSuppressedRetryInput,
  ): Promise<CancelSuppressedRetryResult> {
+    await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
    await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId });
    await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined });
    const [row] = await tx
@@ -917,6 +920,7 @@ export function createPostgresRunDispatchAdapter(
    now: Date,
    providerDispatchNotStarted = false,
  ): Promise<CancelStaleQueuedRunOutcome> {
+      await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
      await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
      await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId });
      const [row] = await tx
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js";
 import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js";
 import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js";
 import { and, eq, inArray } from "drizzle-orm";
@@ -103,6 +104,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu
    },

    async cancelQueueRun({ queueRunId, reason, now }) {
+      await notifyInstructionCollection(tx, { companyId, runId: queueRunId });
      await notifyChatRunMilestones(tx, { companyId, issueId });
      const row = await tx
        .update(heartbeatRuns)
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
 import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js";
 import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
 import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
@@ -48,6 +49,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str
  await notifyScheduledRetryCancellation(tx, { companyId: companyId });
  await notifyInstructionCommit(tx, { companyId: companyId });
  await notifyDirectoryCleanup(tx, { companyId: companyId });
+  await notifyInstructionCollection(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));
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
 import fs from "node:fs/promises";
 import path from "node:path";
 import { and, eq, inArray, isNull, sql } from "drizzle-orm";
@@ -40,6 +41,12 @@ function actor(row: Copy): AuthorizationActor {
 }
 export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string, runId: string) => Promise<Copy | null>, patch: (row: Copy, values: Partial<typeof copies.$inferInsert>) => Promise<Copy>, environmentRuntime?: EnvironmentRuntimeService) {
  const store = agentFileStore(db);
+  async function insertCopy(values: typeof copies.$inferInsert) {
+    return db.transaction(async tx => {
+      await notifyInstructionCollection(tx, { companyId: values.companyId, runId: values.runId });
+      return tx.insert(copies).values(values).returning();
+    });
+  }
  const ownerFile = (row: Copy) => path.join(path.dirname(row.localRoot), "owner.json");
  async function owns(row: Copy) {
    if (row.receipt?.warm !== true) return true;
@@ -104,8 +111,8 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
        const receipt = { ...current.receipt, cleanup: input.target?.kind === "remote" ? { leaseId: input.target.leaseId ?? null, remoteCwd: input.target.remoteCwd } : undefined };
        // Insert before taking the agent-row lock: this foreign key needs an
        // independent transaction, and must survive post-handoff failures.
-        const [successor] = await db.insert(copies).values({ runId: input.runId, companyId: input.companyId, agentId: input.agentId, responsibleUserId: bound.onBehalfOfUserId!,
-          entryFile: current.entryFile, baseHash: current.baseHash, localRoot: current.localRoot, executionRoot: current.executionRoot, location: current.location, state: "prepared", receipt }).returning();
+        const [successor] = await insertCopy({ runId: input.runId, companyId: input.companyId, agentId: input.agentId, responsibleUserId: bound.onBehalfOfUserId!,
+          entryFile: current.entryFile, baseHash: current.baseHash, localRoot: current.localRoot, executionRoot: current.executionRoot, location: current.location, state: "prepared", receipt });
        let handedOff = false;
        try { return await store.locked(input.companyId, input.agentId, bound, false, async (_tx, lockedAgent, canonical) => {
          // The reservation's earlier check is only a hint. Serialize this last
@@ -167,7 +174,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
      nextAttemptAt: null, errorCode: null, errorMessage: null };
    if (row) row = await patch(row, preparing);
    else {
-      await db.insert(copies).values({ runId: input.runId, companyId: input.companyId, agentId: input.agentId, responsibleUserId: bound.onBehalfOfUserId!, ...preparing });
+      await insertCopy({ runId: input.runId, companyId: input.companyId, agentId: input.agentId, responsibleUserId: bound.onBehalfOfUserId!, ...preparing });
      row = (await get(input.companyId, input.runId))!;
    }
    const { snapshot, storageWarning } = await store.locked(input.companyId, input.agentId, bound, false, async (_tx, _agent, canonical) => {
@@ -291,6 +298,8 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
      const current = await inspectAgentDirectory(canonical);
      const canonicalChanged = agentDirectoryBaselineDigest(current.snapshot) !== agentDirectoryBaselineDigest(baseline(prior));
      if (canonicalChanged && !input.allowRetirementHandoff) return null;
+      await notifyInstructionCollection(tx, { companyId: input.companyId, runId: prior.runId });
+      await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
      const [claimed] = await tx.update(copies).set({ receipt: { ...prior.receipt, retainedByRunId: input.runId }, updatedAt: new Date() })
        .where(and(eq(copies.companyId, input.companyId), eq(copies.runId, prior.runId), eq(copies.agentId, input.agentId), eq(copies.state, "unchanged_turn"),
          eq(copies.updatedAt, prior.updatedAt), eq(copies.baseHash, prior.baseHash), eq(copies.localRoot, prior.localRoot),
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection, type InstructionCollectionScope } from "./instruction-collection-notifications.js";
 import { notifyDirectoryCleanup, type DirectoryCleanupScope } from "./directory-cleanup-notifications.js";
 import { notifyInstructionCommit, type InstructionCommitScope } from "./instruction-commit-notifications.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
@@ -9,7 +10,7 @@ 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, gt, inArray, lte, or, isNull, sql } from "drizzle-orm";
+import { and, asc, desc, eq, gt, inArray, lte, or, isNull, sql, type SQL } from "drizzle-orm";
 import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, databaseWorkScopeKey, type Db } from "@paperclipai/db";
 import {
  prepareAdapterExecutionTargetRuntime,
@@ -95,21 +96,23 @@ 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) {
+  async function tryPatch(row: Copy, values: Partial<typeof copies.$inferInsert>, requireDue = false, condition?: SQL) {
+    const signalsCollection = collectionCandidate(row) || collectionCandidate({ ...row, ...values });
    const signalsCleanup = directoryCleanupEntry(row) !== null || directoryCleanupEntry({ ...row, ...values }) !== null;
    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 });
      if (signalsCleanup) await notifyDirectoryCleanup(tx, { companyId: row.companyId, runId: row.runId });
+      if (signalsCollection) await notifyInstructionCollection(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,
+        requireDue ? or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())) : undefined, condition,
      )).returning();
      return updated;
    };
-    return signalsCommit || signalsCleanup ? db.transaction(tx => write(tx as unknown as Db)) : write(db);
+    return signalsCommit || signalsCleanup || signalsCollection ? db.transaction(tx => write(tx as unknown as Db)) : write(db);
  }
  async function patch(row: Copy, values: Partial<typeof copies.$inferInsert>) {
    const updated = await tryPatch(row, values);
@@ -199,12 +202,15 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
      ? path.posix.join(target.remoteCwd, ".paperclip-runtime", `instruction-edits-${input.runId}`, "instructions")
      : localRoot;
    if (!existing) {
-      await db.insert(copies).values({
+      await db.transaction(async tx => {
+        await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
+      await tx.insert(copies).values({
        runId: input.runId, companyId: input.companyId, agentId: input.agentId,
        responsibleUserId: bound.onBehalfOfUserId!, entryFile,
        baseRevisionId: baseline?.revision.id ?? null, baseHash: hash(content),
        localRoot, executionRoot, location, state: "preparing",
      });
+      });
    }
    const row = (await get(input.companyId, input.runId))!;
    await assertInstructionPathSafe(localRoot, entryFile);
@@ -363,53 +369,114 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
  /** Recover only the registered copy after independently recorded process-stop
   * proof. Terminal run status is never a substitute for that proof. Remote
   * receipts mean the environment is stopped; do not restart it just to read. */
-  async function recoverStopped() {
-    const pending = await db.select({ copy: copies, runtimeMode: heartbeatRuns.runtimeMode }).from(copies)
-      .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, copies.companyId), eq(heartbeatRuns.id, copies.runId)))
-      .where(and(or(and(or(inArray(copies.state, ["prepared", "pending_collection", "unchanged_turn", "warm_saved"]),
-          and(eq(copies.state, "preparing"), sql`${copies.receipt}->>'schema' = 'paperclip.agent-files.v1'`)),
-          lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1)),
-        and(eq(copies.state, "unavailable"), isNull(copies.processStoppedAt),
-          sql`${copies.location} like 'remote:%'`,
-          sql`${copies.receipt}->>'schema' = 'paperclip.agent-files.v1'`)),
-        sql`NOT (coalesce(${copies.receipt}, '{}'::jsonb) ? 'retainedByRunId')`,
-        inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]),
-        or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())))).orderBy(asc(copies.updatedAt)).limit(20);
-    for (const { copy: row, runtimeMode } of pending) {
-      if (row.state === "unavailable" && isAgentDirectoryCopy(row)) {
-        await recoverDirectoryCleanup(row, () => directories.recoverUnavailable(row));
-      } else if (row.state === "preparing" && isAgentDirectoryCopy(row)) {
-        // A provider cannot launch until preparation records "prepared". With
-        // the owning run terminal, this is an interrupted staging copy only.
-        await directories.release(await patch(row, { state: "unavailable", processStoppedAt: new Date(),
-          errorCode: "AGENT_FILES_PREPARE_FAILED", errorMessage: "Agent-file preparation was interrupted. The temporary copy was discarded.", nextAttemptAt: null }));
-      } else if (row.location === "local" && (row.processStoppedAt || runtimeMode === "native" &&
-          await hasNativeLocalProcessStop(db, row.companyId, row.runId))) {
-        const result = await collectStopped({ companyId: row.companyId, runId: row.runId });
-        if (result && isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result);
-      } else if (row.location !== "local" && await remoteExecutionHasStopped(db, row.companyId, row.runId)) {
-        if (row.receipt?.warm === true) {
-        const unavailable = row.receipt?.warm === true
-          ? await patch(row, { state: "unavailable", errorCode: "AGENT_FILES_FINAL_COLLECTION_UNAVAILABLE",
-            errorMessage: "The last completed checkpoint remains saved. Files changed afterwards could not be collected after remote termination.", nextAttemptAt: null })
-          : await reportUnavailable(row.companyId, row.runId);
-        if (unavailable && isAgentDirectoryCopy(unavailable)) {
-          await directories.release(await patch(unavailable, { processStoppedAt: unavailable.processStoppedAt ?? new Date() }));
-        }
-        } else {
-        if (isAgentDirectoryCopy(row)) await directories.recoverStoppedRemote(row);
-        else await reportUnavailable(row.companyId, row.runId);
-        }
-      } else if (row.state === "prepared") {
-        await patch(row, { state: "pending_collection", errorCode: "INSTRUCTION_STOP_UNCONFIRMED",
-          errorMessage: "The provider's stop has not been confirmed. Instruction collection is pending; no save is claimed.", nextAttemptAt: new Date(Date.now() + 30_000) });
-      } else if (["warm_saved", "pending_collection", "unchanged_turn"].includes(row.state)) {
-        // Live retained sessions must not occupy every batch and starve stopped
-        // copies from other agents. This does not permit reads without stop proof.
-        await patch(row, { nextAttemptAt: new Date(Date.now() + 30_000) });
+  function collectionCandidate(row: Copy) {
+    return !row.receipt?.retainedByRunId && (
+      row.attempts < MAX_COLLECTION_ATTEMPTS && (["prepared", "pending_collection", "unchanged_turn", "warm_saved"].includes(row.state)
+        || row.state === "preparing" && isAgentDirectoryCopy(row))
+      || row.state === "unavailable" && !row.processStoppedAt && row.location.startsWith("remote:") && isAgentDirectoryCopy(row));
+  }
+  function collectionPredicate() {
+    return and(sql`NOT (coalesce(${copies.receipt}, '{}'::jsonb) ? 'retainedByRunId') AND (
+      (${copies.attempts} < 3 AND (${copies.state} IN ('prepared', 'pending_collection', 'unchanged_turn', 'warm_saved')
+        OR (${copies.state} = 'preparing' AND ${copies.receipt}->>'schema' = 'paperclip.agent-files.v1')))
+      OR (${copies.state} = 'unavailable' AND ${copies.processStoppedAt} IS NULL
+        AND ${copies.location} LIKE 'remote:%' AND ${copies.receipt}->>'schema' = 'paperclip.agent-files.v1'))`,
+      inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]));
+  }
+  async function collectRecovered(row: Copy, runtimeMode: string | null) {
+    if (row.state === "unavailable" && isAgentDirectoryCopy(row)) {
+      await recoverDirectoryCleanup(row, () => directories.recoverUnavailable(row));
+    } else if (row.state === "preparing" && isAgentDirectoryCopy(row)) {
+
+
+      await directories.release(await patch(row, { state: "unavailable", processStoppedAt: new Date(),
+        errorCode: "AGENT_FILES_PREPARE_FAILED", errorMessage: "Agent-file preparation was interrupted. The temporary copy was discarded.", nextAttemptAt: null }));
+    } else if (row.location === "local" && (row.processStoppedAt || runtimeMode === "native" &&
+        await hasNativeLocalProcessStop(db, row.companyId, row.runId))) {
+      const result = await collectStopped({ companyId: row.companyId, runId: row.runId });
+      if (result && isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result);
+    } else if (row.location !== "local" && await remoteExecutionHasStopped(db, row.companyId, row.runId)) {
+      if (row.receipt?.warm === true) {
+      const unavailable = row.receipt?.warm === true
+        ? await patch(row, { state: "unavailable", errorCode: "AGENT_FILES_FINAL_COLLECTION_UNAVAILABLE",
+          errorMessage: "The last completed checkpoint remains saved. Files changed afterwards could not be collected after remote termination.", nextAttemptAt: null })
+        : await reportUnavailable(row.companyId, row.runId);
+      if (unavailable && isAgentDirectoryCopy(unavailable)) {
+        await directories.release(await patch(unavailable, { processStoppedAt: unavailable.processStoppedAt ?? new Date() }));
      }
+      } else {
+      if (isAgentDirectoryCopy(row)) await directories.recoverStoppedRemote(row);
+      else await reportUnavailable(row.companyId, row.runId);
+      }
+    } else if (row.state === "prepared") {
+      await patch(row, { state: "pending_collection", errorCode: "INSTRUCTION_STOP_UNCONFIRMED",
+        errorMessage: "The provider's stop has not been confirmed. Instruction collection is pending; no save is claimed.", nextAttemptAt: new Date(Date.now() + 30_000) });
+    } else if (["warm_saved", "pending_collection", "unchanged_turn"].includes(row.state)) {
+
+
+      await patch(row, { nextAttemptAt: new Date(Date.now() + 30_000) });
    }
-    return pending.length;
+  }
+  type CollectionEntry = { id: string; companyId: string; dueAt: number };
+  async function loadCollection(scope: InstructionCollectionScope | undefined, admitted: () => boolean) {
+    const entries: CollectionEntry[] = [];
+    let cursor: string | undefined;
+    while (admitted()) {
+      const rows = await db.select({ id: copies.runId, companyId: copies.companyId, nextAttemptAt: copies.nextAttemptAt })
+        .from(copies).innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, copies.companyId), eq(heartbeatRuns.id, copies.runId)))
+        .where(and(collectionPredicate(), 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({ id: row.id, companyId: row.companyId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
+      if (rows.length < 100) return entries;
+      cursor = rows[rows.length - 1].id;
+    }
+    throw new Error("Instruction collection reconstruction stopped");
+  }
+  async function readCollection(entry: CollectionEntry) {
+    const [current] = await db.select({ copy: copies, runtimeMode: heartbeatRuns.runtimeMode }).from(copies)
+      .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, copies.companyId), eq(heartbeatRuns.id, copies.runId)))
+      .where(and(collectionPredicate(), scope(entry.companyId, entry.id)));
+    return current;
+  }
+  async function recoverCollection(entry: CollectionEntry, admitted: () => boolean) {
+    const current = await readCollection(entry);
+    if (!current) return null;
+    const dueAt = current.copy.nextAttemptAt?.getTime() ?? 0;
+    if (dueAt > Date.now()) return { ...entry, dueAt };
+    if (!admitted()) throw new Error("Instruction collection recovery stopped");
+    const claimed = await tryPatch(current.copy, { nextAttemptAt: new Date(Date.now() + 30_000) }, true,
+      sql`EXISTS (SELECT 1 FROM ${heartbeatRuns} WHERE ${heartbeatRuns.id} = ${copies.runId}
+        AND ${heartbeatRuns.companyId} = ${copies.companyId}
+        AND ${heartbeatRuns.status} IN ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted'))`);
+    if (claimed) await collectRecovered(claimed, current.runtimeMode);
+    const remaining = await readCollection(entry);
+    if (!remaining) return null;
+    const next = { ...entry, dueAt: remaining.copy.nextAttemptAt?.getTime() ?? 0 };
+    if (next.dueAt <= Date.now()) throw new Error("Instruction collection remains unresolved without a future retry");
+    return next;
+  }
+  async function recoverStopped() {
+    const rows = await db.select({ id: copies.runId, companyId: copies.companyId, nextAttemptAt: copies.nextAttemptAt })
+      .from(copies).innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, copies.companyId), eq(heartbeatRuns.id, copies.runId)))
+      .where(and(collectionPredicate(), or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date()))))
+      .orderBy(asc(copies.updatedAt)).limit(20);
+    for (const row of rows) {
+      try { await recoverCollection({ id: row.id, companyId: row.companyId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }, () => true); }
+      catch (error) { logger.warn({ err: error, runId: row.id }, "Instruction collection recovery deferred"); }
+    }
+    return rows.length;
+  }
+  function registerStoppedCollectionWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+    canRun: () => boolean, canRecover: () => Promise<boolean>) {
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.instructionCollection, batchSize: 20,
+      decodeScope: (hint: string): InstructionCollectionScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (entry: CollectionEntry, scope?: InstructionCollectionScope) => !scope ||
+        entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId),
+      load: async (scope, signal) => await canRecover() ? loadCollection(scope, () => canRun() && !signal?.aborted) : [],
+      check: async (entry, signal) => await canRecover() ? recoverCollection(entry, () => canRun() && !signal?.aborted) : null,
+    }, canRun);
  }

  type DirectoryCleanupEntry = { id: string; companyId: string; dueAt: number };
@@ -566,5 +633,5 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
    }
    return row;
  }
-  return { registerDirectoryCleanupWork, registerCapturedCommitWork, prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
+  return { registerStoppedCollectionWork, registerDirectoryCleanupWork, registerCapturedCommitWork, prepare, checkpointWarm, canReuseWarm: directories.canReuse, adopt, reportRetirementUnconfirmed, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
 }
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
 import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js";
 import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
 import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
@@ -102,6 +103,7 @@ export async function deleteCompany(db: Db, id: string) {
        await notifyScheduledRetryCancellation(tx, { companyId: id });
        await notifyInstructionCommit(tx, { companyId: id });
        await notifyDirectoryCleanup(tx, { companyId: id });
+        await notifyInstructionCollection(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));
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";

 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  instructionCollection: "instruction-collection",
  directoryCleanup: "directory-cleanup",
  instructionCommit: "instruction-commit",
  scheduledRetryCancellation: "scheduled-retry-cancellation",
@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
 import { enableRunFailureReporting } from "../__tests__/helpers/run-failure-reporting.js";
 import { randomUUID } from "node:crypto";
 import { eq } from "drizzle-orm";
@@ -19,7 +20,8 @@ vi.mock("../sentry.js", async () => {

 import { reconcileAbandonedExecutionControl } from "./execution-control-reconciliation.js";

-const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXECUTION_CONTROL_TEST_DATABASE_URL");
+const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;

 describeEmbeddedPostgres("reconcileAbandonedExecutionControl reports a genuine failed transition", () => {
@@ -27,11 +29,12 @@ describeEmbeddedPostgres("reconcileAbandonedExecutionControl reports a genuine f
  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;

  beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("execution-control-reconciliation-");
-    db = createDb(tempDb.connectionString);
+    if (!external) tempDb = await startEmbeddedPostgresTestDatabase("execution-control-reconciliation-");
+    db = createDb(external ?? tempDb!.connectionString);
  }, 30_000);

  afterAll(async () => {
+    await db?.$client.end({ timeout: 1 });
    await tempDb?.cleanup();
  });

@@ -218,10 +221,10 @@ describeEmbeddedPostgres("native review execution reconciliation admission", ()
  let db!: ReturnType<typeof createDb>;
  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
  beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("execution-control-review-");
-    db = createDb(tempDb.connectionString);
+    if (!external) tempDb = await startEmbeddedPostgresTestDatabase("execution-control-review-");
+    db = createDb(external ?? tempDb!.connectionString);
  }, 30_000);
-  afterAll(async () => { await tempDb?.cleanup(); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); });

  it.each([true, false])("only releases the exact admitted review (admitted=%s)", async (admitted) => {
    const companyId = randomUUID();
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
 import { notifyExecutionRunWork } from "./execution-control-notifications.js";
 import { randomUUID } from "node:crypto";
 import { logger } from "../middleware/logger.js";
@@ -183,6 +184,7 @@ export async function reconcileAbandonedExecutionControl(
                updatedAt: now,
              })
              .where(eq(nativeRunFinalizations.runId, run.id));
+          await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
          await notifyExecutionRunWork(tx);
          const [updatedRun] = await tx
            .update(heartbeatRuns)
@@ -4084,6 +4084,14 @@ export function heartbeatService(
    triggerIssueMonitor,

    reportRunActivity: clearDetachedRunWarning,
+    registerStoppedInstructionCollectionWork: (
+      coordinator: Parameters<typeof instructionCopies.registerStoppedCollectionWork>[0], canRun: () => boolean,
+    ) => instructionCopies.registerStoppedCollectionWork(coordinator,
+      () => canRun() && readTaskDrain(new Date()) === null,
+      async () => {
+        const state = await getSchedulingSuppression(true);
+        return !state.suppressed || state.reason === "task_drain";
+      }),
    registerInstructionDirectoryCleanupWork: (
      coordinator: Parameters<typeof instructionCopies.registerDirectoryCleanupWork>[0], canRun: () => boolean,
    ) => instructionCopies.registerDirectoryCleanupWork(coordinator,
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js";
 import { notifyChatResponseAvailability } from "../chat-response-availability.js";
 import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js";
@@ -850,6 +851,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
          ? "Automatic continuation stopped by the committed chat conversation close. Send a new request in chat or on the Board to start fresh work."
          : "Automatic continuation source could not be verified before provider admission. Review the task and send a fresh request; this attempt will not automatically retry.";
        await notifyChatRunMilestones(tx, { companyId: current.companyId, issueId });
+        await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id });
        [terminal] = await tx
          .update(heartbeatRuns)
          .set({
@@ -971,6 +973,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
            .for("update");
          if (!currentRun) return null;
          await notifyChatRunMilestones(tx, { companyId: run.companyId });
+          await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
          const [row] = await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(),
            errorCode: "case_handoff_superseded", error: "This case handoff was cancelled or superseded.", updatedAt: new Date() })
            .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "queued"))).returning();
@@ -1331,6 +1334,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
                  const error =
                    "Automatic continuation stopped by the committed chat conversation close. Send a fresh request in chat or on the Board.";
                  await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId });
+                  await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
                  const [cancelled] = await tx
                    .update(heartbeatRuns)
                    .set({
@@ -1452,6 +1456,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
              if (liveIds.length === 0) {
                const reason = "Queued messages were discarded before dispatch";
                await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId });
+                await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
                const [cancelled] = await tx
                  .update(heartbeatRuns)
                  .set({
@@ -3489,6 +3494,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
            const reason = issueCancelled
              ? "Cancelled because the issue was cancelled before the scheduled retry became due"
              : "Cancelled because the issue was reassigned before the scheduled retry became due";
+            await notifyInstructionCollection(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id });
            await notifyScheduledRetryCancellation(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id });
            await notifyChatRunMilestones(tx, { companyId: scheduledRun.companyId, issueId: issue.id });
            const cancelled = await tx
@@ -3627,6 +3633,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
            issue.assigneeAgentId &&
            activeExecutionRun.agentId !== issue.assigneeAgentId
          ) {
+            await notifyInstructionCollection(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id });
            await notifyChatRunMilestones(tx, { companyId: activeExecutionRun.companyId, issueId: issue.id });
            const cancelled = await tx
              .update(heartbeatRuns)
@@ -1947,14 +1947,14 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD

  let settledProcessesDiscovered = false;

-  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean }) {
+  async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean }) {
    await preserveCancelledNativeWorkspaces(db);
    const staleThresholdMs = opts?.staleThresholdMs ?? 0;
    const now = new Date();
    if (!opts?.skipScheduledRetryCancellation) await scheduledRetryCancellation.sweep();
    // Recovery never launches a provider or infers stopped ownership from
    // terminal status. Uncaptured local copies require durable stop evidence.
-    await instructionCopies.recoverStopped().catch(error => {
+    if (!opts?.skipStoppedInstructionCollection) await instructionCopies.recoverStopped().catch(error => {
      logger.warn({ err: error }, "failed to recover stopped instruction copies");
    });
    await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit, skipDirectoryCleanup: opts?.skipInstructionDirectoryCleanup }).catch(error => {
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js";
 import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
 import { notifyExecutionRunWork } from "../execution-control-notifications.js";
@@ -1023,6 +1024,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
      const tx = transaction as unknown as Db;
      if (previous && (previous.status === "scheduled_retry" || status === "scheduled_retry"))
        await notifyScheduledRetryCancellation(tx, { companyId: previous.companyId, runId: previous.id });
+      if (previous) await notifyInstructionCollection(tx, { companyId: previous.companyId, runId: previous.id });
      if (isHeartbeatRunTerminalStatus(status)) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, companyId);
      await notifyExecutionRunWork(tx);
      if (companyId) await notifyChatRunMilestones(tx, { companyId });
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
 import { notifyDirectoryCleanup } from "./directory-cleanup-notifications.js";
 import { notifyInstructionCommit } from "./instruction-commit-notifications.js";
 import { notifyScheduledRetryCancellation } from "./scheduled-retry-cancellation-notifications.js";
@@ -622,6 +623,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
        if (patch.enableWorktreeRunExecution !== undefined) {
          await notifyInstructionCommit(tx);
          await notifyDirectoryCleanup(tx);
+          await notifyInstructionCollection(tx);
        }
        if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx);
        await notifyToolConnectionHealth(tx, "*");
@@ -0,0 +1,7 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type InstructionCollectionScope = { companyId: string; runId?: string };
+
+export async function notifyInstructionCollection(transaction: object, scope?: InstructionCollectionScope) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.instructionCollection, scope ? JSON.stringify(scope) : "*");
+}
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
 import { notifyExecutionRunWork } from "./execution-control-notifications.js";
 import { hasRequiredWorkspaceRecovery, preserveWorkspaceRestoreRecoveryMetadata, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js";
 import { hasUnrestoredRemoteWorkspace, preserveLegacyWorkspaceRestoreSources, type LegacyWorkspaceRestoreSource } from "./legacy-workspace-restore-recovery.js";
@@ -118,7 +119,8 @@ export async function terminalizeLegacyExecution(input: {
          )
          .for("update")
      : [];
-    const [current] = await tx.select({ runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status,
+    const [current] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
+      runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status,
      resultJson: heartbeatRuns.resultJson, executionStatusDeliveryId: heartbeatRuns.executionStatusDeliveryId,
    }).from(heartbeatRuns).where(and(
      eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
@@ -129,6 +131,7 @@ export async function terminalizeLegacyExecution(input: {
      resultJson: preserveWorkspaceRestoreRecoveryMetadata(current.resultJson,
        input.recordRestoreFailureOnly ? { ...current.resultJson, ...patch.resultJson } : patch.resultJson),
    };
+    await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id });
    await notifyExecutionRunWork(tx);
    let [updated] = await tx
      .update(heartbeatRuns)
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { readRequiredQualityReview } from "../required-quality-review.js";
 import type { WorkspaceWorkers } from "../plugin-workspaces.js";
 import { recoverLegacyUnsafeWorkspaceExports } from "./native-workspace-export-recovery.js";
@@ -515,6 +516,7 @@ export async function claimNativeSessionResumptions(input: {
          nextAttemptAt: null,
          updatedAt: now,
        }).where(eq(nativeRunFinalizations.runId, row.run.id));
+        await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id });
        const [updatedRun] = await tx.update(heartbeatRuns).set({
          status: "failed",
          finishedAt: now,
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { hasNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
 import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
@@ -919,6 +920,7 @@ export async function claimNativeRestartRecoveries(input: {
              eq(nativeRunFinalizations.phase, row.coordinator.phase),
            ),
          );
+        await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id });
        const [updatedRun] = await tx
          .update(heartbeatRuns)
          .set({
@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import { enableRunFailureReporting } from "../../__tests__/helpers/run-failure-reporting.js";
 import { randomUUID } from "node:crypto";
 import { verifyCommittedReplayPayloadReads } from "./__tests__/helpers/committed-replay.js";
@@ -71,7 +72,8 @@ import { commitNativeStatusDecision } from "./status-decision-committer.js";
 import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeStatusDecision } from "./status-arbiter.js";
 import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js";

-const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_FINALIZER_TEST_DATABASE_URL");
+const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;

 if (!embeddedPostgresSupport.supported) {
@@ -89,8 +91,8 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a
  let agentId: string;

  beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-native-finalizer-telemetry-");
-    db = createDb(tempDb.connectionString);
+    if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-native-finalizer-telemetry-");
+    db = createDb(external ?? tempDb!.connectionString);
    companyId = randomUUID();
    agentId = randomUUID();
    await db.insert(companies).values({ id: companyId, name: "Telemetry", issuePrefix: "TLM" });
@@ -104,6 +106,7 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a
  }, 30_000);

  afterAll(async () => {
+    await db?.$client.end({ timeout: 1 });
    await tempDb?.cleanup();
  });

@@ -960,6 +963,26 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a
    expect(agentTaskRunCalls(callsBefore)).toHaveLength(1);
  });

+  it.each([true, false])("signals instruction collection only when native failure projects terminal status (%s)", async projectRunStatus => {
+    const fixture = await seedNativeRun();
+    await driveToCompleteResult(fixture);
+    const settled = vi.fn();
+    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.instructionCollection, settled);
+    try {
+      await recordNativeFinalizationFailure({ db, runId: fixture.runId,
+        error: new Error("native_fixture_permanent_failure"), permanent: true, projectRunStatus });
+      expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)))[0].phase).toBe("terminal_failure");
+      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+      if (projectRunStatus) {
+        expect(run.status).not.toBe("running");
+        expect(settled).toHaveBeenCalled();
+      } else {
+        expect(run.status).toBe("running");
+        expect(settled).not.toHaveBeenCalled();
+      }
+    } finally { unsubscribe(); }
+  });
+
  it("signals durable accounting when native commitment or permanent failure makes a receipt actionable", async () => {
    const settled = vi.fn();
    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.accounting, settled);
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { readRequiredQualityReview } from "../required-quality-review.js";
 import { pendingNativeGovernance } from "./native-governance.js";
 import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
@@ -410,6 +411,7 @@ async function recordRetryableFailure(input: {
    if (projectsTerminalStatus && input.failureScope === "workspace") {
      await preserveNativeWorkspaceExportLease(tx as unknown as Db, input.run, input.coordinator.resultId);
    }
+    if (projectsTerminalStatus) await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id });
    const [updatedRun] = await tx
      .update(heartbeatRuns)
      .set({
@@ -667,6 +669,7 @@ async function projectCommittedRun(input: {
  const now = new Date();
  const [updatedRun] = await input.db.transaction(async transaction => {
    const tx = transaction as unknown as Db;
+    await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id });
    await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
    await notifyExecutionRunWork(tx);
    await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
@@ -1685,6 +1688,7 @@ export async function finalizeNativeRun(input: {
        .returning();
      const [updatedRun] = await input.db.transaction(async transaction => {
        const tx = transaction as unknown as Db;
+        if (input.projectRunStatus) await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
        if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
        await notifyExecutionRunWork(tx);
        await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { providerPackWorkspaceId, recordAdmittedRemoteProviderPack, resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js";
 import { type ProviderPackInstallation, verifyRemoteProviderPack as verifyRemoteProviderPackIdentity } from "./remote-provider-pack-verification.js";
 import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js";
@@ -9817,6 +9818,7 @@ async function executePaperclipNativeSessionWithinScope(
          .from(heartbeatRuns)
          .where(eq(heartbeatRuns.id, input.execution.binding.runId))
          .for("update");
+        if (!ownershipUnverified) await notifyInstructionCollection(tx, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId });
        let [updatedRun] = await tx
          .update(heartbeatRuns)
          .set({
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
 import { scheduledIssueMonitorWait } from "../issue-monitors.js";
 import { onboardingFirstTaskCompletionAllowed } from "../onboarding-first-task-admission.js";
@@ -1066,6 +1067,7 @@ async function materializeDecisionEffect(input: {
          ),
        );
    }
+    await notifyInstructionCollection(input.tx, { companyId: input.companyId, runId: input.runId });
    await notifyDeliveryWork(input.tx, DELIVERY_QUEUES.accounting, input.companyId);
    const [run] = await input.tx
      .update(heartbeatRuns)
@@ -1,3 +1,4 @@
+import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
 import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
 import { boundRunIssueId } from "../run-task-binding.js";
 import { recoveryRunContextColumn, recoveryRunVersionColumn } from "./run-context-read.js";
@@ -6247,6 +6248,7 @@ export function recoveryService(
      // to recheck finalization records committed after this lock was taken.
      const [candidate] = await tx.select().from(heartbeatRuns).where(eligible).for("update");
      if (!candidate) return null;
+      await notifyInstructionCollection(tx, { companyId: candidate.companyId, runId: candidate.id });
      const boundIssueId = boundRunIssueId(candidate);
      if (boundIssueId) {
        await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: boundIssueId });