fix(server): wake chat milestones after orphan run recovery

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.

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.

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<typeof createDb>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let fixtures: ReturnType<typeof createChatMilestoneFixtures>;
+  let f: Awaited<ReturnType<typeof fixtures.create>>;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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);
+  });
+});

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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();
   });
 

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.