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); + } +}