mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 21:49:18 +01:00
fix(server): schedule native pending effects from committed deadlines
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
ba0f0d6ea2
commit
56c3601ea6
26 files changed
+714
-110
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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",
|
||||
]);
|
||||
|
||||
|
||||
@@ -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');
|
||||
+152
-2
@@ -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": [
|
||||
+7
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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'`),
|
||||
|
||||
@@ -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],
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator> | 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<typeof commitNativeStatusDecision>[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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let admitted = true;
|
||||
|
||||
@@ -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<void>((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 {
|
||||
|
||||
+3
-1
@@ -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();
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<typeof registerNativePendingEffectWork>[1], canRun: () => boolean) =>
|
||||
registerNativePendingEffectWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null),
|
||||
registerUnsafeWorkspaceExportWork: (coordinator: Parameters<typeof registerUnsafeWorkspaceExportWork>[1], canRun: () => boolean) =>
|
||||
registerUnsafeWorkspaceExportWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null),
|
||||
registerAutomaticCompletionReviewWork: (coordinator: Parameters<typeof registerAutomaticCompletionReviewWork>[1], canRun: () => boolean) =>
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 }));
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
|
||||
type Entry = { id: string; companyId: string; dueAt: number };
|
||||
async function load(scope?: NativePendingEffectScope, signal?: AbortSignal): Promise<Entry[]> {
|
||||
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);
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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 });
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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(),
|
||||
|
||||
Reference in new issue
Block a user