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 75c7d8f0552..fdbc436c110 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2234,3 +2234,33 @@ Older gates were revalidated: #17490 and #17572 still have E2E shard 3 and aggre 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 index d840bf4ade1..8087ffd645b 100644 --- a/doc/plans/2026-10-25-native-pending-effects-sleep-audit.md +++ b/doc/plans/2026-10-25-native-pending-effects-sleep-audit.md @@ -1,82 +1,160 @@ # Native pending status-effect recovery -Status: owner/producer audit, before implementation. Branch: `codex/sleep-47-native-pending-effects`. Parent: #17687, `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 is distinct from response presentation, -evidence reassessment, and accepted-result/workspace settlement. +candidate processing for that pass. This slice moves only that duty. Response +presentation, evidence reassessment, and accepted-result/workspace settlement +remain separate migrations. -`dispatchPendingNativeStatusEffects` owns the recovery transaction. It locks the -coordinator, checks the exact company/issue/run/decision binding, locks pending -rows, and validates that their materialized targets still exist. Only then does -it mark the ledger delivered, increment attempts, clear retry/error state, update -the run's delivered-effect IDs, and signal chat milestones. It does not create -those targets again and does not perform another status arbitration. +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. -A missing target is an error. Do not turn it into successful settlement merely -to disarm a worker. Retain that error and an owned retry/recovery obligation. +## Source of truth and scheduling -## Current writers +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 production `statusDecisionEffects` insert in `commitNativeStatusDecision` -creates effect targets and their ledger entries in one transaction. It explicitly -inserts `deliveryState: delivered`, `attemptCount: 1`, and `deliveredAt`. The two -production ledger update paths also settle rows to delivered: direct recovery -and the `dispatch_pending_effect` materializer. Repository searches found no -current producer inserting pending/failed ledger rows; the schema default and -historical fixtures still support those states. This is a source audit, not -permission to drop historical recovery or assume independent SQL is supported. +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. -Accordingly, ordinary successful decision commits should not acquire a needless -pending-effects timer. Startup must reconstruct historical unresolved effects. -The next audit must identify coordinator transitions that can make historical -rows eligible again, and decide which existing mutation boundaries need signals. +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. -## Eligibility and ordering to preserve +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. -The existing outer candidate query requires native runtime, a non-null result, -a coordinator outside terminal_failure, a due/null nextAttemptAt, and an absent, -null-expiry, or expired coordinator lease. It also requires the issue join. -Pending effects refer to the coordinator's exact current decision. A worker must -preserve those authority constraints and distinguish a future eligibility -deadline from dormant superseded ledger history. +## Producer audit and shared boundaries -The periodic reconciler must not continue into other finalization work for a -candidate whose pending effects moved to the new owner. Simply deleting its -pending-effect branch would change ordering. Keep direct/manual behavior or -replace its contract explicitly. Coordinate completion with the remaining -finalization owner until that owner is separately migrated. +Notifications carry only company and optional run IDs. They register intent before +the write and use the existing outer-transaction/uncertain-outcome protocol. -Open design questions before coding: +| 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. | -- Which historical pending records are still owed, and which are superseded? -- Which coordinator phase/decision/result/lease changes reactivate dormant work? -- How are saved coordinator retry deadlines combined with pending-effect retry - after a target/read failure without resetting future deadlines on inspection? -- How does a worker stop or drain without losing a first dirty scope or retry? -- Which query/index bounds discovery when delivered history dominates? +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. -Use the existing scoped deadline helper and transaction notifications if they -fit. Keep eligibility and mutation policy in native runtime. No new queue table, -provider branch, generic materialization replay, or host-sleep exemption is needed. +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. -## Baseline and required verification +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. -All 72 cases in `native-status-arbiter-corpus.test.ts` passed on fresh external -PostgreSQL before this slice changed source (`/tmp/sleep47-baseline.log`). The -corpus exercises crash-after-decision-commit recovery and rejects a pending -continuation whose target is absent. It also validates adjacent arbitration, -review, cancellation, and finalization behavior. +## Query and migration evidence -Add focused worker tests for delivered-history quiet, pending/failed startup -recovery, no duplicate targets, missing-target failure, exact binding/tenant -fences, coordinator retry/lease readiness, mutation reactivation, transaction -commit/rollback, drain/resume, concurrent state changes, shutdown, and bounded -multi-page discovery. Re-run the corpus and affected consumers after source -changes. Current tests establish only the baseline, not a migrated worker. +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. + +Migration 0399 uses concurrent IF NOT EXISTS index builds through the existing +nontransactional migration owner. Generated snapshots retain the newest five +(0395–0399). 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. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 0c55c105038..ffd237fca33 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -524,6 +524,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0396_late_cerise.sql", "0397_lumpy_talos.sql", "0398_shocking_molecule_man.sql", + "0399_jazzy_ben_grimm.sql", "0380_funny_adam_warlock.sql", ]); diff --git a/packages/db/src/migrations/0399_jazzy_ben_grimm.sql b/packages/db/src/migrations/0399_jazzy_ben_grimm.sql new file mode 100644 index 00000000000..b0a67f12121 --- /dev/null +++ b/packages/db/src/migrations/0399_jazzy_ben_grimm.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/0394_snapshot.json b/packages/db/src/migrations/meta/0399_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0394_snapshot.json rename to packages/db/src/migrations/meta/0399_snapshot.json index a1c0ba77f16..97b135b3e34 100644 --- a/packages/db/src/migrations/meta/0394_snapshot.json +++ b/packages/db/src/migrations/meta/0399_snapshot.json @@ -1,6 +1,6 @@ { - "id": "90ac76c4-b6d8-4c38-b4b8-d51699ca2ec1", - "prevId": "f6db457e-383c-4518-a0de-7dde2fa6816d", + "id": "5b5659ef-e4b6-4584-9b4f-ed075047ef2f", + "prevId": "d6b708dd-c6d1-4bea-98c3-d84f5cfcf00a", "version": "7", "dialect": "postgresql", "tables": { @@ -2637,6 +2637,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": [ @@ -33627,6 +33671,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": [ @@ -42808,6 +42874,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": [ @@ -51994,6 +52116,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 2a7b0d92366..f1e41ee5c3e 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2780,6 +2780,13 @@ "when": 1792984941591, "tag": "0398_shocking_molecule_man", "breakpoints": true + }, + { + "idx": 399, + "version": "7", + "when": 1792986633369, + "tag": "0399_jazzy_ben_grimm", + "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 b8d7e8dacd5..7789dcc0d97 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 ab830b573fe..0407f6c24a8 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -108,6 +108,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() })), @@ -546,6 +547,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(); @@ -952,7 +960,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 e630972dd55..e2c2bff55d4 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1777,6 +1777,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, @@ -1915,7 +1917,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. trackHeartbeatSchedulerWork(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 105626c4f31..602a3d71c2a 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 c7f27216b78..1d8f1165fcb 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"; @@ -145,6 +146,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 958527f2268..04bb371f9cb 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"; @@ -4066,6 +4067,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 885b2527138..8aa28f5dba6 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1915,7 +1915,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(); @@ -1936,6 +1936,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 e5fa0e5dcb1..1955f8ab0df 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"; @@ -636,6 +637,7 @@ type NativeReconciliationOptions = { skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; + skipNativePendingEffects?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -770,6 +772,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()), @@ -799,6 +802,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 325a0075e8e..8d0e75ecf4f 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"; @@ -212,6 +213,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({ @@ -379,6 +381,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 ef3217d03fc..9714403b1d1 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"; @@ -3813,6 +3814,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({ @@ -3870,6 +3872,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({ @@ -9474,6 +9477,7 @@ async function executePaperclipNativeSessionWithinScope( ) .for("update"); await notifyExecutionRunWork(tx); + await notifyNativePendingEffects(tx, input.execution.binding); const updated = await tx .update(nativeRunFinalizations) .set({ @@ -9808,33 +9812,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 53e60a7d90e..6961ea01515 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 type { UnsafeWorkspaceExportScope } from "../unsafe-workspace-export-notifications.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; @@ -95,6 +96,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 28030a7c022..bf233cd933e 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"; @@ -77,6 +78,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; @@ -84,11 +94,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( @@ -104,6 +120,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, @@ -131,6 +151,7 @@ export async function dispatchPendingNativeStatusEffects(input: { targetId: effect.targetId, }); } + await notifyNativePendingEffects(tx, input); await notifyChatRunMilestones(tx, input); await tx .update(statusDecisionEffects) @@ -1217,6 +1238,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({ @@ -1866,6 +1888,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(),