From 82529741418e138d9d623bf13d2fa22c1ff79d36 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:13:57 -0500 Subject: [PATCH] fix(server): schedule stopped instruction collection (#17674) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Runs can stop while their private instruction edits still need collection. > - Existing copy rows and stop evidence describe that unfinished work. > - The orphan reaper still queries for it on every maintenance pass. > - This PR activates collection from committed copy and run transitions. > - It uses saved retry deadlines when stop proof or collection remains pending. > - Active copies and completed history no longer cause collection polling. ## Linked Issues or Issue Description Refs #17671 and #17670. Both prerequisites are merged into master. Captured saves and post-save release remain separate reasons with their own workers. **What needs to be improved?** `recoverStopped` queries instruction copies on every orphan-reaper pass, even when no terminal run owes collection. A run can become terminal without changing its copy row, so copy-only notifications would miss real work. **Why is this improvement important?** Repeated empty queries prevent idle tenant instances from becoming quiet. Removing them must preserve edits, stop proof, current ownership, and recovery after restart. **Proposed solution** Use the existing transaction signals, scoped deadline worker, and scheduler. Reconstruct eligible copies at startup, signal supported copy and terminal-run writers, and retain saved retry deadlines for unresolved work. Preserve the existing collection and directory owners. Related PR search found the warm agent-file ownership work (#14695), verified stop boundary (#14311), and task-workspace cleanup (#16464). This PR does not replace those contracts. ROADMAP.md has no duplicate migration. ## What Changed - Add scoped stopped-instruction collection with keyset reconstruction and fair batches of 20. - Signal before copy preparation, mutation and adoption writes, and before terminal run writes in their existing transactions. - Cover heartbeat, native finalization/restart/session recovery, execution-control, queue and dispatch producers. - Signal settings changes and company/agent deletion. Preserve drain, standby, restart and shutdown ownership. - Recheck terminal-run eligibility in the due-time claim. Keep terminal status separate from process-stop proof. - Retain existing stop checks, attempt limits, saved candidate bytes and remote-resource fencing. - Skip only this lane in periodic orphan recovery. Preserve manual/startup recovery. - Add concurrent, replay-safe partial index 0408 and retain the five newest snapshots. Add no table. - Record the owner and producer audit in `doc/plans/2026-10-25-stopped-instruction-collection-sleep-audit.md`. ## Verification - Source `cb2818a8f46ef961f744977a258f0e9042e31a42`: all 1469 cases across 14 complete files pass without skips in 141.69 seconds. This includes all 968 native-session cases, filesystem saves, startup, terminal-run producers, dispatch, queue, execution-control, finalization and lifecycle. - Eleven separate fresh PostgreSQL 17 databases migrated through 0408. The embedded-only lifecycle fixture temporarily used its own external database on this IPC-exhausted host and was restored byte for byte. Assertions and timing limits stayed unchanged. This qualifies external-database behavior; it does not claim an unmodified embedded CI run. - Database build, server typecheck/build, module boundaries and exact-current migration order pass. The original concurrent index SQL is unchanged. All prior executable migration history, journal entries and retained snapshot bytes are preserved; the five newest snapshots are 0404 through 0408. - A populated 0407-to0408 upgrade preserves all 406 prior migration receipts and 38 companies. It appends exactly one receipt. Replay adds none. The collection index is valid and ready on fresh and upgraded databases. - The normal merge of actual parent `267842fc3c6e6542deed8a1bcc2c78bec8df7862` preserves both complete patches, all independent sidebar changes from #18117, and the full accepted server/package/test trees. The complete sidebar test file passes all 36 cases on the combined tree. React act warnings remain. - Prior failures and fixture qualifications remain in the audit. Published head `d176d731f083c7edd2ede688cb901d1e66d8f1d8` has Greptile 5/5, all 57 checks passing or skipped, and no unresolved review threads. No Ponytail review is present. These local tests do not establish deployed host/database suspension or Cloud alarm delivery. ## Risks - Process-local signals cover instrumented writers. Independent SQL writers and additional replicas still need integration. - Terminal-run signals can cause one scoped inspection even if no instruction copy exists. They do not create a recurring scan. - Reconstruction and memory scale with actual pending candidates; active candidates are filtered by the run join. - Admitted file/provider operations can outlive a later drain request. Shutdown joins them. Uncertain ownership remains pending rather than being discarded. - The partial index adds storage and write cost. Its concurrent migration uses the existing interrupted-index repair path. - Cloud alarms, sleep eligibility and provider suspension are unchanged. Worker quiet is not a claim of deployed host/database sleep or measured savings. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, shell tools, and PostgreSQL/filesystem tests. The runtime does not expose a more precise model ID or context-window size for this turn. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 16 + ...pped-instruction-collection-sleep-audit.md | 276 ++++++++++++++++++ packages/db/src/client.ts | 1 + .../src/migrations/0408_short_red_skull.sql | 5 + ...{0403_snapshot.json => 0408_snapshot.json} | 261 ++++++++++++++++- packages/db/src/migrations/meta/_journal.json | 7 + .../src/schema/agent_instruction_revisions.ts | 5 + .../agent-directory-working-copies.test.ts | 26 +- .../agent-instruction-working-copies.test.ts | 211 +++++++++++++ .../native-finalization-recovery.test.ts | 17 +- .../server-startup-feedback-export.test.ts | 10 +- server/src/index.ts | 4 +- .../active-run-watchdog/adapters/postgres.ts | 2 + .../modules/run-dispatch/adapters/postgres.ts | 4 + .../adapters/queued-comment-postgres.ts | 2 + server/src/services/agent-deletion.ts | 2 + .../agent-directory-working-copies.ts | 15 +- .../agent-instruction-working-copies.ts | 169 +++++++---- server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + .../execution-control-reconciliation.test.ts | 15 +- .../execution-control-reconciliation.ts | 2 + server/src/services/heartbeat.ts | 8 + server/src/services/heartbeat/queue.ts | 7 + server/src/services/heartbeat/recovery.ts | 4 +- .../src/services/heartbeat/run-lifecycle.ts | 2 + server/src/services/instance-settings.ts | 2 + .../instruction-collection-notifications.ts | 7 + .../src/services/legacy-execution-recovery.ts | 5 +- .../native-finalization-reconciler.ts | 2 + .../native-runtime/native-restart-recovery.ts | 2 + .../native-run-finalizer-telemetry.test.ts | 29 +- .../native-runtime/native-run-finalizer.ts | 4 + .../native-runtime/native-session-executor.ts | 2 + .../status-decision-committer.ts | 2 + server/src/services/recovery/service.ts | 2 + 36 files changed, 1037 insertions(+), 94 deletions(-) create mode 100644 doc/plans/2026-10-25-stopped-instruction-collection-sleep-audit.md create mode 100644 packages/db/src/migrations/0408_short_red_skull.sql rename packages/db/src/migrations/meta/{0403_snapshot.json => 0408_snapshot.json} (99%) create mode 100644 server/src/services/instruction-collection-notifications.ts 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(resolve => { release = resolve; }); + const reached = new Promise(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 | undefined; + try { + const registration = copies.registerStoppedCollectionWork(worker, () => true, async () => true); + await reached; + let complete = false; + stopped = worker.stop().then(() => { complete = true; }); + await new Promise(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>; let db: ReturnType; @@ -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((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 { + 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 { + 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, patch: (row: Copy, values: Partial) => Promise, 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, requireDue = false) { + async function tryPatch(row: Copy, values: Partial, 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) { 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, + canRun: () => boolean, canRecover: () => Promise) { + 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> | 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; let tempDb: Awaited> | 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[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[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 });