From 50fe32684d2d4e1db129e3c1bb455571344a0c09 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 05:10:16 -0500 Subject: [PATCH] fix(server): wake committed native run projection from durable changes (#17752) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A native run has separate result acceptance, task decision, run status, and resource settlement steps. > - A task decision can commit before its run status is saved. > - Recovery currently revisits committed decisions through the periodic orphan pass. > - This PR gives the existing run projection repair its own transaction-driven worker. > - Existing records reconstruct unfinished work after restart. Completed history becomes quiet. > - Other completion and resource obligations keep their existing recovery paths. ## Linked Issues or Issue Description Refs #17748 (stack parent: accepted-result commit), #17734 (workspace finalization), and #17717 (retained session cleanup). Related #17614 preserves terminal notifications. This slice is prepared on current parent #17748 at `b2efdb2dbbbef5b2227ffa2bd2a9a6cfe3867fe2`; the parent has merged as `28e38da7e6bdd6af4092dda56bc2be05bed47af5`. **Current behavior** The native recovery pass revisits committed decisions to repair the heartbeat-run status and metadata. A crash between decision and run writes creates real work, but periodic scanning also revisits retained history. **Proposed behavior** Wake projection repair after committed producer changes. Reconstruct mismatches on startup and settings activation. Retry concrete failures through the shared scheduler. Keep the run projection before resource settlement. **Reason and benefit** The repair worker performs no recurring SQL after its work settles. A real decision, effect settlement, or ownership release reactivates it. This is one worker migration; it does not establish host or database suspension. ## What Changed - Extract the existing projection predicate and write into their native-runtime owner. The finalizer keeps initial metadata enrichment and task decision authority. - Discover accepted, committed, mismatched run projections with company, issue and run bindings. Exclude pending status effects and held runner ownership. Read narrow fields instead of whole result payloads. - Use existing transaction hints, scoped delivery scheduling, failure backoff and lifecycle handling. Add no tables or per-feature timers. - Require issue/run scope for ordinary finalization fanout. Routine company-only hints leave dormant projection quiet; explicit settings and dedicated producer signals keep their existing contract. - Connect finalization, pending effects, lifecycle status writes, issue changes and worktree execution activation. Preserve outer transaction commit and rollback behavior. - Lock the current run before deciding which terminal event and completion time to save. Concurrent live and recovery projection must not announce the same transition twice. - Delegate only already-committed projection from the periodic pass. Its candidate scan uses the same pending predicate to defer remaining consequences and cleanup until projection finishes. - Add real PostgreSQL producer, race and lifecycle tests plus an owner audit. ## Verification - Current local source `a83b9449233a275176e637c765059f414305d7c1` incorporates the parent with a normal merge. All 1,208 parent-only, seven owned-only and ten automatic shared paths are preserved. The only manual conflict keeps new projection fanout inside the ordinary issue/run guard. The complete package tree matches parent profile 80 byte-for-byte. - Controlled real-database regression: the original projection-only unscoped fanout produces three candidate SELECTs for routine company hints. The same zero-query assertion passes with scoped fanout. A real committed result then repairs the run and emits one terminal event before returning to quiet. Production bytes restore exactly in both phases. Twenty cases outside that focused pattern are not counted as passing there. - The complete main command passes **1,830 cases in 19 files**, without skips, in 301.14 seconds. It includes all 21 projection cases, 85 startup cases, required quality review and heartbeat run lifecycle, plus physical-work and result-commit consumers. Separate process recovery passes 410 in 163.58 seconds and external-chat response passes 273 in 53.26 seconds. Total: **2,513 distinct passes in 21 complete files**, with no skips. Normal merge `d4b46fd29cb23126d556cc3aeb883e6b41f1a8a5` adopts the actual parent squash with this entire tested child tree unchanged. - Production server typecheck, build, module boundaries, migration order and source identity checks pass. Seventeen dedicated PostgreSQL 17 databases use the unchanged migration 413 schema. Embedded-only calls use the existing fresh-per-call recovery helper. All temporary imports restore byte-exact and all 21 selected test hashes match after completion. No assertion or timeout is relaxed. - Parent package acceptance remains applicable: 130 passing installation/compatibility/transport cases, 22 installed ACPX/Grok contract cases and 26 native-provider Rust cases. One pre-existing opt-in closed Pi snapshot case is skipped because its distribution root is not configured. No live provider qualification is claimed. - The owner audit at `doc/plans/2026-10-26-native-run-projection-sleep-audit.md` retains prior failures and their disposition. Current-head acceptance: all 57 checks are clean, including protected CI and security. Greptile gives 5/5 on `a49a2dac13b0e1c19da5d3d6b435c781703b1c0b`, with no actionable findings. Old scores do not substitute. No Ponytail verdict is present under the authorized missing-review policy. - These checks do not claim a full local repository suite, deployed suspension, expanded Cloud wake coverage or measured cost savings. ## Risks - Process-local signals cover instrumented application writers. Independent SQL writers and replicas without supported ingress are not newly covered. Durable startup reconstruction remains enabled. - Projection must precede remaining committed resource settlement. The real reconciler test proves the periodic path defers cleanup while projection is pending and resumes it after repair. - Status decisions, external conversation settlement, watchdog resolution, Muse reconciliation and resource release keep their own authority and recovery. A committed result alone is not proof of completed cleanup. - Candidate discovery uses narrow bounded pages. There is no large-fleet performance or deployed sleep claim. Existing Cloud wake coverage and sleep blockers remain unchanged. - Parent #17748 has merged. This PR targets master at `28e38da7e6bdd6af4092dda56bc2be05bed47af5`. Current-head CI and review must pass before the authorized protected merge. ## Model Used OpenAI Codex, GPT-6 family, with repository tools and code execution. The exact serving model ID and context-window size are not exposed in this session. No subagent was used for this slice. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip Co-authored-by: Codex --- ...2026-10-24-recurring-server-sleep-stack.md | 20 ++ ...10-26-native-run-projection-sleep-audit.md | 247 +++++++++++++ .../native-run-projection-work.test.ts | 337 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 18 +- server/src/index.ts | 9 +- .../services/delivery-work-notifications.ts | 1 + server/src/services/heartbeat.ts | 9 + server/src/services/heartbeat/recovery.ts | 3 +- .../src/services/heartbeat/run-lifecycle.ts | 4 +- server/src/services/instance-settings.ts | 2 + server/src/services/issues.ts | 2 + .../native-finalization-notifications.ts | 2 + .../native-run-projection-notifications.ts | 8 + .../native-finalization-reconciler.ts | 8 + .../native-run-finalizer-telemetry.test.ts | 2 +- .../native-runtime/native-run-finalizer.ts | 159 ++------- .../native-run-projection-work.ts | 75 ++++ .../native-runtime/native-run-projection.ts | 172 +++++++++ 18 files changed, 935 insertions(+), 143 deletions(-) create mode 100644 doc/plans/2026-10-26-native-run-projection-sleep-audit.md create mode 100644 server/src/__tests__/native-run-projection-work.test.ts create mode 100644 server/src/services/native-run-projection-notifications.ts create mode 100644 server/src/services/native-runtime/native-run-projection-work.ts create mode 100644 server/src/services/native-runtime/native-run-projection.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 0fc94c07624..26ea1eea102 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -3250,3 +3250,23 @@ recovery. Publication and final local/remote gates remain pending. Master refreshed to `49bd7a4ba6`; subsequent finalization changes must be reconciled through the stack before final acceptance. All earlier gate failures, suppression audits and remaining inventory duties stay open. Nothing was merged. + +## Slice 53 accepted; slice 54 projection repair + +Slice 53 [#17748](https://github.com/paperclipai/paperclip/pull/17748) is verified at +`1fc2e8c5b61173203abbd50b1b03346abc629218`: all 53 remote checks/status entries +passed, Greptile 5/5 and zero unresolved threads. Local evidence includes 250 +passing tests, workspace typechecking and build. The embedded-cluster executor +command was interrupted after PostgreSQL semaphore exhaustion; that run is not +counted as passing. The full local test command was not claimed. The PR remains +draft/open and unmerged. + +Slice 54, `codex/sleep-54-native-run-projection`, moves the existing committed run +projection predicate and write into their owner and drives repair from scoped +committed changes. [Its audit](2026-10-26-native-run-projection-sleep-audit.md) +records the reproduced duplicate-event/stale-time/cleanup-ordering races, 400 +passing local cases, producer coverage and the retained post-commit/resource +obligations. Workspace typechecking passes; the full local suite remains limited +by embedded PostgreSQL startup capacity. Final local +and remote acceptance is pending. Master refreshed to `ad861a7f94`; stack integration, +prior failed gates, suppression audits and all remaining duties stay in scope. diff --git a/doc/plans/2026-10-26-native-run-projection-sleep-audit.md b/doc/plans/2026-10-26-native-run-projection-sleep-audit.md new file mode 100644 index 00000000000..81648080f36 --- /dev/null +++ b/doc/plans/2026-10-26-native-run-projection-sleep-audit.md @@ -0,0 +1,247 @@ +# Committed native run projection: sleep audit + +Date: 2026-10-26. Slice 54 starts at `1fc2e8c5b61173203abbd50b1b03346abc629218`, +[parent #17748](https://github.com/paperclipai/paperclip/pull/17748). +Branch: `codex/sleep-54-native-run-projection`. This is one duty from the +[recurring-server stack](2026-10-24-recurring-server-sleep-stack.md). + +## Obligation and boundary + +A native status decision can commit before its heartbeat-run projection. A crash +between those transactions leaves a real repair obligation. The periodic native +reconciler previously revisited committed decisions to repair that state. + +This slice moves the existing run projection repair to transaction wakes and +concrete failure retries. It does not move all committed completion work. External +conversation settlement, watchdog resolution, Muse reconciliation, workspace +metadata cleanup, environment/runtime release and agent settlement retain their +existing recovery paths. A successful workspace barrier or committed decision +proves none of those other obligations complete. + +`native-run-projection.ts` owns the existing projection predicate and write. The +finalizer and the new worker use that same owner. The finalizer keeps its initial +metadata enrichment and status-decision authority. The worker neither arbitrates +an issue outcome nor invokes provider work. No new table or migration is needed. + +The extracted `projectNativeTerminalRunStatus` function remains re-exported from +the finalizer so existing consumers retain their contract. Shared failure metadata +construction stays beside the projection owner and is reused by live finalization. +Heartbeat and server startup only wire lifecycle and execution policy. + +## Durable truth and quiet condition + +Candidates join the committed coordinator to its issue and heartbeat run. A +correlated existence predicate binds its accepted result by company, issue and +run IDs. The periodic candidate scan reuses this predicate for ordering. The accepted terminal state must +be succeeded, failed or cancelled. The worker excludes pending decision effects +and held native runner ownership. + +The pending predicate is the existing repairable run mismatch: status, completion +time, native phase, finalization metadata or stale failure fields. The write and +discovery share that predicate. Existing terminal-state restrictions remain in +place, including the existing protection for an already-cancelled run. Retained +completed history is not work. The accepted result envelope remains immutable. + +Discovery reads IDs in keyset pages of 100. It does not transfer full result or +run payloads. Due execution reads the small run fields needed by the projection +owner and only the accepted terminal-state scalar. The worker reuses shared +company-fair batches, coalesced dirty scopes and delivery scheduling. There is no +recurring query after an empty or settled result. No fleet-scale query benchmark +is claimed. + +Projection failure leaves the same durable mismatch. Shared per-entry error +backoff starts at 30 seconds and is bounded at five minutes. This is recovery of +concrete unfinished work, not periodic discovery of an empty table. A restart +reconstructs the mismatch immediately; no business deadline is moved. A successful +check that still observes a mismatch retains a scoped 30-second retry. + +## Producers and commit ownership + +| Producer | Contract | +| --- | --- | +| Status decision commit | Existing `notifyNativeFinalizationWork` fanout now includes projection before the coordinator write. This covers the first committed decision and later reassessments. | +| Native interaction bridge | Its existing finalization fanout covers its independent decision/coordinator writes. | +| Pending-effect settlement | The existing finalization fanout wakes projection after the effect owner finishes. | +| Finalizer and projection writes | Register the scoped hint before changing the run. This reconciles any pending in-memory entry after live completion. | +| Heartbeat lifecycle status writer | Native run changes publish a scoped hint, including release of an ownership hold and status/metadata changes after commit. | +| Issue mutations | Relevant status, binding, assignment and dependency changes reconstruct the affected issue scope. | +| Worktree execution setting | Activation emits a global reconstruction hint. | + +The coordinator store's result/claim/retry operations cannot independently create +a committed decision. Its existing finalization fanout covers changes to an +already-decided coordinator. Resultless restart and startup-stop writes cannot +create eligible committed projection work. + +All hints use the existing DB intent and outer-transaction protocol. Awaited intent +registration precedes the relevant write. Nested savepoint release does not wake +the worker before outer commit. Rollback produces no work. Unknown transaction +outcomes remain the shared protocol's responsibility. Process-local signals cover +instrumented application writes in the owning server process; independent SQL +writers and replicas without supported ingress are not newly supported. + +## Concurrency and authority + +The live finalizer can commit its decision and pause before its run write. The +new recovery worker can then project first. Both projection paths now read current +run status, native phase and completion time under the run row lock before deciding +which terminal event is owed. A writer that arrives second does not announce the +same terminal transition again. Initial live projection preserves the completion +time when the current run already has that same terminal status. + +Recovery projection rechecks committed coordinator/result identity and run binding +in its write predicate. Its background caller also requires settled status effects +at that boundary. A candidate snapshot does not authorize a write after these facts +change. The conditional update retains unrelated JSON diagnostics and saves cleared +execution failures as recovery evidence. Metadata-only repair does not emit another +terminal event. Leaving a pending native retry can still require the first terminal +failure report even when the stored status remains failed. + +The real race regression inserts recovery after the real status decision commit +and before the live finalizer resumes. It initially observed two terminal emissions; +a fresh locked snapshot reduces that to one. Another regression supplied a stale +completion-time snapshot and initially overwrote the durable timestamp. The owner +now uses the locked row's timestamp. Concurrent recovery calls use the same guard. + +## Lifecycle and periodic delegation + +Public heartbeat registration uses the existing task-drain, restore and worktree +suppression policy. Suppressed workers do not repeatedly query. Real worktree +activation reconstructs saved work. The delivery coordinator retains changes made +during drain, retries unresolved work, and joins active execution on shutdown. +Startup reconstructs committed decisions whose run projection is still missing. + +The periodic path sets `skipNativeRunProjection`; the finalizer receives the +narrow `skipCommittedRunProjection` option. This skips only the already-committed +projection call. It does not set the broader `projectRunStatus` flag false, because +that would also disable other completion effects. Default direct/manual calls and +new-result initial projection retain their behavior. The resource-settlement +callback still runs on the periodic path after projection completes. The candidate +scan uses the shared pending predicate to defer resource settlement while projection +is outstanding. Pending effects and reassessment retain their prior order. This +adds no separate per-candidate read. A real reconciler regression first observed +premature settlement, then passed after this guard. + +## Verification and preserved failures + +At this checkpoint, 20 worker/owner cases plus 78 startup cases pass: **98 tests**. +Six complete affected suites contribute another 302 passing cases: finalization +recovery, finalizer telemetry, status arbiter corpus, heartbeat lifecycle, +result-commit worker and heartbeat recovery. Total: **400 passing local cases**. +Workspace typechecking passes after the ordering guard. A workspace build passed +before that final guard; the final committed build and remote checks remain pending. +Coverage includes quiet startup and retained success/failure/cancellation history, +real status-decision activation, outer commit/rollback, live/recovery concurrency, +competing recovery calls, stale timestamps, metadata preservation, suppression and +real settings activation, drain/resume, failure/restart, shutdown and later work, +changed coordinator authority, real pending-effect settlement, the real lifecycle +status writer, and retained resource-settlement recovery. + +Initial evidence is retained: + +- Six initial regressions produced four failures and two passes. The missing + producer signal broke normal and outer-commit activation. The live/recovery + race emitted twice. A stale snapshot replaced a saved completion time. +- An intermediate scripted extraction introduced a syntax error; that command + ran no tests. After correction, all six initial cases passed. +- Initial typechecking found the terminal mapper's wider return union included + `running`. The validated accepted terminal scalar is now narrowed at its + boundary; no active result is eligible for committed projection. + +- The broader six-suite run passed 300 cases and failed two diagnostic-race + cases. Their interception watched a full result JSON selection; projection now + selects only the terminal scalar. Updating that interception restored the same + concurrent-write assertion. The 29-case telemetry suite then passed in full. +- An intermediate rerun reused a database with a fixed `TLM` company prefix. + Setup rejected the duplicate; 29 telemetry cases were skipped, not passed. + Fresh dedicated databases passed all 65 telemetry/recovery cases. +- The new ordering regression failed because cleanup ran before delegated run + projection. The shared-predicate guard fixes it; the 98 focused cases pass. + +Local logs: `/tmp/sleep54-initial-red.log`, `/tmp/sleep54-initial-green.log` +(the syntax failure), `/tmp/sleep54-initial-green-fixed.log`, +`/tmp/sleep54-types.log`, and `/tmp/sleep54-lifecycle.log`. +Later evidence: `/tmp/sleep54-affected.log`, `/tmp/sleep54-ordering-red.log`, +`/tmp/sleep54-ordering-green.log` (98 passes plus the fixture setup failure), +`/tmp/sleep54-affected-fixed.log` (65 passes), and `/tmp/sleep54-final-types.log`. +The full local `pnpm test:run` was not attempted: this host's embedded PostgreSQL +cluster startup previously exhausted shared semaphore capacity. No further OS +resources were removed. Fresh-host CI must supply the full-suite gate. Publication, +final build, exact-head CI and review are required before acceptance. + +## Remaining scope + +Master was refreshed to `ad861a7f94`. The stack still needs propagation of later +master changes and fresh integration checks. Relevant native changes include +#17525/#17531 (narrow reads), #17614 (unchanged terminal notifications), and +#17641/#17665 (monitor proof). Keep both contracts when integrating; do not replace +their changes with an older finalizer block. + +All other recurring duties, older failed stack gates and the earlier eight-worker +suppression audit remain open. Cloud alarm coverage, idle-sleep safety and durable +resource blockers are unchanged. This audit establishes worker behavior, not +measured database suspension, host wakeup or cost savings. No merge is authorized. + +## Current parent integration and routine-hint regression (2026-10-29) + +This slice incorporates parent #17748 at +`b2efdb2dbbbef5b2227ffa2bd2a9a6cfe3867fe2` through a normal merge. A +batched tree comparison preserves 1,208 parent-only entries, seven owned-only +entries, 11,524 identical entries and ten automatic shared merges. The sole +manual conflict adds run-projection fanout inside the existing ordinary issue/run +scope guard. Dedicated notifications and explicit settings activation retain their +contracts. The entire package tree matches parent profile 80, including all +historical identities. No provider profile is regenerated for this server change. + +Twenty-four added comment objects in six files were removed with identical +nontrivia TypeScript token streams. No directive was removed. The projection +extraction retains the shared candidate predicate, exact run/result correlation, +atomic metadata preservation, terminal-event idempotency, original completion time, +and physical-owner evidence. Status decisions, external effects and resource +settlement remain with their existing owners. + +The new real-database regression starts an idle projection worker and submits +three routine company-only finalization hints. A controlled baseline moves only +projection fanout outside the guard and keeps all other parent guards intact. +It produces three candidate SELECTs and fails the zero-query assertion. Restoring +the production helper passes the same assertion. A real accepted result and +committed decision then repair the run and emit exactly one terminal event before +returning to quiet. Red fails one selected case in 6.63 seconds; green passes it in +6.55 seconds. Twenty cases outside the focused pattern are not counted as passing +there. Both phases use fresh dedicated PostgreSQL 17 databases and restore the +production helper byte-for-byte. No assertion or timeout is relaxed. + +Complete validation includes the result/physical-work consumers plus run lifecycle +and required quality review, because the projection is extracted from the live +finalizer. It retains the startup held-work regression and the real producer paths. +Current-head CI and fresh review remain separate publication and merge gates. + +### Complete current-source acceptance + +Source `a83b9449233a275176e637c765059f414305d7c1` passes **2,513 distinct +cases in 21 complete files**, with no skips in those files. The 19-file main +command passes 1,830 cases in 301.14 seconds. Process recovery passes 410 in +163.58 seconds; external-chat response passes 273 in 53.26 seconds. The selection +includes all 21 projection cases, 85 startup cases, required quality review, +heartbeat run lifecycle, both accepted-result producers, and physical-work +consumers. Production server typechecking, build, module boundaries, migration +order and the source identity check pass. + +Seventeen dedicated PostgreSQL 17 databases use the unchanged migration 413 schema. +Embedded-only calls use existing fresh-per-call recovery fixtures. Every temporary +import is restored byte-for-byte and all 21 selected test hashes match after +completion. No assertion or timeout is relaxed. The controlled unscoped-hint +baseline remains an intended failing regression; it is not counted as acceptance. +Earlier failures in this audit retain their revision and disposition. + +Parent #17748 merged as `28e38da7e6bdd6af4092dda56bc2be05bed47af5`, after +57 clean current-head checks and Greptile 5/5. Its complete reviewed tree equals +the actual squash. Normal merge `d4b46fd29cb23126d556cc3aeb883e6b41f1a8a5` +adopts that ancestry with this child's complete tested tree unchanged. This +slice can now target master; fresh exact-head CI and review remain required. + +The unchanged package tree retains parent profile 80 acceptance: 130 passing +installation/compatibility/transport cases, 22 installed ACPX/Grok contract cases +and 26 native-provider Rust cases. One existing opt-in closed Pi snapshot case +was skipped because its distribution root was unset. This does not qualify a +live Pi installation. No full local repository suite, live provider qualification, +deployed sleep, expanded Cloud alarm coverage or measured savings is claimed. diff --git a/server/src/__tests__/native-run-projection-work.test.ts b/server/src/__tests__/native-run-projection-work.test.ts new file mode 100644 index 00000000000..26f0a9d239a --- /dev/null +++ b/server/src/__tests__/native-run-projection-work.test.ts @@ -0,0 +1,337 @@ +import { heartbeatService } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js"; +import { reconcileNativeFinalizations } from "../services/native-runtime/native-finalization-reconciler.js"; +import { statusDecisionEffects } from "@paperclipai/db"; +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { nativeRunProjectionWork } from "../services/native-runtime/native-run-projection-work.js"; +import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; +import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; +import { finalizeNativeRun } from "../services/native-runtime/native-run-finalizer.js"; +import * as statusCommitter from "../services/native-runtime/status-decision-committer.js"; +import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; +import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; +import { workspaceOperationService } from "../services/workspace-operations.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { emitAgentTaskRun } from "../services/agent-task-run-telemetry.js"; +vi.mock("../services/agent-task-run-telemetry.js", () => ({ emitAgentTaskRun: vi.fn(async () => undefined) })); +vi.mock("../services/run-failure-report.js", async importOriginal => ({ + ...await importOriginal(), reportRunFailure: vi.fn(async () => undefined), +})); + +describe("committed native run projection lifecycle", () => { + let temporary: Awaited> | undefined; + let db: Db, root: string; + let coordinator: ReturnType | undefined; + let worker: ReturnType["register"]> | undefined; + const companyId = randomUUID(), agentId = randomUUID(); + const priorLogs = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + let admitted = true; + const errors: unknown[] = []; + beforeAll(async () => { + root = await mkdtemp(join(tmpdir(), "native-run-projection-")); + process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = root; + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RUN_PROJECTION_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-run-projection-"); + db = createDb(external ?? temporary!.connectionString); + await db.insert(companies).values({ id: companyId, name: "Result commits", issuePrefix: `RC${randomUUID().slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Result worker", adapterType: "paperclip_runner" }); + }, 30_000); + beforeEach(() => { + admitted = true; errors.length = 0; vi.mocked(emitAgentTaskRun).mockClear(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; worker = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId)); + expect(errors).toEqual([]); + }); + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); await rm(root, { recursive: true, force: true }); + if (priorLogs === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH; + else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorLogs; + }); + async function start(project = projectCommittedRun) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = nativeRunProjectionWork(db, project).register(coordinator, () => admitted); + await worker.ready; + } + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + } + async function quiet() { + const previous = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_id, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); await delay(10); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previous; } + } + async function seed() { + const runId = randomUUID(), issueId = randomUUID(), contractId = randomUUID(), sourceInstanceId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Commit accepted work", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "result-worker-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "result-worker-v1", objective: "Finish", criteria: [{ id: "objective", requirement: "Complete the task" }] }, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, runtimeMode: "native", status: "running", + nativeIssueId: issueId, nativeSessionId: runId, runnerInstanceId: sourceInstanceId, + completionContractId: contractId, completionContractSha256: contractId, contextSnapshot: { issueId } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + return { companyId, issueId, runId, agentId, sessionId: runId, completionContractId: contractId, + completionContractSha256: contractId, sourceInstanceId, controlPlaneSourceInstanceId: "result-worker-test" }; + } + async function complete(binding: Awaited>, producer: "control_plane" | "coordinator", writer: Db = db, terminalStatus: "succeeded" | "failed" | "cancelled" = "succeeded") { + const port = producer === "control_plane" ? new PaperclipControlPlanePort(writer, binding) + : new NativeRunCoordinatorStore(writer, { ...binding, normalizedSessionId: binding.sessionId, + runnerSourceInstanceId: binding.sourceInstanceId, + completionContractRevision: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.contractRevision, + completionContractCriterionIds: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.criteria.map(item => item.criterionId) }); + await port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, runTerminalState: terminalStatus }, + callerResultId: `${binding.runId}:result` }); + } + async function barrier(binding: Awaited>) { + await workspaceOperationService(db).createRecorder({ companyId, issueId: binding.issueId, heartbeatRunId: binding.runId }) + .recordOperation({ phase: "workspace_finalize", cwd: root, run: async () => ({ status: "succeeded", exitCode: 0 }) }); + } + async function state(runId: string) { + return db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)).then(rows => rows[0]); + } + async function read(runId: string) { + return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0]); + } + async function accepted(status: "succeeded" | "failed" | "cancelled" = "succeeded") { + const binding = await seed(); await complete(binding, "control_plane", db, status); await barrier(binding); return binding; + } + it("reconstructs a committed decision with an unfinished run projection and stays quiet after repair", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + expect((await read(binding.runId)).status).toBe("running"); + await start(); await flush(); + expect((await read(binding.runId)).status).toBe("succeeded"); + expect(emitAgentTaskRun).toHaveBeenCalledTimes(1); await quiet(); + const settled = await read(binding.runId); + await coordinator!.stop(); await start(); await flush(); await quiet(); + expect((await read(binding.runId)).executionStatusDeliveryId).toBe(settled.executionStatusDeliveryId); + expect(emitAgentTaskRun).toHaveBeenCalledTimes(1); + }); + it("ignores routine company-only finalization hints and still repairs a committed run", async () => { + await start(); await flush(); await quiet(); + const previous = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_id, query) => { + if (/from\s+"native_run_finalizations"/i.test(query)) queries.push(query); + }; + try { + for (let index = 0; index < 3; index += 1) { + await db.transaction(async tx => { await notifyNativeFinalizationWork(tx, { companyId }); }); + await flush(); + } + expect(queries).toEqual([]); + } finally { db.$client.options.debug = previous; } + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await flush(); + expect((await read(binding.runId)).status).toBe("succeeded"); + expect(emitAgentTaskRun).toHaveBeenCalledTimes(1); + await quiet(); + }); + + it("wakes from the real status decision transaction after startup has become quiet", async () => { + await start(); await flush(); await quiet(); + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await flush(); + expect((await read(binding.runId)).status).toBe("succeeded"); await quiet(); + }); + it.each(["commit", "rollback"])("honors the status writer's outer %s", async outcome => { + await start(); await flush(); + const binding = await accepted(); + const transaction = db.transaction(async tx => { + await finalizeNativeRun({ db: tx as unknown as Db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await vi.advanceTimersByTimeAsync(5000); + expect((await read(binding.runId)).status).toBe("running"); + if (outcome === "rollback") throw new Error("rollback fixture"); + }); + if (outcome === "rollback") await expect(transaction).rejects.toThrow("rollback fixture"); else await transaction; + await flush(); + expect((await read(binding.runId)).status).toBe(outcome === "commit" ? "succeeded" : "running"); await quiet(); + }); + it("does not emit another terminal event when recovery projects before the live finalizer", async () => { + const binding = await accepted(); + const commit = statusCommitter.commitNativeStatusDecision; + const hook = vi.spyOn(statusCommitter, "commitNativeStatusDecision").mockImplementation(async input => { + const outcome = await commit(input); + await projectCommittedRun({ db, run: await read(binding.runId), coordinator: await state(binding.runId) }); + expect((await read(binding.runId)).status).toBe("succeeded"); + return outcome; + }); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect(hook).toHaveBeenCalled(); + expect(emitAgentTaskRun).toHaveBeenCalledTimes(1); + }); + it("uses fresh completion time when the caller's projection snapshot is stale", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const stale = await read(binding.runId); + const finishedAt = new Date("2020-01-01T00:00:00Z"); + await db.update(heartbeatRuns).set({ finishedAt }).where(eq(heartbeatRuns.id, binding.runId)); + await projectCommittedRun({ db, run: stale, coordinator: await state(binding.runId) }); + expect((await read(binding.runId)).finishedAt).toEqual(finishedAt); + }); + it.each(["succeeded", "failed", "cancelled"] as const)("stays quiet with retained %s history", async status => { + const binding = await accepted(status); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const before = await read(binding.runId); + expect(before.status).toBe(status); + const calls = vi.mocked(emitAgentTaskRun).mock.calls.length; + await start(); await flush(); await quiet(); + expect(await read(binding.runId)).toEqual(before); + expect(emitAgentTaskRun).toHaveBeenCalledTimes(calls); + }); + it("serializes competing recovery projections without duplicate terminal events", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const input = { db, run: await read(binding.runId), coordinator: await state(binding.runId) }; + await Promise.all(Array.from({ length: 4 }, () => projectCommittedRun(input))); + expect((await read(binding.runId)).status).toBe("succeeded"); + expect(emitAgentTaskRun).toHaveBeenCalledTimes(1); + }); + it("repairs projection metadata and preserves unrelated diagnostics", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const before = await read(binding.runId); + await db.update(heartbeatRuns).set({ nativePhase: "retryable_failure", error: "old failure", errorCode: "transport_error", + resultJson: { ...before.resultJson, failureCode: "old", nextAttemptAt: "old", unrelatedReceipt: { key: "preserve" } }, + }).where(eq(heartbeatRuns.id, binding.runId)); + const calls = vi.mocked(emitAgentTaskRun).mock.calls.length; + await start(); await flush(); + expect(await read(binding.runId)).toMatchObject({ status: "succeeded", nativePhase: "committed", error: null, errorCode: null, + finishedAt: before.finishedAt, resultJson: { failureCode: null, nextAttemptAt: null, + unrelatedReceipt: { key: "preserve" }, recoveredExecutionFailure: { error: "old failure", errorCode: "transport_error" } } }); + expect(emitAgentTaskRun).toHaveBeenCalledTimes(calls); await quiet(); + }); + it.each(["restore", "worktree"])("respects %s suppression through the public heartbeat registration", async mode => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" } + : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "projection-suppression-test" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + if (mode === "worktree") await settings.updateExperimental({ enableWorktreeRunExecution: false }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = heartbeatService(db, { runtimeEnv }).registerNativeRunProjectionWork(coordinator, () => admitted); + await worker.ready; await flush(); await quiet(); + expect((await read(binding.runId)).status).toBe("running"); + if (mode === "worktree") { + try { + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet(); + } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + } + }); + it("retains the first committed decision during drain and projects on resume", async () => { + await start(); await flush(); admitted = false; + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await vi.advanceTimersByTimeAsync(120_000); await quiet(); + expect((await read(binding.runId)).status).toBe("running"); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await read(binding.runId)).status).toBe("succeeded"); await quiet(); + }); + it("retains a failed projection across drain and restart", async () => { + let fail = true; + const project: typeof projectCommittedRun = input => { + if (fail) { fail = false; throw new Error("projection unavailable"); } + return projectCommittedRun(input); + }; + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await start(project); + expect(errors).toHaveLength(1); errors.length = 0; + expect((await read(binding.runId)).status).toBe("running"); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + admitted = false; await vi.advanceTimersByTimeAsync(120_000); await quiet(); + await coordinator!.stop(); admitted = true; await start(); await flush(); + expect((await read(binding.runId)).status).toBe("succeeded"); await quiet(); + }); + it("joins active projection on shutdown and reconstructs later work", async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const entered = vi.fn(); + const project: typeof projectCommittedRun = async input => { entered(); await gate; return projectCommittedRun(input); }; + await start(project); await flush(); + const first = await accepted(); + await finalizeNativeRun({ db, runId: first.runId, workspaceFinalizeStatus: "succeeded" }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(entered).toHaveBeenCalledTimes(1); }); + const second = await accepted(); + await finalizeNativeRun({ db, runId: second.runId, workspaceFinalizeStatus: "succeeded" }); + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await delay(10); expect(stopped).toBe(false); release(); await stopping; + expect((await read(first.runId)).status).toBe("succeeded"); + expect((await read(second.runId)).status).toBe("running"); + await start(); await flush(); expect((await read(second.runId)).status).toBe("succeeded"); await quiet(); + }); + it("rechecks committed authority after candidate discovery", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const project: typeof projectCommittedRun = async input => { + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, binding.runId)); + return projectCommittedRun(input); + }; + await start(project); await flush(); + expect((await read(binding.runId)).status).toBe("running"); expect(emitAgentTaskRun).not.toHaveBeenCalled(); await quiet(); + }); + it("waits for real pending-effect settlement before projection", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const committed = await state(binding.runId); + const [effect] = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, committed.decisionId!)); + expect(effect).toBeDefined(); + await db.update(statusDecisionEffects).set({ deliveryState: "pending", deliveredAt: null }).where(eq(statusDecisionEffects.id, effect.id)); + await start(); await flush(); await quiet(); expect((await read(binding.runId)).status).toBe("running"); + await statusCommitter.dispatchPendingNativeStatusEffects({ db, ...binding, decisionId: committed.decisionId!, onlyWhenDue: true }); + await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet(); + }); + it("wakes from the real lifecycle status writer when an ownership hold is released", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + await db.update(heartbeatRuns).set({ nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" }) + .where(eq(heartbeatRuns.id, binding.runId)); + await start(); await flush(); await quiet(); + + + const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: vi.fn() } as unknown as HeartbeatLifecycleDependencies); + await lifecycle.setRunStatus(binding.runId, "running", { errorCode: null, nativePhase: "committed" }); + await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet(); + }); + it("waits for delegated run projection before committed resource settlement", async () => { + const binding = await accepted(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + const settled = vi.fn(async () => undefined); + await reconcileNativeFinalizations(db, [binding.runId], { skipNativeRunProjection: true, skipNativeResultCommit: true, + skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled }); + expect((await read(binding.runId)).status).toBe("running"); expect(settled).not.toHaveBeenCalled(); + await start(); await flush(); expect((await read(binding.runId)).status).toBe("succeeded"); await quiet(); + await reconcileNativeFinalizations(db, [binding.runId], { skipNativeRunProjection: true, skipNativeResultCommit: true, + skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled }); + expect(settled).toHaveBeenCalledTimes(1); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bdda480294c..c54b7b2c14e 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -120,6 +120,7 @@ const { registerRetainedNativeSessionCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), registerNativeWorkspaceFinalizationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), registerNativeResultCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerNativeRunProjectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeStatusReassessmentWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeReviewResponseWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeResponsePresentationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), @@ -598,6 +599,13 @@ describe("startServer feedback export wiring", () => { } }); + it.each([false, true])("registers native run projection only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerNativeRunProjectionWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerNativeRunProjectionWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers native result commit only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -1066,7 +1074,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, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: 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, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { @@ -1222,15 +1230,17 @@ describe("startServer feedback export wiring", () => { } }); - it.each(["retained cleanup", "workspace finalization", "native result commit"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { + it.each(["retained cleanup", "workspace finalization", "native run projection", "native result commit"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); let releaseCleanup!: () => void; const ready = new Promise(resolve => { releaseCleanup = resolve; }); const stop = vi.fn(); const register = owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork : owner === "workspace finalization" ? heartbeatServiceMock.registerNativeWorkspaceFinalizationWork + : owner === "native run projection" ? heartbeatServiceMock.registerNativeRunProjectionWork : heartbeatServiceMock.registerNativeResultCommitWork; - if (owner === "native result commit") heartbeatServiceMock.registerNativeResultCommitWork.mockReturnValueOnce({ ready }); + if (owner === "native run projection") heartbeatServiceMock.registerNativeRunProjectionWork.mockReturnValueOnce({ ready }); + else if (owner === "native result commit") heartbeatServiceMock.registerNativeResultCommitWork.mockReturnValueOnce({ ready }); else (owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork : heartbeatServiceMock.registerNativeWorkspaceFinalizationWork).mockReturnValueOnce({ ready, stop }); const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(() => 1 as unknown as ReturnType); @@ -1247,7 +1257,7 @@ describe("startServer feedback export wiring", () => { const server = await starting; stopping = server.shutdown().then(() => { stopped = true; }); await settleStartupHousekeeping(); - if (owner !== "native result commit") expect(stop).toHaveBeenCalledOnce(); + if (owner === "retained cleanup" || owner === "workspace finalization") expect(stop).toHaveBeenCalledOnce(); expect(stopped).toBe(false); } finally { releaseCleanup(); diff --git a/server/src/index.ts b/server/src/index.ts index 7e221c2191d..419138856de 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1842,6 +1842,13 @@ async function startServerWithDatabaseTeardown( server.on("close", worker.stop); + trackHeartbeatSchedulerWork(worker.ready); + } + if (config.heartbeatSchedulerEnabled) { + const worker = heartbeat.registerNativeRunProjectionWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); + + trackHeartbeatSchedulerWork(worker.ready); } if (config.heartbeatSchedulerEnabled) { @@ -2001,7 +2008,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. runPeriodicRecovery(() => heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: 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 ed5a2fd5a4d..4a88a4dada4 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 = { + nativeRunProjection: "native-run-projection", nativeResultCommit: "native-result-commit", nativeWorkspaceFinalization: "native-workspace-finalization", retainedNativeSessionCleanup: "retained-native-session-cleanup", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e204c0e3568..8fd88e51861 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { nativeRunProjectionWork } from "./native-runtime/native-run-projection-work.js"; import { nativeResultCommitWork } from "./native-runtime/native-result-commit-work.js"; import { nativeStatusReassessmentWork } from "./native-runtime/native-status-reassessment-work.js"; import { registerNativeReviewResponseWork } from "./native-runtime/native-review-response-work.js"; @@ -4095,6 +4096,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeRunProjectionWork: ( + coordinator: Parameters["register"]>[0], canRun: () => boolean, + ) => nativeRunProjectionWork(db).register(coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerNativeResultCommitWork: ( coordinator: Parameters["register"]>[0], canRun: () => boolean, ) => nativeResultCommitWork(db).register(coordinator, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index fd7fa043005..9b99ec4ff97 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1967,7 +1967,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; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: 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; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1994,6 +1994,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipNativeStatusReassessment: opts?.skipNativeStatusReassessment, skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization, skipNativeResultCommit: opts?.skipNativeResultCommit, + skipNativeRunProjection: opts?.skipNativeRunProjection, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 43787f4ceea..9c3626ac52f 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -1018,13 +1019,14 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl return recordHeartbeatRunRuntimeProgress(currentRun, update, issueId); } - async function writeRunStatus(status: string, previous: Pick | null, + async function writeRunStatus(status: string, previous: Pick | null, write: (tx: Db) => Promise) { const companyId = previous?.companyId; // Register on the outer transaction before the terminal write. An unknown // commit cannot strand newly-actionable accounting after the live event. return db.transaction(async transaction => { const tx = transaction as unknown as Db; + if (previous?.runtimeMode === "native") await notifyNativeRunProjection(tx, { companyId: previous.companyId, runId: previous.id }); if (previous && previous.status !== status) await notifyUnsafeWorkspaceExportRunChange(tx, previous); if (previous && (previous.status === "scheduled_retry" || status === "scheduled_retry")) await notifyScheduledRetryCancellation(tx, { companyId: previous.companyId, runId: previous.id }); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index e088559962b..94af107ec22 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; @@ -628,6 +629,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDirectoryCleanup(tx); await notifyInstructionCollection(tx); await notifyCancelledWorkspacePreservation(tx); + await notifyNativeRunProjection(tx); await notifyNativeResultCommit(tx); await notifyNativeWorkspaceFinalization(tx); } diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index a33129c30fa..1f9c3cef1af 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; @@ -11504,6 +11505,7 @@ export function issueService(db: Db) { if (patch.status !== undefined || patch.statusVersion !== undefined || patch.executionState !== undefined || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || blockedByIssueIds !== undefined) { + await notifyNativeRunProjection(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeResultCommit(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeWorkspaceFinalization(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeStatusReassessment(tx, { companyId: receiptExisting.companyId, issueId: id }); diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index 1f1952d696a..9cf4d073621 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; @@ -10,6 +11,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { if (scope.issueId || scope.runId) { + await notifyNativeRunProjection(transaction, scope); await notifyNativeResultCommit(transaction, scope); await notifyNativeWorkspaceFinalization(transaction, scope); await notifyRetainedNativeSessionCleanup(transaction, scope); diff --git a/server/src/services/native-run-projection-notifications.ts b/server/src/services/native-run-projection-notifications.ts new file mode 100644 index 00000000000..7e23de97a1f --- /dev/null +++ b/server/src/services/native-run-projection-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeRunProjectionScope = { companyId: string; issueId?: string; runId?: string }; + +export function notifyNativeRunProjection(tx: object, scope?: NativeRunProjectionScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeRunProjection, + scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, 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 819d4ed3fb7..568f540b31d 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 { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js"; import { resolveNativeReconciliationStatus } from "./native-reconciliation-status.js"; export { resolveNativeReconciliationStatus, type NativeReconciliationFacts } from "./native-reconciliation-status.js"; @@ -389,6 +390,7 @@ type NativeReconciliationOptions = { skipNativeStatusReassessment?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; + skipNativeRunProjection?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -507,6 +509,7 @@ async function reconcileNativeFinalizationsPass( issueStatusVersion: issues.statusVersion, issueDecisionId: issues.lastStatusDecisionId, coordinatorPhase: nativeRunFinalizations.phase, + runProjectionPending: options.skipNativeRunProjection ? pendingNativeRunProjectionCondition(db) : sql`false`, resultId: nativeRunFinalizations.resultId, assessmentId: nativeRunFinalizations.assessmentId, decisionId: nativeRunFinalizations.decisionId, @@ -618,6 +621,9 @@ async function reconcileNativeFinalizationsPass( } } } + + + if (options.skipNativeRunProjection && row.coordinatorPhase === "committed" && row.runProjectionPending) continue; const barrier = await db.select({ status: workspaceOperations.status }) .from(workspaceOperations) .where(and( @@ -681,6 +687,7 @@ async function reconcileNativeFinalizationsPass( projectRunStatus: true, skipNativeResponsePresentation: options.skipNativeResponsePresentation, skipNativeReviewResponse: options.skipNativeReviewResponse, + skipCommittedRunProjection: options.skipNativeRunProjection, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( row.nativeWorkspaceSync, @@ -742,6 +749,7 @@ async function reconcileNativeFinalizationsPass( projectRunStatus: true, skipNativeResponsePresentation: options.skipNativeResponsePresentation, skipNativeReviewResponse: options.skipNativeReviewResponse, + skipCommittedRunProjection: options.skipNativeRunProjection, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( row.nativeWorkspaceSync, diff --git a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts index a128f0bb2bf..426996547f9 100644 --- a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts +++ b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts @@ -447,7 +447,7 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a selection: any, ) => { const query = select(selection); - return selection?.resultJson === nativeRunResults.resultJson + return selection?.terminalState ? wrap(query) : query; }) as typeof db.select); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 833cccaca14..f11eacf277c 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,3 +1,5 @@ +import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; +import { projectCommittedRun, recoveredExecutionFailureMetadata } from "./native-run-projection.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; @@ -93,7 +95,6 @@ import { } from "../durable-chat-wakeup.js"; import { isNativeRunnerOwnershipHeld, - nativeRunnerOwnershipNotHeldCondition, } from "./native-runner-ownership.js"; function record(value: unknown): Record { @@ -137,12 +138,7 @@ function isCommittedAuditOnlyRouting(value: unknown) { ); } -/** Server-owned terminal conversion used by live finalization and read models. */ -export function projectNativeTerminalRunStatus( - terminalState: "succeeded" | "failed" | "cancelled" | "active", -) { - return terminalState === "active" ? ("running" as const) : terminalState; -} +export { projectNativeTerminalRunStatus } from "./native-run-projection.js"; /** Fact-based arbitration at the production finalization authority seam. */ export function resolveNativeFinalizerStatus( @@ -614,128 +610,6 @@ async function resolveCommittedFinalizationRecovery(db: Db, run: Pick; - coordinator: typeof nativeRunFinalizations.$inferSelect; -}) { - if (!input.coordinator.resultId) return; - const resultRow = await input.db - .select({ resultJson: nativeRunResults.resultJson }) - .from(nativeRunResults) - .where( - and( - eq(nativeRunResults.id, input.coordinator.resultId), - eq(nativeRunResults.runId, input.run.id), - eq(nativeRunResults.companyId, input.run.companyId), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - const terminalState = record( - record(resultRow?.resultJson).terminal, - ).runTerminalState; - if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) { - throw new Error("native_finalization_invalid"); - } - const projectedStatus = projectNativeTerminalRunStatus( - terminalState as "succeeded" | "failed" | "cancelled", - ); - const needsRepair = and( - eq(heartbeatRuns.id, input.run.id), - eq(heartbeatRuns.runtimeMode, "native"), - or( - inArray(heartbeatRuns.status, ["queued", "running", "failed"]), - and( - eq(heartbeatRuns.status, "succeeded"), - or( - isNotNull(heartbeatRuns.error), - isNotNull(heartbeatRuns.errorCode), - isNull(heartbeatRuns.finishedAt), - sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`, - sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`, - sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`, - sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`, - sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`, - ), - ), - ), - // Reconciliation revisits committed results periodically. Only repair - // a changed projection; rewriting an unchanged failed run would mint a - // fresh status delivery (and failure toast) on every sweep. Check the - // current row so concurrent replays cannot both queue the same repair. - or( - sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`, - isNull(heartbeatRuns.finishedAt), - sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`, - sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`, - sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`, - sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`, - sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`, - ...(terminalState === "succeeded" - ? [isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode)] - : []), - ), - nativeRunnerOwnershipNotHeldCondition(), - ); - const [candidate] = await input.db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(needsRepair).limit(1); - if (!candidate) return; - const now = new Date(); - const [updatedRun] = await input.db.transaction(async transaction => { - const tx = transaction as unknown as Db; - await notifyRetainedNativeSessionCleanup(tx, { companyId: input.run.companyId, runId: input.run.id }); - await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id }); - await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id }); - await notifyUnsafeWorkspaceExportRunChange(tx, input.run); - await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); - await notifyExecutionRunWork(tx); - await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); - return tx - .update(heartbeatRuns) - .set({ - executionStatusDeliveryId: randomUUID(), - status: projectedStatus, - finishedAt: input.run.finishedAt ?? now, - nativePhase: "committed", - nativePhaseUpdatedAt: now, - ...(terminalState === "succeeded" - ? { - error: null, - errorCode: null, - // Capture the row being updated, not the earlier admission read: - // cleanup may have recorded a new diagnostic in the meantime. - // Other result metadata and all physical-owner evidence stay put. - resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) - || ${recoveredExecutionFailureMetadata()} - || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)`, - } - : { resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)` }), - updatedAt: now, - }) - .where(needsRepair) - .returning(); - }); - // Metadata repairs can preserve the terminal status. Only a genuine status - // transition should emit another terminal event. A failed run that leaves a - // pending retry still needs its first failure report. - if (updatedRun && updatedRun.status !== input.run.status) { - await emitAgentTaskRun(input.db, updatedRun); - void reportRunFailure(input.db, updatedRun); - } else if (updatedRun && isNewTerminalRunFailure(input.run, updatedRun)) { - void reportRunFailure(input.db, updatedRun); - } -} - async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) { const [run] = await dbOrTx .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, @@ -1225,6 +1099,7 @@ export async function finalizeNativeRun(input: { projectRunStatus?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; + skipCommittedRunProjection?: boolean; /** Workspace-only replay must not consume the provider recovery budget. */ preserveProviderAttempt?: boolean; failpoint?: NativeStatusCommitFailpoint; @@ -1275,7 +1150,7 @@ export async function finalizeNativeRun(input: { }), ); } - if (input.projectRunStatus) + if (input.projectRunStatus && !input.skipCommittedRunProjection) await projectCommittedRun({ db: input.db, run: admittedRun, coordinator }); if (input.projectRunStatus && !presentationAlreadyMaterialized) await materializeCommittedReviewResponse(input.db, input.runId); @@ -1760,8 +1635,17 @@ export async function finalizeNativeRun(input: { }) .where(eq(heartbeatRuns.id, run.id)) .returning(); - const [updatedRun] = await input.db.transaction(async transaction => { + const projection = await input.db.transaction(async transaction => { const tx = transaction as unknown as Db; + + + + const [before] = await tx.select({ status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))) + .for("update").limit(1); + if (!before) return null; + await notifyNativeRunProjection(tx, { companyId: run.companyId, runId: run.id }); await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id }); if (input.projectRunStatus) await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); @@ -1769,15 +1653,20 @@ export async function finalizeNativeRun(input: { if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); await notifyExecutionRunWork(tx); await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId }); - return projectRun(tx); + const [after] = await projectRun(tx); + return after ? { before, after } : null; }); + const updatedRun = projection?.after; if ( input.projectRunStatus && !alreadyEmittedByCommittedDecision && - updatedRun + projection && projection.after.status !== projection.before.status ) { - await emitAgentTaskRun(input.db, updatedRun); - void reportRunFailure(input.db, updatedRun); + await emitAgentTaskRun(input.db, projection.after); + void reportRunFailure(input.db, projection.after); + } else if (input.projectRunStatus && !alreadyEmittedByCommittedDecision && projection + && isNewTerminalRunFailure(projection.before, projection.after)) { + void reportRunFailure(input.db, projection.after); } if (input.projectRunStatus) await materializeCommittedReviewResponse(input.db, input.runId); diff --git a/server/src/services/native-runtime/native-run-projection-work.ts b/server/src/services/native-runtime/native-run-projection-work.ts new file mode 100644 index 00000000000..54656c61c64 --- /dev/null +++ b/server/src/services/native-runtime/native-run-projection-work.ts @@ -0,0 +1,75 @@ +import { and, asc, eq, gt, not, sql } from "drizzle-orm"; +import { heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativeRunProjectionScope } from "../native-run-projection-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; +import { pendingNativeRunProjectionCondition, projectCommittedRun } from "./native-run-projection.js"; + +type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; + + + +export function nativeRunProjectionWork(db: Db, project = projectCommittedRun) { + function candidates(scope?: NativeRunProjectionScope, after?: string) { + return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: nativeRunFinalizations.issueId }) + .from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) + .where(and(eq(nativeRunFinalizations.phase, "committed"), + pendingNativeRunProjectionCondition(db), + not(pendingNativeStatusEffectsCondition(db)), + scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, + scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, + after ? gt(heartbeatRuns.id, after) : undefined, + )).orderBy(asc(heartbeatRuns.id)).limit(100); + } + async function load(scope?: NativeRunProjectionScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await candidates(scope, after); + entries.push(...rows.map(row => ({ ...row, dueAt: 0 }))); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].id; + } + throw new Error("Native run projection discovery stopped"); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Native run projection stopped"); + const scope = { companyId: entry.companyId, issueId: entry.issueId, runId: entry.id }; + const [pending] = await candidates(scope); + if (!pending) return null; + const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt, nativePhase: heartbeatRuns.nativePhase, + runtimeMode: heartbeatRuns.runtimeMode, nativeIssueId: heartbeatRuns.nativeIssueId, + contextSnapshot: sql>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->>'issueId')`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, entry.id), eq(heartbeatRuns.companyId, entry.companyId))); + const [coordinator] = await db.select({ resultId: nativeRunFinalizations.resultId, issueId: nativeRunFinalizations.issueId }) + .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.runId, entry.id), + eq(nativeRunFinalizations.companyId, entry.companyId), eq(nativeRunFinalizations.issueId, entry.issueId), + eq(nativeRunFinalizations.phase, "committed"))); + if (!run || !coordinator) return null; + if (signal?.aborted) throw new Error("Native run projection stopped"); + await project({ db, run, coordinator, onlyWhenEffectsSettled: true }); + return (await candidates(scope)).length ? { ...entry, dueAt: Date.now() + 30_000 } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise = async () => true) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeRunProjection, batchSize: 25, + decodeScope: (hint: string): NativeRunProjectionScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativeRunProjectionScope) => !scope || entry.companyId === scope.companyId + && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId), + load: async (scope, signal) => await canRecover() ? load(scope, signal, canRun) : [], + check: async (entry, signal) => { + if (!canRun()) throw new Error("Native run projection paused"); + return await canRecover() ? check(entry, signal) : null; + }, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts new file mode 100644 index 00000000000..72edd36f05b --- /dev/null +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -0,0 +1,172 @@ +import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; +import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; +import { randomUUID } from "node:crypto"; +import { and, eq, exists, inArray, isNotNull, isNull, not, or, sql, type SQL } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; +import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; +import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; +import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; +import { notifyExecutionRunWork } from "../execution-control-notifications.js"; +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; +import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; +import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js"; +import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js"; + + +export function projectNativeTerminalRunStatus( + terminalState: "succeeded" | "failed" | "cancelled" | "active", +) { + return terminalState === "active" ? ("running" as const) : terminalState; +} + + + +export function nativeRunProjectionNeedsRepair(projectedStatus: "succeeded" | "failed" | "cancelled" | SQL) { + return and( + eq(heartbeatRuns.runtimeMode, "native"), + or( + inArray(heartbeatRuns.status, ["queued", "running", "failed"]), + and( + eq(heartbeatRuns.status, "succeeded"), + or( + isNotNull(heartbeatRuns.error), + isNotNull(heartbeatRuns.errorCode), + isNull(heartbeatRuns.finishedAt), + sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`, + sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`, + sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`, + sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`, + sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`, + ), + ), + ), + + + + + or( + sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`, + isNull(heartbeatRuns.finishedAt), + sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`, + sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`, + sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`, + sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`, + sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`, + and(sql`${projectedStatus} = 'succeeded'`, or(isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode))), + ), + nativeRunnerOwnershipNotHeldCondition(), + ); +} + + + +export function pendingNativeRunProjectionCondition(db: Db) { + const status = sql`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState'`; + return exists(db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.runId, nativeRunFinalizations.runId), + eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), + eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), + eq(nativeRunResults.schemaStatus, "accepted"), + inArray(status, ["succeeded", "failed", "cancelled"]), + nativeRunProjectionNeedsRepair(status), + ))); +} + +export function recoveredExecutionFailureMetadata() { + + return sql`case when ${heartbeatRuns.error} is not null or ${heartbeatRuns.errorCode} is not null + then jsonb_build_object('recoveredExecutionFailure', jsonb_build_object( + 'schema', 'paperclip.recovered_execution_failure.v1', + 'errorCode', ${heartbeatRuns.errorCode}, 'error', ${heartbeatRuns.error}, + 'observedAt', ${heartbeatRuns.updatedAt} + )) else '{}'::jsonb end`; +} + +export async function projectCommittedRun(input: { + db: Db; + onlyWhenEffectsSettled?: boolean; + run: Pick; + coordinator: Pick; +}) { + if (!input.coordinator.resultId) return; + const resultRow = await input.db + .select({ terminalState: sql`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState'` }) + .from(nativeRunResults) + .where( + and( + eq(nativeRunResults.id, input.coordinator.resultId), + eq(nativeRunResults.runId, input.run.id), + eq(nativeRunResults.companyId, input.run.companyId), + eq(nativeRunResults.issueId, input.coordinator.issueId), + eq(nativeRunResults.schemaStatus, "accepted"), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); + const terminalState = resultRow?.terminalState; + if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) { + throw new Error("native_finalization_invalid"); + } + const projectedStatus = terminalState as "succeeded" | "failed" | "cancelled"; + const needsRepair = and(eq(heartbeatRuns.id, input.run.id), eq(heartbeatRuns.companyId, input.run.companyId), + eq(heartbeatRuns.nativeIssueId, input.coordinator.issueId), nativeRunProjectionNeedsRepair(projectedStatus), + exists(input.db.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, input.run.id), eq(nativeRunFinalizations.companyId, input.run.companyId), + eq(nativeRunFinalizations.issueId, input.coordinator.issueId), eq(nativeRunFinalizations.resultId, input.coordinator.resultId), + eq(nativeRunFinalizations.phase, "committed"), + input.onlyWhenEffectsSettled ? not(pendingNativeStatusEffectsCondition(input.db)) : undefined, + )))); + const now = new Date(); + const projected = await input.db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [before] = await tx.select({ status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt }) + .from(heartbeatRuns).where(needsRepair).for("update").limit(1); + if (!before) return null; + await notifyNativeRunProjection(tx, { companyId: input.run.companyId, runId: input.run.id }); + await notifyRetainedNativeSessionCleanup(tx, { companyId: input.run.companyId, runId: input.run.id }); + await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id }); + await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id }); + await notifyUnsafeWorkspaceExportRunChange(tx, input.run); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId); + await notifyExecutionRunWork(tx); + await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId }); + const [after] = await tx + .update(heartbeatRuns) + .set({ + executionStatusDeliveryId: randomUUID(), + status: projectedStatus, + finishedAt: before.finishedAt ?? now, + nativePhase: "committed", + nativePhaseUpdatedAt: now, + ...(terminalState === "succeeded" + ? { + error: null, + errorCode: null, + + + + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) + || ${recoveredExecutionFailureMetadata()} + || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)`, + } + : { resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)` }), + updatedAt: now, + }) + .where(needsRepair) + .returning(); + return after ? { before, after } : null; + }); + + + + if (projected && projected.after.status !== projected.before.status) { + await emitAgentTaskRun(input.db, projected.after); + void reportRunFailure(input.db, projected.after); + } else if (projected && isNewTerminalRunFailure(projected.before, projected.after)) { + void reportRunFailure(input.db, projected.after); + } +}