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 674fe474361..97cc082599f 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1660,3 +1660,11 @@ GitHub credential tests pass after repairing their external-DB fixture URL; server types/build pass. Recovery terminalization/classification writers are the next audit. Master 34409a92c4 was fetched without disturbing this draft. The worker is still not production-registered and all original stack duties remain. + + +Slice 28 now covers orphan-run terminalization in stale-lock recovery. Both +real-producer missing-notice cases failed before repair and pass afterward; +all 40 existing stale-lock ownership/race cases and server types/build pass. +The conditional update and provider/process authority retain their existing +owners. Adapter-failure classification and remaining producer audits still gate +production registration; this does not migrate the stale-lock polling duty itself. diff --git a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md index 9bd78bf6f06..98fa3579c34 100644 --- a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md @@ -1290,3 +1290,30 @@ Fresh master is 34409a92c4 (native failover backup filtering); it was fetched, not merged into this draft. Recovery service orphan terminalization and adapter failure classification are the next direct run writers to audit. Production registration/poll removal and the original stack's gates remain unfinished. + + +## Orphan-run terminalization + +The stale-lock recovery owner can move a running run to interrupted/cancelled +without passing through normal run lifecycle writers. Its terminal update now +registers an issue-scoped milestone intent in a short transaction. The existing +conditional update still rechecks status, runtime, native ownership/finalization, +and legacy controller generation/lease. Process checks and telemetry remain +outside that transaction; cleanup and audit handling retain their prior order. + +Two real-producer cases (dead process and cancelled issue with a live process) +first exhausted the worker after its working notice, then recovered the run. +Both lacked the terminal notice before this change. Both now publish it once, +return to two-minute SQL quiet, and remain quiet after a repeated empty sweep. +Evidence: /tmp/sleep28-recovery-before.log and /tmp/sleep28-recovery-tests.log. +All 40 existing stale-lock recovery cases pass against a separate test database +(/tmp/sleep28-stale-lock-acceptance.log), including native ownership arrival, +controller/finalization races, shared references, explicit retry ownership and +failed audit insertion. Its test harness now supports the guarded external DB +path; no recovery assertions were weakened. Server typecheck/build and diff +whitespace checks pass (/tmp/sleep28-recovery-{typecheck,build}.log). + +Adapter-failure classification remains a separate producer audit: it changes +errorCode/resultJson/updatedAt after issue-monitor or escalation writes, so an +earlier issue notification cannot by itself prove coverage of the later run +commit. Production registration and removal of the lifecycle poll remain gated. diff --git a/server/src/__tests__/chat-run-milestone-recovery-work.test.ts b/server/src/__tests__/chat-run-milestone-recovery-work.test.ts new file mode 100644 index 00000000000..c99ac6917ab --- /dev/null +++ b/server/src/__tests__/chat-run-milestone-recovery-work.test.ts @@ -0,0 +1,78 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { chatPublications, createDb, heartbeatRuns, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createChatMilestoneFixtures } from "./helpers/chat-milestone-fixtures.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { chatRunMilestoneWork } from "../services/chat-run-milestone-work.js"; +import { recoveryService } from "../services/recovery/service.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MILESTONE_RECOVERY_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("chat milestones after orphan recovery", () => { + let db: ReturnType; + let temporary: Awaited> | undefined; + let fixtures: ReturnType; + let f: Awaited>; + let coordinator: ReturnType | undefined; + const errors: unknown[] = []; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("chat-recovery-milestones-"); + db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + fixtures = createChatMilestoneFixtures(db); + }, 60_000); + beforeEach(async () => { + errors.length = 0; f = await fixtures.create(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); + await fixtures.cleanup(); expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + } + async function quiet() { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + } + const publications = () => db.select().from(chatPublications).where(eq(chatPublications.issueId, f.issueId)); + + it.each(["process_gone", "issue_cancelled"] as const)("publishes the recovered terminal state after %s and becomes quiet", async reason => { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, + status: "running", runtimeMode: "legacy", responsibleUserId: f.userId, + processPid: reason === "process_gone" ? 2_000_000_000 : process.pid, + contextSnapshot: { issueId: f.issueId, source: "chat:slack", endpointId: f.endpointId, wakeCommentId: f.commentId }, + }); + await db.update(issues).set({ executionRunId: runId, + status: reason === "issue_cancelled" ? "cancelled" : "in_progress", + }).where(eq(issues.id, f.issueId)); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) }); + await chatRunMilestoneWork(db).register(coordinator, () => true).ready; + await flush(); + expect(await publications()).toEqual([expect.objectContaining({ idempotencyKey: `run:${runId}:working:${f.endpointId}` })]); + await quiet(); + + const recovery = recoveryService(db, { enqueueWakeup: vi.fn() }); + expect(await recovery.sweepStaleIssueLocks()).toEqual({ cleared: 1, issueIds: [f.issueId], terminalizedRunIds: [runId] }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0].status) + .toBe(reason === "process_gone" ? "interrupted" : "cancelled"); + await flush(); + expect((await publications()).map(row => row.idempotencyKey).sort()).toEqual([ + `run:${runId}:failed:${f.endpointId}`, `run:${runId}:working:${f.endpointId}`, + ].sort()); + await quiet(); + expect(await recovery.sweepStaleIssueLocks()).toEqual({ cleared: 0, issueIds: [], terminalizedRunIds: [] }); + await flush(); await quiet(); expect(await publications()).toHaveLength(2); + }); +}); diff --git a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts index a6168189b0b..5269256d0d1 100644 --- a/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts +++ b/server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; @@ -27,7 +28,8 @@ vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClien import { heartbeatService } from "../services/heartbeat.ts"; import { recoveryService } from "../services/recovery/service.ts"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STALE_LOCK_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -43,8 +45,8 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-stale-lock-sweep-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-stale-lock-sweep-"); + db = createDb(external ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -64,6 +66,7 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 42765bb3606..1058c7fd412 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,5 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; +import { boundRunIssueId } from "../run-task-binding.js"; import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js"; import { onboardingFirstTaskAwaitingUserSql } from "./onboarding-first-task-wait.js"; import { isAiAuthenticationBlocked } from "../ai-auth-failure.js"; @@ -6147,70 +6149,76 @@ export function recoveryService( await deps.beforeOrphanedRunTerminalWrite?.(run.id); const now = new Date(); - const updated = await db - .update(heartbeatRuns) - .set({ - status: terminalStatus, - finishedAt: run.finishedAt ?? now, - error: run.error ?? (terminalStatus === "interrupted" ? message : null), - errorCode: - run.errorCode ?? - (terminalStatus === "interrupted" ? errorCode : null), - updatedAt: now, - }) - .where( - and( - eq(heartbeatRuns.id, run.id), - eq(heartbeatRuns.status, "running"), - eq(heartbeatRuns.runtimeMode, run.runtimeMode), - nativeRunnerOwnershipNotHeldCondition(), - // Provider exit is expected while the native coordinator resumes a - // session or copies its completed workspace back. The coordinator - // owns those retries, including expired leases and future attempts. - // Check at the write so a newly recorded result cannot be orphaned - // using the earlier liveness snapshot. Terminal issue status remains - // the stronger authority. - !issueTerminalStatus && run.runtimeMode === "native" - ? notExists(db.select({ runId: nativeRunFinalizations.runId }) - .from(nativeRunFinalizations).where(and( - eq(nativeRunFinalizations.runId, heartbeatRuns.id), - eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), - or( - and( - isNotNull(nativeRunFinalizations.resultId), - inArray(nativeRunFinalizations.phase, [ - "observed", "workspace_finalizing", "ready_for_assessment", - "arbitrating", "retryable_failure", - ]), - ), - and( - isNull(nativeRunFinalizations.resultId), - or( - eq(nativeRunFinalizations.phase, "retryable_failure"), - and(eq(nativeRunFinalizations.phase, "observed"), gt(nativeRunFinalizations.attempt, 0)), + const updated = await db.transaction(async tx => { + const boundIssueId = boundRunIssueId(run); + if (boundIssueId) { + await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: boundIssueId }); + } + return tx + .update(heartbeatRuns) + .set({ + status: terminalStatus, + finishedAt: run.finishedAt ?? now, + error: run.error ?? (terminalStatus === "interrupted" ? message : null), + errorCode: + run.errorCode ?? + (terminalStatus === "interrupted" ? errorCode : null), + updatedAt: now, + }) + .where( + and( + eq(heartbeatRuns.id, run.id), + eq(heartbeatRuns.status, "running"), + eq(heartbeatRuns.runtimeMode, run.runtimeMode), + nativeRunnerOwnershipNotHeldCondition(), + // Provider exit is expected while the native coordinator resumes a + // session or copies its completed workspace back. The coordinator + // owns those retries, including expired leases and future attempts. + // Check at the write so a newly recorded result cannot be orphaned + // using the earlier liveness snapshot. Terminal issue status remains + // the stronger authority. + !issueTerminalStatus && run.runtimeMode === "native" + ? notExists(tx.select({ runId: nativeRunFinalizations.runId }) + .from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, heartbeatRuns.id), + eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), + or( + and( + isNotNull(nativeRunFinalizations.resultId), + inArray(nativeRunFinalizations.phase, [ + "observed", "workspace_finalizing", "ready_for_assessment", + "arbitrating", "retryable_failure", + ]), + ), + and( + isNull(nativeRunFinalizations.resultId), + or( + eq(nativeRunFinalizations.phase, "retryable_failure"), + and(eq(nativeRunFinalizations.phase, "observed"), gt(nativeRunFinalizations.attempt, 0)), + ), ), ), + ))) + : undefined, + // Recheck ownership in the write: a controller can renew or claim + // the run after the liveness read. An old snapshot cannot end a new + // controller's run, even if that controller's lease later expires. + run.runtimeMode === "legacy" + ? and( + run.controllerBootId + ? eq(heartbeatRuns.controllerBootId, run.controllerBootId) + : isNull(heartbeatRuns.controllerBootId), + or( + isNull(heartbeatRuns.controllerBootId), + sql`${heartbeatRuns.controllerLeaseExpiresAt} <= clock_timestamp()`, ), - ))) - : undefined, - // Recheck ownership in the write: a controller can renew or claim - // the run after the liveness read. An old snapshot cannot end a new - // controller's run, even if that controller's lease later expires. - run.runtimeMode === "legacy" - ? and( - run.controllerBootId - ? eq(heartbeatRuns.controllerBootId, run.controllerBootId) - : isNull(heartbeatRuns.controllerBootId), - or( - isNull(heartbeatRuns.controllerBootId), - sql`${heartbeatRuns.controllerLeaseExpiresAt} <= clock_timestamp()`, - ), - ) - : undefined, - ), - ) - .returning() - .then((rows) => rows[0] ?? null); + ) + : undefined, + ), + ) + .returning() + .then((rows) => rows[0] ?? null); + }); if (!updated) { // Another path finalized the run between the read and this write. Keep // that terminal outcome authoritative.