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(),