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 85017a8f68e..0fc94c07624 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -3231,3 +3231,22 @@ The previous head's startup gate passed, with Greptile 5/5. A separate CI import hook timeout (118 agent-skills cases skipped) passed all 118 locally and received one justified job retry. New producer-fix acceptance remains pending. No merge; all prior gate, suppression-audit and remaining-duty obligations remain open. + +## Slice 52 accepted; slice 53 implementation checkpoint + +Slice 52 [#17734](https://github.com/paperclipai/paperclip/pull/17734) is verified +at `d5c540b1c0df9aba35262ba121c651806bbca161`: all current checks/statuses pass or +intentionally skip, Greptile 5/5, both review threads resolved. The preceding +startup timeout received one justified retry which passed. The interrupted local +full-suite run remains an interruption, not a local pass. The PR is draft/open. + +Slice 53, `codex/sleep-53-native-result-commit`, moves accepted-result commit after +a successful workspace barrier to scoped transaction wakes and saved deadlines. +[Its audit](2026-10-26-native-result-commit-sleep-audit.md) records both real writers, +pending-effect/cancellation handoffs, atomic claim deferrals and 96 focused +passing cases. Committed projection and resource settlement retain existing +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. diff --git a/doc/plans/2026-10-26-native-result-commit-sleep-audit.md b/doc/plans/2026-10-26-native-result-commit-sleep-audit.md new file mode 100644 index 00000000000..fb27fb28c95 --- /dev/null +++ b/doc/plans/2026-10-26-native-result-commit-sleep-audit.md @@ -0,0 +1,240 @@ +# Accepted native result commits: sleep audit + +Date: 2026-10-26. Slice 53 starts at `d5c540b1c0df9aba35262ba121c651806bbca161`, +parent [#17734](https://github.com/paperclipai/paperclip/pull/17734). The branch is +`codex/sleep-53-native-result-commit`. This is one duty from the +[recurring-server inventory](2026-10-24-recurring-server-sleep-stack.md). + +## Problem and bounded change + +The aggregate orphan recovery pass can finish three different obligations: + +1. Commit an accepted native result after successful workspace finalization. +2. Repair the run projection and consequences of an already committed decision. +3. Release runtime resources and settle the agent after finalization. + +This slice moves only the first duty to transaction signals and concrete saved +deadlines. The native finalizer remains the authority for claiming, classifying, +arbitrating and committing results. The new worker does not arbitrate statuses or +implement provider behavior. No table, migration, provider loop or external API +is added. Committed projections and resource settlement remain in existing +recovery until their own durable pending predicates and producers are migrated. + +The current default branch was refreshed during this work. Inspected master is +`49bd7a4ba6` and contains subsequent native finalization/recovery changes, +including #17525, #17531, #17614, #17641 and #17665. Full-stack integration with +master and repeat verification remain acceptance work; this branch keeps its +requested stack parent rather than silently dropping or rewriting prior slices. + +## Trigger, durable truth and quiet predicate + +`nativeResultCommitWork` reads the existing coordinator, accepted result, +heartbeat run, issue and successful workspace operation. All joins bind company, +issue and run. Eligibility requires: + +- native runtime and accepted result; +- coordinator phase other than `committed` or `terminal_failure`; +- no held native runner ownership and no pending native status effects; +- at least one successful scoped `workspace_finalize` receipt. + +A later stale failure receipt does not invalidate an earlier successful receipt. +The shared pending-effect worker owns pending effect delivery. This worker waits +quietly until that owner publishes settlement. A workspace receipt proves the +copy-back barrier; it does not prove that runtime cleanup finished. + +Discovery reads narrow candidate fields in keyset pages of 100. Execution reads +the run profile only when a candidate is due. There is no recurring scan for +empty work or retained committed/terminal history. Discovery still costs SQL at +startup and after notifications. No large-fleet query benchmark is claimed. + +The next deadline is the later of saved `nextAttemptAt` and a held lease's saved +expiry. Inspection does not advance either date. A missing lease expiry follows +the existing finalizer claim contract; it is not evidence of physical process +termination. If an invocation leaves a concrete eligible result without a saved +future deadline, only that unfinished result retains a 30-second fallback. + +## Producers and transaction ownership + +Signals use the existing DB intent/outer-commit protocol. Each write registers +and awaits its intent before the write. Durable rows remain truth; notifications +are scoped hints. Savepoint release does not dispatch before outer commit. +Rollback does not create work. Ambiguous outcomes remain owned by shared DB work +signals rather than worker-local counters or new recovery tables. + +| Producer | Boundary and purpose | +| --- | --- | +| `PaperclipControlPlanePort.completeRun` | Signal before the first accepted-result insert. Covers result arriving after an existing workspace barrier. | +| `NativeRunCoordinatorStore.completeRun` | Same contract for the PRP coordinator's independent accepted-result writer. Terminal-event reconciliation delegates here. | +| Workspace operation recorder | Signal before workspace-finalization receipt writes, including the successful barrier arriving after the accepted result. | +| Direct workspace recovery | Signal the repaired successful receipt in the same transaction. | +| `notifyNativeFinalizationWork` | Existing fanout now covers this duty: status decision, effect settlement, saved failure/retry, native execution release and other finalization transitions. | +| Initial finalizer/store claim | First-result lease writes now signal even when no decision exists. Claims with a decision use the existing fanout. | +| Workspace ownership clear/recovery | Signal the affected run when durable ownership changes. This does not grant physical ownership to the commit worker. | +| Issue mutation | Relevant status, binding, assignment and dependency changes reconstruct scoped candidates. | +| Instance execution setting | Worktree execution activation publishes a global reconstruction signal. | +| Native retry cancellation | Signal before changing the retry coordinator to terminal failure so its saved wake is cancelled immediately after commit. | + +The startup/restart recovery path for provider adoption selects resultless runs; +it does not create accepted-result work. The stopped-before-first-turn terminal +write likewise requires a null result ID. Those writes cannot enter this worker's +candidate set. Normal accepted-result execution release already uses the shared +finalization fanout. + +The supported producer model remains the application's instrumented writes in +one owning server process. Independent SQL writers and cross-process mutations +without supported ingress are not covered by a process-local notification. +Startup reconstructs durable work. This slice does not add a cross-replica bus. + +## Claim and lifecycle boundaries + +The public finalizer gains optional `onlyWhenDue`. Under the existing coordinator +row lock it rejects a future retry before acquiring a lease or charging an +attempt. A typed deferral distinguishes a live lease or moved retry from an +execution failure. Existing callers retain immediate behavior unless they opt in. +The worker rereads durable state after deferral, commit or recorded failure. + +The worker reuses `registerScopedDeadlineWork` and the delivery coordinator for +coalescing, serialized execution, scheduled deadlines, retained dirty scopes, +drain/resume and shutdown. It processes one candidate at a time. It introduces no +separate timer or event bus. Heartbeat is only the lifecycle/policy facade. + +Public registration checks task drain and the existing restore/worktree execution +suppression policy. A drain keeps its wake for resume. Suppression does not launch +recovery or keep querying. Worktree activation reconstructs from durable records. +Server registration remains under `heartbeatSchedulerEnabled` and the existing +warm-standby/shutdown gates. Coordinator shutdown joins in-flight work; startup +reconstructs any additional work committed while shutdown began. + +The periodic orphan pass sets `skipNativeResultCommit`. The reconciler skips +uncommitted result execution only. Its committed projection and resource paths +remain enabled. Direct/manual callers retain default behavior. In particular, +`phase=committed` cannot be treated as proof that every consequence is settled: +the status decision and run projection occur in separate transactions. + +## Verification and preserved failures + +The real-DB worker suite has 20 passing cases. Together with 76 startup cases, +the focused command passes **96 tests**. It covers: + +- no work and retained committed history with zero recurring SQL after settlement; +- both real accepted-result producers, barrier before and after the result; +- caller-owned outer commit and rollback for both writers; +- startup reconstruction and a successful barrier followed by stale failure; +- future retry and lease deadlines without early attempt charges; +- a retry moved between candidate discovery and the atomic claim; +- a real finalizer status-commit failpoint, rolled-back decision and saved retry; +- real pending-effect settlement waking accepted-result commit; +- first work during drain and subsequent resume; +- public restore/worktree suppression and real settings activation; +- shutdown joining active execution and restart recovering later work; +- real native cancellation disarming an existing retry; +- committed projection remaining covered by the existing reconciler. + +The projection test commits without projecting and then calls the existing +reconciler. It proves the retained recovery boundary, not a deployed crash test. +Shared transaction uncertainty behavior is inherited from its existing tests; +this slice does not claim a new network-fault qualification. + +Preserved failures: + +- Initial producer regressions: six failures and four passes. Missing signals + prevented the first result, later barrier, outer commit and drain/resume from + reactivating the worker. Wiring the real producer transactions made all ten pass. +- Cancellation regression: the test timed out at 15 seconds waiting for the + cancelled ten-minute deadline to clear. The cancellation transaction had no + signal. Adding its awaited pre-write signal makes that case and all 96 focused + cases pass. The failed command remains evidence; it is not counted as a pass. + +Logs for this local session: `/tmp/sleep53-producer-red.log`, +`/tmp/sleep53-producer-green.log`, `/tmp/sleep53-cancellation-red.log`, and +`/tmp/sleep53-cancellation-green.log`. Initial server typechecking passed before +the cancellation follow-up. Broader affected suites, final full commands, remote +checks and review are still pending at this checkpoint. + +## Limits and follow-up + +This is worker quiet, not host suspension. Cloud alarm coverage, admission fencing, +idle-sleep safety and resource blockers are unchanged. No database-suspension, +deployed-wake or fleet-savings claim is made. The aggregate orphan scan still has +other duties and still performs queries. + +Next distinct duties include committed run projection/consequences and native +resource settlement. Keep uncertainty visible until each has an owning pending +contract. The prior eight-worker suppression audit, older stack gate failures, +full-stack master propagation and every remaining inventory duty remain open. +No merge is authorized by this task. + +## Startup readiness follow-up (2026-10-27) + +The initial result-commit sweep can wait on a finalization lock. Registration now +tracks its readiness promise through the existing heartbeat shutdown owner instead +of awaiting it before HTTP startup completes. The delivery coordinator still owns +worker cancellation; no independent subscription or timer is introduced. A held +worker regression failed on the old wiring, then passed with HTTP ready, subsequent +recovery registered, and shutdown waiting until the held work settles. All 80 +startup tests pass before and after integrating pinned master +`8548173fc9efdc0fa07644fb92dcc0b6851a95e5`; production server types pass at the +integrated revision. These lifecycle checks do not establish deployed host sleep. + +## Current parent integration and routine-hint regression (2026-10-29) + +This slice now incorporates parent #17734 at +`7df5c9c44ea9e171be3c61b4430d3727b10f3337` through a normal merge. A +batched tree comparison preserves 1,203 parent-only entries, nine owned-only +entries, 11,520 identical entries and 12 automatic shared merges. The sole +manual conflict adds result-commit notification inside the existing ordinary +issue/run scope guard. Explicit company settings and dedicated result-state +notifications retain their existing scopes. The entire package tree matches +the parent, including combined Runner profile 80 and all historical fixtures. + +Six added comments in three files were removed with identical nontrivia +TypeScript token streams; no directive was removed. This includes the readiness +repair already present in the prior branch: result recovery's first `ready` +promise is owned by the existing idle/shutdown tracker. It does not block HTTP +readiness. Its held-callback startup regression remains in the complete startup +suite. The earlier Greptile 4/5 on commit `656f6722be1aedf3d52b7f853956e15838204278` +reported that blocking await; it is historical failure evidence, not an unresolved +finding to waive. The subsequent Greptile service error is not a review pass. +A fresh review on the final published head is still required. + +The new real-database regression submits three routine company-only finalization +hints to an idle result worker. A controlled baseline moves only result fanout +outside the guard and leaves all other parent guards intact. The worker issues +three candidate SELECTs, so the zero-query assertion fails. Restoring production +bytes passes the same assertion. The real workspace recorder and accepted-result +writer then produce exactly one committed decision and return to quiet. The +red phase fails one selected case in 7.35 seconds; green passes it in 6.66 seconds. +Twenty cases outside the pattern are not included in that focused count. Both +phases use fresh dedicated PostgreSQL 17 databases and restore the production +helper byte-for-byte. No test assertion or timeout is relaxed. + +### Complete current-source acceptance + +Source `2af88fa55f961a74340880240019f6cd31e21bf1` passes **2,429 distinct +cases in 18 complete files**, with no skips in those files. The 16-file main +command passes 1,746 cases in 277.97 seconds. Process recovery passes 410 in +162.42 seconds. External-chat response passes 273 in 53.41 seconds. This includes +all 21 result-worker cases and 82 startup cases, including held recovery, +nonblocking readiness and shutdown joins. Production server typechecking, build, +module boundaries, migration order and the source identity check pass. + +Fifteen dedicated PostgreSQL 17 databases use the unchanged migration 413 schema. +Embedded-only invocations use the existing fresh-per-call recovery helper. All +temporary imports are restored byte-for-byte and every selected test hash matches +after completion. No assertion or timeout is relaxed. The readiness thread was +resolved after matching its existing remote fix to the passing complete startup +suite. No old review score or service error is treated as current acceptance. + +Parent #17734 merged as `a019d00a393b57f531afe8dd8ab0b8a1820d1c4c`, after +57 clean current-head checks and Greptile 5/5. Its complete reviewed tree equals +the landed squash. Normal merge `998239b019defe52e7b5d64f884de4e871ccd736` +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 parent package tree retains its 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. Earlier failed commands remain in the record. No full local +repository suite, live provider qualification, deployed suspension, expanded Cloud +wake coverage or measured savings is claimed. diff --git a/server/src/__tests__/native-result-commit-work.test.ts b/server/src/__tests__/native-result-commit-work.test.ts new file mode 100644 index 00000000000..25edb2561e1 --- /dev/null +++ b/server/src/__tests__/native-result-commit-work.test.ts @@ -0,0 +1,340 @@ +import { cancelNativeSession } from "../services/native-runtime/native-session-executor.js"; +import { dispatchPendingNativeStatusEffects } from "../services/native-runtime/status-decision-committer.js"; +import { reconcileNativeFinalizations } from "../services/native-runtime/native-finalization-reconciler.js"; +import { notifyNativeResultCommit } from "../services/native-result-commit-notifications.js"; +import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +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, + statusDecisionEffects, statusDecisions, workspaceOperations, type Db } from "@paperclipai/db"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { nativeResultCommitWork } from "../services/native-runtime/native-result-commit-work.js"; +import { finalizeNativeRun, NativeFinalizationDeferredError } from "../services/native-runtime/native-run-finalizer.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"; + +describe("accepted result commit 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-result-commit-")); + process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = root; + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RESULT_COMMIT_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-result-commit-"); + 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.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(finalize = finalizeNativeRun) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = nativeResultCommitWork(db, finalize).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: 15_000 }); + } + 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) { + 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, + 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]); + } + it.each(["restore", "worktree"])("respects %s suppression and reactivation through heartbeat", async mode => { + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" } + : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "result-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 }).registerNativeResultCommitWork(coordinator, () => admitted); + await worker.ready; await flush(); await quiet(); + expect((await state(binding.runId)).phase).not.toBe("committed"); + if (mode === "worktree") { + try { + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await flush(); expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + } + }); + it.each(["control_plane", "coordinator"] as const)("wakes on the first %s accepted result and becomes quiet with committed history", async producer => { + await start(); await flush(); await quiet(); + const binding = await seed(); await barrier(binding); await flush(); + await complete(binding, producer); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, binding.runId)); + expect(decisions).toHaveLength(1); + await quiet(); await coordinator!.stop(); await start(); await flush(); await quiet(); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, binding.runId))).toHaveLength(1); + }); + it("ignores routine company-only finalization hints and still commits accepted work", 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 seed(); await barrier(binding); await flush(); + await complete(binding, "control_plane"); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, binding.runId))).toHaveLength(1); + await quiet(); + }); + + it("wakes when the real workspace recorder commits a later successful barrier", async () => { + await start(); await flush(); + const binding = await seed(); await complete(binding, "control_plane"); await flush(); await quiet(); + expect((await state(binding.runId)).phase).not.toBe("committed"); + await barrier(binding); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + it.each([ + ["commit", "control_plane"], ["rollback", "control_plane"], + ["commit", "coordinator"], ["rollback", "coordinator"], + ] as const)("honors outer %s for %s results", async (outcome, producer) => { + await start(); await flush(); + const binding = await seed(); await barrier(binding); await flush(); + const transaction = db.transaction(async tx => { + await complete(binding, producer, tx as unknown as Db); + await vi.advanceTimersByTimeAsync(5000); + expect(await state(binding.runId)).toBeUndefined(); + 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 state(binding.runId))?.phase).toBe(outcome === "commit" ? "committed" : undefined); + await quiet(); + }); + it("reconstructs unfinished work at startup using any successful scoped barrier", async () => { + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + await db.insert(workspaceOperations).values({ companyId, issueId: binding.issueId, heartbeatRunId: binding.runId, + phase: "workspace_finalize", status: "failed", createdAt: new Date(Date.now() + 1000) }); + await start(); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + it("atomically rejects a future retry without consuming an attempt", async () => { + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + const retryAt = new Date(Date.now() + 60_000); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: retryAt }).where(eq(nativeRunFinalizations.runId, binding.runId)); + await expect(finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", onlyWhenDue: true })) + .rejects.toBeInstanceOf(NativeFinalizationDeferredError); + expect((await state(binding.runId)).attempt).toBe(0); + await start(); + expect(coordinator!.nextWakeAt()).toBe(retryAt.getTime()); + await vi.advanceTimersByTimeAsync(59_000); + expect((await state(binding.runId)).attempt).toBe(0); + await vi.advanceTimersByTimeAsync(1000); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + it("preserves a first result committed during drain until resume", async () => { + await start(); await flush(); admitted = false; + const binding = await seed(); await barrier(binding); await complete(binding, "coordinator"); + await vi.advanceTimersByTimeAsync(120_000); + expect((await state(binding.runId)).phase).not.toBe("committed"); + await quiet(); admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + + it("disarms a saved retry as soon as durable cancellation commits", async () => { + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + const retryAt = new Date(Date.now() + 600_000); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", nextAttemptAt: retryAt }) + .where(eq(nativeRunFinalizations.runId, binding.runId)); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, binding.runId)); + await start(); expect(coordinator!.nextWakeAt()).toBe(retryAt.getTime()); + await cancelNativeSession(binding.runId, "Stop", { db, scope: "run" }); + expect((await state(binding.runId)).phase).toBe("terminal_failure"); + await flush(); await quiet(); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, binding.runId))).toHaveLength(0); + }); + + it("honors an existing finalizer lease without charging a failed attempt", async () => { + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + const leaseExpiresAt = new Date(Date.now() + 60_000); + await db.update(nativeRunFinalizations).set({ leaseOwner: "other-controller", leaseExpiresAt }) + .where(eq(nativeRunFinalizations.runId, binding.runId)); + await expect(finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", onlyWhenDue: true })) + .rejects.toBeInstanceOf(NativeFinalizationDeferredError); + expect((await state(binding.runId)).attempt).toBe(0); + await start(); expect(coordinator!.nextWakeAt()).toBe(leaseExpiresAt.getTime()); + await vi.advanceTimersByTimeAsync(59_000); expect((await state(binding.runId)).attempt).toBe(0); + await vi.advanceTimersByTimeAsync(1000); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + it("reconstructs a saved retry after restart without advancing its deadline", async () => { + const binding = await seed(); await complete(binding, "coordinator"); await barrier(binding); + const retryAt = new Date(Date.now() + 60_000); + const store = new NativeRunCoordinatorStore(db, { ...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 store.claimFinalization({ leaseOwner: "retry-owner" }); + await store.markFinalizationRetry({ leaseOwner: "retry-owner", failureCode: "transient_storage_failure", retryAfterMs: 60_000 }); + const saved = (await state(binding.runId)).nextAttemptAt!.getTime(); + expect(saved).toBeGreaterThanOrEqual(retryAt.getTime()); + await start(); expect(coordinator!.nextWakeAt()).toBe(saved); await coordinator!.stop(); + await start(); expect(coordinator!.nextWakeAt()).toBe(saved); + await vi.advanceTimersByTimeAsync(saved - Date.now()); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + it("joins an active finalizer on shutdown and reconstructs work arriving during it", async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const entered = vi.fn(); + const finalize: typeof finalizeNativeRun = async input => { entered(); await gate; return finalizeNativeRun(input); }; + await start(finalize); await flush(); + const first = await seed(); await complete(first, "control_plane"); await barrier(first); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(entered).toHaveBeenCalledTimes(1); }); + const second = await seed(); await complete(second, "coordinator"); await barrier(second); + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await delay(10); expect(stopped).toBe(false); expect(entered).toHaveBeenCalledTimes(1); + release(); await stopping; + expect((await state(first.runId)).phase).toBe("committed"); + expect((await state(second.runId)).phase).not.toBe("committed"); + await start(); await flush(); + expect((await state(second.runId)).phase).toBe("committed"); await quiet(); + }); + + it("does not consume an attempt if a retry moves after candidate discovery", async () => { + let rescheduled = false; + let retryAt = 0; + const finalize: typeof finalizeNativeRun = async input => { + if (!rescheduled) { + rescheduled = true; retryAt = Date.now() + 60_000; + await db.transaction(async tx => { + await notifyNativeResultCommit(tx, { companyId, runId: input.runId }); + await tx.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(retryAt) }) + .where(eq(nativeRunFinalizations.runId, input.runId)); + }); + } + return finalizeNativeRun(input); + }; + const binding = await seed(); await complete(binding, "coordinator"); await barrier(binding); + await start(finalize); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(coordinator!.nextWakeAt()).toBe(retryAt); }); + expect((await state(binding.runId)).attempt).toBe(0); + await vi.advanceTimersByTimeAsync(retryAt - Date.now()); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + it("retries a real atomic status-commit failure at its saved deadline", async () => { + let fail = true; + const finalize: typeof finalizeNativeRun = input => { + const failpoint = fail ? "status_projection" as const : undefined; fail = false; + return finalizeNativeRun({ ...input, failpoint }); + }; + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + await start(finalize); + const failed = await state(binding.runId); + expect(failed.phase).toBe("retryable_failure"); + expect(failed.nextAttemptAt!.getTime()).toBeGreaterThan(Date.now()); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, binding.runId))).toHaveLength(0); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(coordinator!.nextWakeAt()).toBe(failed.nextAttemptAt!.getTime()); }); + await vi.advanceTimersByTimeAsync(failed.nextAttemptAt!.getTime() - Date.now()); await flush(); + expect((await state(binding.runId)).phase).toBe("committed"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, binding.runId))).toHaveLength(1); await quiet(); + }); + it("waits quietly for pending effects and wakes from their real settlement", async () => { + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + 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 db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, binding.runId)); + await start(); await flush(); await quiet(); + expect((await state(binding.runId)).phase).toBe("retryable_failure"); + await dispatchPendingNativeStatusEffects({ db, ...binding, decisionId: committed.decisionId!, onlyWhenDue: true }); + await flush(); expect((await state(binding.runId)).phase).toBe("committed"); await quiet(); + }); + it("delegates only uncommitted results and keeps committed projection repair available", async () => { + const binding = await seed(); await complete(binding, "control_plane"); await barrier(binding); + await reconcileNativeFinalizations(db, [binding.runId], { skipNativeResultCommit: true, skipNativeWorkspaceFinalization: true }); + expect((await state(binding.runId)).phase).toBe("workspace_finalizing"); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded" }); + expect((await state(binding.runId)).phase).toBe("committed"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, binding.runId)))[0].status).toBe("running"); + await start(); await flush(); await quiet(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, binding.runId)))[0].status).toBe("running"); + await reconcileNativeFinalizations(db, [binding.runId], { skipNativeResultCommit: true, skipNativeWorkspaceFinalization: true }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, binding.runId)))[0].status).toBe("succeeded"); + await flush(); await quiet(); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 3747c5275ed..bdda480294c 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -119,6 +119,7 @@ const { registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), 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() })), 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() })), @@ -597,6 +598,13 @@ describe("startServer feedback export wiring", () => { } }); + 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(); + expect(heartbeatServiceMock.registerNativeResultCommitWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerNativeResultCommitWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers native reassessment only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -1058,7 +1066,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 }); + 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.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { @@ -1214,14 +1222,17 @@ describe("startServer feedback export wiring", () => { } }); - it.each(["retained cleanup", "workspace finalization"] as const)("becomes ready while %s is held but joins it at shutdown", async owner => { + it.each(["retained cleanup", "workspace finalization", "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 - : heartbeatServiceMock.registerNativeWorkspaceFinalizationWork; - register.mockReturnValueOnce({ ready, stop }); + : owner === "workspace finalization" ? heartbeatServiceMock.registerNativeWorkspaceFinalizationWork + : heartbeatServiceMock.registerNativeResultCommitWork; + 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); let started = false; const starting = startServer().then(server => { started = true; return server; }); @@ -1236,7 +1247,7 @@ describe("startServer feedback export wiring", () => { const server = await starting; stopping = server.shutdown().then(() => { stopped = true; }); await settleStartupHousekeeping(); - expect(stop).toHaveBeenCalledOnce(); + if (owner !== "native result commit") expect(stop).toHaveBeenCalledOnce(); expect(stopped).toBe(false); } finally { releaseCleanup(); diff --git a/server/src/index.ts b/server/src/index.ts index ef3038a3288..7e221c2191d 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1842,6 +1842,12 @@ async function startServerWithDatabaseTeardown( server.on("close", worker.stop); + trackHeartbeatSchedulerWork(worker.ready); + } + if (config.heartbeatSchedulerEnabled) { + const worker = heartbeat.registerNativeResultCommitWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); + trackHeartbeatSchedulerWork(worker.ready); } if (config.heartbeatSchedulerEnabled) await heartbeat.registerNativeStatusReassessmentWork(deliveryWork, @@ -1995,7 +2001,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 }) + .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 }) .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 9c8c9fe6e48..ed5a2fd5a4d 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 = { + nativeResultCommit: "native-result-commit", nativeWorkspaceFinalization: "native-workspace-finalization", retainedNativeSessionCleanup: "retained-native-session-cleanup", nativeStatusReassessment: "native-status-reassessment", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index a1333fdc22b..e204c0e3568 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +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"; import { registerNativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js"; @@ -4094,6 +4095,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeResultCommitWork: ( + coordinator: Parameters["register"]>[0], canRun: () => boolean, + ) => nativeResultCommitWork(db).register(coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerNativeWorkspaceFinalizationWork: ( coordinator: Parameters[0], canRun: () => boolean, ) => registerNativeWorkspaceFinalizationWork(coordinator, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 4e755f92582..fd7fa043005 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 }) { + 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 }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1993,6 +1993,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipNativeReviewResponse: opts?.skipNativeReviewResponse, skipNativeStatusReassessment: opts?.skipNativeStatusReassessment, skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization, + skipNativeResultCommit: opts?.skipNativeResultCommit, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 6e034f42303..e088559962b 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; @@ -627,6 +628,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDirectoryCleanup(tx); await notifyInstructionCollection(tx); await notifyCancelledWorkspacePreservation(tx); + await notifyNativeResultCommit(tx); await notifyNativeWorkspaceFinalization(tx); } if (patch.enableChatConnectors !== undefined) await notifyVoiceChange(tx); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index c492a313e6d..a33129c30fa 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; @@ -11503,6 +11504,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 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 29b5479b571..1f1952d696a 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +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"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; @@ -9,6 +10,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { if (scope.issueId || scope.runId) { + await notifyNativeResultCommit(transaction, scope); await notifyNativeWorkspaceFinalization(transaction, scope); await notifyRetainedNativeSessionCleanup(transaction, scope); await notifyNativeStatusReassessment(transaction, scope); diff --git a/server/src/services/native-result-commit-notifications.ts b/server/src/services/native-result-commit-notifications.ts new file mode 100644 index 00000000000..1adadbbf4a8 --- /dev/null +++ b/server/src/services/native-result-commit-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeResultCommitScope = { companyId: string; runId?: string; issueId?: string }; + +export function notifyNativeResultCommit(tx: object, scope?: NativeResultCommitScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeResultCommit, + scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"); +} diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index ad0d46bf05b..819d4ed3fb7 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -388,6 +388,7 @@ type NativeReconciliationOptions = { skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipNativeWorkspaceFinalization?: boolean; + skipNativeResultCommit?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -671,6 +672,7 @@ async function reconcileNativeFinalizationsPass( } continue; } + if (options.skipNativeResultCommit && row.coordinatorPhase !== "committed") continue; try { const finalized = await finalizeNativeRun({ db, @@ -723,6 +725,7 @@ async function reconcileNativeFinalizationsPass( } continue; } + if (options.skipNativeResultCommit && row.coordinatorPhase !== "committed") continue; const replayDecision = resolveNativeReconciliationStatus({ facts: { canonicalReplay: true }, priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, diff --git a/server/src/services/native-runtime/native-result-commit-work.ts b/server/src/services/native-runtime/native-result-commit-work.ts new file mode 100644 index 00000000000..0a221e1c53e --- /dev/null +++ b/server/src/services/native-runtime/native-result-commit-work.ts @@ -0,0 +1,101 @@ +import { and, asc, eq, exists, gt, not, notInArray } from "drizzle-orm"; +import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativeResultCommitScope } from "../native-result-commit-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; +import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js"; +import { finalizeNativeRun, NativeFinalizationDeferredError, recordNativeFinalizationFailure } from "./native-run-finalizer.js"; +import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; + +type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; + + + +export function nativeResultCommitWork(db: Db, finalize = finalizeNativeRun) { + function candidates(scope?: NativeResultCommitScope, after?: string) { + return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, + issueId: nativeRunFinalizations.issueId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + }).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))) + .innerJoin(nativeRunResults, 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"))) + .where(and(eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition(), + notInArray(nativeRunFinalizations.phase, ["committed", "terminal_failure"]), + not(pendingNativeStatusEffectsCondition(db)), + exists(db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, nativeRunFinalizations.companyId), + eq(workspaceOperations.issueId, nativeRunFinalizations.issueId), eq(workspaceOperations.heartbeatRunId, nativeRunFinalizations.runId), + eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded"), + ))), + scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, + scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined, + after ? gt(nativeRunFinalizations.runId, after) : undefined, + )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); + } + function entry(row: Awaited>[number]): Entry { + return { id: row.id, companyId: row.companyId, issueId: row.issueId, + dueAt: Math.max(row.nextAttemptAt?.getTime() ?? 0, row.leaseOwner ? row.leaseExpiresAt?.getTime() ?? 0 : 0) }; + } + async function load(scope?: NativeResultCommitScope, 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(entry)); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].id; + } + throw new Error("Native result commit discovery stopped"); + } + async function check(current: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Native result commit stopped"); + const scope = { companyId: current.companyId, issueId: current.issueId, runId: current.id }; + const [row] = await candidates(scope); + if (!row) return null; + const due = entry(row); + if (due.dueAt > Date.now()) return due; + const [run] = await db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, current.companyId), eq(heartbeatRuns.id, current.id))); + if (!run) return null; + if (signal?.aborted) throw new Error("Native result commit stopped"); + try { + await finalize({ db, runId: current.id, workspaceFinalizeStatus: "succeeded", projectRunStatus: true, + onlyWhenDue: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, + preserveProviderAttempt: Boolean(readNativeWorkspaceSyncReference(run.profile?.nativeWorkspaceSync)), + }); + } catch (error) { + + if (!(error instanceof NativeFinalizationDeferredError)) { + await recordNativeFinalizationFailure({ db, runId: current.id, error, projectRunStatus: true }); + } + } + const [remaining] = await candidates(scope); + if (!remaining) return null; + const next = entry(remaining); + + + return next.dueAt > Date.now() ? next : { ...next, dueAt: Date.now() + 30_000 }; + } + function register(coordinator: ReturnType, canRun: () => boolean, + canRecover: () => Promise = async () => true) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeResultCommit, batchSize: 1, + decodeScope: (hint: string): NativeResultCommitScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (item: Entry, scope?: NativeResultCommitScope) => !scope || item.companyId === scope.companyId + && (!scope.issueId || item.issueId === scope.issueId) && (!scope.runId || item.id === scope.runId), + load: async (scope, signal) => await canRecover() ? load(scope, signal, canRun) : [], + check: async (item, signal) => { + if (!canRun()) throw new Error("Native result commit paused"); + return await canRecover() ? check(item, signal) : null; + }, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-run-coordinator-store.ts b/server/src/services/native-runtime/native-run-coordinator-store.ts index 36077aa71a6..6bbff21cfdb 100644 --- a/server/src/services/native-runtime/native-run-coordinator-store.ts +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; @@ -392,6 +393,7 @@ export class NativeRunCoordinatorStore { return { disposition: "duplicate" as const, resultId: existing.id }; } + await notifyNativeResultCommit(tx, this.#binding); await notifyNativeWorkspaceFinalization(tx, this.#binding); const [inserted] = await tx .insert(nativeRunResults) @@ -530,6 +532,7 @@ export class NativeRunCoordinatorStore { throw new Error("native_finalization_lease_conflict"); } if (row.decisionId) await notifyNativeFinalizationWork(tx, this.#binding); + else await notifyNativeResultCommit(tx, this.#binding); if ( row.leaseOwner === input.leaseOwner && row.leaseExpiresAt && diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 5eff107290c..833cccaca14 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; @@ -173,9 +174,16 @@ async function acceptedInteractionFromRun(input: { .then((rows) => rows[0] ?? null); } +export class NativeFinalizationDeferredError extends Error { + constructor(readonly reason: "lease_busy" | "retry_not_due") { + super(`native_finalization_${reason}`); + } +} + async function claimCoordinator(input: { db: Db; runId: string; + onlyWhenDue?: boolean; preserveProviderAttempt?: boolean; }) { const leaseOwner = `native-finalizer:${randomUUID()}`; @@ -210,14 +218,17 @@ async function claimCoordinator(input: { } if (coordinator.phase === "terminal_failure") throw new Error("native_finalization_terminal_failure"); + if (input.onlyWhenDue && coordinator.nextAttemptAt && coordinator.nextAttemptAt > now) + throw new NativeFinalizationDeferredError("retry_not_due"); if ( coordinator.leaseOwner && coordinator.leaseExpiresAt && coordinator.leaseExpiresAt > now && coordinator.leaseOwner !== leaseOwner ) - throw new Error("native_finalization_lease_busy"); + throw new NativeFinalizationDeferredError("lease_busy"); if (coordinator.decisionId) await notifyNativeFinalizationWork(tx, coordinator); + else await notifyNativeResultCommit(tx, coordinator); const [updated] = await tx .update(nativeRunFinalizations) .set({ @@ -1208,6 +1219,8 @@ export async function finalizeNativeRun(input: { db: Db; runId: string; workspaceFinalizeStatus: "succeeded" | "failed"; + + onlyWhenDue?: boolean; /** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */ projectRunStatus?: boolean; skipNativeResponsePresentation?: boolean; @@ -1232,6 +1245,7 @@ export async function finalizeNativeRun(input: { db: input.db, runId: input.runId, preserveProviderAttempt: input.preserveProviderAttempt, + onlyWhenDue: input.onlyWhenDue, }); const coordinator = claim.coordinator; if (!claim.leaseOwner && coordinator.phase === "committed") { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index be78a042cff..69b8b5149d9 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { releaseRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import type { NativeCleanupOutcome } from "./retained-native-session-cleanup.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -7460,6 +7461,7 @@ export async function cancelNativeSession( if (!written) throw new Error("native_cancellation_binding_changed"); // Cancellation is also an authority fence for a durable retry whose // preceding provider has already failed. No in-memory session is needed. + await notifyNativeResultCommit(tx, { companyId: cancellationContext.companyId, runId }); await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts index 3226c359f3a..c908a7d004b 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization, releaseNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { beginRunPhase } from "../execution-preparation-timing.js"; @@ -298,6 +299,7 @@ export async function withNativeWorkspaceFinalizationOwnership( joinedOwnerTokens.add(owner.token); try { await input.db.transaction(async tx => { + await notifyNativeResultCommit(tx, { companyId: input.companyId, runId: input.runId }); await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, runId: input.runId }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.runId }); await tx.update(heartbeatRuns).set({ diff --git a/server/src/services/native-runtime/native-workspace-owner-recovery.ts b/server/src/services/native-runtime/native-workspace-owner-recovery.ts index 59dc5a43db0..60a92a72b89 100644 --- a/server/src/services/native-runtime/native-workspace-owner-recovery.ts +++ b/server/src/services/native-runtime/native-workspace-owner-recovery.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -67,6 +68,7 @@ export async function resumeNativeWorkspaceAfterOwnerStop(input: { const now = new Date(); const proof = { owner: prior, resultId: result.id, actorId: input.actorId, stopEvidence: confirmation.stopEvidence, confirmedAt: now.toISOString() }; + await notifyNativeResultCommit(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId }); await tx.update(heartbeatRuns).set({ diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index 9b58bea9cc8..47faa2e5231 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "../native-workspace-finalization-notifications.js"; import { createAgentIdentityRedactor } from "../agent-identity-redaction.js"; import { and, asc, eq, gt, or } from "drizzle-orm"; @@ -409,6 +410,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { ignoredAttentionRequests: normalizationDiagnostics.ignoredAttentionRequests, }, } as unknown as Record; + await notifyNativeResultCommit(tx, this.#binding); await notifyNativeWorkspaceFinalization(tx, this.#binding); const inserted = await tx.insert(nativeRunResults).values({ companyId: this.#binding.companyId, diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index e2e3b81aa52..7bd7e2ab658 100644 --- a/server/src/services/workspace-operations.ts +++ b/server/src/services/workspace-operations.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { randomUUID } from "node:crypto"; @@ -481,6 +482,8 @@ export function workspaceOperationService(db: Db) { async function writeState(write: (database: Db) => Promise): Promise { if (recordInput.phase !== "workspace_finalize" || !input.heartbeatRunId) return write(db); return db.transaction(async tx => { + await notifyNativeResultCommit(tx, { companyId: input.companyId, + runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined }); await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId, runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined }); await notifyNativeStatusReassessment(tx, { companyId: input.companyId, diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index e82cf5cc0e7..ee185f8be91 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -1,3 +1,4 @@ +import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -248,6 +249,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ const next = { ...current!, revision: current!.revision + 1, phase: "reconcile_effects" as const, state: "complete" as const, nextAttemptAt: null }; const activity = await db.transaction(async tx => { + await notifyNativeResultCommit(tx, input); await notifyNativeWorkspaceFinalization(tx, input); await notifyNativeStatusReassessment(tx, input); await tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,