From a8a038e659c50c1c5e3ce532d13072a29a5cf951 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 00:28:00 -0500 Subject: [PATCH] fix(server): schedule native pending effect recovery (#17690) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native status decisions store effect targets and a delivery ledger. > - Historical pending or failed effects still need recovery. > - The periodic finalizer queries that ledger even when work is complete. > - This pull request gives pending effects their own commit-driven worker and saved deadlines. > - Completed history becomes quiet while incomplete effects remain recoverable. ## Linked Issues or Issue Description Refs #17687 (merged). This is slice 47 of the sleep migration stack, now based on master. **What problem does this solve?** Periodic native finalization repeatedly checks effect delivery. A tenant with completed history should not perform recurring SQL for this duty. Historical unresolved effects must still recover without duplicating their targets. **Proposed solution** Reconstruct current pending effects at startup. Wake from committed coordinator changes. Use the saved retry and ownership deadlines. Keep the existing dispatcher as the only recovery mutation owner. **Alternatives considered** A slower unconditional scan still creates idle traffic. A new queue table duplicates existing durable state. Replaying materialization can duplicate effects. This change uses the existing ledger, transaction signals, and shared scheduler. ## What Changed - Add a scoped worker with 100-row discovery pages and fair 25-entry dispatch batches. Delivered, superseded, and terminal-failure history has no recurring scan. - Preserve exact company/issue/run/decision binding and validate materialized targets before settling effects. Recheck phase, saved result, retry, and lease under the coordinator lock. Missing targets retain recovery. - Notify from decision replay, accepted-result recovery, coordinator claim/retry/release, attention replay, and execution-control recovery. Only two DB-only producer writes need a new transaction wrapper. Nested transactions use the existing outer-commit boundary. - Exclude pending-effect candidates from periodic finalization so it cannot skip ahead of their owner. Keep explicit and startup recovery behavior compatible. Respect maintenance disablement, drain, standby, and shutdown. - Add concurrent, idempotent migration 0411 with partial indexes for pending effects and their current coordinator binding. Retain the five newest schema snapshots. - Document producers, query evidence, failure history, and scope in `doc/plans/2026-10-25-native-pending-effects-sleep-audit.md`. ## Verification - Source `ba38ee465df02b6026286b5d5e15ae49fe07a367` passes all 2284 distinct cases across 25 complete files, without skips. The main selection passes 2225 in 203.94 seconds; safe-replacement passes 59 in 26.03 seconds. Coverage includes all 968 native-session cases, the 94-case status-authority/pending-effect corpus, actual decision/lease producers, Stop, queue, interaction, reconstruction and Slack consumers. The audit gives per-file counts. - The complete 94-case corpus passes again on the same database in 16.03 seconds. Retained PSC/OUT rows and an unrelated sentinel remain unchanged. The repeat is not counted again as distinct coverage. - Twenty-one separate fresh PostgreSQL 17 databases migrate through 0411 before setup. Four embedded-only fixture imports temporarily use their own external database on this IPC-exhausted host and are restored byte for byte. Assertions and timing limits stay unchanged. This qualifies external-database behavior; it does not claim an unmodified embedded CI run. - Database build, server typecheck/build, module boundaries and exact-head migration order pass. Original runtime/schema edits and unrelated upstream blobs are preserved. Both original concurrent index statements, all prior executable history, journal entries and retained snapshot bytes are unchanged; five snapshots 0407 through 0411 remain. - Populated 0410-to0411 upgrade preserves all 409 prior migration receipts and 38 companies, appends one receipt and adds none on replay. Both indexes are valid and ready on fresh and upgraded databases. - Adopting the latest prepared prerequisite changes ancestry and parent audits only. All accepted core trees remain identical. The audit retains earlier failures and qualifications. Actual landed-parent ancestry is now reconciled; fresh published-head CI and exact-head Greptile 5/5 remain required. No deployed sleep, Cloud-wake or fleet-cost claim is made. - Current parent context: integration `5f7cd7bc589ad0169dfb5f33df611ead56510ca3` changes only the parent's GitHub observation bound and independent cleanup assertion. All production code, packages and 25 accepted test files remain unchanged. Both focused GitHub idle cases pass in 15.26 seconds and the complete six-case cleanup suite passes in 8.47 seconds on two fresh 0411 PostgreSQL fixtures. This does not claim the full parent workflow was rerun here. ## Risks - A missing producer can delay work until restart. The audit covers current supported writers. Independent SQL and other processes remain outside the local notification contract. - Recovery validates existing targets; it cannot recreate a missing target or silently mark it delivered. Such work retains an error and retry. - The indexes add write/storage cost and use the existing restart-safe concurrent migration runner. - Other finalization duties still poll until their separate migrations. This PR does not prove host/database suspension, extend Cloud alarms, or claim fleet savings. - The actual prerequisite squash is adopted through a normal merge. Its complete tree equals the reviewed parent, and the complete accepted child tree is preserved. Fresh published-head CI and review remain required. ## Model Used OpenAI Codex, based on GPT-6, with reasoning and code-execution tools. The exact serving model ID and context-window size are not exposed in this session. ## 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 Co-authored-by: Codex --- ...2026-10-24-recurring-server-sleep-stack.md | 53 ++++ ...0-25-native-pending-effects-sleep-audit.md | 276 ++++++++++++++++++ packages/db/src/client.ts | 1 + .../db/src/migrations/0411_real_scrambler.sql | 2 + ...{0406_snapshot.json => 0411_snapshot.json} | 154 +++++++++- packages/db/src/migrations/meta/_journal.json | 7 + .../db/src/schema/native_run_finalizations.ts | 3 + .../db/src/schema/status_decision_effects.ts | 5 + .../native-status-arbiter-corpus.test.ts | 185 +++++++++++- .../server-startup-feedback-export.test.ts | 10 +- server/src/index.ts | 4 +- .../services/delivery-work-notifications.ts | 1 + .../execution-control-reconciliation.ts | 2 + server/src/services/heartbeat.ts | 3 + server/src/services/heartbeat/recovery.ts | 3 +- .../native-pending-effect-notifications.ts | 8 + .../native-finalization-reconciler.ts | 4 + .../native-interaction-bridge.ts | 44 +-- .../native-pending-effect-work.ts | 71 +++++ .../native-run-coordinator-store.ts | 3 + .../native-runtime/native-run-finalizer.ts | 3 + .../native-runtime/native-session-executor.ts | 58 ++-- .../native-workspace-export-recovery.ts | 2 + .../native-workspace-export-retry.ts | 2 + .../status-decision-committer.ts | 23 ++ .../services/preserved-result-finalization.ts | 2 + 26 files changed, 877 insertions(+), 52 deletions(-) create mode 100644 doc/plans/2026-10-25-native-pending-effects-sleep-audit.md create mode 100644 packages/db/src/migrations/0411_real_scrambler.sql rename packages/db/src/migrations/meta/{0406_snapshot.json => 0411_snapshot.json} (99%) create mode 100644 server/src/services/native-pending-effect-notifications.ts create mode 100644 server/src/services/native-runtime/native-pending-effect-work.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 73a6a22c916..14a2777304e 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2662,3 +2662,56 @@ producer coverage, fixture failures, and the supported writer boundary. Workspac typechecks and build passed; remote acceptance remains outstanding. Origin/master was fetched at `6336c34041`; the full stack still needs final reconciliation with master. No PR in this stack has been merged, and the original remaining scope is unchanged. + +## Slice 46 published; slice 47 pending-effects audit + +Unsafe-export recovery is draft [#17687](https://github.com/paperclipai/paperclip/pull/17687) +at `b9fe9e2e40ab48104cf18822b5302a740d1957a7`, stacked on #17680. Workspace types +and build passed. Greptile reviewed that exact head at 5/5 with no actionable +findings; full remote CI remains active. This is not acceptance or merge proof. +The attempted full local `pnpm test:run` hit embedded PostgreSQL initialization +failures (15 native-session failures plus database-suite skips) and was stopped +with its log preserved at `/tmp/sleep46-full-tests.log`. It did not pass. The +626 focused external-database cases remain the local acceptance evidence. + +Slice 47 begins with native pending status effects. The current writer already +materializes targets and records delivered effects atomically; historical pending +or failed rows still require recovery. The existing owner validates targets and +settles the ledger without recreating effects. The audit and open producer/ +ordering questions are in [the pending-effects audit](2026-10-25-native-pending-effects-sleep-audit.md). +Its 72-case status-arbiter baseline passed. No source migration is implemented yet. + +Older gates were revalidated: #17490 and #17572 still have E2E shard 3 and aggregate +failures at their recorded heads. #17576 additionally reports an action-required +check. #17580 has successful/neutral CI, but its recorded review findings still +require resolution. These remain part of the full requested stack finish line. + + +## Slice 46 accepted; slice 47 locally verified + +#17687 at `b9fe9e2e40ab48104cf18822b5302a740d1957a7` has all 53 checks passing, +Greptile 5/5, no unresolved review threads, and a mergeable base. It remains +unmerged. Its full-local-suite PostgreSQL failure above remains part of the record. + +Slice 47 implements native pending-effect recovery with exact coordinator binding, +commit-aware producers, saved lease/retry deadlines, lock-time eligibility checks, +and pending-only indexed discovery. Periodic finalization skips this duty and +preserves ordering; explicit/startup recovery remains compatible. No provider turn, +new queue table, or host-sleep exemption is introduced. See the +[pending-effects audit](2026-10-25-native-pending-effects-sleep-audit.md). + +The expanded consumer run passed 680 tests across 15 files. The final producer +audit found a lease-shortening wake gap; its new real-store regression failed +before the fix and passed afterward. Final owner/startup tests pass 158 cases; +the finalizer rerun passes 29 on a fresh database (683 distinct local cases across +these runs). The initial finalizer rerun reused a fixed company prefix and failed +setup; that failure is preserved. Index discovery was measured with 10,000 delivered +effects and 10,000 completed coordinators. Migration replay and valid/ready checks +pass. Final workspace and remote gates must be recorded before acceptance. + +The remaining original scope, older CI/review issues, full master reconciliation, +and all final-head PR gates remain open. No PR has been merged by this stack work. + +Final slice 47 source passes workspace typecheck/build and 158 worker/startup +cases, including exact near-term concurrent deadlines. Remote publication and +acceptance are the next gates; local success does not satisfy the full stack. diff --git a/doc/plans/2026-10-25-native-pending-effects-sleep-audit.md b/doc/plans/2026-10-25-native-pending-effects-sleep-audit.md new file mode 100644 index 00000000000..83347984ffc --- /dev/null +++ b/doc/plans/2026-10-25-native-pending-effects-sleep-audit.md @@ -0,0 +1,276 @@ +# Native pending status-effect recovery + +Branch: `codex/sleep-47-native-pending-effects`. +Current published parent: #17687, `b28399babac72ba9db1c4a4ef882c5e87aded38c`. +Original draft parent: `b9fe9e2e40ab48104cf18822b5302a740d1957a7`. +Status: implemented and locally verified; remote PR acceptance remains required. + +## Duty and existing ownership + +The native finalization reconciler reads pending/failed `status_decision_effects` +for each candidate's current decision before other finalization work. When any +exist, it calls `dispatchPendingNativeStatusEffects` and skips the remaining +candidate processing for that pass. This slice moves only that duty. Response +presentation, evidence reassessment, and accepted-result/workspace settlement +remain separate migrations. + +The existing dispatcher owns the recovery transaction. It locks the coordinator, +checks the exact company/issue/run/decision binding, locks pending effects, and +checks that their materialized targets exist. It then marks the ledger delivered, +increments attempts, clears retry/error state, updates delivered-effect IDs, and +signals chat milestones. It does not recreate targets or arbitrate status again. +A missing target remains an error with an owned retry, never successful cleanup. + +## Source of truth and scheduling + +The only current production insert creates effect targets and delivered ledger +entries atomically in `commitNativeStatusDecision`. Both production effect-update +paths also settle to delivered. No current pending/failed producer was found. +Historical rows and the schema default still require recovery. Ordinary successful +commits therefore discover no pending-effects timer. + +The worker reconstructs existing pending/failed effects belonging to the exact +current coordinator decision. It requires native runtime, an existing issue, +a saved result, and a phase other than terminal_failure. Superseded decisions and +terminal-failure history remain dormant. Its deadline is the later of the saved +coordinator retry and a held lease's expiry. An owner without an expiry preserves +the old reconciler's immediately eligible behavior. The ledger's nextAttemptAt +was not an eligibility fence in the old dispatcher and this slice does not invent +one. Actual errors use the shared bounded retry backoff. + +Discovery uses keyset pages of 100. The existing scoped deadline helper coalesces +commit hints, handles dirty generations/drain/failures, and fairly dispatches 25 +entries per batch. Empty, delivered, and dormant history leaves no recurring SQL. +The dispatcher rechecks eligibility under its existing coordinator lock so a +concurrent lease or retry extension cannot be bypassed. + +Startup registration respects the existing maintenance toggle and task-drain, +shutdown, and warm-standby gates. Periodic finalization excludes pending-effect +candidates and retains a late pending-effect guard to preserve ordering. Explicit +and startup recovery retain the old dispatcher default. The remaining finalizer +can process a candidate after its effects settle; this slice does not remove its +other polling duties. + +## Producer audit and shared boundaries + +Notifications carry only company and optional run IDs. They register intent before +the write and use the existing outer-transaction/uncertain-outcome protocol. + +| Mutation owner | Why it matters | +| --- | --- | +| Status decision commit/replay | Can restore the current decision and phase; new effects already delivered remain quiet. | +| Pending-effect dispatcher/materializer | Retires an existing obligation and refreshes remembered deadlines. | +| Preserved-result finalization | Reactivates accepted-result history. | +| Workspace export retry and historical export recovery | Restore accepted-result eligibility. | +| Persisted-attention replay | Rebinds an existing decision and clears its lease/retry. Its DB-only write now has an explicit transaction. | +| Finalizer claim and retryable failure | Claim can clear a later retry; failure releases ownership and changes the next retry. | +| Coordinator-store claim/renewal and retry | An existing owner can shorten its lease; release can move eligibility earlier. | +| Retained-session cleanup success/failure | Releases the coordinator lease. | +| Native execution failure/release | Can leave a saved result eligible and release its lease. The final DB-only release now has an explicit transaction. | +| Abandoned execution-control recovery | Releases ownership of an already accepted result. | + +The audit enumerated 34 coordinator writes that change phase, result, decision, +lease, or retry fields, then inspected their guards. Provider execution claims, +restart recovery, session resumption, and stopped-execution checks reject saved +results or explicitly require resultId null. Terminal-failure writes disable +eligibility; safe replacement keeps that terminal phase. Retained cleanup claims +and normal session renewals only postpone eligibility, which is rechecked at the +remembered deadline. Claim/renewal paths that can shorten it are signaled. + +The accepted-result insertion cannot expose an existing pending decision without +a result: schema checks require decision -> assessment -> result, and replay of +an existing accepted result returns before replacing the coordinator. Runtime-mode +writes occur in initial native preparation or the signaled status-decision owner. +Deleting records removes work; a scheduled lookup drops a missing candidate. + +This remains the supported single-owner client notification model. Independent +SQL writes and other processes do not produce these local signals; startup can +reconstruct their durable records but this change does not add live cross-process +delivery. No new queue table, provider protocol, or host-sleep exemption is added. + +## Query and migration evidence + +Captured the actual worker SQL using the PostgreSQL client's debug hook. A fixture +added 10,000 delivered effects and 10,000 completed coordinators, with one eligible +pending effect. The committed fixture matters: an earlier rolled-back experiment +could not consistently use the newly built partial index under its old snapshot. + +| Discovery | Existing indexes | Pending-effect index only | Both new indexes | +| --- | ---: | ---: | ---: | +| Global startup | 2.247 ms | 1.080 ms | 0.036 ms | +| Company scope | 1.807 ms | 1.048 ms | 0.015 ms | + +Without the indexes, discovery filtered 10,809 delivered effects and scanned 10,353 +coordinators. The partial effect index selects pending/failed rows; the partial +coordinator index resolves their current company/issue/decision owner. Both plans +used index access for those relations after the change. This fixture establishes +history cost for this query, not universal tenant-size or fleet performance. + +Current migration 0411 uses concurrent IF NOT EXISTS index builds through the existing +nontransactional migration owner. Generated snapshots retain the newest five +(0407–0411). Fresh schema application, migration replay, direct SQL replay, and +valid/ready index checks pass. The first replay command failed before SQL because +the tsx evaluator rejected top-level await; the corrected promise-chain command +passed. Logs: `/tmp/sleep47-query-plan-committed.log`, +`/tmp/sleep47-migration-replay-v2.log`. + +## Verification and remaining gates + +The pre-change corpus passed 72 cases. Initial worker verification passed 86. +Two initial test failures compared epoch scheduler deadlines to Date objects; +the assertions were corrected and passed without weakening production behavior. +Worker/startup wiring passed 154 cases. The expanded owner and consumer selection +passed 680 cases across 15 files on fresh external PostgreSQL. Workspace types +and build passed. The later producer audit identified a shortened-lease signal +gap; its regression and final verification are recorded below before acceptance. + +Tests exercise delivered/empty/terminal history quiet, pending/failed recovery, +no duplicate decisions or targets, missing-target failure, future retry/lease +quiet, concurrent deadline extension, real replay, outer commit/rollback, real +lease release, company-scope isolation, drain/resume, restart reconstruction, +103-candidate paging, and periodic/manual owner ordering. Existing consumers cover +finalization, export recovery, execution control, lifecycle, and chat delivery. + +The full local suite is not claimed green. The parent attempt encountered embedded +PostgreSQL initialization failures and was intentionally stopped; that failure is +preserved in `/tmp/sleep46-full-tests.log`. This slice uses fresh external databases +for affected suites. Remote CI/review must pass at the final PR head. Actual host +or database suspension and the remaining recurring-server duties are not covered +by this worker's quiet proof. + +## Final local acceptance + +The real-store shortened-lease regression failed before the missing claim signal: +recovery waited beyond the old deadline. After notifying at the two claim owners, +the regression passes. A concurrently saved ten-second deadline now remains ten +seconds rather than being rounded up to the fallback retry. The saved lease-owner +check preserves SQL null semantics, including legacy empty-string owners. + +The final worker/startup selection passes 158 tests (94 corpus, 64 startup), +including all 22 cases added to the corpus by this slice. The finalizer selection +passes 29 tests on a fresh database. A first rerun reused the TLM company-prefix +fixture and failed setup with companies_issue_prefix_idx; it did not pass, and +its log remains `/tmp/sleep47-owner-final.log`. The fresh rerun is +`/tmp/sleep47-finalizer-final.log`. Combined with unaffected consumer cases from +the 680-case run, 683 distinct cases have passed. The final worker log is +`/tmp/sleep47-deadlines-final.log`. + +Final source verification: `pnpm typecheck` and `pnpm build` pass after the near-term +deadline repair (`/tmp/sleep47-deadlines-typecheck.log`, +`/tmp/sleep47-deadlines-build.log`). The final worker/startup run now passes 158 +cases, giving 683 distinct passing cases with the earlier unaffected consumers. + + +## Current migration411 integration and complete acceptance + +Only this slice's unlanded 0403 migration artifacts were deferred before a +normal merge of the verified current unsafe-export parent. No conflicts occur. +The complete original runtime/schema patch and every unrelated upstream blob +are preserved. The migration directory and client matched the exact parent +before generation. `pnpm db:generate` produced `0411_real_scrambler.sql`. Both +concurrent, idempotent index statements are byte-identical to the original. +All earlier SQL, journal entries and retained snapshot bytes remain unchanged, +with five snapshots 0407 through 0411. No new code comments are added. + +Source `ba38ee465df02b6026286b5d5e15ae49fe07a367` passes **2284 distinct +cases in 25 complete files**, without skips. The main selection passes 2225 +cases in 203.94 seconds and safe-replacement passes 59 in 26.03 seconds. +The complete 94-case status-authority corpus then passes again on the same +database in 16.03 seconds. The repeat is not double-counted. Three preexisting +PSC/OUT/sentinel company rows remain unchanged in their inspected ID/name/prefix +fields after both runs; each invocation appends two distinct UUID-derived +fixture prefixes. Current per-file counts: + +| Complete file | Passing cases | +| --- | ---: | +| src/__tests__/issue-recovery-actions.test.ts | 89 | +| src/__tests__/native-finalization-recovery.test.ts | 32 | +| src/__tests__/native-interaction-bridge.test.ts | 7 | +| src/__tests__/native-status-arbiter-corpus.test.ts | 94 | +| src/__tests__/native-workspace-sync.test.ts | 10 | +| src/__tests__/recovery-stale-issue-lock-sweep.test.ts | 40 | +| src/__tests__/server-startup-feedback-export.test.ts | 67 | +| src/__tests__/slack-conversation-lifecycle.test.ts | 41 | +| src/modules/run-dispatch/adapters/postgres.test.ts | 69 | +| src/modules/wake-queue/adapters/postgres.test.ts | 93 | +| src/modules/wake-queue/adapters/queued-comment-postgres.test.ts | 5 | +| src/services/execution-control-reconciliation.test.ts | 10 | +| src/services/explicit-native-continuation.test.ts | 335 | +| src/services/heartbeat/queue.test.ts | 32 | +| src/services/heartbeat/recovery.test.ts | 45 | +| src/services/heartbeat/run-control.test.ts | 51 | +| src/services/heartbeat/run-lifecycle.test.ts | 31 | +| src/services/heartbeat/upgrade-readiness.test.ts | 7 | +| src/services/legacy-execution-recovery.test.ts | 15 | +| src/services/native-runtime/native-restart-recovery.test.ts | 25 | +| src/services/native-runtime/native-run-finalizer-telemetry.test.ts | 29 | +| src/services/native-runtime/native-safe-replacement.test.ts | 59 | +| src/services/native-runtime/native-session-executor.test.ts | 968 | +| src/services/native-runtime/native-workspace-export-retry.test.ts | 80 | +| src/services/native-runtime/native-workspace-reconstruction.test.ts | 50 | + +Twenty-one separate fresh PostgreSQL 17 databases migrate through 0411 before +setup. Four embedded-only fixture imports temporarily use their own external +database on this IPC-exhausted host and are restored byte for byte: interaction +bridge, workspace reconstruction, Slack lifecycle and wake-queue PostgreSQL. +All assertions and timing limits remain unchanged. Other suites use their +existing supported external-database paths. This is qualified external-database +acceptance, not an unmodified embedded CI reproduction. Actual database build, +server types/build, module boundaries and exact-head migration order pass. + +A populated 0410-to0411 upgrade preserves all 409 prior migration receipts and +all 38 companies, appends exactly one receipt and appends none on replay. Both +partial indexes are valid and ready on fresh and upgraded databases. Earlier +failures above retain their original scope and causes. These checks verify +current worker and producer/consumer behavior, not deployed host/database +suspension, external Cloud wakes or fleet savings. + +Normal merge `6cc9e0f4a05ac7aee225b8ac7dce119c31799257` adopts prepared parent +`d64d307484633485716124b24ad8bbdbea61f399`, carrying the published automatic-review +ancestry and parent documentation. Complete server, package, UI and test trees +remain identical to the accepted source, including the whole 0411 database +tree. Publication stays held for the actual unsafe-export prerequisite merge +and ancestry reconciliation. Fresh published-head CI and reviews remain required. + + +## Inherited CI correction checked in native-effect context + +Normal integration `5f7cd7bc589ad0169dfb5f33df611ead56510ca3` carries the +upstream GitHub observation-bound correction and independent cleanup assertion. +Only those two previously unselected test files and parent audits change. +All production source, all packages including migration 0411, and all 25 files +in the accepted 2284-case selection remain byte-identical. That acceptance and +its 94-case same-database repeat retain their original source scope. + +Both focused native-presentation GitHub idle variants pass in 15.26 seconds. +All six cases in the complete cleanup-removal suite pass in 8.47 seconds. Two +separate fresh PostgreSQL 17 databases migrate through 0411; supported external +URLs require no fixture source adaptation. The other 1779 cases in the large +chat file are outside this focused selection. The full parent 274-case workflow +is not claimed as executed here. The parent audit preserves the original CI +failure and controlled comparison, including its limit on exact CI causality. +Actual prerequisite ancestry and fresh published-head CI/reviews remain required. + + +## Published unsafe-export parent + +Normal merge `01ff4da4457bf2afc3d2e5e65fa33c16d56a4e6f` adopts published prerequisite +`b28399babac72ba9db1c4a4ef882c5e87aded38c`, including the actual automatic-review +squash ancestry. Complete server, package, UI and test trees remain identical +to the accepted child. The whole 0411 database tree and all earlier acceptance +scopes remain unchanged. Publication remains held until the unsafe-export +parent lands and its actual ancestry is reconciled. Fresh current-head CI +and review are still required. + + +## Actual unsafe-export squash ancestry + +Unsafe-export prerequisite #17687 merged as +`4e715eb7ea56d49a099530c3dce59e61e698077c`, reviewed at +`b28399babac72ba9db1c4a4ef882c5e87aded38c`. Its entire squash tree equals +the reviewed parent, which was already an ancestor of this prepared child. +Normal merge `45a1f20c8830efb0d60794f01fd7a5013ff41198` resolves duplicate +squash ancestry while preserving the complete held child tree byte for byte. +All accepted source, all migrations through 0411, and all earlier validation +scopes remain unchanged. The only following edit records this proof. Fresh +published-head CI and exact-head Greptile 5/5 are still required. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 27dc4736d5a..6e562381f7c 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -560,6 +560,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0408_short_red_skull.sql", "0409_uneven_zuras.sql", "0410_freezing_kinsey_walden.sql", + "0411_real_scrambler.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0411_real_scrambler.sql b/packages/db/src/migrations/0411_real_scrambler.sql new file mode 100644 index 00000000000..b0a67f12121 --- /dev/null +++ b/packages/db/src/migrations/0411_real_scrambler.sql @@ -0,0 +1,2 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "native_run_finalizations_pending_effect_owner_idx" ON "native_run_finalizations" USING btree ("company_id","issue_id","decision_id") WHERE "native_run_finalizations"."decision_id" is not null AND "native_run_finalizations"."result_id" is not null AND "native_run_finalizations"."phase" <> 'terminal_failure';--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "status_decision_effects_pending_delivery_idx" ON "status_decision_effects" USING btree ("company_id","issue_id","decision_id") WHERE "status_decision_effects"."delivery_state" in ('pending', 'failed'); diff --git a/packages/db/src/migrations/meta/0406_snapshot.json b/packages/db/src/migrations/meta/0411_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0406_snapshot.json rename to packages/db/src/migrations/meta/0411_snapshot.json index 866cacb35c5..084f5255a1d 100644 --- a/packages/db/src/migrations/meta/0406_snapshot.json +++ b/packages/db/src/migrations/meta/0411_snapshot.json @@ -1,6 +1,6 @@ { - "id": "c34fcd91-0d57-4241-b8b4-399bc9f6dcb0", - "prevId": "c87ad360-f306-459b-b814-fde37327c6da", + "id": "ff56ee93-8cfc-465a-b26a-1c12d10773ff", + "prevId": "c02426e4-de1a-4ce3-a541-99bb970c272a", "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": [ @@ -33727,6 +33771,28 @@ } }, "indexes": { + "heartbeat_runs_cancelled_workspace_preservation_idx": { + "name": "heartbeat_runs_cancelled_workspace_preservation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" = 'cancelled'\n AND \"heartbeat_runs\".\"runner_profile_json\"->'nativeWorkspaceSync'->>'state' = 'prepared'\n AND \"heartbeat_runs\".\"result_json\"->'nativeCancellation'->>'dispatchState' = 'acknowledged'\n AND (\"heartbeat_runs\".\"result_json\"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'\n AND \"heartbeat_runs\".\"result_json\"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_retained_recovery_adoption_idx": { "name": "heartbeat_runs_retained_recovery_adoption_idx", "columns": [ @@ -43042,6 +43108,62 @@ } }, "indexes": { + "native_run_finalizations_pending_effect_owner_idx": { + "name": "native_run_finalizations_pending_effect_owner_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"decision_id\" is not null AND \"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"phase\" <> 'terminal_failure'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_unsafe_export_recovery_idx": { + "name": "native_run_finalizations_unsafe_export_recovery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"phase\" = 'terminal_failure' AND \"native_run_finalizations\".\"failure_code\" = 'native_workspace_sync_out_unsafe_archive'", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ @@ -52346,6 +52468,34 @@ } }, "indexes": { + "status_decision_effects_pending_delivery_idx": { + "name": "status_decision_effects_pending_delivery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"status_decision_effects\".\"delivery_state\" in ('pending', 'failed')", + "concurrently": false, + "method": "btree", + "with": {} + }, "status_decision_effects_decision_ordinal_uq": { "name": "status_decision_effects_decision_ordinal_uq", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 9647f3b918e..f02c1c52a93 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2864,6 +2864,13 @@ "when": 1793244017955, "tag": "0410_freezing_kinsey_walden", "breakpoints": true + }, + { + "idx": 411, + "version": "7", + "when": 1793246172969, + "tag": "0411_real_scrambler", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/native_run_finalizations.ts b/packages/db/src/schema/native_run_finalizations.ts index 874c27393dc..aaadc6fc97c 100644 --- a/packages/db/src/schema/native_run_finalizations.ts +++ b/packages/db/src/schema/native_run_finalizations.ts @@ -50,6 +50,9 @@ export const nativeRunFinalizations = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + pendingEffectOwnerIdx: index("native_run_finalizations_pending_effect_owner_idx") + .on(table.companyId, table.issueId, table.decisionId) + .where(sql`${table.decisionId} is not null AND ${table.resultId} is not null AND ${table.phase} <> 'terminal_failure'`), unsafeExportRecoveryIdx: index("native_run_finalizations_unsafe_export_recovery_idx") .on(table.companyId, table.issueId, table.runId) .where(sql`${table.phase} = 'terminal_failure' AND ${table.failureCode} = 'native_workspace_sync_out_unsafe_archive'`), diff --git a/packages/db/src/schema/status_decision_effects.ts b/packages/db/src/schema/status_decision_effects.ts index 989c70d1818..ff89d1259db 100644 --- a/packages/db/src/schema/status_decision_effects.ts +++ b/packages/db/src/schema/status_decision_effects.ts @@ -1,5 +1,7 @@ +import { sql } from "drizzle-orm"; import { foreignKey, + index, pgTable, uuid, text, @@ -34,6 +36,9 @@ export const statusDecisionEffects = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + pendingDeliveryIdx: index("status_decision_effects_pending_delivery_idx") + .on(table.companyId, table.issueId, table.decisionId) + .where(sql`${table.deliveryState} in ('pending', 'failed')`), issueCompanyFk: foreignKey({ columns: [table.companyId, table.issueId], foreignColumns: [issues.companyId, issues.id], diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 8fea007c2e5..cf330f4105a 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -1,3 +1,5 @@ +import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; +import { registerNativePendingEffectWork } from "../services/native-runtime/native-pending-effect-work.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { dismissAutomaticCompletionReviews, registerAutomaticCompletionReviewWork } from "../services/native-runtime/automatic-completion-reviews.js"; @@ -417,7 +419,8 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); - async function seedFixture(fixture: Fixture) { + async function seedFixture(fixture: Fixture, tenant = { companyId, agentId }) { + const { companyId, agentId } = tenant; const issueId = randomUUID(); const runId = randomUUID(); const contractId = randomUUID(); @@ -2050,6 +2053,186 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } + describe("native pending-effect scheduling", () => { + let coordinator: ReturnType | undefined; + let admitted = true; + const errors: unknown[] = []; + beforeEach(async () => { + await db.update(statusDecisionEffects).set({ deliveryState: "delivered" }).where(eq(statusDecisionEffects.companyId, companyId)); + admitted = true; errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); expect(errors).toEqual([]); + }); + async function seedEffects(state = "pending", tenant = { companyId, agentId }) { + const { companyId } = tenant; + const template = corpus.fixtures.find(candidate => candidate.mode === "native")!; + const f = await seedFixture({ ...template, id: `pending-effects-${randomUUID()}`, + given: { ...template.given, priorIssueStatus: "in_progress", completionState: "pending_effect_recovery" } }, tenant); + const commitInput: Parameters[0] = { db, companyId, issueId: f.issueId, runId: f.runId, + assessmentId: f.assessmentId, priorStatus: "in_progress", priorStatusVersion: 0, priorDecisionId: null, + decision: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "preserve", toStatus: "in_progress", + reasonCode: "live_continuation_registered", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] } }; + const committed = await commitNativeStatusDecision(commitInput); + const [effect] = await db.update(statusDecisionEffects).set({ deliveryState: state }) + .where(eq(statusDecisionEffects.decisionId, committed.decision.id)).returning(); + return { ...f, decisionId: committed.decision.id, effect, commitInput }; + } + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + await registerNativePendingEffectWork(db, coordinator, () => admitted).ready; + } + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 15_000 }); + } + async function quiet(milliseconds = 120_000) { + const read = vi.spyOn(db, "select"), write = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(milliseconds); expect(read).not.toHaveBeenCalled(); expect(write).not.toHaveBeenCalled(); } + finally { read.mockRestore(); write.mockRestore(); } + } + const effect = async (id: string) => (await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.id, id)))[0]; + it.each(["empty", "delivered"])("stays quiet with %s history", async state => { + if (state === "delivered") await seedEffects(state); + await start(); await flush(); await quiet(); + }); + it.each(["pending", "failed"])("recovers %s effects without materializing another target or decision", async state => { + const f = await seedEffects(state); + await start(); await flush(); + expect(await effect(f.effect.id)).toMatchObject({ deliveryState: "delivered", attemptCount: f.effect.attemptCount + 1 }); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, f.issueId))).toHaveLength(1); + expect(await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.issueId, f.issueId))).toHaveLength(1); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_progress"); + await quiet(); + }); + it("leaves terminal-failure history dormant", async () => { + const f = await seedEffects(); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); await flush(); expect((await effect(f.effect.id)).deliveryState).toBe("pending"); await quiet(); + }); + it.each(["retry", "lease", "empty lease owner"])("honors the saved %s deadline without polling before it", async kind => { + const f = await seedEffects(); const due = new Date(Date.now() + 180_000); + await db.update(nativeRunFinalizations).set(kind === "retry" ? { nextAttemptAt: due } : { leaseOwner: kind === "empty lease owner" ? "" : "current-owner", leaseExpiresAt: due }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); expect(coordinator!.nextWakeAt()).toBe(due.getTime()); + await quiet(120_000); expect((await effect(f.effect.id)).deliveryState).toBe("pending"); + await vi.advanceTimersByTimeAsync(60_000); await flush(); + expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + it.each([["retry", 180_000], ["lease", 180_000], ["retry", 10_000]] as const)("rechecks a concurrent %s extension of %i ms under the coordinator lock", async (kind, delay) => { + const f = await seedEffects(); + const due = new Date(Date.now() + delay); + const original = db.transaction.bind(db); + const transaction = vi.spyOn(db, "transaction").mockImplementationOnce(async (callback, config) => { + await db.update(nativeRunFinalizations).set(kind === "retry" + ? { nextAttemptAt: due } : { leaseOwner: "new-owner", leaseExpiresAt: due }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + return original(callback, config); + }); + await start(); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(due.getTime())); + transaction.mockRestore(); + expect((await effect(f.effect.id)).deliveryState).toBe("pending"); + await quiet(delay / 2); + await vi.advanceTimersByTimeAsync(delay / 2); await flush(); + expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + it("leaves pending effects to their owner during periodic finalization and preserves explicit recovery", async () => { + const f = await seedEffects(); + expect(await reconcileNativeFinalizations(db, [f.runId], { skipNativePendingEffects: true })).toEqual([]); + expect((await effect(f.effect.id)).deliveryState).toBe("pending"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, f.issueId))).toHaveLength(1); + await reconcileNativeFinalizations(db, [f.runId]); + expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, f.issueId))).toHaveLength(1); + }); + it("limits a committed run wake to its owner and recovers other durable work on restart", async () => { + const local = await seedEffects(); + const other = await seedEffects("pending", { companyId: outsideCompanyId, agentId: outsideAgentId }); + for (const f of [local, other]) await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); await flush(); + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, other.runId)); + await commitNativeStatusDecision(local.commitInput); await flush(); + expect((await effect(local.effect.id)).deliveryState).toBe("delivered"); + expect((await effect(other.effect.id)).deliveryState).toBe("pending"); await quiet(); + await coordinator!.stop(); coordinator = undefined; + await start(); await flush(); + expect((await effect(other.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + it("drains more than one discovery page and stops querying delivered history", async () => { + const fixtures = []; + for (let index = 0; index < 103; index++) fixtures.push(await seedEffects()); + await start(); await flush(); + for (const f of fixtures) expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); + await quiet(); + }, 60_000); + it("preserves failed target verification and retries only that obligation", async () => { + const f = await seedEffects(); + await db.update(statusDecisionEffects).set({ targetId: randomUUID() }).where(eq(statusDecisionEffects.id, f.effect.id)); + await start(); await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + expect((await effect(f.effect.id)).deliveryState).toBe("pending"); errors.length = 0; + await db.update(statusDecisionEffects).set({ targetId: f.effect.targetId }).where(eq(statusDecisionEffects.id, f.effect.id)); + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + it("reactivates dormant history when the real decision replay restores its binding", async () => { + const f = await seedEffects(); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); await flush(); await quiet(); + expect((await commitNativeStatusDecision(f.commitInput)).replayed).toBe(true); + await flush(); expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + it.each(["commit", "rollback"] as const)("defers decision reactivation until the outer transaction %s", async outcome => { + const f = await seedEffects(); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); await flush(); + const writing = db.transaction(async tx => { + await commitNativeStatusDecision({ ...f.commitInput, db: tx as unknown as typeof db }); + await vi.advanceTimersByTimeAsync(1000); + expect((await effect(f.effect.id)).deliveryState).toBe("pending"); + if (outcome === "rollback") throw new Error("roll back decision replay"); + }); + if (outcome === "rollback") await expect(writing).rejects.toThrow("roll back decision replay"); + else await writing; + await flush(); expect((await effect(f.effect.id)).deliveryState).toBe(outcome === "rollback" ? "pending" : "delivered"); await quiet(); + }); + it.each(["releases", "shortens"])("moves a future ownership deadline earlier when the real store %s its lease", async change => { + const f = await seedEffects(); + const due = new Date(Date.now() + 300_000); + await db.update(nativeRunFinalizations).set({ leaseOwner: "release-owner", leaseExpiresAt: due }).where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); expect(coordinator!.nextWakeAt()).toBe(due.getTime()); await quiet(); + const store = new NativeRunCoordinatorStore(db, { companyId, issueId: f.issueId, runId: f.runId, agentId, + normalizedSessionId: "pending-effects", runnerSourceInstanceId: "pending-effects", completionContractId: f.contractId!, + completionContractSha256: "test", completionContractRevision: "test", completionContractCriterionIds: [] }); + if (change === "shortens") await store.claimFinalization({ leaseOwner: "release-owner", leaseTtlMs: 1000 }); + else await store.markFinalizationRetry({ leaseOwner: "release-owner", failureCode: "temporary_failure", retryAfterMs: 1000 }); + await flush(); expect(Date.now()).toBeLessThan(due.getTime()); + expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + it("retains a first reactivation arriving during drain without passive reads", async () => { + const f = await seedEffects(); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await start(); await flush(); admitted = false; + await commitNativeStatusDecision(f.commitInput); await quiet(); + expect((await effect(f.effect.id)).deliveryState).toBe("pending"); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + it("retains startup work through drain and reconstructs an owed retry after shutdown", async () => { + const f = await seedEffects(); admitted = false; + await start(); await quiet(); expect((await effect(f.effect.id)).deliveryState).toBe("pending"); + await coordinator!.stop(); coordinator = undefined; + admitted = true; await start(); await flush(); + expect((await effect(f.effect.id)).deliveryState).toBe("delivered"); await quiet(); + }); + }); + describe.each(["obsolete policy", "automatic completion"] as const)("%s review scheduling", kind => { let coordinator: ReturnType | undefined; let admitted = true; diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 95afd9989f9..62bc485a0c2 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -116,6 +116,7 @@ const { 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() })), + registerNativePendingEffectWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerUnsafeWorkspaceExportWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerAutomaticCompletionReviewWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerObsoletePolicyReviewWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -557,6 +558,13 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers native pending effects only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerNativePendingEffectWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerNativePendingEffectWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers unsafe export recovery only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -997,7 +1005,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, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index ea4cee55de3..719ea5ee7c6 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.registerNativePendingEffectWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerUnsafeWorkspaceExportWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerAutomaticCompletionReviewWork(deliveryWork, @@ -1962,7 +1964,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, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 50152f2c127..aa160fed3dc 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 = { + nativePendingEffects: "native-pending-effects", unsafeWorkspaceExportRecovery: "unsafe-workspace-export-recovery", automaticCompletionReviews: "automatic-completion-reviews", obsoletePolicyReviews: "obsolete-policy-reviews", diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index b423f8205d8..7c0f0c4eb18 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyNativePendingEffects } from "./native-pending-effect-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; @@ -146,6 +147,7 @@ export async function reconcileAbandonedExecutionControl( } // A persisted result belongs to the existing finalizer, not a replacement provider. if (coordinator?.resultId) { + await notifyNativePendingEffects(tx, { companyId: run.companyId, runId: run.id }); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 1bcecebabd9..7d4630a41f2 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerNativePendingEffectWork } from "./native-runtime/native-pending-effect-work.js"; import { registerUnsafeWorkspaceExportWork } from "./native-runtime/native-workspace-export-recovery.js"; import { registerAutomaticCompletionReviewWork } from "./native-runtime/automatic-completion-reviews.js"; import { registerObsoletePolicyReviewWork } from "./native-runtime/obsolete-policy-reviews.js"; @@ -4088,6 +4089,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativePendingEffectWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerNativePendingEffectWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerUnsafeWorkspaceExportWork: (coordinator: Parameters[1], canRun: () => boolean) => registerUnsafeWorkspaceExportWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerAutomaticCompletionReviewWork: (coordinator: Parameters[1], canRun: () => boolean) => diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 335532eafa1..2e1b8675580 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1947,7 +1947,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1968,6 +1968,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews, skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews, skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports, + skipNativePendingEffects: opts?.skipNativePendingEffects, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/native-pending-effect-notifications.ts b/server/src/services/native-pending-effect-notifications.ts new file mode 100644 index 00000000000..e9c28d79782 --- /dev/null +++ b/server/src/services/native-pending-effect-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativePendingEffectScope = { companyId: string; runId?: string }; + +export async function notifyNativePendingEffects(transaction: object, scope: NativePendingEffectScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativePendingEffects, + JSON.stringify({ companyId: scope.companyId, runId: scope.runId })); +} diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index bfab2c0025a..4797142552e 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 { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; @@ -642,6 +643,7 @@ type NativeReconciliationOptions = { skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; + skipNativePendingEffects?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -776,6 +778,7 @@ async function reconcileNativeFinalizationsPass( eq(heartbeatRuns.runtimeMode, "native"), isNotNull(nativeRunFinalizations.resultId), notInArray(nativeRunFinalizations.phase, ["terminal_failure"]), + options.skipNativePendingEffects ? not(pendingNativeStatusEffectsCondition(db)) : undefined, or( isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, new Date()), @@ -805,6 +808,7 @@ async function reconcileNativeFinalizationsPass( )) : []; if (row.decisionId && pendingEffects.length > 0) { + if (options.skipNativePendingEffects) continue; const decision = resolveNativeReconciliationStatus({ facts: { undeliveredEffectCount: pendingEffects.length }, priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index abe854fad6b..680ab867033 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -1,3 +1,4 @@ +import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import type { Db } from "@paperclipai/db"; import type { NativeInteractionResponseEnvelope } from "../../vendor/paperclip-runner/index.js"; @@ -631,26 +632,29 @@ export async function routePersistedNativeResultAttention(input: { const effects = Array.isArray(decisionJson.effects) ? decisionJson.effects.map(record) : []; const delegation = effects.find((effect) => effect.kind === "create_delegated_issue"); const finalizationError = effects.find((effect) => effect.kind === "record_finalization_error"); - await input.db.update(nativeRunFinalizations).set({ - phase: finalizationError ? "retryable_failure" : "committed", - assessmentId: existingAssessment.id, - decisionId: existingDecision.id, - leaseOwner: null, - leaseExpiresAt: null, - recoveryState: null, - recoveryRequestId: null, - failureCode: nonEmptyText(finalizationError?.cause), - failureDetail: finalizationError ? { - originalFailureCode: finalizationError.cause, - nextAction: finalizationError.nextAction, - } : null, - nextAttemptAt: finalizationError ? new Date(Date.now() + 30_000) : null, - updatedAt: new Date(), - }).where(and( - eq(nativeRunFinalizations.runId, input.runId), - eq(nativeRunFinalizations.companyId, binding.companyId), - eq(nativeRunFinalizations.issueId, binding.issueId), - )); + await input.db.transaction(async tx => { + await notifyNativePendingEffects(tx, { companyId: binding.companyId, runId: input.runId }); + await tx.update(nativeRunFinalizations).set({ + phase: finalizationError ? "retryable_failure" : "committed", + assessmentId: existingAssessment.id, + decisionId: existingDecision.id, + leaseOwner: null, + leaseExpiresAt: null, + recoveryState: null, + recoveryRequestId: null, + failureCode: nonEmptyText(finalizationError?.cause), + failureDetail: finalizationError ? { + originalFailureCode: finalizationError.cause, + nextAction: finalizationError.nextAction, + } : null, + nextAttemptAt: finalizationError ? new Date(Date.now() + 30_000) : null, + updatedAt: new Date(), + }).where(and( + eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.companyId, binding.companyId), + eq(nativeRunFinalizations.issueId, binding.issueId), + )); + }); receipts.push({ requestId, route: candidate.route, diff --git a/server/src/services/native-runtime/native-pending-effect-work.ts b/server/src/services/native-runtime/native-pending-effect-work.ts new file mode 100644 index 00000000000..5893f2eadf3 --- /dev/null +++ b/server/src/services/native-runtime/native-pending-effect-work.ts @@ -0,0 +1,71 @@ +import { and, asc, eq, exists, gt, inArray, isNotNull, ne } from "drizzle-orm"; +import { heartbeatRuns, issues, nativeRunFinalizations, statusDecisionEffects, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativePendingEffectScope } from "../native-pending-effect-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { dispatchPendingNativeStatusEffects, nativePendingEffectDueAt } from "./status-decision-committer.js"; + +export function pendingNativeStatusEffectsCondition(db: Db) { + return exists(db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, nativeRunFinalizations.companyId), + eq(statusDecisionEffects.issueId, nativeRunFinalizations.issueId), + eq(statusDecisionEffects.decisionId, nativeRunFinalizations.decisionId), + inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + ))); +} + +function findPendingEffects(db: Db, scope?: NativePendingEffectScope, after?: string) { + return db.select({ + id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, + issueId: nativeRunFinalizations.issueId, decisionId: nativeRunFinalizations.decisionId, + phase: nativeRunFinalizations.phase, resultId: nativeRunFinalizations.resultId, + nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId))) + .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) + .where(and(eq(heartbeatRuns.runtimeMode, "native"), isNotNull(nativeRunFinalizations.resultId), + isNotNull(nativeRunFinalizations.decisionId), ne(nativeRunFinalizations.phase, "terminal_failure"), + scope ? and(eq(nativeRunFinalizations.companyId, scope.companyId), scope.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined) : undefined, + after ? gt(nativeRunFinalizations.runId, after) : undefined, + pendingNativeStatusEffectsCondition(db), + )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); +} + +export function registerNativePendingEffectWork(db: Db, coordinator: ReturnType, canRun: () => boolean) { + type Entry = { id: string; companyId: string; dueAt: number }; + async function load(scope?: NativePendingEffectScope, signal?: AbortSignal): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await findPendingEffects(db, scope, after); + for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, dueAt: nativePendingEffectDueAt(row)! }); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].id; + } + throw new Error("Native pending effect discovery stopped"); + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativePendingEffects, batchSize: 25, + decodeScope: (hint: string): NativePendingEffectScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativePendingEffectScope) => !scope || entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId), + load, + check: async (entry: Entry, signal?: AbortSignal) => { + if (!canRun() || signal?.aborted) throw new Error("Native pending effect recovery stopped"); + const scope = { companyId: entry.companyId, runId: entry.id }; + const [current] = await findPendingEffects(db, scope); + if (!current) return null; + const dueAt = nativePendingEffectDueAt(current)!; + if (dueAt > Date.now()) return { ...entry, dueAt }; + if (!canRun() || signal?.aborted) throw new Error("Native pending effect recovery stopped"); + await dispatchPendingNativeStatusEffects({ db, companyId: current.companyId, issueId: current.issueId, + runId: current.id, decisionId: current.decisionId!, onlyWhenDue: true }); + const [pending] = await findPendingEffects(db, scope); + if (!pending) return null; + const nextDueAt = nativePendingEffectDueAt(pending)!; + return { ...entry, dueAt: nextDueAt > Date.now() ? nextDueAt : Date.now() + 30_000 }; + }, + }, canRun); +} diff --git a/server/src/services/native-runtime/native-run-coordinator-store.ts b/server/src/services/native-runtime/native-run-coordinator-store.ts index de6955c3c44..593f5c0b4b2 100644 --- a/server/src/services/native-runtime/native-run-coordinator-store.ts +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -1,3 +1,4 @@ +import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { isSafeNativeChatProgressEventType } from "../safe-native-chat-progress.js"; import { notifyChatNativeProgress } from "../chat-native-progress-notifications.js"; @@ -522,6 +523,7 @@ export class NativeRunCoordinatorStore { ) { throw new Error("native_finalization_lease_conflict"); } + if (row.decisionId) await notifyNativePendingEffects(tx, this.#binding); if ( row.leaseOwner === input.leaseOwner && row.leaseExpiresAt && @@ -564,6 +566,7 @@ export class NativeRunCoordinatorStore { throw new Error("native_finalization_retry_invalid"); } const [updated] = await this.#db.transaction(async tx => { + await notifyNativePendingEffects(tx, this.#binding); if (input.failureCode === "native_workspace_sync_out_unsafe_archive") { await notifyUnsafeWorkspaceExportRecovery(tx, this.#binding); } diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index ed1c10648ce..f6b93c9699f 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 { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -214,6 +215,7 @@ async function claimCoordinator(input: { coordinator.leaseOwner !== leaseOwner ) throw new Error("native_finalization_lease_busy"); + if (coordinator.decisionId) await notifyNativePendingEffects(tx, coordinator); const [updated] = await tx .update(nativeRunFinalizations) .set({ @@ -381,6 +383,7 @@ async function recordRetryableFailure(input: { ? "native_workspace_sync_out_retry_exhausted" : "native_finalization_retry_exhausted" : input.failureCode; + await notifyNativePendingEffects(tx, { companyId: input.run.companyId, runId: input.run.id }); if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 7ec1a87225e..c854d1a2e45 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 { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; @@ -3897,6 +3898,7 @@ export async function reconcileRetainedNativeSessionCleanup( renameSync(owned.root, archive); } renameSync(copy, owned.root); + await notifyNativePendingEffects(tx, { companyId: owned.run.companyId, runId: owned.run.id }); await tx .update(nativeRunFinalizations) .set({ @@ -3954,6 +3956,7 @@ export async function reconcileRetainedNativeSessionCleanup( .limit(1) .then((rows) => rows[0]); if (!current || current.leaseOwner !== leaseOwner) return; + await notifyNativePendingEffects(tx, { companyId: owned.run.companyId, runId: owned.run.id }); await tx .update(nativeRunFinalizations) .set({ @@ -9697,6 +9700,7 @@ async function executePaperclipNativeSessionWithinScope( ) .for("update"); await notifyExecutionRunWork(tx); + await notifyNativePendingEffects(tx, input.execution.binding); const updated = await tx .update(nativeRunFinalizations) .set({ @@ -10031,33 +10035,35 @@ async function executePaperclipNativeSessionWithinScope( } } const releaseNow = new Date(); - const released = await timeController("coordinator_release", () => input.db - .update(nativeRunFinalizations) - .set({ - leaseOwner: null, - leaseExpiresAt: null, - recoveryState: null, - recoveryRequestId: null, - updatedAt: releaseNow, - }) - .where( - and( - eq(nativeRunFinalizations.runId, input.execution.binding.runId), - eq(nativeRunFinalizations.companyId, input.execution.binding.companyId), - eq(nativeRunFinalizations.issueId, input.execution.binding.issueId), - eq(nativeRunFinalizations.leaseOwner, leaseOwner), - eq(nativeRunFinalizations.attempt, attempt), - eq(nativeRunFinalizations.controllerBootId, controller.bootId), - eq(nativeRunFinalizations.controllerPid, controller.pid), - eq( - nativeRunFinalizations.controllerProcessStartedAt, - controller.processStartedAt, + const released = await timeController("coordinator_release", () => input.db.transaction(async tx => { + await notifyNativePendingEffects(tx, input.execution.binding); + return tx.update(nativeRunFinalizations) + .set({ + leaseOwner: null, + leaseExpiresAt: null, + recoveryState: null, + recoveryRequestId: null, + updatedAt: releaseNow, + }) + .where( + and( + eq(nativeRunFinalizations.runId, input.execution.binding.runId), + eq(nativeRunFinalizations.companyId, input.execution.binding.companyId), + eq(nativeRunFinalizations.issueId, input.execution.binding.issueId), + eq(nativeRunFinalizations.leaseOwner, leaseOwner), + eq(nativeRunFinalizations.attempt, attempt), + eq(nativeRunFinalizations.controllerBootId, controller.bootId), + eq(nativeRunFinalizations.controllerPid, controller.pid), + eq( + nativeRunFinalizations.controllerProcessStartedAt, + controller.processStartedAt, + ), + gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), ), - gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), - ), - ) - .returning({ runId: nativeRunFinalizations.runId }) - .then((rows) => rows[0] ?? null)); + ) + .returning({ runId: nativeRunFinalizations.runId }) + .then((rows) => rows[0] ?? null); + })); if (!released) throw new Error("native_session_lease_lost"); const finalization: NativeFinalizationResult = { schema: "paperclip.native-finalization.v1", diff --git a/server/src/services/native-runtime/native-workspace-export-recovery.ts b/server/src/services/native-runtime/native-workspace-export-recovery.ts index e64bc4037c4..8974e68a6e6 100644 --- a/server/src/services/native-runtime/native-workspace-export-recovery.ts +++ b/server/src/services/native-runtime/native-workspace-export-recovery.ts @@ -1,3 +1,4 @@ +import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; import { alias } from "drizzle-orm/pg-core"; import { notifyUnsafeWorkspaceExportRecovery, type UnsafeWorkspaceExportScope } from "../unsafe-workspace-export-notifications.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; @@ -100,6 +101,7 @@ async function recoverCandidate(db: Db, candidate: { runId: string; companyId: s phase: "workspace_finalize", status: "succeeded", exitCode: 0, finishedAt: now, metadata: { owningService: "native_workspace_finalizer", workspaceSync: { omitted: true, reason: "restore_unsafe_archive", legacy: true } }, }); + await notifyNativePendingEffects(tx, { companyId: run.companyId, runId: run.id }); await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, failureDetail: { workspaceFinalizeAttempt: 0, legacyUnsafeExportOmitted: { resultId: result.id, recoveredAt: now.toISOString() } }, updatedAt: now, }).where(eq(nativeRunFinalizations.runId, run.id)); diff --git a/server/src/services/native-runtime/native-workspace-export-retry.ts b/server/src/services/native-runtime/native-workspace-export-retry.ts index 4c1fe7c8474..69485cb4d74 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.ts @@ -1,3 +1,4 @@ +import { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { randomUUID } from "node:crypto"; import type { Environment, EnvironmentLease } from "@paperclipai/shared"; @@ -134,6 +135,7 @@ export async function retryNativeWorkspaceExport(input: { const now = new Date(); const retry = { ...receipt, requestId: randomUUID(), actorId: input.actorId, repairNote: input.repairNote, requestedAt: now.toISOString(), stoppedProvider: initial.lease.metadata?.remoteExecutionTermination }; + await notifyNativePendingEffects(tx, { companyId: current.run.companyId, runId: current.run.id }); await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null, failureDetail: { ...current.coordinator.failureDetail, workspaceFinalizeAttempt: 0, workspaceExportRetry: retry }, updatedAt: now }).where(eq(nativeRunFinalizations.runId, current.run.id)); await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 1606994752e..751734d70fb 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 { notifyNativePendingEffects } from "../native-pending-effect-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js"; import { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js"; @@ -80,6 +81,15 @@ export class NativeStatusRaceError extends Error { } } +export function nativePendingEffectDueAt(coordinator: { + phase: string; resultId: string | null; nextAttemptAt: Date | null; + leaseOwner: string | null; leaseExpiresAt: Date | null; +}): number | null { + if (coordinator.phase === "terminal_failure" || coordinator.resultId === null) return null; + return Math.max(coordinator.nextAttemptAt?.getTime() ?? 0, + coordinator.leaseOwner !== null ? coordinator.leaseExpiresAt?.getTime() ?? 0 : 0); +} + /** Reconciliation delivery resumes ledger work only; it never re-arbitrates. */ export async function dispatchPendingNativeStatusEffects(input: { db: Db; @@ -87,11 +97,17 @@ export async function dispatchPendingNativeStatusEffects(input: { issueId: string; decisionId: string; runId: string; + onlyWhenDue?: boolean; }) { return input.db.transaction(async (tx) => { const coordinator = await tx .select({ decisionId: nativeRunFinalizations.decisionId, + phase: nativeRunFinalizations.phase, + resultId: nativeRunFinalizations.resultId, + nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + leaseOwner: nativeRunFinalizations.leaseOwner, + leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, }) .from(nativeRunFinalizations) .where( @@ -107,6 +123,10 @@ export async function dispatchPendingNativeStatusEffects(input: { if (!coordinator || coordinator.decisionId !== input.decisionId) { throw new Error("native_pending_effect_decision_binding_mismatch"); } + if (input.onlyWhenDue) { + const dueAt = nativePendingEffectDueAt(coordinator); + if (dueAt === null || dueAt > Date.now()) return []; + } const pending = await tx .select({ id: statusDecisionEffects.id, @@ -134,6 +154,7 @@ export async function dispatchPendingNativeStatusEffects(input: { targetId: effect.targetId, }); } + await notifyNativePendingEffects(tx, input); await notifyChatRunMilestones(tx, input); await tx .update(statusDecisionEffects) @@ -1220,6 +1241,7 @@ async function materializeDecisionEffect(input: { ) .for("update"); if (pending.length === 0) throw new Error("native_pending_effect_missing"); + await notifyNativePendingEffects(input.tx, { companyId: input.companyId }); await input.tx .update(statusDecisionEffects) .set({ @@ -1909,6 +1931,7 @@ export async function commitNativeStatusDecision(input: { effect => effect.kind === "record_finalization_error", ); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.companyId); + await notifyNativePendingEffects(tx, input); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/preserved-result-finalization.ts b/server/src/services/preserved-result-finalization.ts index 6073896feeb..083e84fefe6 100644 --- a/server/src/services/preserved-result-finalization.ts +++ b/server/src/services/preserved-result-finalization.ts @@ -1,3 +1,4 @@ +import { notifyNativePendingEffects } from "./native-pending-effect-notifications.js"; import { and, desc, eq, gte, ne, or, sql } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { readSystemRecovery } from "./system-recovery-state.js"; @@ -45,6 +46,7 @@ export async function resumePreservedResultFinalization(db: Db, companyId: strin const generationId = intent?.id ?? recovery.generationId; if (coordinator.recoveryHistory.some(entry => entry.kind === "preserved_result_finalization" && entry.generationId === generationId)) return false; + await notifyNativePendingEffects(tx, { companyId, runId }); await tx.update(nativeRunFinalizations).set({ phase: coordinator.assessmentId ? "arbitrating" : "workspace_finalizing", nextAttemptAt: new Date(), failureCode: null, failureDetail: null, updatedAt: new Date(),