diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md
index 5cd374773d8..9ac4abb3b2b 100644
--- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md
+++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md
@@ -2585,3 +2585,19 @@ Local verification currently has 200 passing cases across directory, startup,
 captured instruction and heartbeat release suites, plus workspace typecheck/build.
 Remote acceptance is pending. Stopped collection and the remaining
 original inventory and older stack review/CI gates remain outstanding.
+
+## Slice 41 acceptance and slice 42 collection
+
+Parent directory cleanup PR #17671 at
+`854244092a993b621ab8efd2ca9f80a0e3b0b9d4` has all 53 checks successful,
+Greptile 5/5 at that head, and no unresolved review findings. It remains open.
+
+Slice 42 migrates stopped instruction collection, including terminal-run writers
+that make unchanged copy rows eligible. The producer, lifecycle, partial-index,
+and verification audit is in
+[the stopped collection audit](2026-10-25-stopped-instruction-collection-sleep-audit.md).
+The initial eight-suite acceptance selection passed 269 tests; directory-owner
+coverage passed 101 separately. Exact-head remote acceptance is still required.
+Other native recovery, dispatch, retention, integration and aggregate-tick duties
+remain in scope. The older CI/review gates recorded above are not discharged by
+this slice's local tests.
diff --git a/doc/plans/2026-10-25-stopped-instruction-collection-sleep-audit.md b/doc/plans/2026-10-25-stopped-instruction-collection-sleep-audit.md
new file mode 100644
index 00000000000..c65231a2635
--- /dev/null
+++ b/doc/plans/2026-10-25-stopped-instruction-collection-sleep-audit.md
@@ -0,0 +1,276 @@
+# Stopped instruction collection
+
+Date: 2026-10-25. Parent: directory cleanup PR #17671 at
+`854244092a993b621ab8efd2ca9f80a0e3b0b9d4`.
+
+## Obligation and owner
+
+The orphan reaper previously rediscovered stopped instruction copies on every
+maintenance pass. This slice migrates that one reason to committed signals and
+saved retry deadlines. Captured candidate commits (#17670) and post-save
+physical release (#17671) remain separate workers. No new table is needed.
+
+The instruction working-copy service owns eligibility, reconstruction and
+collection scheduling. The directory service still owns directory locks,
+materialization identity, warm handoff, stop proof and physical operations.
+Native and legacy run owners emit a normalized company/run signal before their
+status writes. They do not inspect instruction receipts or perform file work.
+Startup only wires the existing coordinator and scheduler.
+
+## Existing durable predicate
+
+A copy is eligible when its same-company run is terminal (succeeded, failed,
+cancelled, timed_out or interrupted), its receipt has no retainedByRunId key,
+and either:
+
+- It has fewer than three collection attempts and is prepared,
+  pending_collection, unchanged_turn, warm_saved, or an agent-files.v1 copy in
+  preparing state.
+- It is an unavailable remote agent-files.v1 copy without processStoppedAt.
+  This existing cleanup-proof branch is not subject to the collection cap.
+
+Active prepared copies and retained warm owners are excluded. Exhausted or
+completed copy history is passive. Terminal run status is eligibility for
+inspection, not proof that the provider has stopped.
+
+Startup loads scalar run IDs, company IDs and nextAttemptAt in keyset pages of
+100. The shared worker keeps outstanding IDs and selects fair batches of 20.
+Each due check rereads eligibility, honors a later saved deadline, and reserves
+30 seconds through the existing copy CAS before I/O. The claim also verifies
+that the same-company run is still terminal. A resumed run cannot be collected
+merely because it was in an earlier discovery result.
+
+The existing collection owner then applies its rules:
+
+- Interrupted directory preparation can be discarded because launch requires
+  prepared state first.
+- Local collection requires saved stop proof or a valid server-authored native
+  stop event, including its existing launch-order validation.
+- Remote collection retains exact lease and termination checks. Unknown ownership
+  stays pending. Recovery never starts a replacement environment to fetch files.
+- Failed capture retains candidate/error state and the existing attempt limit.
+- Captured bytes hand off to the existing commit owner. Directory cleanup keeps
+  its own receipt and owner.
+
+The worker rereads durable state after the attempt. Settled entries disarm;
+unresolved entries retain their saved future deadline. A failure or unresolved
+entry without a future deadline uses the shared bounded error backoff. Later
+stop evidence is covered by the already-pending retry; it needs no recurring
+scan of unrelated process events or environment leases.
+
+## Producer audit and transaction ownership
+
+Copy writes alone are insufficient. Run terminalization creates eligibility
+without changing the copy row. The audit inspected run inserts/updates as well
+as both instruction-copy owners.
+
+| Producer family | Integration |
+| --- | --- |
+| Common instruction-copy patch | Signal changes into, within and out of potential collection work, inside the short existing transaction |
+| Legacy copy preparation | Signal before the preparing insert; stage files outside the new short transaction |
+| Directory preparation and warm successor insert | Signal in a short insert transaction; preserve insertion before the agent-row handoff lock |
+| Directory adoption | Signal prior and successor scopes inside the existing handoff transaction |
+| Heartbeat run lifecycle and legacy terminalization | Signal exact company/run inside existing status transaction |
+| Heartbeat queue and run-dispatch adapter | Signal cancellation, supersession and other terminal writes inside their existing transactions |
+| Queued-comment adapter and active-run watchdog | Signal exact terminalized run in the existing transaction |
+| Execution-control and task recovery | Signal after candidate eligibility checks, before terminal write |
+| Native finalizer, restart recovery and finalization reconciler | Signal terminal projection and terminal recovery writes in their owning transactions |
+| Native session executor and status-decision committer | Signal verified terminal failure/cancellation; retain unverified-ownership behavior |
+| Agent/company deletion | Invalidate company scope before cascades |
+| Worktree execution policy | Signal globally through the existing settings transaction |
+
+Metadata-only run writes do not create eligibility. A reset to running is
+rechecked before execution; the subsequent terminal transition signals again.
+Deletion of an unsuccessful directory reservation cannot create new collection
+work. An already-known entry is invalidated at its next scheduled check.
+
+Signals join existing outer transaction settlement, including nested writes and
+ambiguous commit outcomes. No filesystem/provider call is wrapped in a new
+transaction. Signal handling can inspect the exact run even if it has no copy;
+this is bounded producer-time work, not recurring empty discovery. The supported
+model is one process with instrumented writers. Independent SQL writers and
+additional replicas still need explicit integration.
+
+## Lifecycle and index
+
+The heartbeat facade retains task-drain and worktree-execution admission.
+Committed changes during drain/standby remain pending without passive SQL.
+Startup/resume reconstructs existing debt. Shutdown joins admitted work and
+retires subscriptions and deadlines. Manual/startup recoverStopped remains
+available; only the migrated periodic caller skips it.
+
+Migration `0408_short_red_skull.sql` adds the concurrent, replay-safe partial collection index on
+(run_id, company_id). Its predicate excludes ineligible copy history; terminal
+run eligibility remains a same-company join. The migration is registered with
+the existing nontransactional runner and retains snapshots 0404 through 0408.
+The original draft used 0396; later regeneration history remains recorded below.
+Memory and startup cost scale with actual pending copy candidates, including
+active candidates filtered by the run join; this is not a constant-cost claim.
+
+## Verification and preserved failures
+
+- 269 tests passed across instruction copies, startup wiring, run dispatch,
+  queued-comment dispatch, heartbeat queue/recovery, execution-control recovery,
+  and native finalization recovery. No tests were skipped in this selection.
+- The instruction suite has 45 cases, including 12 stopped-worker cases. These
+  cover quiet active/history states, real terminalization without a copy write,
+  nested rollback/commit, restart, drain/standby, saved deadlines, delayed stop
+  evidence, resumed-run claim races, 103 candidates, a failed claim, and shutdown
+  during a save. Filesystem and PostgreSQL effects are exercised together.
+- The directory owner suite passed all 101 cases after updating the existing
+  stale-collector race injector to intercept the actual transaction boundary.
+- Workspace typecheck and build passed.
+- Native session/restart/reconciler tests passed 814 cases. Fifteen embedded-only
+  remote warm ownership cases were filtered out locally. Native finalizer tests
+  initially skipped all 27 cases because embedded PostgreSQL was unavailable;
+  the suite now accepts a dedicated external URL and passed all 29 cases,
+  including two real terminal-projection notification cases. These skips are
+  not counted as passes. Full CI must exercise the embedded-only fixtures.
+- Fresh migration application, migration-runner replay, direct index replay and
+  migration numbering/safety checks passed.
+- A rolled-back PostgreSQL fixture added 10,000 completed-history rows and one
+  pending terminal copy. The plan used the partial collection index and run PK
+  join, returned one row, and used 12 buffer hits (0.026 ms local execution).
+  Dead index entries existed in this transaction. This is evidence of index
+  selection, not a production latency or one-index-tuple claim.
+
+The first broad run passed 187 tests and failed one existing directory race
+test: its direct-update spy no longer injected adoption after the mutation moved
+into a transaction. The repaired test injects at that transaction and retains
+its original ownership, untouched-file, attempt and stop-proof assertions. The
+subsequent instruction/directory run passed 143 tests. The later eight-suite run
+includes three additional collection lifecycle tests and passed 269 tests.
+
+A first added native-finalizer notification test passed 27 cases and failed one:
+it expected a collection wake for a metadata-only finalizer failure. That API
+only writes terminal run status when projectRunStatus is enabled. The final test
+separates both cases: metadata-only failure leaves the run running and emits no
+collection signal; terminal projection emits the signal. All 29 finalizer cases
+then passed on a fresh database. No production behavior was weakened.
+
+The embedded PostgreSQL support probe failed before suite startup on this host.
+Tests used fresh disposable PostgreSQL 17 databases through the existing
+validated external-test-database helper. Native finalization, native finalizer and execution-control
+suites now accept opt-in dedicated URLs; normal embedded CI remains unchanged.
+The failed harness log and all prior test logs remain available locally.
+
+Full remote CI and review are required for this PR's exact head. This slice does
+not establish Cloud alarm coverage, deployed database/host suspension, fleet
+savings, or completion of the remaining orphan-reaper migrations.
+
+
+## Current parent and complete-file acceptance (2026-10-28)
+
+Current source `28037085a01306e8e06ce8ffeacc112d781bf1fd` passes **1469 cases
+across 14 complete files**, no skips, in 142.21 seconds. The selection uses no
+case-name filter and includes the complete native session executor and restart
+recovery suites. The earlier filtered or skipped results remain recorded above.
+
+| Complete file | Passing cases |
+| --- | ---: |
+| src/services/native-runtime/native-session-executor.test.ts | 968 |
+| src/__tests__/agent-directory-working-copies.test.ts | 101 |
+| src/services/heartbeat/run-lifecycle.test.ts | 31 |
+| src/services/heartbeat/queue.test.ts | 32 |
+| src/__tests__/agent-instruction-working-copies.test.ts | 50 |
+| src/modules/run-dispatch/adapters/postgres.test.ts | 69 |
+| src/services/heartbeat/recovery.test.ts | 45 |
+| src/__tests__/server-startup-feedback-export.test.ts | 57 |
+| src/services/native-runtime/native-run-finalizer-telemetry.test.ts | 29 |
+| src/__tests__/native-finalization-recovery.test.ts | 32 |
+| src/services/execution-control-reconciliation.test.ts | 10 |
+| src/modules/wake-queue/adapters/queued-comment-postgres.test.ts | 5 |
+| src/services/native-runtime/native-restart-recovery.test.ts | 25 |
+| src/services/legacy-execution-recovery.test.ts | 15 |
+
+Ten distinct fresh PostgreSQL databases were created and migrated before
+fixture setup, covering the filesystem, recovery, dispatch, queue, execution
+control and native finalization boundaries. The preserved embedded fixtures
+within the selected files were exercised as supported; no suite was replaced
+with a mock to count a pass. Database build, server typecheck/build, module
+boundaries, target migration order, and all ten fresh migrations pass.
+
+Only this slice's unpublished 0400 migration artifacts were deferred before a
+normal merge of the current directory-cleanup parent. The original concurrent,
+idempotent collection index SQL was regenerated through `pnpm db:generate` as
+`0406_gigantic_sunfire.sql`, with identical SQL bytes. All prior SQL, journal
+entries, and retained snapshot bytes remain unchanged; five snapshots 0402-0406
+remain. The historical snapshot merge conflict used exact parent bytes.
+Original domain/schema edits and unrelated upstream changes are preserved.
+
+A populated parent database upgraded 0405 to 0406, preserving all 404 old
+migration receipts and appending exactly one. Replay appends none. The collection
+index is valid and ready on both fresh and upgraded databases. This is local
+worker, producer and migration evidence; deployed host/database suspension and
+external Cloud wake coverage are separate. Actual landed parent ancestry and
+fresh current-head CI and reviews are still required before merge.
+
+
+## Reviewed parent and comment compliance
+
+The latest prepared directory-cleanup parent carries the published captured-save
+review correction. A normal merge changes parent audits and comment text only;
+every changed code file retains an identical non-comment TypeScript token stream
+and the complete database source tree remains unchanged. Four newly added
+comment lines in this slice's instruction-copy owner are removed under
+CONTRIBUTING.md. The staging/terminal-proof and fair-page explanations remain
+in this audit. No guard, query, assertion or deadline changes. The 1469-case
+acceptance retains its original scope; final actual-parent ancestry, current
+head CI and reviews are still required before publication and merge.
+
+## Current parent, migration408 and complete native-runtime acceptance
+
+The current directory-cleanup parent carries the new Coordinator/Runner behavior,
+bounded retry cancellation and actual landed instruction-recovery ancestry. Only
+this slice's unlanded 0406 artifacts were deferred before the normal merge. The
+two native finalization source conflicts were import-only: both the collection
+notification and required-quality-review imports remain. Complete original
+runtime/schema edits and every unrelated upstream blob are preserved. The
+migration tree and client matched the exact parent before generation.
+
+The original concurrent, idempotent index SQL was generated as
+`0408_short_red_skull.sql`. All earlier executable SQL, prior journal entries,
+and retained snapshot bytes remain unchanged. The five snapshots are
+0404 through 0408. No new code comments were added relative to the parent.
+
+Source `cb2818a8f46ef961f744977a258f0e9042e31a42` passes all **1469 cases in
+fourteen complete files**, without skips, in 141.69 seconds. This includes the
+full 968-case native-session suite and the collection, directory, startup,
+recovery, dispatch, queued-comment, execution-control, finalization and lifecycle
+consumers. Eleven separate fresh PostgreSQL 17 databases migrated through 0408
+isolate destructive fixtures. The embedded-only lifecycle fixture temporarily
+uses its own external database on this IPC-exhausted host, then is restored byte
+for byte. All assertions and original timing limits stay unchanged. This is
+qualified external-database acceptance, not an unmodified embedded CI run.
+Actual database build, server types/build, boundaries and exact-current migration
+order pass with commands selected by name.
+
+A populated 0407-to0408 upgrade preserves every one of the 406 prior migration
+receipts and all 38 companies, adds exactly one receipt, and adds none on replay.
+The collection index is valid and ready on upgraded and fresh databases.
+
+Adopting published parent `210fcb1215a3b5e5ed84e420ddb73ab5952d9365` changes
+only parent documentation and ancestry. Integration
+`a4dfd30786f30f755273c86515c17644c4f15471` preserves complete server, package,
+UI and test trees from the accepted source, including the whole 0408 database
+tree. It also records the actual 0406 squash ancestry. The parent directory PR
+still must merge, then its actual 0407 squash ancestry must be reconciled before
+this slice publishes. Old draft review scores and local acceptance do not prove
+current-head remote gates or deployed sleep. All earlier failures stay recorded.
+
+
+## Landed directory parent and independent UI preservation
+
+Directory cleanup landed as `267842fc3c6e6542deed8a1bcc2c78bec8df7862`.
+Its server, package and test trees match reviewed parent `210fcb1215a3b5e5ed84e420ddb73ab5952d9365`.
+Master also contains the independent sidebar popover fix from #18117 in three UI
+paths. A normal merge uses the proven combination of both complete patches.
+It preserves every owned stopped-collection edit and every upstream UI blob.
+The resulting server, package and test trees remain identical to the accepted
+1469-case source. The whole 0408 database tree remains unchanged.
+
+Integration `c3a2ad6bdf` records the actual landed-parent ancestry. All 36 tests
+in the complete sidebar account-menu file pass on the combined tree in 2.02
+seconds. Existing React act warnings are present. No timing limits, backend
+assertions or runtime behavior changed for this ancestry reconciliation. Fresh
+CI and exact-current-head reviews must still pass before merge.
diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts
index 6a44691da1c..c327dfe02a1 100644
--- a/packages/db/src/client.ts
+++ b/packages/db/src/client.ts
@@ -557,6 +557,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
   "0403_natural_cerise.sql",
   "0406_zippy_war_machine.sql",
   "0407_greedy_star_brand.sql",
+  "0408_short_red_skull.sql",
 
 
   "0384_funny_adam_warlock.sql",
diff --git a/packages/db/src/migrations/0408_short_red_skull.sql b/packages/db/src/migrations/0408_short_red_skull.sql
new file mode 100644
index 00000000000..82f185ad86c
--- /dev/null
+++ b/packages/db/src/migrations/0408_short_red_skull.sql
@@ -0,0 +1,5 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "agent_instruction_copies_collection_idx" ON "agent_instruction_working_copies" USING btree ("run_id","company_id") WHERE NOT (coalesce("agent_instruction_working_copies"."receipt", '{}'::jsonb) ? 'retainedByRunId') AND (
+      ("agent_instruction_working_copies"."attempts" < 3 AND ("agent_instruction_working_copies"."state" IN ('prepared', 'pending_collection', 'unchanged_turn', 'warm_saved')
+        OR ("agent_instruction_working_copies"."state" = 'preparing' AND "agent_instruction_working_copies"."receipt"->>'schema' = 'paperclip.agent-files.v1')))
+      OR ("agent_instruction_working_copies"."state" = 'unavailable' AND "agent_instruction_working_copies"."process_stopped_at" IS NULL
+        AND "agent_instruction_working_copies"."location" LIKE 'remote:%' AND "agent_instruction_working_copies"."receipt"->>'schema' = 'paperclip.agent-files.v1'));
\ No newline at end of file
diff --git a/packages/db/src/migrations/meta/0403_snapshot.json b/packages/db/src/migrations/meta/0408_snapshot.json
similarity index 99%
rename from packages/db/src/migrations/meta/0403_snapshot.json
rename to packages/db/src/migrations/meta/0408_snapshot.json
index 957177038ec..195b96fbb8e 100644
--- a/packages/db/src/migrations/meta/0403_snapshot.json
+++ b/packages/db/src/migrations/meta/0408_snapshot.json
@@ -1,6 +1,6 @@
 {
-  "id": "e21e38de-5a82-473f-b81e-e96eff5ef4c8",
-  "prevId": "6ce76555-38d9-4074-873f-db02c546e2c6",
+  "id": "6bd67ad8-0d80-4163-873b-8b8eacf32bb2",
+  "prevId": "856a147c-73dc-4908-a8d9-4f19cab7dbf6",
   "version": "7",
   "dialect": "postgresql",
   "tables": {
@@ -2665,6 +2665,50 @@
         }
       },
       "indexes": {
+        "agent_instruction_copies_collection_idx": {
+          "name": "agent_instruction_copies_collection_idx",
+          "columns": [
+            {
+              "expression": "run_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "NOT (coalesce(\"agent_instruction_working_copies\".\"receipt\", '{}'::jsonb) ? 'retainedByRunId') AND (\n      (\"agent_instruction_working_copies\".\"attempts\" < 3 AND (\"agent_instruction_working_copies\".\"state\" IN ('prepared', 'pending_collection', 'unchanged_turn', 'warm_saved')\n        OR (\"agent_instruction_working_copies\".\"state\" = 'preparing' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1')))\n      OR (\"agent_instruction_working_copies\".\"state\" = 'unavailable' AND \"agent_instruction_working_copies\".\"process_stopped_at\" IS NULL\n        AND \"agent_instruction_working_copies\".\"location\" LIKE 'remote:%' AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1'))",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "agent_instruction_copies_cleanup_idx": {
+          "name": "agent_instruction_copies_cleanup_idx",
+          "columns": [
+            {
+              "expression": "run_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"agent_instruction_working_copies\".\"state\" IN ('saved', 'unchanged', 'resolved', 'unavailable')\n      AND \"agent_instruction_working_copies\".\"process_stopped_at\" IS NOT NULL\n      AND \"agent_instruction_working_copies\".\"receipt\"->>'schema' = 'paperclip.agent-files.v1'\n      AND coalesce(\"agent_instruction_working_copies\".\"receipt\"->>'retainedByRunId', '') = ''\n      AND \"agent_instruction_working_copies\".\"receipt\"->>'cleanupPending' IS DISTINCT FROM 'false'\n      AND (\"agent_instruction_working_copies\".\"receipt\" ? 'baseline' OR \"agent_instruction_working_copies\".\"receipt\"->>'cleanupPending' = 'true')",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "agent_instruction_copies_pending_idx": {
           "name": "agent_instruction_copies_pending_idx",
           "columns": [
@@ -33743,6 +33787,22 @@
           "method": "btree",
           "with": {}
         },
+        "heartbeat_runs_scheduled_retry_cancellation_idx": {
+          "name": "heartbeat_runs_scheduled_retry_cancellation_idx",
+          "columns": [
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"heartbeat_runs\".\"status\" = 'scheduled_retry' and jsonb_typeof(\"heartbeat_runs\".\"result_json\"->'startupCancellation') = 'object'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "heartbeat_runs_cost_accounting_pending_idx": {
           "name": "heartbeat_runs_cost_accounting_pending_idx",
           "columns": [
@@ -39111,6 +39171,140 @@
       "checkConstraints": {},
       "isRLSEnabled": false
     },
+    "public.issue_thread_interaction_idempotency_keys": {
+      "name": "issue_thread_interaction_idempotency_keys",
+      "schema": "",
+      "columns": {
+        "id": {
+          "name": "id",
+          "type": "uuid",
+          "primaryKey": true,
+          "notNull": true,
+          "default": "gen_random_uuid()"
+        },
+        "company_id": {
+          "name": "company_id",
+          "type": "uuid",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "issue_id": {
+          "name": "issue_id",
+          "type": "uuid",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "interaction_id": {
+          "name": "interaction_id",
+          "type": "uuid",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "idempotency_key": {
+          "name": "idempotency_key",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "created_at": {
+          "name": "created_at",
+          "type": "timestamp with time zone",
+          "primaryKey": false,
+          "notNull": true,
+          "default": "now()"
+        }
+      },
+      "indexes": {
+        "issue_thread_interaction_idempotency_keys_scope_key_uq": {
+          "name": "issue_thread_interaction_idempotency_keys_scope_key_uq",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "issue_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "idempotency_key",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": true,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "issue_thread_interaction_idempotency_keys_interaction_idx": {
+          "name": "issue_thread_interaction_idempotency_keys_interaction_idx",
+          "columns": [
+            {
+              "expression": "interaction_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        }
+      },
+      "foreignKeys": {
+        "issue_thread_interaction_idempotency_keys_company_id_companies_id_fk": {
+          "name": "issue_thread_interaction_idempotency_keys_company_id_companies_id_fk",
+          "tableFrom": "issue_thread_interaction_idempotency_keys",
+          "tableTo": "companies",
+          "columnsFrom": [
+            "company_id"
+          ],
+          "columnsTo": [
+            "id"
+          ],
+          "onDelete": "cascade",
+          "onUpdate": "no action"
+        },
+        "issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk": {
+          "name": "issue_thread_interaction_idempotency_keys_issue_id_issues_id_fk",
+          "tableFrom": "issue_thread_interaction_idempotency_keys",
+          "tableTo": "issues",
+          "columnsFrom": [
+            "issue_id"
+          ],
+          "columnsTo": [
+            "id"
+          ],
+          "onDelete": "cascade",
+          "onUpdate": "no action"
+        },
+        "issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk": {
+          "name": "issue_thread_interaction_idempotency_keys_interaction_id_issue_thread_interactions_id_fk",
+          "tableFrom": "issue_thread_interaction_idempotency_keys",
+          "tableTo": "issue_thread_interactions",
+          "columnsFrom": [
+            "interaction_id"
+          ],
+          "columnsTo": [
+            "id"
+          ],
+          "onDelete": "cascade",
+          "onUpdate": "no action"
+        }
+      },
+      "compositePrimaryKeys": {},
+      "uniqueConstraints": {},
+      "policies": {},
+      "checkConstraints": {},
+      "isRLSEnabled": false
+    },
     "public.issue_thread_interactions": {
       "name": "issue_thread_interactions",
       "schema": "",
@@ -59963,6 +60157,69 @@
       "checkConstraints": {},
       "isRLSEnabled": false
     },
+    "public.user_workspace_preferences": {
+      "name": "user_workspace_preferences",
+      "schema": "",
+      "columns": {
+        "id": {
+          "name": "id",
+          "type": "uuid",
+          "primaryKey": true,
+          "notNull": true,
+          "default": "gen_random_uuid()"
+        },
+        "user_id": {
+          "name": "user_id",
+          "type": "text",
+          "primaryKey": false,
+          "notNull": true
+        },
+        "coordinator_workspace_enabled": {
+          "name": "coordinator_workspace_enabled",
+          "type": "boolean",
+          "primaryKey": false,
+          "notNull": true,
+          "default": false
+        },
+        "created_at": {
+          "name": "created_at",
+          "type": "timestamp with time zone",
+          "primaryKey": false,
+          "notNull": true,
+          "default": "now()"
+        },
+        "updated_at": {
+          "name": "updated_at",
+          "type": "timestamp with time zone",
+          "primaryKey": false,
+          "notNull": true,
+          "default": "now()"
+        }
+      },
+      "indexes": {
+        "user_workspace_preferences_user_uq": {
+          "name": "user_workspace_preferences_user_uq",
+          "columns": [
+            {
+              "expression": "user_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": true,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        }
+      },
+      "foreignKeys": {},
+      "compositePrimaryKeys": {},
+      "uniqueConstraints": {},
+      "policies": {},
+      "checkConstraints": {},
+      "isRLSEnabled": false
+    },
     "public.work_assessments": {
       "name": "work_assessments",
       "schema": "",
diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json
index cc5297bfcae..d193710eb2c 100644
--- a/packages/db/src/migrations/meta/_journal.json
+++ b/packages/db/src/migrations/meta/_journal.json
@@ -2843,6 +2843,13 @@
       "when": 1793236707492,
       "tag": "0407_greedy_star_brand",
       "breakpoints": true
+    },
+    {
+      "idx": 408,
+      "version": "7",
+      "when": 1793240590542,
+      "tag": "0408_short_red_skull",
+      "breakpoints": true
     }
   ]
 }
\ No newline at end of file
diff --git a/packages/db/src/schema/agent_instruction_revisions.ts b/packages/db/src/schema/agent_instruction_revisions.ts
index f13e13e9910..f2849346721 100644
--- a/packages/db/src/schema/agent_instruction_revisions.ts
+++ b/packages/db/src/schema/agent_instruction_revisions.ts
@@ -114,6 +114,11 @@ export const agentInstructionWorkingCopies = pgTable(
   },
   (t) => ({
     owner: foreignKey({ columns: [t.companyId, t.agentId], foreignColumns: [agents.companyId, agents.id] }).onDelete("cascade"),
+    collection: index("agent_instruction_copies_collection_idx").on(t.runId, t.companyId).where(sql`NOT (coalesce(${t.receipt}, '{}'::jsonb) ? 'retainedByRunId') AND (
+      (${t.attempts} < 3 AND (${t.state} IN ('prepared', 'pending_collection', 'unchanged_turn', 'warm_saved')
+        OR (${t.state} = 'preparing' AND ${t.receipt}->>'schema' = 'paperclip.agent-files.v1')))
+      OR (${t.state} = 'unavailable' AND ${t.processStoppedAt} IS NULL
+        AND ${t.location} LIKE 'remote:%' AND ${t.receipt}->>'schema' = 'paperclip.agent-files.v1'))`),
     directoryCleanup: index("agent_instruction_copies_cleanup_idx").on(t.runId, t.companyId).where(sql`${t.state} IN ('saved', 'unchanged', 'resolved', 'unavailable')
       AND ${t.processStoppedAt} IS NOT NULL
       AND ${t.receipt}->>'schema' = 'paperclip.agent-files.v1'
diff --git a/server/src/__tests__/agent-directory-working-copies.test.ts b/server/src/__tests__/agent-directory-working-copies.test.ts
index a88c87121d3..190fef24d99 100644
--- a/server/src/__tests__/agent-directory-working-copies.test.ts
+++ b/server/src/__tests__/agent-directory-working-copies.test.ts
@@ -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,
diff --git a/server/src/__tests__/agent-instruction-working-copies.test.ts b/server/src/__tests__/agent-instruction-working-copies.test.ts
index 85654fcb793..f0f08638ed6 100644
--- a/server/src/__tests__/agent-instruction-working-copies.test.ts
+++ b/server/src/__tests__/agent-instruction-working-copies.test.ts
@@ -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();
diff --git a/server/src/__tests__/native-finalization-recovery.test.ts b/server/src/__tests__/native-finalization-recovery.test.ts
index 90d0be71741..f54811a36d1 100644
--- a/server/src/__tests__/native-finalization-recovery.test.ts
+++ b/server/src/__tests__/native-finalization-recovery.test.ts
@@ -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(
diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts
index 936218bbd3a..4aba603e719 100644
--- a/server/src/__tests__/server-startup-feedback-export.test.ts
+++ b/server/src/__tests__/server-startup-feedback-export.test.ts
@@ -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 {
diff --git a/server/src/index.ts b/server/src/index.ts
index 591ff902e1c..96e027f6834 100644
--- a/server/src/index.ts
+++ b/server/src/index.ts
@@ -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();
diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts
index c2e45643050..26c2a9cd517 100644
--- a/server/src/modules/active-run-watchdog/adapters/postgres.ts
+++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts
@@ -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({
diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts
index 547356e34fd..aef69b49971 100644
--- a/server/src/modules/run-dispatch/adapters/postgres.ts
+++ b/server/src/modules/run-dispatch/adapters/postgres.ts
@@ -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
diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts
index 242f9089a3d..01f0ef27acb 100644
--- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts
+++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts
@@ -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)
diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts
index bcced794117..0ed1a03ae40 100644
--- a/server/src/services/agent-deletion.ts
+++ b/server/src/services/agent-deletion.ts
@@ -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));
diff --git a/server/src/services/agent-directory-working-copies.ts b/server/src/services/agent-directory-working-copies.ts
index 6c54dc381a4..0fdd11d4239 100644
--- a/server/src/services/agent-directory-working-copies.ts
+++ b/server/src/services/agent-directory-working-copies.ts
@@ -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),
diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts
index 04c428be98e..e0309bc8792 100644
--- a/server/src/services/agent-instruction-working-copies.ts
+++ b/server/src/services/agent-instruction-working-copies.ts
@@ -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 };
 }
diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts
index 50c293cd0e8..0ccd4499d77 100644
--- a/server/src/services/company-deletion.ts
+++ b/server/src/services/company-deletion.ts
@@ -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));
diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts
index e023e149690..ae7f4a22c36 100644
--- a/server/src/services/delivery-work-notifications.ts
+++ b/server/src/services/delivery-work-notifications.ts
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
 
 /** Topics for existing durable queues; register intent before their writes. */
 export const DELIVERY_QUEUES = {
+  instructionCollection: "instruction-collection",
   directoryCleanup: "directory-cleanup",
   instructionCommit: "instruction-commit",
   scheduledRetryCancellation: "scheduled-retry-cancellation",
diff --git a/server/src/services/execution-control-reconciliation.test.ts b/server/src/services/execution-control-reconciliation.test.ts
index 1b873eb4e1e..10923cd5bc1 100644
--- a/server/src/services/execution-control-reconciliation.test.ts
+++ b/server/src/services/execution-control-reconciliation.test.ts
@@ -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();
diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts
index dd82e02b56a..1c4ad3d0885 100644
--- a/server/src/services/execution-control-reconciliation.ts
+++ b/server/src/services/execution-control-reconciliation.ts
@@ -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)
diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts
index 91a832ce103..f56ad8b529c 100644
--- a/server/src/services/heartbeat.ts
+++ b/server/src/services/heartbeat.ts
@@ -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,
diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts
index aee0ca9e90e..f3c2366c219 100644
--- a/server/src/services/heartbeat/queue.ts
+++ b/server/src/services/heartbeat/queue.ts
@@ -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)
diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts
index 3ab69cbff7e..28fc3077f09 100644
--- a/server/src/services/heartbeat/recovery.ts
+++ b/server/src/services/heartbeat/recovery.ts
@@ -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 => {
diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts
index d93a57f8fb3..8729a0f2ab8 100644
--- a/server/src/services/heartbeat/run-lifecycle.ts
+++ b/server/src/services/heartbeat/run-lifecycle.ts
@@ -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 });
diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts
index bb6fd525c81..2d54950c51b 100644
--- a/server/src/services/instance-settings.ts
+++ b/server/src/services/instance-settings.ts
@@ -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, "*");
diff --git a/server/src/services/instruction-collection-notifications.ts b/server/src/services/instruction-collection-notifications.ts
new file mode 100644
index 00000000000..7ce151cad52
--- /dev/null
+++ b/server/src/services/instruction-collection-notifications.ts
@@ -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) : "*");
+}
diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts
index 9a3041c88d8..8252a2ecd56 100644
--- a/server/src/services/legacy-execution-recovery.ts
+++ b/server/src/services/legacy-execution-recovery.ts
@@ -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)
diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts
index 8964e24ed5a..9505af6f60f 100644
--- a/server/src/services/native-runtime/native-finalization-reconciler.ts
+++ b/server/src/services/native-runtime/native-finalization-reconciler.ts
@@ -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,
diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts
index a906c14c56e..1325a45d8c4 100644
--- a/server/src/services/native-runtime/native-restart-recovery.ts
+++ b/server/src/services/native-runtime/native-restart-recovery.ts
@@ -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({
diff --git a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts
index 5d1ed177987..a128f0bb2bf 100644
--- a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts
+++ b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts
@@ -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);
diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts
index 6d283970f1f..4fc1345e17d 100644
--- a/server/src/services/native-runtime/native-run-finalizer.ts
+++ b/server/src/services/native-runtime/native-run-finalizer.ts
@@ -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 });
diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts
index fa23f49cc08..d40029b85a4 100644
--- a/server/src/services/native-runtime/native-session-executor.ts
+++ b/server/src/services/native-runtime/native-session-executor.ts
@@ -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({
diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts
index b0d0913a8b3..157be7c0abe 100644
--- a/server/src/services/native-runtime/status-decision-committer.ts
+++ b/server/src/services/native-runtime/status-decision-committer.ts
@@ -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)
diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts
index 81c0ecb3347..e8f708ed602 100644
--- a/server/src/services/recovery/service.ts
+++ b/server/src/services/recovery/service.ts
@@ -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 });