From 31955507a38e8ca536f2d540b819dee57f8819fa Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 20:34:30 -0500 Subject: [PATCH] Drive live run and resource recovery from scoped durable deadlines Co-Authored-By: Paperclip --- doc/plans/2026-10-30-live-recovery-sleep.md | 26 ++ ...artbeat-active-run-output-watchdog.test.ts | 12 +- .../src/__tests__/live-recovery-work.test.ts | 413 ++++++++++++++++++ .../native-settled-process-release.test.ts | 13 +- .../server-startup-feedback-export.test.ts | 27 +- server/src/index.ts | 69 +-- .../active-run-watchdog/adapters/postgres.ts | 3 +- .../environment-cleanup-notifications.ts | 10 +- .../services/heartbeat/live-recovery-work.ts | 6 +- server/src/services/heartbeat/recovery.ts | 23 +- server/src/services/instance-settings.ts | 5 +- .../live-run-recovery-notifications.ts | 13 +- .../src/services/native-local-process-stop.ts | 9 +- 13 files changed, 536 insertions(+), 93 deletions(-) create mode 100644 doc/plans/2026-10-30-live-recovery-sleep.md create mode 100644 server/src/__tests__/live-recovery-work.test.ts diff --git a/doc/plans/2026-10-30-live-recovery-sleep.md b/doc/plans/2026-10-30-live-recovery-sleep.md new file mode 100644 index 00000000000..10dfec0ff76 --- /dev/null +++ b/doc/plans/2026-10-30-live-recovery-sleep.md @@ -0,0 +1,26 @@ +# Live run and resource recovery deadlines + +This is the final recovery slice after run dispatch, issue state recovery, and deferred execution receipts. It removes the automatic aggregate orphan and silence scans from `index.ts`. The aggregate API remains for explicit compatibility callers. Startup native classification and hot adoption still run first. Native classification failure still closes startup. Unrelated session-goal startup errors no longer skip independent recovery registrations. + +| Duty | Durable truth and pending predicate | Producers | Deadline and exit | +| --- | --- | --- | --- | +| Resultless native session | Native running/failed run; finalization `retryable_failure`; no result | Execution/finalization writes, checkpoint writes, restart claim writes, process identity events | Maximum saved retry/claim lease deadline. Fresh claim owner validates provider/session state. Accepted result, terminal failure, or explicit ownership gate removes the entry. | +| Native question cancellation | Cancellable run with committed `nativeQuestionCancellation` marker | Authorized task-completion interaction transaction | Immediate. Existing cancellation owner consumes exact company/run marker. Terminal status removes debt. | +| Active execution | Running run | Run lifecycle, controller renewal/revocation, native authority and process evidence writes | Fresh native/legacy controller expiry, saved five-minute preparation grace, or bounded physical-process probe. Local execution promises retain execution ownership. Explicit evidence/manual gates park until their owning producer changes truth. | +| Settled native process identity | Terminal native failure/cancellation with persisted process identity | Terminal execution writes, server process identity/stop receipts, environment settlement | Immediate inspection. Only matched local host/PID/group/birth or confirmed remote stop evidence clears identity. A live matched process or local execution owner keeps a bounded probe. Missing authority waits for committed evidence. Startup reconstructs rows independently of the old capped process Map. Successful release removes any obsolete compatibility Map entry. | +| Orphan active environment lease | Active lease with NULL run reference or terminal same-company run | Environment owning transactions and indexed physical-resource peer notifications | Saved lease updatedAt plus five minutes, re-read before due work. Existing export, shutdown, and shared-resource fences remain. Exact-lease dispatch does not mutate timestamps for broad-page fairness. Successful CAS hands work to the existing pending-cleanup owner. | +| Legacy shutdown settlement | Exact lease with pending shutdown marker | Begin, renew, finish, expiry settlement owning transactions | Exact saved marker deadline. Existing owner revalidates boot/controller, source, competing allocations and stop/copyback authority. Expiry records preserved workspace recovery; it never infers file export from process loss. | +| Stranded agent status | Ready agent with status running | Agent configuration/lifecycle/instruction settlement and run/resource transitions | Saved updatedAt plus five minutes, then actual surviving controller expiry when present. Fresh settlement rechecks live execution, native controller, working copy, and lease truth. Settled status removes debt. | +| Active output silence | Running run and coalesced last output/process start/run start/created timestamp | Actual output progress transaction, terminal source mutation, same-run terminal activity commit, watchdog decisions, execution state | Five-minute threshold, exact saved snooze/continue expiry, or physical termination probe. Fresh check re-reads output and decision. Existing watchdog alone owns completion evidence, process cleanup and fold. Suppressed/manual/blocked history does not poll. | + +All reconstruction queries use company/run/lease/agent scopes and keyset pages. Run and resource topics discard irrelevant fields before coalescing. The generic scheduler's pending/deadline reads use memory. Ordinary writes register intent before mutation in the same owning transaction. Nested commit, outer rollback and uncertain transaction reconciliation retain the shared database-work semantics. + +The existing retained-workspace, native finalization, native result, native environment, instruction, review, conversation, status and pending-cleanup owners remain registered. Startup native recovery explicitly skips their aggregate prefix. The residual owners do not call that prefix or `reapOrphanedRuns`. + +## Proof + +The real PostgreSQL matrix uses the actual heartbeat service and all recovery, dispatch, retry-cancellation and deferred-receipt registrations together. It verifies durable status/output, competing owners, current lease authority, exact saved deadlines, more than 100 guarded/suppressed rows, late host/birth evidence, deleted run references, nested transactions, task drain, warm standby, feature suppression and a one-shot PostgreSQL write failure. Each fully settled fixture requires no running/queued run, no scheduler deadline, and no idle hold before twenty virtual minutes with zero actual SQL. The live-run pagination fixture keeps real controller obligations and proves their exact remaining deadline separately. + +The legacy/controller process retry executes on real JavaScript and PostgreSQL clocks. Other deadline cases install their virtual clock before scheduler registration. Installing fake timers after a real scheduler timer has armed is not a valid deadline test. + +The initial matrix passed12/12 in97.89s. Startup registration, scheduler-disabled and unrelated-owner failure contracts passed111/111. Final expanded receipts and current-head CI/review results are recorded in the PR. diff --git a/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts b/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts index 42d8ad30150..3300ccd12a4 100644 --- a/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts +++ b/server/src/__tests__/heartbeat-active-run-output-watchdog.test.ts @@ -1,5 +1,6 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { activityLog, @@ -39,7 +40,8 @@ vi.mock("../services/heartbeat-run-events.js", async (importOriginal) => { const mockedAppendHeartbeatRunEvent = vi.mocked(appendHeartbeatRunEvent); const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); -const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_OUTPUT_WATCHDOG_TEST_DATABASE_URL"); +const describeEmbeddedPostgres = external || embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { console.warn( @@ -52,8 +54,9 @@ describeEmbeddedPostgres("active-run output watchdog", () => { let db: ReturnType; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-active-run-output-watchdog-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-active-run-output-watchdog-"); + db = createDb(external ?? tempDb!.connectionString); + if (external) await db.execute(sql`truncate companies cascade`); }, 30_000); afterEach(async () => { @@ -65,6 +68,7 @@ describeEmbeddedPostgres("active-run output watchdog", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/live-recovery-work.test.ts b/server/src/__tests__/live-recovery-work.test.ts new file mode 100644 index 00000000000..82cc2f87251 --- /dev/null +++ b/server/src/__tests__/live-recovery-work.test.ts @@ -0,0 +1,413 @@ +import { randomUUID } from "node:crypto"; +import os from "node:os"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { setImmediate as immediate } from "node:timers/promises"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, authUsers, companies, companyMemberships, createDb, closeLivenessDb, closeCheckpointDb, + heartbeatRuns, heartbeatRunEvents, nativeRunFinalizations, environmentLeases, issueThreadInteractions, + issueRecoveryActions, issues, activityLog } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { notifyLiveRunRecovery, notifyLiveResourceRecovery, notifyStrandedAgentStatus } from "../services/live-run-recovery-notifications.js"; +import { notifyExecutionRunWork } from "../services/execution-control-notifications.js"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { writeEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; +import { beginLegacyShutdownWorkspaceSettlement } from "../services/legacy-shutdown-workspace-settlement.js"; +import { legacyControllerBootId } from "../services/legacy-controller-lease.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { createHeartbeatRunOutput } from "../services/heartbeat/run-output.js"; +import { recoveryService } from "../services/recovery/service.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { persistActivity } from "../services/activity-log.js"; +import { issueService } from "../services/issues.js"; + +vi.setConfig({ testTimeout: 90_000 }); +describe("residual recovery production matrix", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let heartbeat: ReturnType; + let coordinator: ReturnType; + let draining = false, standby = false; + const errors: unknown[] = [], passes = new Set>(); + const deadlines = new Map number | null>(); + const timeline: unknown[] = []; + const extraStops: Array<() => Promise> = []; + beforeAll(async () => { + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_LIVE_RECOVERY_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("live-recovery-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + }); + beforeEach(async () => { + await db.execute(sql`truncate companies cascade`); errors.length = 0; deadlines.clear(); timeline.length = 0; draining = false; standby = false; + heartbeat = heartbeatService(db); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !standby, canReconcile: () => !standby, + onError: error => errors.push(error) }); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => { deadlines.set(queue, () => task.nextRunAt?.() ?? null); return register(queue, { ...task, onSettled: hints => { timeline.push(["settled", queue, Date.now(), [...hints ?? []]]); task.onSettled?.(hints); }, run: signal => { + timeline.push(["pass", queue, Date.now()]); + const pass = Promise.resolve(task.run(signal)); passes.add(pass); + void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass; + } }); }; + }); + afterEach(async () => { + for (const stop of extraStops.splice(0)) await stop(); + await coordinator.stop(); await heartbeat.drainActiveRunExecutions(); + vi.useRealTimers(); vi.restoreAllMocks(); expect(errors).toEqual([]); + }); + afterAll(async () => { + await closeLivenessDb(db, 1); await closeCheckpointDb(db, 1); + await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); + }); + async function fixture() { + const companyId = randomUUID(), actorId = "live-recovery-user"; + await db.insert(authUsers).values({ id: actorId, name: "Recovery owner", email: `${actorId}@example.test`, + createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companies).values({ id: companyId, name: "Live recovery", issuePrefix: `LR${companyId.slice(0, 6)}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: actorId }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: actorId, + status: "active", membershipRole: "owner" }); + const [agent] = await db.insert(agents).values({ companyId, name: "Recovery process", status: "idle", adapterType: "process", + adapterConfig: { command: process.execPath, args: ["-e", "console.log('live-recovery-proof')"], cwd: process.cwd() }, + runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } } }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Recovered work", status: "done", assigneeAgentId: agent.id }).returning(); + return { companyId, agent, issue, actorId }; + } + async function start(realClock = false) { + if (!realClock && !vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const admitted = () => !draining && !standby; + await heartbeat.registerResidualRecoveryWork(coordinator, admitted).ready; + await heartbeat.registerActiveRunSilenceWork(coordinator, admitted).ready; + await heartbeat.registerIssueStateRecoveryWork(coordinator, admitted).ready; + await heartbeat.registerDeferredReceiptWork(coordinator, admitted).ready; + await heartbeat.registerRunDispatchWork(coordinator, admitted).ready; + await heartbeat.registerScheduledRetryCancellationWork(coordinator, admitted).ready; + } + async function advance(ms: number) { + if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const until = Date.now() + ms; + while (Date.now() < until) { + await vi.advanceTimersByTimeAsync(Math.min(500, until - Date.now())); + await Promise.allSettled([...passes]); await immediate(); + } + } + async function quiet() { + await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); + const quietAt = Date.now(); + if (vi.isFakeTimers()) await advance(31_000); + else await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 35_000 }); + console.info("quiet-drain", JSON.stringify({ quietAt, now: Date.now(), active: idleWorkSnapshot().active, next: [...deadlines].map(([q, at]) => [q, at()]), timeline })); + await vi.waitFor(() => { expect(coordinator.nextWakeAt(), JSON.stringify([...deadlines].map(([name, at]) => [name, at()]))).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + const queries: string[] = [], prior = db.$client.options.debug; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await vi.advanceTimersByTimeAsync(20 * 60_000); await immediate(); expect(queries).toEqual([]); } + finally { db.$client.options.debug = prior; } + } + async function getRun(id: string) { return (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]; } + async function getLease(id: string) { return (await db.select().from(environmentLeases).where(eq(environmentLeases.id, id)))[0]; } + it("leaves configured idle and completed native history without a timer or SQL", async () => { + const f = await fixture(); + const runs = await db.insert(heartbeatRuns).values(Array.from({ length: 150 }, () => ({ companyId: f.companyId, + agentId: f.agent.id, status: "failed", runtimeMode: "native", nativeIssueId: f.issue.id, finishedAt: new Date() }))).returning(); + await db.insert(nativeRunFinalizations).values(runs.map(run => ({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, phase: "committed" }))); + await start(); await quiet(); + }); + it("catches up native resultless retry debt through its real claim owner without replaying missing state", async () => { + const f = await fixture(); await db.update(issues).set({ status: "todo" }).where(eq(issues.id, f.issue.id)); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "failed", runtimeMode: "native", nativeIssueId: f.issue.id, contextSnapshot: { issueId: f.issue.id }, + finishedAt: new Date(), errorCode: "original_native_failure" }).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, + phase: "retryable_failure", nextAttemptAt: new Date(Date.now() - 1_000) }); + await start(); + await vi.waitFor(async () => { + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0]) + .toMatchObject({ phase: "terminal_failure", nextAttemptAt: null, leaseOwner: null }); + }); + expect((await getRun(run.id)).errorCode).toBe("original_native_failure"); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issue.id))) + .toEqual([expect.objectContaining({ ownerType: "board", status: "active" })]); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + await quiet(); + }); + it("consumes the durable native question cancellation from ordinary task completion", async () => { + const f = await fixture(); await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id)); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "running", + runtimeMode: "native", nativeIssueId: f.issue.id, contextSnapshot: { issueId: f.issue.id }, nativePhase: "awaiting_input" }).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, + phase: "observed", recoveryState: "awaiting_evidence", leaseExpiresAt: new Date(Date.now() + 60_000) }); + await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issue.id, kind: "ask_user_questions", + sourceRunId: run.id, status: "pending", continuationPolicy: "none", + idempotencyKey: `paperclip-runner-question:${run.id}:question-1`, payload: { + version: 1, questions: [{ id: "continue", prompt: "Continue?", selectionMode: "single", required: true, allowOther: false, options: [{ id: "yes", label: "Yes" }] }], + questionSet: { schema: "paperclip.question_set.v1", questions: [{ id: "continue", prompt: "Continue?", required: true, answerMode: "single_select", options: [{ id: "yes", label: "Yes" }] }] }, runtimeRequestId: "question-1", + } }); + await start(); await issueService(db).update(f.issue.id, { status: "done" }); + await vi.waitFor(async () => expect((await getRun(run.id)).status).toBe("cancelled"), { timeout: 15_000 }); + expect((await getRun(run.id)).resultJson).toMatchObject({ cancelledByIssueStatus: "done", cancelledIssueId: f.issue.id }); + await quiet(); + }); + it("preserves a fresh foreign controller until exact expiry and admits one process retry across two owners", async () => { + const f = await fixture(), expiry = new Date(Date.now() + 3_000); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "running", + runtimeMode: "legacy", responsibleUserId: f.actorId, controllerBootId: randomUUID(), controllerLeaseExpiresAt: expiry, + executionStage: "preparing", updatedAt: new Date(Date.now() - 6 * 60_000), startedAt: new Date(Date.now() - 6 * 60_000) }).returning(); + await start(true); + const other = heartbeatService(db), competing = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await other.registerResidualRecoveryWork(competing, () => true).ready; extraStops.push(() => competing.stop()); + expect((await getRun(run.id)).status).toBe("running"); expect(coordinator.nextWakeAt()).toBeLessThanOrEqual(expiry.getTime()); + await vi.waitFor(async () => expect((await getRun(run.id)).status).toBe("failed"), { timeout: 10_000 }); + let retryId: string; + await vi.waitFor(async () => { + const retries = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run.id)); + expect(retries).toHaveLength(1); retryId = retries[0].id; + await heartbeat.drainActiveRunExecutions(); await other.drainActiveRunExecutions(); + expect((await getRun(retryId)).status).toBe("succeeded"); + expect((await getRun(retryId)).stdoutExcerpt).toContain("live-recovery-proof"); + }, { timeout: 45_000 }); + await competing.stop(); extraStops.pop(); await quiet(); + }); + it("releases a settled native process only after matching host and birth evidence arrives", async () => { + const f = await fixture(), startedAt = new Date(); + const child = spawn(process.execPath, ["-e", "console.log('identity-proof')"], { stdio: "ignore" }); + const pid = child.pid!; await once(child, "exit"); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, runtimeMode: "native", + nativeIssueId: f.issue.id, status: "failed", finishedAt: new Date(), processPid: pid, processStartedAt: startedAt, + runnerProfileJson: { sessionCheckpoint: { sessionId: "preserved-native-session" } } }).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, phase: "committed" }); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, agentId: f.agent.id, runId: run.id, + eventType: "native.process_identity_recorded", payload: { targetKind: "local", hostname: os.hostname(), + processPid: pid, processGroupId: null, processStartedAt: new Date(startedAt.getTime() - 1).toISOString() } }); + await start(); expect((await getRun(run.id)).processPid).toBe(pid); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, agentId: f.agent.id, runId: run.id, + eventType: "native.process_identity_recorded", payload: { targetKind: "local", hostname: os.hostname(), + processPid: pid, processGroupId: null, processStartedAt: startedAt.toISOString() } }); + await vi.waitFor(async () => expect((await getRun(run.id)).processPid).toBeNull(), { timeout: 40_000 }); + expect(await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, "native.local_process_stopped")))).toHaveLength(1); + await quiet(); + }); + it("recovers orphan active leases after restart, including a deleted heartbeat reference, without starving behind 100 guarded rows", async () => { + const f = await fixture(), foreign = await fixture(), resource = randomUUID(), old = new Date(Date.now() - 6 * 60_000); + await db.insert(environmentLeases).values({ companyId: foreign.companyId, status: "retained", provider: "fixture", providerLeaseId: resource, updatedAt: old }); + const guarded = await db.insert(environmentLeases).values(Array.from({ length: 105 }, () => ({ companyId: f.companyId, + status: "active", provider: "fixture", providerLeaseId: resource, updatedAt: old }))).returning(); + const [deleted] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "succeeded", finishedAt: old }).returning(); + const [orphan] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: deleted.id, + status: "active", provider: "local", updatedAt: old }).returning(); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, deleted.id)); + expect((await getLease(orphan.id)).heartbeatRunId).toBeNull(); await start(); + await vi.waitFor(async () => expect((await getLease(orphan.id)).status).toBe("pending_cleanup"), { timeout: 15_000 }); + expect(await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, f.companyId), + eq(environmentLeases.status, "pending_cleanup")))).toHaveLength(1); + for (const row of guarded) expect((await getLease(row.id)).status).toBe("active"); + await quiet(); + }); + it("reactivates an orphan when a different company's physical resource owner releases it", async () => { + const f = await fixture(), other = await fixture(), old = new Date(Date.now() - 6 * 60_000), resource = randomUUID(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: other.companyId, agentId: other.agent.id, status: "succeeded", finishedAt: old }).returning(); + const [owner] = await db.insert(environmentLeases).values({ companyId: other.companyId, heartbeatRunId: run.id, + status: "retained", provider: "fixture", providerLeaseId: resource, updatedAt: old }).returning(); + const [orphan] = await db.insert(environmentLeases).values({ companyId: f.companyId, status: "active", + provider: "fixture", providerLeaseId: resource, updatedAt: old }).returning(); + await start(); expect((await getLease(orphan.id)).status).toBe("active"); + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "released", releasedAt: new Date() }) + .where(eq(environmentLeases.id, owner.id)), { companyId: other.companyId, runId: run.id, leaseId: owner.id }); + await vi.waitFor(async () => expect((await getLease(orphan.id)).status).toBe("pending_cleanup"), { timeout: 40_000 }); + await quiet(); + }); + it("expires the exact shutdown settlement marker into preserved recovery instead of inferring file export", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, issueId: f.issue.id, + status: "failed", runtimeMode: "legacy", contextSnapshot: { issueId: f.issue.id }, finishedAt: old }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id, + issueId: f.issue.id, status: "active", provider: "fixture", providerLeaseId: randomUUID(), updatedAt: old, + metadata: { driver: "sandbox" } }).returning(); + await beginLegacyShutdownWorkspaceSettlement(db, run, new Date(Date.now() + 1_500)); await start(); + expect((await getLease(lease.id)).metadata?.legacyShutdownWorkspaceSettlement).toMatchObject({ state: "pending" }); + await vi.waitFor(async () => expect((await getLease(lease.id)).metadata?.legacyShutdownWorkspaceSettlement) + .toMatchObject({ state: "expired" }), { timeout: 10_000 }); + expect((await getLease(lease.id)).leasePolicy).toBe("retain_on_failure"); + expect((await getLease(lease.id)).metadata?.workspaceRestoreRecovery).toBeTruthy(); + expect((await getLease(lease.id)).providerLeaseId).toBe(lease.providerLeaseId); + expect((await getLease(lease.id)).metadata).toMatchObject({ workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", runId: run.id } }); + await quiet(); + }); + it("settles stranded agent status at its saved grace", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "succeeded", finishedAt: old }); + await db.transaction(async tx => { + await notifyStrandedAgentStatus(tx, { companyId: f.companyId, agentId: f.agent.id }); + await tx.update(agents).set({ status: "running", updatedAt: old }).where(eq(agents.id, f.agent.id)); + }); + await start(); await vi.waitFor(async () => expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("idle")); + await quiet(); + }); + it("moves the silence threshold on actual output and folds same-run completion only at strict snooze expiry", async () => { + const f = await fixture(), now = new Date(), old = new Date(now.getTime() - 10 * 60_000); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id)); + await db.update(agents).set({ status: "running" }).where(eq(agents.id, f.agent.id)); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "running", runtimeMode: "legacy", contextSnapshot: { issueId: f.issue.id }, + responsibleUserId: f.actorId, startedAt: old, processStartedAt: old, updatedAt: old, + controllerBootId: randomUUID(), controllerLeaseExpiresAt: new Date(now.getTime() + 60 * 60_000) }).returning(); + const output = createHeartbeatRunOutput(db, { run, runId: run.id, issueId: f.issue.id, + runLogStore: { begin: async () => ({ store: "local_file", logRef: "fixture-output.ndjson" }), append: async (_handle, event) => Buffer.byteLength(event.chunk) }, + identityRedactor: { chunk: (_stream, chunk) => chunk, redact: value => value }, secretKeys: new Set(), + appendRunEvent: async (_run, event) => { await appendHeartbeatRunEvent(db, { companyId: f.companyId, agentId: f.agent.id, runId: run.id, ...event }); }, + getCurrentUserRedactionOptions: async () => ({ enabled: false }) }); + await (await output.begin(run)).onLog("stdout", "fresh actual output"); + const progressed = await getRun(run.id), snoozedUntil = new Date(progressed.lastOutputAt!.getTime() + 5 * 60_000 + 2_000); + expect(progressed.lastOutputSeq).toBe(1); expect(progressed.lastOutputAt!.getTime()).toBeGreaterThan(old.getTime()); + await issueService(db).update(f.issue.id, { status: "done" }); + await persistActivity(db, { companyId: f.companyId, actorType: "agent", actorId: f.agent.id, agentId: f.agent.id, + runId: run.id, entityType: "issue", entityId: f.issue.id, action: "issue.updated", details: { status: "done" } }); + await recoveryService(db, { enqueueWakeup: heartbeat.wakeup }).recordWatchdogDecision({ runId: run.id, + actor: { type: "board" }, decision: "snooze", snoozedUntil, reason: "Wait for the exact saved deadline" }); + await start(); + await advance(snoozedUntil.getTime() - Date.now() - 1); + expect((await getRun(run.id)).status).toBe("running"); + await advance(1); + await vi.waitFor(async () => expect((await getRun(run.id)).status).toBe("succeeded")); + expect((await getRun(run.id)).resultJson).toMatchObject({ sourceResolvedWatchdogFold: { sourceIssueStatus: "done", sameRunEvidenceKind: "activity" } }); + await quiet(); + }); + it("re-reads a concurrently moved stranded-agent grace before settlement", async () => { + const f = await fixture(), initialAt = new Date(); + await db.update(agents).set({ status: "running", updatedAt: initialAt }).where(eq(agents.id, f.agent.id)); + await start(); await advance(60_000); + const moved = new Date(); + await db.update(agents).set({ updatedAt: moved }).where(eq(agents.id, f.agent.id)); + await advance(initialAt.getTime() + 5 * 60_000 - Date.now()); + expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("running"); + await advance(moved.getTime() + 5 * 60_000 - Date.now()); + await vi.waitFor(async () => expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("idle")); + await quiet(); + }); + it("reaches a silent run behind more than 100 suppressed live runs and leaves only real controller deadlines", async () => { + const f = await fixture(), now = new Date(), old = new Date(now.getTime() - 20 * 60_000); + await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issue.id)); + const expiry = new Date(now.getTime() + 60 * 60_000); + const suppressed = await db.insert(heartbeatRuns).values(Array.from({ length: 105 }, () => ({ companyId: f.companyId, + agentId: f.agent.id, status: "running", runtimeMode: "legacy", contextSnapshot: { issueId: f.issue.id }, + startedAt: old, updatedAt: old, controllerBootId: randomUUID(), controllerLeaseExpiresAt: expiry }))).returning(); + const [source] = await db.insert(issues).values({ companyId: f.companyId, status: "done", title: "Actual completed source", assigneeAgentId: f.agent.id }).returning(); + const [target] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "running", runtimeMode: "legacy", contextSnapshot: { issueId: source.id }, + startedAt: new Date(old.getTime() + 60_000), updatedAt: old, + controllerBootId: randomUUID(), controllerLeaseExpiresAt: expiry }).returning(); + await db.insert(activityLog).values({ companyId: f.companyId, actorType: "agent", actorId: f.agent.id, + agentId: f.agent.id, runId: target.id, entityType: "issue", entityId: source.id, action: "issue.updated", details: { status: "done" } }); + await start(); await advance(31_000); + await vi.waitFor(async () => expect((await getRun(target.id)).status).toBe("succeeded")); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "running"))).toHaveLength(suppressed.length); + expect(deadlines.get("active-run-silence")!()).toBeNull(); + const [agentState] = await db.select().from(agents).where(eq(agents.id, f.agent.id)); + const grace = agentState.updatedAt.getTime() + 5 * 60_000; + expect(agentState.status).toBe("running"); + expect(deadlines.get("stranded-agent-status")!()).toBe(grace); + await advance(grace - Date.now() + 1); + expect(deadlines.get("stranded-agent-status")!()).toBe(expiry.getTime()); + expect(coordinator.nextWakeAt(), JSON.stringify([...deadlines].map(([name, at]) => [name, at()]))).toBe(expiry.getTime()); + const queries: string[] = [], prior = db.$client.options.debug; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await advance(20 * 60_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = prior; } + }); + it("reconciles nested rollback and dispatches only the outer committed orphan producer", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000), rolledBack = randomUUID(); + await start(); await quiet(); + await expect(db.transaction(async outer => { + await outer.transaction(async inner => { + await notifyLiveResourceRecovery(inner, { companyId: f.companyId, leaseId: rolledBack }); + await inner.insert(environmentLeases).values({ id: rolledBack, companyId: f.companyId, status: "active", provider: "local", updatedAt: old }); + }); + throw new Error("abort outer producer"); + })).rejects.toThrow("abort outer producer"); + await advance(31_000); expect(await getLease(rolledBack)).toBeUndefined(); await quiet(); + const committed = randomUUID(); + await db.transaction(async outer => { + await outer.transaction(async inner => { + await notifyLiveResourceRecovery(inner, { companyId: f.companyId, leaseId: committed }); + await inner.insert(environmentLeases).values({ id: committed, companyId: f.companyId, status: "active", provider: "local", updatedAt: old }); + }); + expect(await getLease(committed)).toBeUndefined(); + }); + await advance(1_000); + await vi.waitFor(async () => expect((await getLease(committed)).status).toBe("pending_cleanup")); + await quiet(); + }); + it("retains and re-reads a moved native controller deadline while awaiting recovery evidence", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000), expiry = new Date(Date.now() + 3_000); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, + status: "running", runtimeMode: "native", nativeIssueId: f.issue.id, contextSnapshot: { issueId: f.issue.id }, + startedAt: old, updatedAt: old }).returning(); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, + phase: "observed", recoveryState: "awaiting_evidence", controllerBootId: randomUUID(), controllerPid: process.pid + 100_000, + controllerProcessStartedAt: old, leaseExpiresAt: expiry }); + await start(); expect(deadlines.get("active-execution-recovery")!()).toBe(expiry.getTime()); + const moved = new Date(expiry.getTime() + 10_000); + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: moved }).where(eq(nativeRunFinalizations.runId, run.id)); + await advance(expiry.getTime() - Date.now()); + expect((await getRun(run.id)).status).toBe("running"); expect((await getRun(run.id)).errorCode).toBeNull(); + expect(deadlines.get("active-execution-recovery")!()).toBe(moved.getTime()); + await db.transaction(async tx => { + await notifyExecutionRunWork(tx, { companyId: f.companyId, runId: run.id, agentId: f.agent.id }); + await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.id)); + await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", recoveryState: "blocked", leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, run.id)); + }); + await quiet(); + }); + it("retains failed orphan work with bounded backoff after a real PostgreSQL write failure", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, status: "active", provider: "local", updatedAt: old }).returning(); + await db.execute(sql`create sequence live_recovery_failure_once`); + await db.execute(sql.raw(`create function live_recovery_failure_once() returns trigger language plpgsql as $$ + begin if new.id = '${lease.id}'::uuid and new.status = 'pending_cleanup' and nextval('live_recovery_failure_once') = 1 + then raise exception 'fixture cleanup persistence unavailable'; end if; return new; end $$`)); + await db.execute(sql`create trigger live_recovery_failure_once before update on environment_leases for each row execute function live_recovery_failure_once()`); + extraStops.push(async () => { + await db.execute(sql`drop trigger if exists live_recovery_failure_once on environment_leases`); + await db.execute(sql`drop function if exists live_recovery_failure_once()`); + await db.execute(sql`drop sequence if exists live_recovery_failure_once`); + }); + await start(); + expect(errors).toHaveLength(1); expect(errors.shift()).toBeInstanceOf(AggregateError); + expect((await getLease(lease.id)).status).toBe("active"); + await advance(29_000); expect((await getLease(lease.id)).status).toBe("active"); + await advance(1_000); + await vi.waitFor(async () => expect((await getLease(lease.id)).status).toBe("pending_cleanup")); + await quiet(); + }); + it("parks suppressed worktree recovery without probes and reconstructs it on the committed feature change", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + const settings = instanceSettingsService(db); + await settings.updateExperimental({ enableWorktreeRunExecution: false }); + heartbeat = heartbeatService(db, { runtimeEnv: { PAPERCLIP_IN_WORKTREE: "true" } }); + const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, status: "active", provider: "local", updatedAt: old }).returning(); + await start(); expect((await getLease(lease.id)).status).toBe("active"); await quiet(); + try { + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await advance(1_000); + await vi.waitFor(async () => expect((await getLease(lease.id)).status).toBe("pending_cleanup")); + } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + await quiet(); + }); + it("retains first producer debt during drain and warm standby, then resumes without duplicate cleanup", async () => { + const f = await fixture(), old = new Date(Date.now() - 6 * 60_000); + draining = true; await start(); + const [lease] = await writeEnvironmentCleanup(db, tx => tx.insert(environmentLeases).values({ companyId: f.companyId, + status: "active", provider: "local", updatedAt: old }).returning(), { companyId: f.companyId }); + await advance(1_000); await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect((await getLease(lease.id)).status).toBe("active"); + standby = true; draining = false; await advance(60_000); expect((await getLease(lease.id)).status).toBe("active"); + standby = false; await advance(30_000); + await vi.waitFor(async () => expect((await getLease(lease.id)).status).toBe("pending_cleanup")); await quiet(); + }); +}); diff --git a/server/src/__tests__/native-settled-process-release.test.ts b/server/src/__tests__/native-settled-process-release.test.ts index d2a60b89ccb..50a52e40e90 100644 --- a/server/src/__tests__/native-settled-process-release.test.ts +++ b/server/src/__tests__/native-settled-process-release.test.ts @@ -1,9 +1,10 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { randomBytes, randomUUID } from "node:crypto"; import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; import { spawn } from "node:child_process"; import os from "node:os"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, @@ -39,8 +40,9 @@ vi.mock("../adapters/index.js", async () => { return { ...actual, getServerAdapter: vi.fn(() => ({ supportsLocalAgentJwt: false, execute })) }; }); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_SETTLED_PROCESS_TEST_DATABASE_URL"); const support = await getEmbeddedPostgresTestSupport(); -const suite = support.supported ? describe : describe.skip; +const suite = external || support.supported ? describe : describe.skip; function startDetached() { const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { detached: true, stdio: "ignore" }); @@ -57,14 +59,15 @@ async function exitedPid() { suite("settled native run process identity", () => { let db: ReturnType; - let temporary: Awaited>; + let temporary: Awaited> | undefined; let companyId: string; let agentId: string; const children: ReturnType[] = []; beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("native-settled-process-"); - db = createDb(temporary.connectionString); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-settled-process-"); + db = createDb(external ?? temporary!.connectionString); + if (external) await db.execute(sql`truncate companies cascade`); companyId = randomUUID(); agentId = randomUUID(); await db.insert(companies).values({ id: companyId, name: "Settled processes", issuePrefix: "SPR", defaultResponsibleUserId: "owner" }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 55d261f91c0..371801f79dd 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -134,6 +134,8 @@ const { })), reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })), reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), + registerResidualRecoveryWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerActiveRunSilenceWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerIssueStateRecoveryWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerDeferredReceiptWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -930,6 +932,8 @@ describe("startServer feedback export wiring", () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); expect(heartbeatServiceMock.registerIssueStateRecoveryWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(heartbeatServiceMock.registerResidualRecoveryWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + expect(heartbeatServiceMock.registerActiveRunSilenceWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.registerDeferredReceiptWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.resumeQueuedRuns).not.toHaveBeenCalled(); @@ -1344,7 +1348,8 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled(); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipNativeStatusWakeIntents: true, 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, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled(); + expect(heartbeatServiceMock.scanSilentActiveRuns).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled(); expect(retiredDetector).not.toHaveBeenCalled(); @@ -1611,20 +1616,30 @@ describe("startServer feedback export wiring", () => { } }); - it("does not replay hot-restart adoption when the orphan reaper retries", async () => { + it("keeps residual recovery independent after a partial hot-restart adoption", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true, heartbeatSchedulerIntervalMs: 30000, })); heartbeatServiceMock.reconcileHotRestartAdoption.mockRejectedValueOnce(new Error("partial adoption")); - heartbeatServiceMock.reapOrphanedRuns - .mockRejectedValueOnce(new Error("transient reap failure")) - .mockResolvedValueOnce({ reaped: 0, runIds: [] }); + await startServer(); expect(heartbeatServiceMock.reconcileHotRestartAdoption).toHaveBeenCalledTimes(1); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenCalledTimes(2); + expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled(); + expect(heartbeatServiceMock.registerResidualRecoveryWork).toHaveBeenCalledTimes(1); + expect(heartbeatServiceMock.registerActiveRunSilenceWork).toHaveBeenCalledTimes(1); + }); + + it.each(["recoverPendingSessionGoalActions", "recoverActiveSessionGoals"] as const)("continues independent recovery after %s startup failure", async method => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); + heartbeatServiceMock[method].mockRejectedValueOnce(new Error("goal owner unavailable")); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerResidualRecoveryWork).toHaveBeenCalledTimes(1); + expect(heartbeatServiceMock.registerActiveRunSilenceWork).toHaveBeenCalledTimes(1); + expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(1); + expect(fakeServer.close).not.toHaveBeenCalled(); }); it("closes the bound listener when native startup recovery fails", async () => { diff --git a/server/src/index.ts b/server/src/index.ts index fffac2a7972..5a3ff5ff900 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -62,7 +62,6 @@ import { chatCompletionDeliveryService } from "./services/chat-completion-delive // instrumentationReady before opening DB connections or constructing the // HTTP server, so trace coverage does not depend on incidental timing. import { instrumentationReady, shutdownInstrumentation } from "./instrumentation.js"; -import { createTrackedSingleFlight } from "./lib/single-flight.js"; import { sentryReady, shutdownSentry, captureException } from "./sentry.js"; import { setSentryRunLinkResolver } from "./sentry-links.js"; import { createSentryRunLinkResolver } from "./services/sentry-run-links.js"; @@ -1262,16 +1261,6 @@ async function startServerWithDatabaseTeardown( }); heartbeatSchedulerInFlight.add(tracked); }; - const runPeriodicRecovery = createTrackedSingleFlight({ - trackWork: trackHeartbeatSchedulerWork, - onSkipped: (inFlightForMs) => { - if (inFlightForMs >= 3 * config.heartbeatSchedulerIntervalMs) { - logger.error({ inFlightForMs }, "periodic heartbeat recovery remains in flight; duplicate pass suppressed"); - } else { - logger.warn({ inFlightForMs }, "skipped periodic heartbeat recovery tick; previous chain still in flight"); - } - }, - }); const waitForHeartbeatSchedulerIdle = async () => { while (heartbeatSchedulerInFlight.size > 0) { await Promise.allSettled([...heartbeatSchedulerInFlight]); @@ -1715,7 +1704,7 @@ async function startServerWithDatabaseTeardown( // workers before cleanup or retry admission, including unmanaged installs. await timeStartupRecovery("plugin_workers", () => app.locals.bundledPluginsStartup); try { - const nativeRecovery = await timeStartupRecovery("native_recovery", () => heartbeat.recoverNativeRunsAfterRestart()); + const nativeRecovery = await timeStartupRecovery("native_recovery", () => heartbeat.recoverNativeRunsAfterRestart(new Date(), undefined, { skipDelegatedWork: true })); if (nativeRecovery.dispositions.length > 0) { logger.info( { @@ -1750,31 +1739,14 @@ async function startServerWithDatabaseTeardown( } catch (err) { logger.error( { err }, - "startup hot-restart adoption reconciliation failed - orphan reaper will serve as degraded backstop", + "startup hot-restart adoption reconciliation failed - recovery deadlines remain registered", ); } - for (let attempt = 1; attempt <= 2; attempt++) { - try { - const result = await timeStartupRecovery("orphan_reaping", () => heartbeat.reapOrphanedRuns()); - logger.info( - { reaped: result.reaped, runIds: result.runIds }, - "startup reap of orphaned heartbeat runs complete", - ); - break; - } catch (err) { - if (attempt < 2) { - logger.warn({ err, attempt }, "startup reap failed, retrying"); - } else { - logger.error( - { err }, - "startup reap of orphaned heartbeat runs failed after retry - periodic reaper will serve as degraded backstop", - ); - } - } - } - - const recoveredGoalActions = await timeStartupRecovery("goal_actions", () => heartbeat.recoverPendingSessionGoalActions()); + const recoveredGoalActions = await timeStartupRecovery("goal_actions", () => heartbeat.recoverPendingSessionGoalActions()).catch(err => { + logger.error({ err }, "startup session-goal action recovery failed"); + return { enqueued: 0, invalid: 0 }; + }); if ( recoveredGoalActions.enqueued > 0 || recoveredGoalActions.invalid > 0 @@ -1784,17 +1756,16 @@ async function startServerWithDatabaseTeardown( "startup session-goal action outbox recovery reconciled pending controls", ); } - const recoveredGoals = await timeStartupRecovery("active_goals", () => heartbeat.recoverActiveSessionGoals()); + const recoveredGoals = await timeStartupRecovery("active_goals", () => heartbeat.recoverActiveSessionGoals()).catch(err => { + logger.error({ err }, "startup active session-goal recovery failed"); + return { enqueued: 0 }; + }); if (recoveredGoals.enqueued > 0) { logger.warn( recoveredGoals, "startup session-goal recovery resumed durable agent goals", ); } - const scanned = await heartbeat.scanSilentActiveRuns(); - if (scanned.created > 0 || scanned.escalated > 0) { - logger.warn({ ...scanned }, "startup active-run output watchdog created review work"); - } })().catch((err) => { logger.error({ err }, "startup heartbeat recovery failed"); throw err; @@ -1803,6 +1774,10 @@ async function startServerWithDatabaseTeardown( await startupHeartbeatRecovery; } + if (config.heartbeatSchedulerEnabled) await heartbeat.registerResidualRecoveryWork(deliveryWork, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; + if (config.heartbeatSchedulerEnabled) await heartbeat.registerActiveRunSilenceWork(deliveryWork, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerIssueStateRecoveryWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerDeferredReceiptWork(deliveryWork, @@ -2005,22 +1980,6 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "status-card scheduler tick failed"); })); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { - // 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, skipNativeStatusWakeIntents: true, 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, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }) - .then(async () => { - const scanned = await heartbeat.scanSilentActiveRuns(); - if (scanned.created > 0 || scanned.escalated > 0) { - logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); - } - }) - .catch((err) => { - logger.error({ err }, "periodic heartbeat recovery failed"); - })); - } })().catch((err) => { logger.error({ err }, "heartbeat scheduler tick failed"); })); diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 2fdc6f68f3c..151c57e7a16 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -1,4 +1,4 @@ -import { notifyActiveRunSilence } from "../../../services/live-run-recovery-notifications.js"; +import { notifyActiveRunSilence, notifyLiveRunRecovery } from "../../../services/live-run-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; @@ -348,6 +348,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch }; const transactionResult = await db.transaction(async (tx) => { + await notifyLiveRunRecovery(tx, input.run); await notifyInstructionCollection(tx, { companyId, runId: input.run.id }); await notifyExternalAgentSettlement(tx, { companyId, runId: input.run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id }); diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index bd26350e3ab..ce0ded6fcc0 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -5,13 +5,15 @@ import { notifyNativeEnvironmentSettlement, type NativeEnvironmentSettlementScop import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyEnvironmentCleanup(transaction: object, scope?: NativeEnvironmentSettlementScope): Promise { +export type EnvironmentCleanupWorkScope = NativeEnvironmentSettlementScope & { leaseId?: string }; + +export async function notifyEnvironmentCleanup(transaction: object, scope?: EnvironmentCleanupWorkScope): Promise { await notifyLiveResourceRecovery(transaction, scope); - if (scope?.runId && typeof (transaction as Db).select === "function") { + if (scope && (scope.runId || scope.leaseId) && typeof (transaction as Db).select === "function") { const peers = await (transaction as Db).select({ id: environmentLeases.id, companyId: environmentLeases.companyId, runId: environmentLeases.heartbeatRunId }).from(environmentLeases).where(and(eq(environmentLeases.status, "active"), sql`exists (select 1 from environment_leases changed where changed.company_id = ${scope.companyId} - and changed.heartbeat_run_id = ${scope.runId} and changed.provider_lease_id is not null + and ${scope.leaseId ? sql`changed.id = ${scope.leaseId}` : sql`changed.heartbeat_run_id = ${scope.runId}`} and changed.provider_lease_id is not null and changed.provider = ${environmentLeases.provider} and changed.provider_lease_id = ${environmentLeases.providerLeaseId})`)); for (const peer of peers) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.orphanActiveLeases, JSON.stringify({ companyId: peer.companyId, leaseId: peer.id })); @@ -21,7 +23,7 @@ export async function notifyEnvironmentCleanup(transaction: object, scope?: Nati await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); } -export function writeEnvironmentCleanup(db: Db, write: (tx: Db) => PromiseLike, scope: NativeEnvironmentSettlementScope): Promise { +export function writeEnvironmentCleanup(db: Db, write: (tx: Db) => PromiseLike, scope: EnvironmentCleanupWorkScope): Promise { return db.transaction(async tx => { await notifyEnvironmentCleanup(tx, scope); return write(tx as unknown as Db); diff --git a/server/src/services/heartbeat/live-recovery-work.ts b/server/src/services/heartbeat/live-recovery-work.ts index d2d9ab78a5b..a15cbeed18c 100644 --- a/server/src/services/heartbeat/live-recovery-work.ts +++ b/server/src/services/heartbeat/live-recovery-work.ts @@ -66,10 +66,12 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType'nativeQuestionCancellation' is null`) : kind === "question" ? and(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is not null`) - : kind === "active" ? eq(heartbeatRuns.status, "running") + : kind === "active" ? and(eq(heartbeatRuns.status, "running"), + sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`) : and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["failed", "cancelled", "timed_out", "interrupted"]), isNotNull(heartbeatRuns.finishedAt), or(isNotNull(heartbeatRuns.processPid), isNotNull(heartbeatRuns.processGroupId), isNotNull(heartbeatRuns.processStartedAt))), diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 75afd5d881c..089a4ab8f66 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -916,8 +916,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }; } - async function recoverNativeRunsAfterRestart(now = new Date(), runIds?: string[]) { - if (!runIds) { + async function recoverNativeRunsAfterRestart(now = new Date(), runIds?: string[], options?: { skipDelegatedWork?: boolean }) { + if (!runIds && !options?.skipDelegatedWork) { await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments); await reconcileNativeFinalizations(db, undefined, { environmentRuntime, @@ -1529,11 +1529,11 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD if (lease.heartbeatRunId && !activeRunExecutions.has(lease.heartbeatRunId) && !adapterExecutionControls.has(lease.heartbeatRunId) && await preserveOrphanedNativeWorkspace(db, lease)) recovered += 1; - await deferOrphanedActiveLease(lease.id); + if (!opts.leaseId) await deferOrphanedActiveLease(lease.id); continue; } if (!(await allowLegacyShutdownWorkspaceCleanup(db, lease))) { - await deferOrphanedActiveLease(lease.id); + if (!opts.leaseId) await deferOrphanedActiveLease(lease.id); continue; } // A provider resource id names one physical sandbox. A different lease @@ -1560,7 +1560,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // Defer this row so the fixed-size page reaches the rows behind // it next tick, instead of re-selecting the same guarded rows // forever. - await deferOrphanedActiveLease(lease.id); + if (!opts.leaseId) await deferOrphanedActiveLease(lease.id); continue; } } @@ -1582,7 +1582,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD eq(environmentLeases.status, "active"), ), ) - .returning({ id: environmentLeases.id }), { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined }); + .returning({ id: environmentLeases.id }), { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined, leaseId: lease.id }); if (flipped.length > 0) recovered += 1; } @@ -2292,6 +2292,10 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD runningProcesses.has(run.id) || activeRunExecutions.has(run.id) || coordinatorOwnedByCurrentController; + if (nativeRun && !locallyTracked && nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now) { + scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime()); + continue; + } if ( nativeRun && ([ @@ -2302,6 +2306,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ].includes(nativeRecoveryState ?? "") || nativeCoordinatorPhase === "retryable_failure") ) { + if (nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now) + scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime()); continue; } const observedOwnerUnverified = @@ -2691,7 +2697,10 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD active: scope => recoverActiveExecutions(scope), settled: async scope => { if (!scope.runId || activeRunExecutions.has(scope.runId) || runningProcesses.has(scope.runId) - || adapterExecutionControls.has(scope.runId)) return; + || adapterExecutionControls.has(scope.runId)) { + scope.onDeferredDeadline(Date.now() + 30_000); + return; + } await releaseSettledNativeRunProcess(db, scope.runId, scope); }, orphanLease: scope => sweepOrphanedActiveLeases({ ...scope, backoffMs: 5 * 60_000 }), diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index a3e2b2f3b14..ad47e240014 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,4 +1,4 @@ -import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; +import { notifyLiveRunRecovery, notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js"; @@ -635,7 +635,8 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService if (patch.enableWorktreeRunExecution !== undefined) { await notifyRunDispatch(tx); await notifyDeferredExecutionEligibility(tx); - await notifyActiveRunSilence(tx); + await notifyLiveRunRecovery(tx); + await notifyLiveResourceRecovery(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyExternalAgentSettlement(tx); await notifyInstructionCommit(tx); diff --git a/server/src/services/live-run-recovery-notifications.ts b/server/src/services/live-run-recovery-notifications.ts index b270e3b9564..05ec8842d15 100644 --- a/server/src/services/live-run-recovery-notifications.ts +++ b/server/src/services/live-run-recovery-notifications.ts @@ -8,10 +8,12 @@ export async function notifyActiveRunSilence(tx: object, scope?: LiveRunRecovery } export async function notifyLiveRunRecovery(tx: object, scope?: LiveRunRecoveryScope) { - const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, agentId: scope.agentId }) : "*"; + const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId }) : "*"; for (const queue of [DELIVERY_QUEUES.nativeProcessSessions, DELIVERY_QUEUES.nativeQuestionCancellations, - DELIVERY_QUEUES.activeExecutionRecovery, DELIVERY_QUEUES.settledNativeProcesses, DELIVERY_QUEUES.strandedAgentStatus]) + DELIVERY_QUEUES.activeExecutionRecovery, DELIVERY_QUEUES.settledNativeProcesses]) await notifyDeliveryWork(tx, queue, hint); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAgentStatus, scope + ? JSON.stringify({ companyId: scope.companyId, agentId: scope.agentId }) : "*"); await notifyActiveRunSilence(tx, scope); } @@ -19,9 +21,12 @@ export type LiveResourceRecoveryScope = { companyId: string; runId?: string; lea export async function notifyLiveResourceRecovery(tx: object, scope?: LiveResourceRecoveryScope) { const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, leaseId: scope.leaseId }) : "*"; - for (const queue of [DELIVERY_QUEUES.orphanActiveLeases, DELIVERY_QUEUES.legacyShutdownSettlement, - DELIVERY_QUEUES.strandedAgentStatus, DELIVERY_QUEUES.settledNativeProcesses]) + for (const queue of [DELIVERY_QUEUES.orphanActiveLeases, DELIVERY_QUEUES.legacyShutdownSettlement]) await notifyDeliveryWork(tx, queue, hint); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.settledNativeProcesses, scope + ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId }) : "*"); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAgentStatus, scope + ? JSON.stringify({ companyId: scope.companyId }) : "*"); } export async function notifyStrandedAgentStatus(tx: object, scope: { companyId: string; agentId: string }) { diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index e30c18e9cdc..69630667313 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -147,7 +147,7 @@ export async function readNativeLocalProcessStop(db: Db, companyId: string, runI type SettledProcessRelease = "released" | "pending" | "skipped"; export async function releaseSettledNativeRunProcess(db: Db, runId: string, options: { onDeferredDeadline?: (at: number) => void } = {}): Promise { - return db.transaction(async transaction => { + const outcome = await db.transaction(async transaction => { const tx = transaction as unknown as Db; const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, finishedAt: heartbeatRuns.finishedAt, @@ -173,7 +173,8 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string, opti eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED), isNull(heartbeatRunEvents.sourceEventId), )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); if (identity?.payload?.targetKind !== "local" || identity.payload.hostname !== os.hostname() || - identity.payload.processPid !== run.processPid || identity.payload.processGroupId !== run.processGroupId) return "skipped"; + identity.payload.processPid !== run.processPid || identity.payload.processGroupId !== run.processGroupId || + !run.processStartedAt || identity.payload.processStartedAt !== run.processStartedAt.toISOString()) return "skipped"; if (!(await recordNativeLocalProcessStop(tx, run))) { options.onDeferredDeadline?.(Date.now() + 30_000); return "pending"; @@ -184,8 +185,10 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string, opti await tx.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, run.id)); - return "released"; + return "released" as const; }); + if (outcome !== "pending") pendingSettledProcessReleases.delete(runId); + return outcome; } const MAX_SETTLED_PROCESS_RELEASE_ATTEMPTS = 20;