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 26ea1eea102..ef7369d178c 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -3270,3 +3270,20 @@ 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. + +## Slice 54 accepted; slice 55 recovery-notice resolution + +Slice 54 [#17752](https://github.com/paperclipai/paperclip/pull/17752) is verified at +`f98e6b5b57558c78ba111170ab3d8d19ff667827`: all 53 checks/status entries passed, +Greptile 5/5, zero unresolved threads, 400 passing local cases, workspace types +and build. The PR remains draft/open and unmerged. Intermediate failures and the +full-local-suite limitation remain recorded in its audit and PR body. + +Slice 55, `codex/sleep-55-native-recovery-notices`, moves only committed native +finalization recovery-notice resolution to scoped transaction wakes. The existing +action rows remain truth; the owner rechecks reused action identities under +lifecycle locks. [Its audit](2026-10-26-native-recovery-notice-sleep-audit.md) +records producer coverage, 102 passing worker/startup cases and 340 passing +affected cases (442 total), plus workspace typechecking. Build and remote +acceptance remain pending. All other inventory, prior gate failures, +suppression audits and master propagation remain active obligations. diff --git a/doc/plans/2026-10-26-native-recovery-notice-sleep-audit.md b/doc/plans/2026-10-26-native-recovery-notice-sleep-audit.md new file mode 100644 index 00000000000..1f4f4fbff48 --- /dev/null +++ b/doc/plans/2026-10-26-native-recovery-notice-sleep-audit.md @@ -0,0 +1,275 @@ +# Committed native recovery notices: sleep audit + +Date: 2026-10-26. Slice 55 starts at +`f98e6b5b57558c78ba111170ab3d8d19ff667827`, +[parent #17752](https://github.com/paperclipai/paperclip/pull/17752). +Branch: `codex/sleep-55-native-recovery-notices`. The full scope remains in the +[recurring-server migration ledger](2026-10-24-recurring-server-sleep-stack.md). + +## One obligation and its owner + +A native finalization may commit while its recovery notice remains active. The +old periodic finalizer replay called `resolveCommittedFinalizationRecovery` to +resolve that notice. This slice extracts that duty into `native-recovery-notices` +and schedules it from committed changes. The recovery-action service still owns +the actual resolution write and its existing notifications. Other watchdogs, +external conversation settlement, Muse reconciliation and physical resource +settlement are separate obligations. + +The existing `issue_recovery_actions` rows are durable truth. No new table, +per-feature timer or provider branch is introduced. Direct/manual finalizer calls +retain notice resolution by default. The periodic path passes a narrow +`skipNativeRecoveryNotices` flag; all its other completion/resource behavior +remains enabled. A real reconciler test verifies the resource-settlement callback +still runs when notice resolution is delegated. + +## Pending predicate and quiet states + +A candidate binds company, issue and run across the native coordinator, heartbeat +run and issue. The coordinator must be committed and have a result. Discovery +excludes held runner ownership, pending status effects and unfinished run +projection. A correlated notice lookup requires: + +- The same company, source issue and `evidence.runId`. +- Kind `active_run_watchdog` and status `active` or `escalated`. +- `wakePolicy.kind = resume_native_run`, or the existing special cause + `native_workspace_finalization_owner_unverified`. + +An exhausted owner-verification notice can still need resolution even after its +retry policy is cleared. Unrelated notices, uncommitted runs, resolved history, +and otherwise idle configuration do not arm a recurring scan. Discovery reads +IDs in keyset pages of 100 and the shared worker executes company-fair batches of +25. Real resolution failures retain shared per-entry backoff, starting at 30 +seconds and bounded at five minutes. A successful attempt with remaining work +retains a scoped 30-second retry. Startup reconstructs durable unfinished notices. + +The owner first checks for a matching notice to avoid taking lifecycle locks on +the normal already-settled path. That read is an optimization, not authority. +Actual resolution locks coordinator, issue, run, then matching actions, following +native recovery's existing order. It rechecks the current committed coordinator, +run binding and physical ownership. The background caller also checks effect and +projection settlement under those locks. The existing action service performs +the resolution inside that transaction. + +## Why the action must be read again + +`upsertSourceScoped` can reuse an active action ID and change its run, cause, +fingerprint or policy. Resolving a captured ID alone can erase a newer incident. +The owner selects matching actions only after acquiring lifecycle locks and locks +them before calling the resolution service. The regression holds the run row lock, +starts resolution, changes the notice through the real upsert service, and then +releases the lock. The new notice stays active. A separate four-writer regression +observes exactly one recovery-action UPDATE. + +## Producer and boundary audit + +| Producer | Wake contract | +| --- | --- | +| Recovery-action insert, supersession, in-place update and budget exhaustion | Scoped notice hint before the existing transaction write. This includes changes to kind or wake policy alone, not just run/cause changes. | +| Recovery-action resolution | Signal only after a matching pending row is found and before the write. An empty resolution call must not wake this worker. | +| Native decision/finalization and pending-effect settlement | Existing finalization notification fanout now includes notices. The committed transaction remains authoritative. | +| Run projection and heartbeat lifecycle status writers | Scoped signal before the write; finishing projection or releasing an ownership hold can make a notice actionable. | +| Relevant issue changes | Existing issue transaction sends an issue-scoped hint. | +| Worktree execution activation | Global reconstruction hint through the existing settings transaction. | +| Recovery route reopening a resolved action | Explicit issue-scoped hint in the reopening transaction. | + +`native-workspace-export-retry` already uses the finalization fanout when it changes +the action to `resume_native_run`; its coordinator becomes `result_accepted`, so +notice resolution waits for the later commit. Native workspace owner recovery +likewise queues result finalization; the subsequent commit supplies the activation. + +Other direct action writers were reviewed. Legacy execution admission and +historical backfill create different causes without a native-resume policy. +Backfill, system recovery, workspace owner progress and queue continuation writes +preserve the top-level run/cause/kind/status eligibility while adding evidence. +The wake-queue adapter adds failure-block evidence. Explicit continuation creates +resolved authorization receipts or resolves previous actions. Safe replacement, +workspace reconstruction/export recovery and execution-recovery resolution remove +active obligations. Their existing queued candidate is rechecked and disarmed; +they cannot create new eligible notice work. Recovery-service monitor/deadline and +board-disposition writes do not introduce the matching native policy. New eligible +notice creation remains with the instrumented action service. + +All producer hints use the existing database intent and outer transaction protocol. +Nested savepoint release is not commit. Unknown outcomes stay with that shared +protocol. Process-local signals cover instrumented writers in this process; +independent SQL writers and replicas without supported ingress are not newly +supported. No database LISTEN connection is added. + +The notification boundary explicitly copies only company, issue and run IDs. A +structurally compatible caller can carry additional internal fields; serializing +its entire object is not safe. A real finalization regression exposed this during +implementation when a caller's database object caused circular JSON serialization. +The corrected hint uses the same narrow payload contract as adjacent workers. + +## Lifecycle and verification + +Public heartbeat registration preserves task drain, restore and worktree execution +suppression. Settings activation reconstructs work. The coordinator retains a +first write during drain and a failed retry through drain/restart, stops future +subscription work on shutdown and joins active resolution. A new startup discovers +notices written while stopped. + +At this checkpoint, **102 tests pass**: 22 new worker/owner cases and 80 startup +cases. Coverage includes startup, zero idle SQL, resolved history, real first and +late producers, outer commit/rollback, finalizer commit, policy/kind-only changes, +supersession, escalated owner notices, concurrent resolution, action reuse while +waiting for a real row lock, no-op resolution, pending effects, projection, +ownership release, suppression/activation, drain/retry/restart, shutdown and +retained periodic resource settlement. + +Preserved failures: + +- Before producer wiring, six tests produced three failures: late notice, outer + commit and later finalizer commit did not wake the worker. The other three passed. +- The first wiring run failed all six notice tests while all 80 startup cases + passed. A stronger fixture assertion exposed `side_effect_planning_failed` with + circular JSON serialization. The hint now copies its three supported IDs. All + initial and expanded cases pass with the real finalizer. + +Local logs: `/tmp/sleep55-initial-red.log`, `/tmp/sleep55-producers-green.log` +(the failed intermediate run), `/tmp/sleep55-producer-diagnostic.log`, +`/tmp/sleep55-lifecycle.log` (96 passes), and `/tmp/sleep55-final-lifecycle.log` +(102 passes). Six complete affected suites also pass: 340 cases across native finalization +recovery, finalizer telemetry, recovery actions, the status arbiter corpus, +result-commit work and run projection. Total: **442 passing local cases**. +Workspace typechecking passes. Evidence is in `/tmp/sleep55-affected.log` and +`/tmp/sleep55-types.log`. Build, publication and exact-head CI/review remain +pending at this checkpoint. The full local suite is not claimed: this host's +embedded PostgreSQL startup previously exhausted semaphore capacity. No further +shared OS cleanup was performed; fresh-host CI is required for full acceptance. + +## Remaining work + +Master was refreshed; `origin/master` was subsequently observed at `125ddcabe5`. +Full-stack integration of later master changes and fresh gates remains required. +The older failed gates and eight-worker restore/worktree suppression audit remain +open. All other duties in the ledger remain in scope. This establishes worker +behavior, not deployed database suspension, host sleep or fleet savings. The stack +is unmerged, and this request does not authorize merging it. + +## Reply and notice contention repair + +The committed-response owner previously locked issue, run, then coordinator. +Notice resolution, status arbitration, and physical workspace recovery lock +coordinator, issue, then run. Concurrent reply and notice repair could therefore +deadlock even though both were authorized to settle the same committed result. +The reply owner now acquires and validates its exact committed, unleased +coordinator first, then retains its existing issue/run and presentation checks. +No publication authority, completion policy, retry cadence, or transaction scope +is broadened. The change belongs with this notice worker because introducing +that concurrent owner exposed the conflicting order. + +The regression uses a real committed Telegram reply, real notice creation, and +the real resolution/publication owners. It holds the notice transaction at its +first coordinator lock, observes the reply's actual PostgreSQL blocking state, +then continues notice resolution with a bounded local lock timeout. Before the +fix PostgreSQL rejected reply recovery with `40P01` (deadlock detected). After +the fix both operations commit, the notice resolves, and exactly one accepted +comment/publication exists; replay publishes nothing. + +The focused regression passed, followed by 73 selected committed/accepted-response +chat cases (1,570 unrelated chat cases filtered out) and all 141 cases in the +notice, finalizer telemetry, startup, and response-authority suites. Logs: +`/tmp/sleep55-lock-order-{red,green,chat,consumers}.log`. This is 214 distinct +passing cases across five files, not a full chat-file or full repository test +run. Integrated typecheck, build, and fresh remote checks remain required. + +## Projection fixture follow-through after the lock-order repair + +The coordinator-first response-recovery guard exposed stale projection fixtures. +Thirteen cases inserted a native run without a committed finalization record. +Recovery correctly stopped before loading that run, so the expected measured +projection was absent. The failure reproduced locally against a new disposable +PostgreSQL database: 13 failures and four passes, matching the CI failures on +later stack branches. + +The fixture now uses the existing accepted-result and finalization setup. It +retains a future retry while comparing narrow and full-row reads, so both paths +observe the same state without publishing between comparisons. The selected +answer guard, typed context flags, non-object result evidence, payload byte +limits, unchanged-row assertions, company/issue scope checks, and separate real +publication comparison remain. All 17 cases now pass; server typechecking also +passes. The existing external-test-database convention permits a real database +replay without requiring another embedded PostgreSQL instance. No production +lock, recovery eligibility, retry policy, or projection was weakened. + +## Startup readiness follow-up (2026-10-27) + +A held recovery-notice sweep reproduced HTTP startup waiting for task locks. +Registration now tracks its readiness promise through the existing heartbeat +shutdown owner instead of awaiting it on the readiness path. The coordinator +continues to own worker cancellation. The regression proves HTTP readiness and +subsequent registration proceed while shutdown waits for the held work. All 86 +startup cases pass on the master-integrated branch; no timer or persistence is +added. Full integrated typechecking and head-specific CI remain separate gates. + +## Current parent integration and routine-hint regression (2026-10-29) + +This slice incorporates parent #17752 at +`a49a2dac13b0e1c19da5d3d6b435c781703b1c0b` through a normal merge. A +batched tree comparison preserves 1,209 parent-only entries, eight owned-only +entries, 11,524 identical entries and 12 automatic shared merges. Two manual +paths remain explicit: ordinary notice fanout is inside the existing issue/run +guard; the chat test's leading import conflict retains all four independent +imports and removes only the three marker lines. Every other automatically +merged byte in that test remains intact before comment removal. The complete +package tree matches parent profile 80, including all historical identities. + +Thirteen added comment objects in five files were removed with identical nontrivia +TypeScript token streams. No directive was removed. Notice settlement retains +coordinator, issue, run and action lock ordering; exact current incident matching; +existing action-resolution side effects; and the separate status/effect/resource +owners. A discovered action ID alone is not authority to resolve a later incident. + +The new real-database regression submits three routine company-only finalization +hints after startup becomes quiet. A controlled baseline moves only notice fanout +outside the guard and keeps every other parent guard. It issues three candidate +SELECTs and fails the zero-query assertion. Restoring production bytes passes the +same assertion. The real notice writer then creates and resolves an action for a +committed run, returns outcome `restored`, and settles back to quiet. Red fails one +selected case in 7.46 seconds; green passes that same case. Twenty-two cases outside +the focused pattern are not counted as passing there. Both phases use fresh +dedicated PostgreSQL 17 databases and restore the production helper exactly. + +The complete main command passes 1,873 cases in 21 files in 309.69 seconds, with +no skips. Process recovery and external-chat response run in separate database +phases. The giant chat integration file is intentionally a focused selection, +including the real reply/notice row-lock regression and related provider and +current-authority cases; unselected chat cases do not constitute acceptance. +All test assertions and timeouts are unchanged. + +### Complete current-source acceptance + +Source `730e00a7b007a36d188ca0a328ef050d3a25d1ee` passes **2,556 cases +in 23 complete files**, with no skips in those files. The main command passes +1,873 cases in 21 files in 309.69 seconds. Process recovery passes 410 in 163.54 +seconds; external-chat response passes 273 in 53.48 seconds. A separate focused +chat selection passes 18 cases in 13.44 seconds, including the actual reply/notice +row-lock race, Slack/Telegram recovery, current-authority refusals, canonical +response binding and internal GitHub navigation. The other 1,796 cases in that +chat file remain unselected. Total selected passing cases: 2,574. This is not full +chat-suite or repository acceptance. + +The complete files include all 23 notice-worker cases, 17 presentation/ownership +read cases and 88 startup cases. Production server typechecking, build, module +boundaries, migration order and the source identity check pass. Twenty 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 24 selected file hashes match after completion. +No assertion or timeout is relaxed. The focused scope green takes 6.69 seconds; +the intended failing unscoped baseline is not counted as acceptance. + +Parent #17752 merged as `50fe32684d2d4e1db129e3c1bb455571344a0c09`, after +57 clean current-head checks and Greptile 5/5. Its complete reviewed tree equals +the actual squash. Normal merge `a0f2b3f4348b078219a53d8cc135c84202ba505e` +adopts that ancestry with this child's entire tested tree unchanged. This slice +can now target master. Fresh exact-head CI and review remain separate gates. + +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. Earlier failures in this audit retain their revision and +subsequent disposition. No deployed sleep, expanded Cloud alarm coverage or +measured savings is claimed. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index c24d5eb5051..c8fea0a9b05 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,6 @@ +import { resolveCommittedNativeRecoveryNotices, pendingNativeRecoveryNotices } from "../services/native-runtime/native-recovery-notices.js"; +import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; import { nativeResponsePresentationWork } from "../services/native-runtime/native-response-presentation-work.js"; import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js"; @@ -74469,6 +74472,54 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("repairs a committed reply and its recovery notice without reversing row locks", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const scope = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId }; + let response: Promise | undefined; + try { + const [run] = await readHeartbeatRunsById(context.runId); + await projectCommittedRun({ db, run, coordinator: context.coordinator }); + const notice = await issueRecoveryActionService(db).upsertSourceScoped({ + companyId: scope.companyId, sourceIssueId: scope.issueId, kind: "active_run_watchdog", + ownerType: "system", cause: "native_workspace_sync_out_failed", fingerprint: scope.runId, + evidence: { runId: scope.runId }, nextAction: "Retry finalization", + wakePolicy: { kind: "resume_native_run", runId: scope.runId }, + }); + expect(await pendingNativeRecoveryNotices(db, scope)).toHaveLength(1); + await db.transaction(async tx => { + + + await tx.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, scope.runId)).for("update"); + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`) as unknown as Array<{ pid: number }>; + response = recoverCommittedNativeChatResponse(db, { ...scope, onlyWhenDue: true }); + + void response.catch(() => {}); + await vi.waitFor(async () => { + const [state] = await db.execute(sql`select exists ( + select 1 from pg_stat_activity where datname = current_database() + and ${backend.pid} = any(pg_blocking_pids(pid)) + ) as blocked`) as unknown as Array<{ blocked: boolean }>; + expect(state.blocked).toBe(true); + }); + await tx.execute(sql`set local lock_timeout = '1s'`); + await resolveCommittedNativeRecoveryNotices(tx as unknown as Db, scope, true); + }); + await expect(response).resolves.toBe(true); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, notice.id)))[0]) + .toMatchObject({ status: "resolved", outcome: "restored" }); + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, scope.runId)); + expect(comments).toHaveLength(1); + expect(comments[0].body).toBe(context.result.summary); + expect(await readChatPublicationsByCommentId(comments[0].id)).toHaveLength(1); + await expect(recoverCommittedNativeChatResponse(db, scope)).resolves.toBe(false); + expect(await pendingNativeRecoveryNotices(db, scope)).toEqual([]); + } finally { + await response?.catch(() => {}); + await context.service.shutdown(); + } + }); + it.each(["slack", "telegram"] as const)( "recovers a committed %s answer once without rerunning or changing review/quarantine evidence", async (provider) => { diff --git a/server/src/__tests__/native-recovery-notice-work.test.ts b/server/src/__tests__/native-recovery-notice-work.test.ts new file mode 100644 index 00000000000..ad70c1b6005 --- /dev/null +++ b/server/src/__tests__/native-recovery-notice-work.test.ts @@ -0,0 +1,345 @@ +import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; +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 { nativeRecoveryNoticeWork } from "../services/native-runtime/native-recovery-notice-work.js"; +import { resolveCommittedNativeRecoveryNotices } from "../services/native-runtime/native-recovery-notices.js"; +import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; +import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { issueRecoveryActions } from "@paperclipai/db"; +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 recovery notice 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-recovery-notice-")); + process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = root; + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RECOVERY_NOTICE_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-recovery-notice-"); + 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 = resolveCommittedNativeRecoveryNotices) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + worker = nativeRecoveryNoticeWork(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; + } + async function committed() { + const binding = await accepted(); + const result = await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect(result.phase, JSON.stringify(await state(binding.runId))).toBe("committed"); + return binding; + } + async function notice(binding: Awaited>, writer: Db = db, + patch: Partial["upsertSourceScoped"]>[0]> = {}) { + return issueRecoveryActionService(writer).upsertSourceScoped({ companyId, sourceIssueId: binding.issueId, + kind: "active_run_watchdog", ownerType: "system", cause: "native_workspace_sync_out_failed", + fingerprint: binding.runId, evidence: { runId: binding.runId }, nextAction: "Retry finalization", + wakePolicy: { kind: "resume_native_run", runId: binding.runId }, ...patch }); + } + async function action(id: string) { + return db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)).then(rows => rows[0]); + } + it("reconstructs pending notices and leaves resolved history quiet across restart", async () => { + const binding = await committed(); const pending = await notice(binding); + await start(); await flush(); expect(await action(pending.id)).toMatchObject({ status: "resolved", outcome: "restored" }); + await quiet(); const settled = await action(pending.id); + await coordinator!.stop(); await start(); await flush(); await quiet(); expect(await action(pending.id)).toEqual(settled); + }); + it("ignores routine company-only finalization hints and still resolves a real notice", 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 committed(); + const pending = await notice(binding); await flush(); + expect(await action(pending.id)).toMatchObject({ status: "resolved", outcome: "restored" }); + await quiet(); + }); + + it("wakes from a real notice created after startup is quiet", async () => { + const binding = await committed(); await start(); await flush(); await quiet(); + const pending = await notice(binding); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it.each(["commit", "rollback"])("honors the notice writer's outer %s", async outcome => { + const binding = await committed(); await start(); await flush(); + let actionId = ""; + const transaction = db.transaction(async tx => { + actionId = (await notice(binding, tx as unknown as Db)).id; + await vi.advanceTimersByTimeAsync(5000); + expect(await action(actionId)).toBeUndefined(); + if (outcome === "rollback") throw new Error("rollback notice"); + }); + if (outcome === "rollback") await expect(transaction).rejects.toThrow("rollback notice"); else await transaction; + await flush(); + if (outcome === "rollback") expect(await action(actionId)).toBeUndefined(); + else expect((await action(actionId)).status).toBe("resolved"); + await quiet(); + }); + it("wakes when the real finalizer commits after a notice was already present", async () => { + const binding = await accepted(); const pending = await notice(binding); + await start(); await flush(); await quiet(); + await finalizeNativeRun({ db, runId: binding.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true, + skipNativeRecoveryNotices: true }); + await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("preserves a notice rebound to a new run after discovery", async () => { + const binding = await committed(); const pending = await notice(binding); + const differentRun = randomUUID(); + const resolve: typeof resolveCommittedNativeRecoveryNotices = async (database, scope, settled) => { + const updated = await notice({ ...binding, runId: differentRun }); expect(updated.id).toBe(pending.id); + await resolveCommittedNativeRecoveryNotices(database, scope, settled); + }; + await start(resolve); await flush(); + expect(await action(pending.id)).toMatchObject({ status: "active", evidence: { runId: differentRun } }); await quiet(); + }); + it("does not wake for an already-resolved notice", async () => { + const binding = await committed(); const pending = await notice(binding); + await start(); await flush(); + const queries: string[] = []; const previous = db.$client.options.debug; + db.$client.options.debug = (_id, query) => { queries.push(query); }; + try { + await issueRecoveryActionService(db).resolveActiveForIssue({ companyId, sourceIssueId: binding.issueId, + actionId: pending.id, status: "resolved", outcome: "restored" }); + await flush(); + expect(queries.filter(query => query.includes('from "native_run_finalizations"'))).toEqual([]); + } finally { db.$client.options.debug = previous; } + await quiet(); + }); + it.each(["restore", "worktree"])("preserves %s suppression and settings activation", async mode => { + const binding = await committed(); const pending = await notice(binding); + const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" } + : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "notice-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 }).registerNativeRecoveryNoticeWork(coordinator, () => admitted); + await worker.ready; await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active"); + if (mode === "worktree") { + try { await settings.updateExperimental({ enableWorktreeRunExecution: true }); await flush(); + expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + } + }); + it("retains the first notice committed during drain", async () => { + const binding = await committed(); await start(); await flush(); admitted = false; + const pending = await notice(binding); await vi.advanceTimersByTimeAsync(120_000); await quiet(); + expect((await action(pending.id)).status).toBe("active"); admitted = true; + await vi.advanceTimersByTimeAsync(30_000); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("retains a real failure retry through drain and restart", async () => { + const binding = await committed(); const pending = await notice(binding); + const resolve: typeof resolveCommittedNativeRecoveryNotices = async () => { throw new Error("notice unavailable"); }; + await start(resolve); expect(errors).toHaveLength(1); errors.length = 0; + expect(coordinator!.nextWakeAt()).not.toBeNull(); admitted = false; + await vi.advanceTimersByTimeAsync(120_000); await quiet(); expect((await action(pending.id)).status).toBe("active"); + await coordinator!.stop(); admitted = true; await start(); await flush(); + expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("joins in-flight resolution on shutdown and reconstructs later notices", async () => { + let release!: () => void; const gate = new Promise(resolve => { release = resolve; }); const entered = vi.fn(); + const resolve: typeof resolveCommittedNativeRecoveryNotices = async (...args) => { + entered(); await gate; return resolveCommittedNativeRecoveryNotices(...args); + }; + const first = await committed(); const second = await committed(); await start(resolve); await flush(); + const pending = await notice(first); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(entered).toHaveBeenCalledTimes(1); }); + const later = await notice(second); let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await delay(20); expect(stopped).toBe(false); release(); await stopping; + expect((await action(pending.id)).status).toBe("resolved"); expect((await action(later.id)).status).toBe("active"); + await start(); await flush(); expect((await action(later.id)).status).toBe("resolved"); await quiet(); + }); + it("waits for real pending-effect settlement", async () => { + const binding = await committed(); const pending = await notice(binding); const current = await state(binding.runId); + const [effect] = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, current.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 action(pending.id)).status).toBe("active"); + await statusCommitter.dispatchPendingNativeStatusEffects({ db, ...binding, decisionId: current.decisionId!, onlyWhenDue: true }); + await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("waits for projection and wakes from the real projection owner", async () => { + const binding = await committed(); const pending = await notice(binding); + await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, binding.runId)); + await start(); await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active"); + await projectCommittedRun({ db, run: await read(binding.runId), coordinator: await state(binding.runId) }); + await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("wakes after the real lifecycle writer releases an ownership hold", async () => { + const binding = await committed(); const pending = await notice(binding); + await db.update(heartbeatRuns).set({ status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" }) + .where(eq(heartbeatRuns.id, binding.runId)); + await start(); await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active"); + const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: vi.fn() } as unknown as HeartbeatLifecycleDependencies); + await lifecycle.setRunStatus(binding.runId, "succeeded", { nativePhase: "committed", errorCode: null }); + await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("delegates only notice repair and preserves periodic resource settlement", async () => { + const binding = await committed(); const pending = await notice(binding); const settled = vi.fn(async () => undefined); + await reconcileNativeFinalizations(db, [binding.runId], { skipNativeRecoveryNotices: true, + skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled }); + expect((await action(pending.id)).status).toBe("active"); expect(settled).toHaveBeenCalledTimes(1); + await start(); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it.each(["wakePolicy", "kind"])("activates an existing notice after only its %s changes", async field => { + const binding = await committed(); + const pending = await notice(binding, db, field === "kind" ? { kind: "missing_disposition" } : { wakePolicy: null }); + await start(); await flush(); await quiet(); expect((await action(pending.id)).status).toBe("active"); + const updated = await notice(binding); expect(updated.id).toBe(pending.id); + await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("resolves the escalated owner-verification notice without a retry policy", async () => { + const binding = await committed(); + const pending = await notice(binding, db, { cause: "native_workspace_finalization_owner_unverified", maxAttempts: 1 }); + await notice(binding, db, { cause: "native_workspace_finalization_owner_unverified", maxAttempts: 1 }); + expect(await action(pending.id)).toMatchObject({ status: "escalated", wakePolicy: null }); + await start(); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("wakes for a new identity while preserving the superseded notice", async () => { + const binding = await committed(); const old = await notice({ ...binding, runId: randomUUID() }); + await start(); await flush(); await quiet(); + const pending = await notice(binding, db, { supersedeOnIdentityChange: true }); + expect(pending.id).not.toBe(old.id); await flush(); + expect((await action(old.id)).status).toBe("cancelled"); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); + }); + it("writes a resolution once across competing live/recovery owners", async () => { + const binding = await committed(); const pending = await notice(binding); + const queries: string[] = []; const previous = db.$client.options.debug; + db.$client.options.debug = (_id, query) => { queries.push(query); }; + try { + await Promise.all(Array.from({ length: 4 }, () => resolveCommittedNativeRecoveryNotices(db, binding, true))); + expect(queries.filter(query => query.startsWith('update "issue_recovery_actions"'))).toHaveLength(1); + } finally { db.$client.options.debug = previous; } + expect((await action(pending.id)).status).toBe("resolved"); + }); + it("rechecks the action after waiting for the run lock", async () => { + const binding = await committed(); const pending = await notice(binding); + let release!: () => void; let locked!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { locked = resolve; }); + const holding = db.transaction(async tx => { + await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.id, binding.runId)).for("update"); + locked(); await gate; + }); + await ready; + let waiting = false; const previous = db.$client.options.debug; + db.$client.options.debug = (_id, query) => { if (query.includes('from "heartbeat_runs"') && query.includes('for update')) waiting = true; }; + const resolving = resolveCommittedNativeRecoveryNotices(db, binding, true); + const replacementRun = randomUUID(); + try { + await vi.waitFor(() => { expect(waiting).toBe(true); }); + const updated = await notice({ ...binding, runId: replacementRun }); expect(updated.id).toBe(pending.id); + } finally { release(); db.$client.options.debug = previous; await holding; await resolving; } + expect(await action(pending.id)).toMatchObject({ status: "active", evidence: { runId: replacementRun } }); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index c54b7b2c14e..21bb94b71de 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() })), + registerNativeRecoveryNoticeWork: 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() })), @@ -599,6 +600,13 @@ describe("startServer feedback export wiring", () => { } }); + it.each([false, true])("registers native recovery notices only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerNativeRecoveryNoticeWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerNativeRecoveryNoticeWork.mock.calls[0][1]()).toBe(true); + }); + 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(); @@ -1074,7 +1082,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, skipNativeRunProjection: 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, skipNativeRecoveryNotices: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { @@ -1230,7 +1238,7 @@ describe("startServer feedback export wiring", () => { } }); - 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 => { + it.each(["retained cleanup", "workspace finalization", "native run projection", "native result commit", "native recovery notices"] 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; }); @@ -1238,9 +1246,11 @@ describe("startServer feedback export wiring", () => { const register = owner === "retained cleanup" ? heartbeatServiceMock.registerRetainedNativeSessionCleanupWork : owner === "workspace finalization" ? heartbeatServiceMock.registerNativeWorkspaceFinalizationWork : owner === "native run projection" ? heartbeatServiceMock.registerNativeRunProjectionWork - : heartbeatServiceMock.registerNativeResultCommitWork; + : owner === "native result commit" ? heartbeatServiceMock.registerNativeResultCommitWork + : heartbeatServiceMock.registerNativeRecoveryNoticeWork; if (owner === "native run projection") heartbeatServiceMock.registerNativeRunProjectionWork.mockReturnValueOnce({ ready }); else if (owner === "native result commit") heartbeatServiceMock.registerNativeResultCommitWork.mockReturnValueOnce({ ready }); + else if (owner === "native recovery notices") heartbeatServiceMock.registerNativeRecoveryNoticeWork.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); diff --git a/server/src/index.ts b/server/src/index.ts index 419138856de..f2a1ce42bcc 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.registerNativeRecoveryNoticeWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); + trackHeartbeatSchedulerWork(worker.ready); } if (config.heartbeatSchedulerEnabled) { @@ -2008,7 +2014,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, skipNativeRunProjection: 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, skipNativeRecoveryNotices: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index e191452d97a..781977b1924 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1,3 +1,4 @@ +import { notifyNativeRecoveryNotices } from "../services/native-recovery-notice-notifications.js"; import { notifyNativeStatusReassessment } from "../services/native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../services/unsafe-workspace-export-notifications.js"; import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js"; @@ -9933,6 +9934,7 @@ export function issueRoutes( ); } await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: lockedIssue.companyId, issueId: lockedIssue.id }); + await notifyNativeRecoveryNotices(tx, { companyId: lockedIssue.companyId, issueId: lockedIssue.id }); const [reopened] = await tx .update(issueRecoveryActions) .set({ status: "active", outcome: null, resolvedAt: null }) diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 4a88a4dada4..91bd37db4b3 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 = { + nativeRecoveryNotices: "native-recovery-notices", nativeRunProjection: "native-run-projection", nativeResultCommit: "native-result-commit", nativeWorkspaceFinalization: "native-workspace-finalization", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 8fd88e51861..4bb88fc631a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js"; 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"; @@ -4096,6 +4097,14 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeRecoveryNoticeWork: ( + coordinator: Parameters["register"]>[0], canRun: () => boolean, + ) => nativeRecoveryNoticeWork(db).register(coordinator, + () => canRun() && readTaskDrain(new Date()) === null, + async () => { + const state = await getSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }), registerNativeRunProjectionWork: ( coordinator: Parameters["register"]>[0], canRun: () => boolean, ) => nativeRunProjectionWork(db).register(coordinator, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 9b99ec4ff97..6967b01ba09 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; skipNativeRunProjection?: 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; skipNativeRecoveryNotices?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1995,6 +1995,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization, skipNativeResultCommit: opts?.skipNativeResultCommit, skipNativeRunProjection: opts?.skipNativeRunProjection, + skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices, 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 9c3626ac52f..c495988bc3c 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; 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"; @@ -1026,6 +1027,7 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl // 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 notifyNativeRecoveryNotices(tx, { companyId: previous.companyId, runId: previous.id }); 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")) diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 94af107ec22..b10e1ed5fd1 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; @@ -629,6 +630,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDirectoryCleanup(tx); await notifyInstructionCollection(tx); await notifyCancelledWorkspacePreservation(tx); + await notifyNativeRecoveryNotices(tx); await notifyNativeRunProjection(tx); await notifyNativeResultCommit(tx); await notifyNativeWorkspaceFinalization(tx); diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 93562d691df..5fb28bcb6d8 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -1,3 +1,4 @@ +import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { writeExecutionControl } from "./execution-control-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -321,6 +322,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ): Promise { try { const created = await db.transaction(async (tx) => { + await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition); await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); @@ -411,6 +413,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { Math.min(nextAttemptCount, effectiveMaxAttempts), ); const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); if ((!input.preserveExistingOwner && existing.cause !== input.cause) || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); @@ -471,6 +474,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { return toReadModel(exhausted); } const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); if ((!input.preserveExistingOwner && existing.cause !== input.cause) || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); @@ -544,6 +548,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { try { const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); return tx .insert(issueRecoveryActions) @@ -590,6 +595,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { const pending = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions) .where(and(...predicates)).limit(1); if (!pending.length) return []; + await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); return tx .update(issueRecoveryActions) diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 1f9c3cef1af..8cc00af2da1 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; @@ -11505,6 +11506,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 notifyNativeRecoveryNotices(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeRunProjection(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeResultCommit(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyNativeWorkspaceFinalization(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 9cf4d073621..79e7c45cc35 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; @@ -11,6 +12,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { if (scope.issueId || scope.runId) { + await notifyNativeRecoveryNotices(transaction, scope); await notifyNativeRunProjection(transaction, scope); await notifyNativeResultCommit(transaction, scope); await notifyNativeWorkspaceFinalization(transaction, scope); diff --git a/server/src/services/native-recovery-notice-notifications.ts b/server/src/services/native-recovery-notice-notifications.ts new file mode 100644 index 00000000000..d586cdee07f --- /dev/null +++ b/server/src/services/native-recovery-notice-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeRecoveryNoticeScope = { companyId: string; issueId?: string; runId?: string }; + +export function notifyNativeRecoveryNotices(tx: object, scope?: NativeRecoveryNoticeScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeRecoveryNotices, + 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 568f540b31d..3c3ac77fc59 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -391,6 +391,7 @@ type NativeReconciliationOptions = { skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; + skipNativeRecoveryNotices?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -688,6 +689,7 @@ async function reconcileNativeFinalizationsPass( skipNativeResponsePresentation: options.skipNativeResponsePresentation, skipNativeReviewResponse: options.skipNativeReviewResponse, skipCommittedRunProjection: options.skipNativeRunProjection, + skipNativeRecoveryNotices: options.skipNativeRecoveryNotices, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( row.nativeWorkspaceSync, @@ -750,6 +752,7 @@ async function reconcileNativeFinalizationsPass( skipNativeResponsePresentation: options.skipNativeResponsePresentation, skipNativeReviewResponse: options.skipNativeReviewResponse, skipCommittedRunProjection: options.skipNativeRunProjection, + skipNativeRecoveryNotices: options.skipNativeRecoveryNotices, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( row.nativeWorkspaceSync, diff --git a/server/src/services/native-runtime/native-recovery-notice-work.ts b/server/src/services/native-runtime/native-recovery-notice-work.ts new file mode 100644 index 00000000000..50be3c7403b --- /dev/null +++ b/server/src/services/native-runtime/native-recovery-notice-work.ts @@ -0,0 +1,43 @@ +import type { Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativeRecoveryNoticeScope } from "../native-recovery-notice-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { pendingNativeRecoveryNotices, resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; + +type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; + +export function nativeRecoveryNoticeWork(db: Db, resolve = resolveCommittedNativeRecoveryNotices) { + async function load(scope?: NativeRecoveryNoticeScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await pendingNativeRecoveryNotices(db, 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 recovery notice discovery stopped"); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Native recovery notice resolution stopped"); + const scope = { companyId: entry.companyId, issueId: entry.issueId, runId: entry.id }; + await resolve(db, scope, true); + return (await pendingNativeRecoveryNotices(db, 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.nativeRecoveryNotices, batchSize: 25, + decodeScope: (hint: string): NativeRecoveryNoticeScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativeRecoveryNoticeScope) => !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 recovery notice resolution paused"); + return await canRecover() ? check(entry, signal) : null; + }, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-recovery-notices.ts b/server/src/services/native-runtime/native-recovery-notices.ts new file mode 100644 index 00000000000..defba4b33f8 --- /dev/null +++ b/server/src/services/native-runtime/native-recovery-notices.ts @@ -0,0 +1,73 @@ +import { and, asc, eq, gt, inArray, isNotNull, not, or, sql } from "drizzle-orm"; +import { heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import type { NativeRecoveryNoticeScope } from "../native-recovery-notice-notifications.js"; +import { nativeRunnerOwnershipNotHeldCondition } from "./native-runner-ownership.js"; +import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; +import { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; + +function matchingNotice(runId: string | typeof nativeRunFinalizations.runId) { + return and( + eq(issueRecoveryActions.kind, "active_run_watchdog"), + inArray(issueRecoveryActions.status, ["active", "escalated"]), + sql`${issueRecoveryActions.evidence}->>'runId' = ${runId}::text`, + or(sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`, + eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified")), + ); +} + +export function pendingNativeRecoveryNotices(db: Db, scope?: NativeRecoveryNoticeScope, after?: string) { + return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.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"), isNotNull(nativeRunFinalizations.resultId), + eq(heartbeatRuns.runtimeMode, "native"), nativeRunnerOwnershipNotHeldCondition(), + not(pendingNativeStatusEffectsCondition(db)), not(pendingNativeRunProjectionCondition(db)), + sql`exists (${db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, nativeRunFinalizations.companyId), + eq(issueRecoveryActions.sourceIssueId, nativeRunFinalizations.issueId), matchingNotice(nativeRunFinalizations.runId)))})`, + 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); +} + + + +export async function resolveCommittedNativeRecoveryNotices(db: Db, + scope: { companyId: string; issueId: string; runId: string }, onlyWhenSettled = false) { + const [pending] = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, scope.companyId), eq(issueRecoveryActions.sourceIssueId, scope.issueId), + matchingNotice(scope.runId))).limit(1); + if (!pending) return; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + + + const [coordinator] = await tx.select({ phase: nativeRunFinalizations.phase, resultId: nativeRunFinalizations.resultId }) + .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, scope.companyId), + eq(nativeRunFinalizations.issueId, scope.issueId), eq(nativeRunFinalizations.runId, scope.runId))).for("update"); + if (!coordinator || coordinator.phase !== "committed" || !coordinator.resultId) return; + const [issue] = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, scope.companyId), eq(issues.id, scope.issueId))).for("update"); + if (!issue) return; + const [run] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, scope.runId), eq(heartbeatRuns.companyId, scope.companyId), + eq(heartbeatRuns.nativeIssueId, scope.issueId), eq(heartbeatRuns.runtimeMode, "native"), + nativeRunnerOwnershipNotHeldCondition())).for("update"); + if (!run) return; + if (onlyWhenSettled && !(await pendingNativeRecoveryNotices(tx, scope)).length) return; + const actions = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, scope.companyId), eq(issueRecoveryActions.sourceIssueId, scope.issueId), + matchingNotice(scope.runId))).for("update"); + for (const action of actions) await issueRecoveryActionService(tx).resolveActiveForIssue({ + companyId: scope.companyId, sourceIssueId: scope.issueId, actionId: action.id, + status: "resolved", outcome: "restored", + resolutionNote: "The accepted native result and workspace finalization committed successfully; this run needs no further finalization retry.", + }); + }); +} diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index f11eacf277c..448d5dad3f9 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 { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; +import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; import { projectCommittedRun, recoveredExecutionFailureMetadata } from "./native-run-projection.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; @@ -37,7 +39,6 @@ import { heartbeatRuns, heartbeatRunEvents, issueComments, - issueRecoveryActions, issueThreadInteractions, issues, nativeRunFinalizations, @@ -594,21 +595,6 @@ export async function recordNativeFinalizationFailure(input: { }); } -async function resolveCommittedFinalizationRecovery(db: Db, run: Pick, issueId: string) { - const actions = await db.select().from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId), - eq(issueRecoveryActions.kind, "active_run_watchdog"), inArray(issueRecoveryActions.status, ["active", "escalated"]), - sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, - or( - sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`, - eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified"), - ), - )); - for (const action of actions) await issueRecoveryActionService(db).resolveActiveForIssue({ - companyId: run.companyId, sourceIssueId: issueId, actionId: action.id, - status: "resolved", outcome: "restored", resolutionNote: "The accepted native result and workspace finalization committed successfully; this run needs no further finalization retry.", - }); -} async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) { const [run] = await dbOrTx @@ -754,6 +740,27 @@ export async function recoverCommittedNativeChatResponse( try { materialized = await db.transaction(async (transaction) => { const tx = transaction as unknown as Db; + + + const [coordinator] = await tx + .select() + .from(nativeRunFinalizations) + .where( + and( + eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.companyId, input.companyId), + eq(nativeRunFinalizations.issueId, input.issueId), + eq(nativeRunFinalizations.phase, "committed"), + isNull(nativeRunFinalizations.leaseOwner), + ), + ) + .for("update"); + if ( + !coordinator?.resultId || + !coordinator.decisionId || + !coordinator.assessmentId + ) + return false; const [issue] = await tx .select() .from(issues) @@ -816,25 +823,6 @@ export async function recoverCommittedNativeChatResponse( record(run.contextSnapshot).skipIssueComment === true ) return false; - const [coordinator] = await tx - .select() - .from(nativeRunFinalizations) - .where( - and( - eq(nativeRunFinalizations.runId, run.id), - eq(nativeRunFinalizations.companyId, input.companyId), - eq(nativeRunFinalizations.issueId, input.issueId), - eq(nativeRunFinalizations.phase, "committed"), - isNull(nativeRunFinalizations.leaseOwner), - ), - ) - .for("update"); - if ( - !coordinator?.resultId || - !coordinator.decisionId || - !coordinator.assessmentId - ) - return false; if (input.onlyWhenDue) { if (coordinator.nextAttemptAt && coordinator.nextAttemptAt.getTime() > Date.now()) return false; const [pendingEffect] = await tx.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( @@ -1100,6 +1088,7 @@ export async function finalizeNativeRun(input: { skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipCommittedRunProjection?: boolean; + skipNativeRecoveryNotices?: boolean; /** Workspace-only replay must not consume the provider recovery budget. */ preserveProviderAttempt?: boolean; failpoint?: NativeStatusCommitFailpoint; @@ -1165,7 +1154,7 @@ export async function finalizeNativeRun(input: { logger.warn({ err, runId: admittedRun.id }, "Slack conversation settlement deferred to reconciliation"); }); } - await resolveCommittedFinalizationRecovery(input.db, admittedRun, coordinator.issueId); + if (!input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: admittedRun.companyId, runId: admittedRun.id, issueId: coordinator.issueId }); if (input.projectRunStatus) { const [projectedRun] = await input.db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, @@ -1645,6 +1634,7 @@ export async function finalizeNativeRun(input: { .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))) .for("update").limit(1); if (!before) return null; + await notifyNativeRecoveryNotices(tx, { companyId: run.companyId, runId: run.id }); 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 }); @@ -1681,7 +1671,7 @@ export async function finalizeNativeRun(input: { logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation"); }); } - if (finalizationPhase === "committed") await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId); + if (finalizationPhase === "committed" && !input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: run.companyId, runId: run.id, issueId: coordinator.issueId }); if (input.projectRunStatus && updatedRun) await reconcileMuseNativeTerminalRun(input.db, updatedRun); return { ...coordinator, diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 72edd36f05b..746aecc663b 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -1,3 +1,4 @@ +import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; import { randomUUID } from "node:crypto"; @@ -126,6 +127,7 @@ export async function projectCommittedRun(input: { nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt }) .from(heartbeatRuns).where(needsRepair).for("update").limit(1); if (!before) return null; + await notifyNativeRecoveryNotices(tx, { companyId: input.run.companyId, runId: input.run.id }); 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 }); diff --git a/server/src/services/presentation-ownership-read.test.ts b/server/src/services/presentation-ownership-read.test.ts index 08ee36f558e..427eb1edb86 100644 --- a/server/src/services/presentation-ownership-read.test.ts +++ b/server/src/services/presentation-ownership-read.test.ts @@ -6,15 +6,18 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { NativeRunCoordinatorStore } from "./native-runtime/native-run-coordinator-store.js"; import { getConversationOwnershipBlocker } from "./conversation-continuation.js"; import { isExternalChatPresentationContext } from "./heartbeat-run-summary.js"; -import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js"; +import { finalizeNativeRun, recoverCommittedNativeChatResponse, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js"; -const support = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PRESENTATION_READ_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read projections", () => { - let database: Awaited>; + let database: Awaited> | undefined; let db: ReturnType; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("presentation-ownership-read-"); - db = createDb(database.connectionString); + if (!external) database = await startEmbeddedPostgresTestDatabase("presentation-ownership-read-"); + db = createDb(external ?? database!.connectionString); }, 60_000); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); @@ -113,8 +116,8 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read async function compareRepair(f: Awaited>) { const baseline = captureReads(true), projected = captureReads(); const input = { companyId: f.companyId, issueId: f.issueId, runId: f.run.id }; - expect(await repairCommittedNativeChatResponse(projected.recorded, input)) - .toBe(await repairCommittedNativeChatResponse(baseline.recorded, input)); + expect(await recoverCommittedNativeChatResponse(projected.recorded, { ...input, onlyWhenDue: true })) + .toBe(await recoverCommittedNativeChatResponse(baseline.recorded, { ...input, onlyWhenDue: true })); expect(projected.rows.flat()).toHaveLength(1); expect(Buffer.byteLength(JSON.stringify(projected.rows))).toBeLessThan(4096); expect(Buffer.byteLength(JSON.stringify(baseline.rows))).toBeGreaterThan(128 * 1024); @@ -123,7 +126,7 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read } it("preserves selected answers without downloading prompts, profiles, or excerpts", async () => { - const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: { + const f = await repairFixture({ resultJson: { presentationDecision: { commentId: randomUUID() }, unknownFutureReceipt: { preserved: true } } }); const row = await compareRepair(f); expect(row.resultJson).toEqual(f.run.resultJson); @@ -167,6 +170,18 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read return { ...f, before, reply }; } + async function repairFixture(patch: Partial) { + + + const f = await unpublishedAcceptedReply(); + const [run] = await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.run.id)).returning(); + + + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(Date.now() + 60_000) }) + .where(eq(nativeRunFinalizations.runId, run.id)); + return { ...f, run }; + } + it("publishes the same accepted reply and decision from equivalent full-row and projected starting states", async () => { const outcomes: unknown[] = []; for (const fullRows of [true, false]) { @@ -200,7 +215,7 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read { source: "chat:fixture", skipIssueComment: true, paperclipWake: { externalInteractionContinuation: "true" } }, { source: "internal", paperclipWake: { externalInteractionContinuation: true } }, ].map(context => ({ context })))("preserves typed publication flags for $context", async ({ context }) => { - const f = await fixture({ runtimeMode: "native", status: "succeeded", contextSnapshot: context as Record | null }); + const f = await repairFixture({ contextSnapshot: context as Record | null }); const row = await compareRepair(f); expect(isExternalChatPresentationContext(row.contextSnapshot)).toBe(isExternalChatPresentationContext(context)); const original = context && typeof context === "object" && !Array.isArray(context) ? context : {}; @@ -208,7 +223,7 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read }); it.each([null, ["stdout", { receipt: true }], "scalar", 42].map(result => ({ result })))("preserves non-object result evidence $result", async ({ result }) => { - const f = await fixture({ runtimeMode: "native", status: "succeeded", resultJson: result as Record | null }); + const f = await repairFixture({ resultJson: result as Record | null }); expect((await compareRepair(f)).resultJson).toEqual(result); }); });