Merge commit 'b4b18c9a37188c912507e18feae9677ba1b21ffa' into codex/sleep-final-qualified-18382

* commit 'b4b18c9a37188c912507e18feae9677ba1b21ffa':
  test: align deferred recovery clocks with real execution
  Keep recovery acceptance timers on their owning clock
  fix(server): recheck reactive sources after status summary commit
  test: measure retention page fairness on a controlled clock
  fix(heartbeat): resume saved chat work on feature enablement
  test: await resolved interaction execution before outcome assertions
  Wake saved conversation assignments when Agent Chat becomes enabled

server/src/__tests__/deferred-execution-receipt-work.test.ts

@@ -1,5 +1,6 @@
 import { conflict, HttpError } from "../errors.js";
 import { setImmediate as immediate } from "node:timers/promises";
+import { setTimeout as realTimeout, clearTimeout as clearRealTimeout } from "node:timers";
 import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
 import { budgetService } from "../services/budgets.js";
@@ -25,6 +26,25 @@ import { issueService } from "../services/issues.js";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 
 vi.setConfig({ testTimeout: 90_000 });
+// Vitest's waitFor advances fake time on every poll. Keep PostgreSQL leases
+// and actual child processes on wall time until an explicit idle deadline step.
+function waitForReal(assertion: () => unknown | Promise<unknown>, { timeout = 1000 } = {}) {
+  return new Promise<void>((resolve, reject) => {
+    let settled = false, lastError: unknown;
+    let retry: ReturnType<typeof realTimeout> | undefined;
+    const deadline = realTimeout(() => finish(lastError ?? new Error("Expectation did not complete before its deadline")), timeout);
+    function finish(error?: unknown) {
+      if (settled) return;
+      settled = true; clearRealTimeout(deadline); if (retry) clearRealTimeout(retry);
+      if (error) reject(error); else resolve();
+    }
+    async function attempt() {
+      try { await assertion(); finish(); }
+      catch (error) { lastError = error; if (!settled) retry = realTimeout(attempt, 50); }
+    }
+    void attempt();
+  });
+}
 describe("deferred receipt durable dispatch", () => {
   let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   let db: ReturnType<typeof createDb>;
@@ -46,6 +66,9 @@ describe("deferred receipt durable dispatch", () => {
   });
   beforeEach(async () => {
     await db.execute(sql`truncate companies cascade`); drain = false; errors.length = 0; errorTimes.length = 0;
+    // Install one clock before any worker arms a timer. It follows wall time
+    // during actual process execution; accelerated steps only settle idle work.
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
     heartbeat = heartbeatService(db);
     coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true,
       onError: error => { errors.push(error); errorTimes.push(Date.now()); } });
@@ -78,7 +101,8 @@ describe("deferred receipt durable dispatch", () => {
     return { companyId, agent, issue, source, actorId };
   }
   async function quiet() {
-    await vi.waitFor(async () => {
+    await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions();
+    await waitForReal(async () => {
       if (vi.isFakeTimers()) {
         // Fake-clock cases have no live execution. Settle finite cleanup
         // deadlines explicitly instead of advancing at waitFor's wall speed.
@@ -89,7 +113,6 @@ describe("deferred receipt durable dispatch", () => {
     }, { timeout: 40_000 });
     const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute");
     const selected = reads.mock.calls.length, executed = executes.mock.calls.length;
-    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     const count = statements.length;
     await vi.advanceTimersByTimeAsync(25 * 60_000);
     expect(statements).toHaveLength(count);
@@ -97,12 +120,11 @@ describe("deferred receipt durable dispatch", () => {
   }
   async function finish(issueId: string) {
     await issueService(db).update(issueId, { status: "done" });
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions();
       expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
     }, { timeout: 15_000 });
     const saved = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"));
-    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     if (saved.length) {
       const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime()));
       while (Date.now() <= due) {
@@ -110,10 +132,16 @@ describe("deferred receipt durable dispatch", () => {
         await Promise.allSettled([...passes]); await immediate();
       }
     }
-    await vi.waitFor(async () => {
-      await vi.advanceTimersByTimeAsync(500); await Promise.allSettled([...passes]);
+    await waitForReal(async () => {
+      // A committed cancellation may be coalesced behind an already-armed
+      // worker deadline. Follow that deadline rather than a guessed 500ms step.
+      const next = coordinator.nextWakeAt();
+      await vi.advanceTimersByTimeAsync(next === null ? 500 : Math.max(1, next - Date.now()));
+      await Promise.allSettled([...passes]); await immediate();
       await heartbeat.drainActiveRunExecutions();
-      expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"))).toHaveLength(0);
+      const remaining = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"));
+      expect(remaining, JSON.stringify({ next: coordinator.nextWakeAt(), at: Date.now(),
+        runs: remaining.map(run => ({ id: run.id, at: run.scheduledRetryAt, error: run.errorCode })) })).toHaveLength(0);
     });
   }
   it("is quiet after empty reconstruction and terminal receipt history", async () => {
@@ -131,11 +159,11 @@ describe("deferred receipt durable dispatch", () => {
       observedRunId: f.source.id, idempotencyKey: randomUUID() });
     expect(receipt.accepted).toBe(true);
     let runId: string | undefined;
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!));
       expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId!;
     }, { timeout: 15_000 });
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       await heartbeat.drainActiveRunExecutions();
       const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
       expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
@@ -150,7 +178,7 @@ describe("deferred receipt durable dispatch", () => {
     expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)).toHaveLength(1);
     await quiet();
   });
-  async function blockedRetry(gate: "budget" | "project budget" | "paused agent", clocked = false) {
+  async function blockedRetry(gate: "budget" | "project budget" | "paused agent") {
     const f = await fixture(), lifecycle = createAgentLifecycle(db), budgets = budgetService(db);
     // Host lifecycle completion is controlled; all state, admission and the
     // eventual process adapter execution use their production implementations.
@@ -170,7 +198,6 @@ describe("deferred receipt durable dispatch", () => {
       await budgets.upsertPolicy(f.companyId, policy, f.actorId);
     } else await lifecycle.pauseAgent(f.agent.id);
     await lifecycleWorker.process(f.agent.id);
-    if (clocked) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
     await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
     await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
@@ -178,7 +205,7 @@ describe("deferred receipt durable dispatch", () => {
       observedRunId: f.source.id, idempotencyKey: randomUUID() });
     expect(request.accepted).toBe(true);
     const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0];
-    await vi.waitFor(async () => expect((await read()).payload?.executionWait).toMatchObject({
+    await waitForReal(async () => expect((await read()).payload?.executionWait).toMatchObject({
       reason: gate !== "paused agent" ? "budget.blocked" : "agent.not_invokable",
     }), { timeout: 10_000 });
     const refusals: unknown[] = [];
@@ -191,7 +218,7 @@ describe("deferred receipt durable dispatch", () => {
   }
 
   it.each(["budget", "project budget", "paused agent"] as const)("parks real task Retry behind unchanged %s without recurring SQL", async gate => {
-    const { request, read } = await blockedRetry(gate, true);
+    const { request, read } = await blockedRetry(gate);
     // The clock was installed before worker registration and every timer.
     // No execution exists while we observe actual driver SQL across25min.
     for (let n = 0; n < 3; n++) {
@@ -205,16 +232,16 @@ describe("deferred receipt durable dispatch", () => {
 
   it.each(["budget", "project budget", "paused agent"] as const)("releases real %s Retry into exactly one saved process execution", async gate => {
     const { f, lifecycle, budgets, policy, lifecycleWorker, request, read } = await blockedRetry(gate);
-    // This independent warm-worker fixture stays on the real PostgreSQL clock
-    // through gate release, controller authority and actual child-process exit.
+    // Keep the clock advancing with wall time through gate release and actual
+    // child-process exit. Accelerate only after that execution has settled.
     if (gate !== "paused agent") await budgets.upsertPolicy(f.companyId, { ...policy, amount: 1000 }, f.actorId);
     else await lifecycle.resumeAgent(f.agent.id);
     await lifecycleWorker.process(f.agent.id);
     let runId: string | null = null;
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       const wake = await read(); expect(wake.runId).toBeTruthy(); runId = wake.runId;
     }, { timeout: 40_000 });
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       await heartbeat.drainActiveRunExecutions();
       const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
       expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
@@ -248,6 +275,9 @@ describe("deferred receipt durable dispatch", () => {
   });
 
   it("retains failure backoff for a real wait-persistence database error", async () => {
+    // This case observes the millisecond before retry. Freeze before any
+    // fixture or worker can arm a timer; it never starts a child process.
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false });
     const f = await fixture();
     await createAgentLifecycle(db).pauseAgent(f.agent.id);
     const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
@@ -257,18 +287,27 @@ describe("deferred receipt durable dispatch", () => {
         raise exception 'C1 injected wait persistence failure'; end if; return new; end $$`);
     await db.execute(sql`create trigger c1_refusal_fault before update on agent_wakeup_requests
       for each row execute function c1_refusal_fault()`);
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     try {
       await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
-      await vi.waitFor(() => expect(errors).toHaveLength(1));
+      await waitForReal(() => expect(errors).toHaveLength(1));
       const failure = errors[0] as AggregateError;
       expect(failure).toBeInstanceOf(AggregateError);
       expect(String(failure.errors[0]?.errors[0]?.cause?.message)).toContain("C1 injected wait persistence failure");
       const due = errorTimes[0] + 30_000;
-      await Promise.allSettled([...passes]); await immediate();
-      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(due));
+      // Other owners may publish a coalesced deadline after their real SQL
+      // settles. Run those saved deadlines without reaching the failed retry.
+      await waitForReal(async () => {
+        await Promise.allSettled([...passes]); await immediate();
+        const next = coordinator.nextWakeAt();
+        if (next !== null && next < due) {
+          await vi.advanceTimersByTimeAsync(Math.max(1, next - Date.now() + 1));
+          await Promise.allSettled([...passes]); await immediate();
+        }
+        expect(errors).toHaveLength(1);
+        expect(coordinator.nextWakeAt()).toBe(due);
+      });
       expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].payload?.executionWait).toBeUndefined();
-      await vi.advanceTimersByTimeAsync(29_999);
+      await vi.advanceTimersByTimeAsync(due - Date.now() - 1);
       expect(errors).toHaveLength(1);
       await db.execute(sql`drop trigger c1_refusal_fault on agent_wakeup_requests`);
       await db.execute(sql`drop function c1_refusal_fault()`);
@@ -290,18 +329,19 @@ describe("deferred receipt durable dispatch", () => {
     const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
       source: "on_demand", reason: "task_retry", status: "deferred_issue_execution" }).returning();
     const failure = conflict("Unexpected admission conflict");
-    let fail = true;
-    const resume = vi.fn(async (scope: { onError?: (error: unknown) => void }) => { if (fail) scope.onError?.(failure); });
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    let fail = true, failedAt = 0;
+    const resume = vi.fn(async (scope: { onError?: (error: unknown) => void }) => {
+      if (fail) { failedAt = Date.now(); scope.onError?.(failure); }
+    });
     await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, dispatchNative: async () => {},
       schedulingSuppressed: async () => false }, () => !drain).ready;
     expect(errors).toHaveLength(1);
     expect((errors[0] as AggregateError).errors[0].errors).toEqual([failure]);
-    expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000);
+    expect(coordinator.nextWakeAt()).toBe(failedAt + 30_000);
     expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0].status).toBe("deferred_issue_execution");
     errors.length = 0; fail = false;
     await vi.advanceTimersByTimeAsync(30_000);
-    await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(2));
+    await waitForReal(() => expect(resume).toHaveBeenCalledTimes(2));
     await quiet();
   });
 
@@ -311,7 +351,6 @@ describe("deferred receipt durable dispatch", () => {
       source: "automation", status: native ? "claimed" : "deferred_issue_execution", claimedAt: updatedAt, updatedAt,
       payload: native ? {} : { executionWait: { reason: "recovery" } },
       requestedByActorType: "system", requestedByActorId: native ? "native-status-committer" : null }).returning();
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     const check = vi.fn(async (scope: { companyId?: string; requestId?: string }) => {
       await db.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, scope.requestId!));
     });
@@ -321,7 +360,7 @@ describe("deferred receipt durable dispatch", () => {
     expect(coordinator.nextWakeAt()).toBe(due); expect(check).not.toHaveBeenCalled();
     await vi.advanceTimersByTimeAsync(due - Date.now() - 1); expect(check).not.toHaveBeenCalled();
     await vi.advanceTimersByTimeAsync(1);
-    await vi.waitFor(() => expect(check).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId: wake.id })));
+    await waitForReal(() => expect(check).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId: wake.id })));
     await quiet();
   });
   it("reconciles the first drained producer and waits for resume without domain SQL", async () => {
@@ -334,12 +373,11 @@ describe("deferred receipt durable dispatch", () => {
       await tx.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id, source: "on_demand",
         status: "deferred_issue_execution", reason: "task_retry" });
     });
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     await vi.advanceTimersByTimeAsync(1000);
     expect(resume).not.toHaveBeenCalled(); expect(dispatchNative).not.toHaveBeenCalled();
-    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+    await waitForReal(() => expect(idleWorkSnapshot().active).toBe(0));
     drain = false; await vi.advanceTimersByTimeAsync(30_000);
-    await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(1)); await quiet();
+    await waitForReal(() => expect(resume).toHaveBeenCalledTimes(1)); await quiet();
   });
   it.each([
     { name: "saved comment", payload: { _paperclipWakeContext: { wakeCommentIds: ["saved-comment"] } } },
@@ -363,7 +401,7 @@ describe("deferred receipt durable dispatch", () => {
       });
       expect(resume).not.toHaveBeenCalled();
     });
-    await vi.waitFor(() => expect(resume).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId })), { timeout: 5_000 });
+    await waitForReal(() => expect(resume).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId })), { timeout: 5_000 });
     expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, other.id)))[0].status).toBe("deferred_issue_execution");
     await quiet();
   });
@@ -379,6 +417,105 @@ describe("deferred receipt durable dispatch", () => {
     expect(resume).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ companyId: f.companyId, requestId: wake.id }));
     await quiet();
   });
+  it("parks an accepted conversation Retry while chat is disabled and resumes it on enablement", async () => {
+    const settings = instanceSettingsService(db), original = await settings.getExperimental();
+    await settings.updateExperimental({ enableAgentChat: true });
+    try {
+      const f = await fixture();
+      await db.update(issues).set({ status: "in_review", conversationAgentId: f.agent.id,
+        conversationUserId: f.actorId, conversationState: "waiting" }).where(eq(issues.id, f.issue.id));
+      const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id,
+        actorId: f.actorId, observedRunId: f.source.id, idempotencyKey: randomUUID() });
+      expect(request.accepted).toBe(true);
+      await settings.updateExperimental({ enableAgentChat: false });
+      await expect(requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id,
+        actorId: f.actorId, observedRunId: f.source.id, idempotencyKey: randomUUID() }))
+        .rejects.toMatchObject({ status: 404, message: "Agent Chat is disabled" });
+      const priorCursor = new Date(Date.now() - 31_000);
+      await db.update(agentWakeupRequests).set({ updatedAt: priorCursor }).where(eq(agentWakeupRequests.id, request.requestId!));
+      await heartbeat.resumeExecutionWaitComments({ companyId: f.companyId });
+      const [parked] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
+      expect(parked.updatedAt.getTime()).toBeGreaterThan(priorCursor.getTime());
+      expect(parked).toMatchObject({ status: "deferred_issue_execution", runId: null });
+      await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+      await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
+      await Promise.allSettled([...passes]);
+      expect(errors).toEqual([]);
+      const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0];
+      expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null });
+      await waitForReal(async () => {
+        await Promise.allSettled([...passes]);
+        expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
+      }, { timeout: 40_000 });
+      const idleStatements = statements.length;
+      await new Promise(resolve => setTimeout(resolve, 31_000));
+      expect(statements).toHaveLength(idleStatements);
+      await settings.updateExperimental({ enableAgentChat: true });
+      let runId: string | null = null;
+      await waitForReal(async () => {
+        const saved = await read(); expect(saved.runId).toBeTruthy(); runId = saved.runId;
+      }, { timeout: 5_000 });
+      await waitForReal(async () => {
+        await heartbeat.drainActiveRunExecutions();
+        const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
+        expect(run.status).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("receipt-proof");
+      }, { timeout: 15_000 });
+      expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(1);
+      await heartbeat.drainActiveRunExecutions();
+    } finally { await settings.updateExperimental({ enableAgentChat: original.enableAgentChat }); }
+  });
+  it("resumes saved conversation input only after chat is re-enabled and its transaction commits", async () => {
+    const settings = instanceSettingsService(db), original = await settings.getExperimental();
+    await settings.updateExperimental({ enableAgentChat: true });
+    try {
+      const f = await fixture();
+      await db.update(issues).set({ status: "in_review", conversationAgentId: f.agent.id,
+        conversationUserId: f.actorId, conversationState: "waiting" }).where(eq(issues.id, f.issue.id));
+      const comment = await issueService(db).addComment(f.issue.id, "Continue with this saved input", { userId: f.actorId });
+      const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
+        source: "on_demand", status: "deferred_issue_execution", reason: "issue_commented",
+        requestedByActorType: "user", requestedByActorId: f.actorId, updatedAt: new Date(Date.now() - 31_000),
+        payload: { issueId: f.issue.id, queuedCommentInterrupt: { actorId: f.actorId },
+          _paperclipWakeContext: { issueId: f.issue.id, wakeCommentIds: [comment.id] } } }).returning();
+      await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(Date.now() - 31_000),
+        resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false },
+          queuedCommentInterruptQueueId: wake.id } }).where(eq(heartbeatRuns.id, f.source.id));
+      await settings.updateExperimental({ enableAgentChat: false });
+      await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+      await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
+      await waitForReal(async () => {
+        await Promise.allSettled([...passes]);
+        expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
+      }, { timeout: 5_000 });
+      const idleStatements = statements.length;
+      await new Promise(resolve => setTimeout(resolve, 31_000));
+      expect(statements).toHaveLength(idleStatements);
+      const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0];
+      expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null });
+      await expect(db.transaction(async tx => {
+        await instanceSettingsService(tx as unknown as typeof db).updateExperimental({ enableAgentChat: true });
+        throw new Error("rolled back chat enablement");
+      })).rejects.toThrow("rolled back chat enablement");
+      await Promise.allSettled([...passes]);
+      expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null });
+      expect((await settings.getExperimental()).enableAgentChat).toBe(false);
+      await settings.updateExperimental({ enableAgentChat: true });
+      let runId: string | null = null;
+      await waitForReal(async () => {
+        const saved = await read(); expect(saved.runId).toBeTruthy(); runId = saved.runId;
+      }, { timeout: 5_000 });
+      await waitForReal(async () => {
+        await heartbeat.drainActiveRunExecutions();
+        const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
+        expect(run.status).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("receipt-proof");
+      }, { timeout: 15_000 });
+      const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
+        sql`${heartbeatRuns.id} <> ${f.source.id}`));
+      expect(runs).toHaveLength(1);
+      expect(runs[0].contextSnapshot?.wakeCommentIds).toEqual([comment.id]);
+      await quiet();
+    } finally { await settings.updateExperimental({ enableAgentChat: original.enableAgentChat }); }
+  });
   it("settles a processed chat receipt without restaging it until an actual eligibility commit", async () => {
     const f = await fixture();
     await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
@@ -402,13 +539,13 @@ describe("deferred receipt durable dispatch", () => {
         providerActionId: randomUUID(), kind: "failed_run_retry", status: "processed", updatedAt: new Date(),
         payload: { issueId: f.issue.id, taskRetryIntentId: intentId, failedRunId: f.source.id } });
     });
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       const [intent] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, intentId));
       expect(intent.updatedAt.getTime()).toBeGreaterThan(previous.getTime());
     }, { timeout: 5_000 });
     expect((await db.select().from(chatActions).where(eq(chatActions.id, actionId)))[0].status).toBe("processed");
     await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId }));
-    await vi.waitFor(async () => expect((await db.select().from(chatActions).where(eq(chatActions.id, actionId)))[0].status)
+    await waitForReal(async () => expect((await db.select().from(chatActions).where(eq(chatActions.id, actionId)))[0].status)
       .toBe("issued"), { timeout: 5_000 });
     await quiet();
   });
@@ -421,7 +558,6 @@ describe("deferred receipt durable dispatch", () => {
       _paperclipWakeContext: { wakeCommentIds: comments.map(row => row.id) } };
     const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
       source: "on_demand", status: "deferred_issue_execution", payload, updatedAt: initialAt }).returning();
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     const deps = createHeartbeatQueueTestDependencies(db), queue = createHeartbeatQueue(db, deps);
     await registerDeferredReceiptWork(db, coordinator, { resume: scope => queue.resumeDeferredExecutionReceipts(scope),
       settleTaskChat: async () => {}, dispatchNative: async () => {}, schedulingSuppressed: async () => false }, () => !drain).ready;
@@ -434,11 +570,11 @@ describe("deferred receipt durable dispatch", () => {
       async (_locked, tx) => { await tx.updateWakeQueuedCommentIds({ wakeId: wake.id, payload,
         ids: comments.map(row => row.id).reverse(), updatedAt: movedAt }); });
     await vi.advanceTimersByTimeAsync(initialAt.getTime() + 30_000 - Date.now());
-    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+    await waitForReal(() => expect(idleWorkSnapshot().active).toBe(0));
     expect(deps.resumeQueuedCommentInterrupt).not.toHaveBeenCalled();
     expect(coordinator.nextWakeAt()).toBe(movedAt.getTime() + 30_000);
     await vi.advanceTimersByTimeAsync(movedAt.getTime() + 30_000 - Date.now());
-    await vi.waitFor(() => expect(deps.resumeQueuedCommentInterrupt).toHaveBeenCalledExactlyOnceWith(f.companyId, wake.id));
+    await waitForReal(() => expect(deps.resumeQueuedCommentInterrupt).toHaveBeenCalledExactlyOnceWith(f.companyId, wake.id));
     await quiet();
   });
   it("reactivates a held task Retry when the authorized subtree hold is released without wakeAgents", async () => {
@@ -451,20 +587,20 @@ describe("deferred receipt durable dispatch", () => {
     const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
       observedRunId: f.source.id, idempotencyKey: randomUUID() });
     expect(request.requestId).toBeTruthy();
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
       expect(wake.payload?.executionWait).toMatchObject({ reason: "issue_tree_hold_active" });
       expect(wake.runId).toBeNull();
     }, { timeout: 5_000 });
-    await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 });
+    await waitForReal(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 });
     await tree.releaseHold(f.companyId, f.issue.id, hold.hold.id, { actor, reason: "Resume saved Retry" });
     let runId: string | null = null;
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
       expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId;
     }, { timeout: 40_000 });
     // Observe this Retry before closing the task and draining its follow-ups.
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
       expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
       expect(run.stdoutExcerpt).toContain("receipt-proof");
@@ -477,7 +613,6 @@ describe("deferred receipt durable dispatch", () => {
     const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
       source: "automation", status: "queued", error: "Transient dispatch failure", updatedAt: initialAt,
       requestedByActorType: "system", requestedByActorId: "native-status-committer" }).returning();
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     const dispatchNative = vi.fn(async () => {});
     await registerDeferredReceiptWork(db, coordinator, { resume: async () => {}, settleTaskChat: async () => {},
       dispatchNative, schedulingSuppressed: async () => false }, () => !drain).ready;
@@ -485,10 +620,10 @@ describe("deferred receipt durable dispatch", () => {
     const movedAt = new Date();
     await db.update(agentWakeupRequests).set({ updatedAt: movedAt }).where(eq(agentWakeupRequests.id, wake.id));
     await vi.advanceTimersByTimeAsync(initialAt.getTime() + 30_000 - Date.now());
-    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+    await waitForReal(() => expect(idleWorkSnapshot().active).toBe(0));
     expect(dispatchNative).not.toHaveBeenCalled(); expect(coordinator.nextWakeAt()).toBe(movedAt.getTime() + 30_000);
     await vi.advanceTimersByTimeAsync(movedAt.getTime() + 30_000 - Date.now());
-    await vi.waitFor(() => expect(dispatchNative).toHaveBeenCalledTimes(1)); await quiet();
+    await waitForReal(() => expect(dispatchNative).toHaveBeenCalledTimes(1)); await quiet();
   });
   it("arms one new wait reason grace and stays quiet when that same gate is still closed", async () => {
     const f = await fixture(); let reason = "operator_pause";
@@ -499,17 +634,16 @@ describe("deferred receipt durable dispatch", () => {
       schedulingSuppressed: async () => false }, () => !drain).ready;
     const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
       source: "on_demand", status: "deferred_issue_execution", reason: "task_retry" }).returning();
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     for (const nextReason of ["operator_pause", "budget_pause"]) {
       reason = nextReason;
       await writeDeferredExecutionWait(db, { companyId: f.companyId, requestId: wake.id }, { reason });
       const [current] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));
       const due = current.updatedAt.getTime() + 30_000;
       await vi.advanceTimersByTimeAsync(1_000);
-      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(due));
+      await waitForReal(() => expect(coordinator.nextWakeAt()).toBe(due));
       const checked = resume.mock.calls.length;
       await vi.advanceTimersByTimeAsync(due - Date.now());
-      await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(checked + 1));
+      await waitForReal(() => expect(resume).toHaveBeenCalledTimes(checked + 1));
       await quiet();
     }
     expect(resume).toHaveBeenCalledTimes(2);
@@ -526,15 +660,15 @@ describe("deferred receipt durable dispatch", () => {
     await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
     await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
     await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
-    await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5_000 });
+    await waitForReal(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5_000 });
     expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].runId).toBeNull();
     await commitEligibility();
     let runId: string | null = null;
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
       expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId;
     }, { timeout: 15_000 });
-    await vi.waitFor(async () => {
+    await waitForReal(async () => {
       await heartbeat.drainActiveRunExecutions();
       const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
       expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");

server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts

@@ -1804,8 +1804,22 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => {
     const interactionId = randomUUID();
     const wakeCommentId = randomUUID();
     let claimedIssue: { status: string; executionRunId: string | null } | null = null;
+    let resumedRunId: string | undefined;
+    let signalAdapterStarted!: () => void;
+    const adapterStarted = new Promise<void>(resolve => { signalAdapterStarted = resolve; });
+    const execute = mockAdapterExecute.getMockImplementation()!;
+    mockAdapterExecute.mockImplementation(async (...args: unknown[]) => {
+      if ((args[0] as { runId: string }).runId === resumedRunId) {
+        signalAdapterStarted();
+        // A claimed continuation can finish after the old terminal poll.
+        if (interactionStatus === "rejected" && interactionKind === "connection_intent") {
+          await new Promise(resolve => setTimeout(resolve, 3100));
+        }
+      }
+      return execute();
+    });
     afterContinuationDispatchCheck = async ({ runId: checkedRunId, issueId: checkedIssueId }) => {
-      if (checkedIssueId !== issueId) return;
+      if (checkedIssueId !== issueId || checkedRunId !== resumedRunId) return;
       claimedIssue = await db
         .select({ status: issues.status, executionRunId: issues.executionRunId })
         .from(issues)
@@ -1839,13 +1853,14 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => {
       contextExtras: { interactionId, interactionKind, interactionStatus, wakeCommentId,
         originCommentIds: [wakeCommentId],
         interactionResolvedAt: new Date().toISOString(), mutation: "interaction", source: `${interactionKind}.resolved`, forceFreshSession: true } });
+    resumedRunId = runId;
     await heartbeat.resumeQueuedRuns();
-    expect(await waitForCondition(async () => (await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]?.status === "succeeded")).toBe(true);
-    // Terminal status precedes completion bookkeeping. Drain those writes before
-    // afterEach truncates the fixture, otherwise PostgreSQL can deadlock.
+    // Adapter entry guarantees this exact execution is registered. Await its
+    // owning completion barrier before terminal assertions and fixture cleanup.
+    await adapterStarted;
     await heartbeat.waitForRunExecutionDrain(runId);
+    expect((await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]?.status).toBe("succeeded");
     expect(countExecuteCallsForRun(runId)).toBe(1);
-    await waitForCondition(async () => claimedIssue !== null);
     expect(claimedIssue).toEqual({ status: "in_progress", executionRunId: runId });
   });
 });

server/src/__tests__/issue-recovery-deadlines.test.ts

@@ -1,12 +1,16 @@
+import { createServer } from "node:http";
 import { setImmediate as immediate } from "node:timers/promises";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
-import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issueRecoveryActions, issueRelations, issues, issueWatchdogs } from "@paperclipai/db";
+import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issueRecoveryActions, issueRelations, issues, issueWatchdogs, issueComments } from "@paperclipai/db";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { heartbeatService } from "../services/heartbeat.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
+import { undeliveredConversationComments } from "../services/agent-conversations.js";
+import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
 import { issueService } from "../services/issues.js";
 import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
 import { registerIssueRecoveryWork } from "../services/recovery/work.js";
@@ -71,6 +75,12 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
   async function compete() {
     const other = heartbeatService(db);
     const competing = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const register = competing.register.bind(competing);
+    competing.register = (queue, task) => register(queue, { ...task, run: signal => {
+      const pass = task.run(signal); passes.add(pass);
+      void pass.then(() => passes.delete(pass), () => passes.delete(pass));
+      return pass;
+    } });
     const work = other.registerIssueStateRecoveryWork(competing, () => true);
     extraStops.push(() => competing.stop()); extraDeadlines.push(competing.nextWakeAt); await work.ready;
   }
@@ -86,8 +96,32 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     }).returning();
     return { companyId, agentId: agent.id };
   }
+  async function completingProvider(f: { companyId: string; agentId: string }, beforeComplete?: () => Promise<unknown>) {
+    const failures: unknown[] = [];
+    const listener = createServer(async (request, response) => {
+      try {
+        let body = "";
+        for await (const chunk of request) body += chunk;
+        const { runId } = JSON.parse(body) as { runId: string };
+        const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId),
+          eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.agentId, f.agentId)));
+        if (!run?.contextSnapshot?.issueId) throw new Error("Unknown controlled task run");
+        await beforeComplete?.();
+        await issueService(db).update(String(run.contextSnapshot.issueId), { status: "done" });
+        response.end("ok");
+      } catch (error) { failures.push(error); response.statusCode = 500; response.end("completion failed"); }
+    });
+    await new Promise<void>(resolve => listener.listen(0, "127.0.0.1", resolve));
+    extraStops.push(async () => { listener.closeAllConnections(); await new Promise<void>(resolve => listener.close(() => resolve())); });
+    const port = (listener.address() as { port: number }).port;
+    const script = `fetch("http://127.0.0.1:${port}/complete", {method:"POST", body:JSON.stringify({runId:process.env.PAPERCLIP_RUN_ID})}).then(async response=>{if(!response.ok)throw new Error(await response.text());console.log("recovery-proof")}).catch(error=>{console.error(error);process.exitCode=1})`;
+    await db.update(agents).set({ adapterConfig: { command: process.execPath, args: ["-e", script], cwd: process.cwd() } })
+      .where(eq(agents.id, f.agentId));
+    return failures;
+  }
   async function settled() {
     await vi.waitFor(async () => {
+      await Promise.allSettled([...passes]);
       await heartbeat.drainActiveRunExecutions();
       const wake = coordinator.nextWakeAt();
       const detail = wake === null ? "" : JSON.stringify({ now: Date.now(), tasks: await Promise.all([...tasks].map(async ([queue, task]) => ({ queue, at: task.nextRunAt?.(), pending: await task.hasPending() }))),
@@ -102,11 +136,19 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
     const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute");
     const selected = reads.mock.calls.length, executed = executes.mock.calls.length;
+    const statements: string[] = [], priorDebug = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query) => { statements.push(query); };
     if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
-    try { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed); }
-    finally { vi.useRealTimers(); reads.mockRestore(); executes.mockRestore(); }
+    try {
+      await vi.advanceTimersByTimeAsync(20 * 60_000);
+      expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed);
+      expect(statements).toEqual([]);
+    } finally { vi.useRealTimers(); db.$client.options.debug = priorDebug; reads.mockRestore(); executes.mockRestore(); }
   }
   async function finish(issueId: string) {
+    const retries = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
+      sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
+    const before = await retries();
     await issueService(db).update(issueId, { status: "done" });
     // Closing one blocker can legitimately start its dependent. Wait for this
     // task's dispatch and execution; quiet() verifies the whole fixture later.
@@ -118,26 +160,18 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
         sql`${heartbeatRuns.status} in ('queued', 'running')`,
       ))).toHaveLength(0);
     }, { timeout: 15_000 });
-    const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
-      sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
-    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    const saved = [...new Map([...before, ...await retries()].map(run => [run.id, run])).values()];
     if (saved.length > 0) {
-      const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime()));
-      // Keep PostgreSQL I/O and coordinator settlement ahead of virtual time.
-      // A single leap can otherwise occur before the retry deadline is armed.
-      while (Date.now() <= due) {
-        await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1));
-        await Promise.allSettled([...passes]);
-        await immediate();
-      }
+      const trace: unknown[] = [];
       await vi.waitFor(async () => {
-        await vi.advanceTimersByTimeAsync(500);
         await Promise.allSettled([...passes]);
         await heartbeat.drainActiveRunExecutions();
-        const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
-            sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
-        expect(remaining).toHaveLength(0);
-      });
+        const remaining = await retries();
+        if (remaining.length > 0) trace.push({ now: Date.now(),
+          deadlines: [coordinator.nextWakeAt(), ...extraDeadlines.map(next => next())],
+          remaining: remaining.map(run => ({ id: run.id, at: run.scheduledRetryAt })) });
+        expect(remaining, JSON.stringify(trace)).toHaveLength(0);
+      }, { timeout: Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())) - Date.now() + 30_000 });
       for (const retry of saved) {
         const [settledRetry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retry.id));
         expect(settledRetry.status).toBe("cancelled");
@@ -160,6 +194,93 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     expect(run.stdoutExcerpt).toContain("recovery-proof");
     return run;
   }
+  async function savedConversation(withMessage = true) {
+    const f = await fixture(), settings = instanceSettingsService(db);
+    await settings.updateExperimental({ enableAgentChat: true, enableConferenceRoomChat: false });
+    const replies: string[] = [], failures: unknown[] = [];
+    const listener = createServer(async (request, response) => {
+      try {
+        let body = "";
+        for await (const chunk of request) body += chunk;
+        const { runId } = JSON.parse(body) as { runId: string };
+        const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId),
+          eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.agentId, f.agentId)));
+        if (!run) throw new Error("Unknown controlled conversation run");
+        await issueService(db).addComment(String(run.contextSnapshot?.issueId), "Saved chat delivered", {
+          agentId: f.agentId, runId: run.id,
+        });
+        replies.push(run.id); response.end("ok");
+      } catch (error) { failures.push(error); response.statusCode = 500; response.end("controlled reply failed"); }
+    });
+    await new Promise<void>(resolve => listener.listen(0, "127.0.0.1", resolve));
+    extraStops.push(async () => { listener.closeAllConnections(); await new Promise<void>(resolve => listener.close(() => resolve())); });
+    const port = (listener.address() as { port: number }).port;
+    const script = `fetch("http://127.0.0.1:${port}/reply", {method:"POST", body:JSON.stringify({runId:process.env.PAPERCLIP_RUN_ID})}).then(async response=>{if(!response.ok)throw new Error(await response.text());console.log("recovery-proof")}).catch(error=>{console.error(error);process.exitCode=1})`;
+    await db.update(agents).set({ adapterConfig: { command: process.execPath, args: ["-e", script], cwd: process.cwd() } })
+      .where(eq(agents.id, f.agentId));
+    const chat = await issueService(db).create(f.companyId, { title: "Saved conversation", status: "in_review",
+      conversationAgentId: f.agentId, conversationUserId: "recovery-user", assigneeAgentId: f.agentId,
+      conversationState: "waiting" });
+    const message = withMessage ? await issueService(db).addComment(chat.id, "Keep this message", { userId: "recovery-user" }, { clientRequestId: randomUUID() }) : null;
+    await settings.updateExperimental({ enableAgentChat: false });
+    await start(); await quiet();
+    expect(replies).toEqual([]);
+    return { ...f, settings, chat, message, replies, failures };
+  }
+  it("dispatches a saved conversation only from the committed Agent Chat enable transition", async () => {
+    const f = await savedConversation();
+    expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toHaveLength(1);
+    await f.settings.updateExperimental({ enableAgentChat: true });
+    expect(coordinator.nextWakeAt()).not.toBeNull();
+    await vi.waitFor(async () => expect(await db.select().from(heartbeatRuns)
+      .where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1), { timeout: 40_000 });
+    const run = await completed(f.chat.id);
+    expect(f.replies).toEqual([run.id]); expect(f.failures).toEqual([]);
+    expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toEqual([]);
+    expect((await db.select().from(issues).where(eq(issues.id, f.chat.id)))[0])
+      .toMatchObject({ conversationState: "waiting", executionRunId: null });
+    expect(await db.select().from(issueComments).where(and(eq(issueComments.createdByRunId, run.id),
+      eq(issueComments.authorAgentId, f.agentId)))).toHaveLength(1);
+    await quiet();
+    let hints = 0;
+    const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.strandedAssignments, () => { hints++; });
+    try { await f.settings.updateExperimental({ enableAgentChat: true }); expect(hints).toBe(0); await quiet(); }
+    finally { stop(); }
+    expect(f.replies).toEqual([run.id]);
+  });
+  it("keeps disabled saved conversations parked after rollback, unrelated flags and unchanged values", async () => {
+    const f = await savedConversation();
+    await expect(db.transaction(async tx => {
+      await instanceSettingsService(tx as unknown as typeof db).updateExperimental({ enableAgentChat: true });
+      throw new Error("chat flag rollback");
+    })).rejects.toThrow("chat flag rollback");
+    await quiet();
+    expect((await f.settings.getExperimental()).enableAgentChat).toBe(false);
+    let hints = 0;
+    const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.strandedAssignments, () => { hints++; });
+    try {
+      await f.settings.updateExperimental({ enableConferenceRoomChat: true });
+      await f.settings.updateExperimental({ enableAgentChat: false });
+      expect(hints).toBe(0); await quiet();
+    } finally { stop(); }
+    expect(f.replies).toEqual([]); expect(f.failures).toEqual([]);
+    expect(await db.select().from(heartbeatRuns)).toEqual([]);
+    expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toHaveLength(1);
+  });
+  it("does not dispatch an accepted saved message after Agent Chat is disabled", async () => {
+    const f = await savedConversation(false);
+    await f.settings.updateExperimental({ enableAgentChat: true }); await quiet();
+    draining = true;
+    await issueService(db).addComment(f.chat.id, "Retain while disabled", { userId: "recovery-user" }, { clientRequestId: randomUUID() });
+    let hints = 0;
+    const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.strandedAssignments, () => { hints++; });
+    try { await f.settings.updateExperimental({ enableAgentChat: false }); expect(hints).toBe(0); }
+    finally { stop(); }
+    draining = false; await quiet();
+    expect(f.replies).toEqual([]); expect(f.failures).toEqual([]);
+    expect(await db.select().from(heartbeatRuns)).toEqual([]);
+    expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toHaveLength(1);
+  });
   it("keeps empty, configured idle, and terminal history quiet", async () => {
     await fixture(); await start(); await quiet();
   });
@@ -281,18 +402,39 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     const blocker = await issueService(db).create(f.companyId, { title: "Blocker", status: "todo" });
     const dependent = await issueService(db).create(f.companyId, { title: "Dependent", status: "blocked", assigneeAgentId: f.agentId,
       blockedByIssueIds: [blocker.id] });
+    const failures = await completingProvider(f);
     await start(); await quiet();
     await compete();
     await issueService(db).update(blocker.id, { status: "done" });
     const run = await completed(dependent.id);
     expect(run.contextSnapshot?.wakeReason).toBe("issue_blockers_resolved");
+    expect(failures).toEqual([]);
     await finish(dependent.id); await quiet();
     expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
       sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'issue_blockers_resolved'`))).toHaveLength(1);
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
   }, 90_000);
+  it("settles saved disposition debt on its original real-clock timers before measuring idle", async () => {
+    const f = await fixture();
+    const blocker = await issueService(db).create(f.companyId, { title: "Saved debt blocker", status: "todo" });
+    const issue = await issueService(db).create(f.companyId, { title: "Saved disposition debt", status: "blocked",
+      assigneeAgentId: f.agentId, blockedByIssueIds: [blocker.id] });
+    await start(); await compete();
+    await issueService(db).update(blocker.id, { status: "done" });
+    await completed(issue.id);
+    let retryId: string | undefined;
+    await vi.waitFor(async () => {
+      const [retry] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
+        sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`));
+      expect(retry).toBeDefined(); retryId = retry!.id;
+    }, { timeout: 40_000 });
+    await finish(issue.id); await quiet();
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryId!)))[0].status).toBe("cancelled");
+  }, 120_000);
   it("uses ordinary watchdog configuration and subtree closure to dispatch a reusable review", async () => {
     const f = await fixture();
     const source = await issueService(db).create(f.companyId, { title: "Finished subtree", status: "done" });
+    const failures = await completingProvider(f, () => taskWatchdogService(db).disableForIssue(f.companyId, source.id));
     await start(); await quiet();
     await compete();
     const watchdog = await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId });
@@ -302,6 +444,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
       expect(row.watchdogIssueId).toBeTruthy(); reviewId = row.watchdogIssueId!;
     }, { timeout: 15_000 });
     await completed(reviewId!);
+    expect(failures).toEqual([]);
     expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
       sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'task_watchdog_stopped_subtree'`))).toHaveLength(1);
     await finish(reviewId!);

server/src/__tests__/skill-version-retention.test.ts

@@ -83,6 +83,7 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
   let worker: ReturnType<typeof registerSkillVersionRetentionWork> | undefined;
   let admitted = true;
+  const passes = new Set<Promise<unknown>>();
   afterEach(async () => {
     worker?.unsubscribe();
     await coordinator?.stop();
@@ -157,11 +158,17 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
     return dir;
   }
 
-  async function startRetention(graceMs = 20_000, service = skillVersionRetentionService(db, { graceMs })) {
+  async function startRetention(graceMs = 20_000, service = skillVersionRetentionService(db, { graceMs }), clock: { shouldAdvanceTime?: boolean; now?: number } = {}) {
     await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select 1`)));
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true, ...clock });
     coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => true,
       onError: error => { throw error; } });
+    const register = coordinator.register.bind(coordinator);
+    coordinator.register = (queue, task) => register(queue, { ...task, run: signal => {
+      const pass = task.run(signal); passes.add(pass);
+      void pass.then(() => passes.delete(pass), () => passes.delete(pass));
+      return pass;
+    } });
     worker = registerSkillVersionRetentionWork(coordinator, service, () => admitted);
     await worker.ready;
     return service;
@@ -577,7 +584,6 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
   it("keeps directory pages moving while another company's grace deadline is due", async () => {
     const dueCompany = await createCompany(), pagedCompany = await createCompany();
     const dueSkill = await createSkill(dueCompany, 2), pagedSkill = await createSkill(pagedCompany, 1);
-    const expiring = await writeSnapshot(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!, 0);
     const parent = path.join(resolveCompanySkillVersionsRoot(pagedCompany), pagedSkill.skillId);
     await fs.mkdir(parent, { recursive: true });
     for (let index = 0; index < 410; index++) await fs.mkdir(path.join(parent, `operator-${index}`));
@@ -588,7 +594,27 @@ describeEmbeddedPostgres("skill version snapshot retention", () => {
       if (id === pagedCompany) pageTimes.push(Date.now());
       return prune(id, options);
     };
-    await startRetention(5000, service); await settled();
+    const expiring = await writeSnapshot(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!, 0);
+    const completeIo = async (work: Promise<unknown>) => {
+      let done = false;
+      void work.then(() => { done = true; }, () => { done = true; });
+      while (!done) {
+        if (vi.isFakeTimers()) await vi.advanceTimersByTimeAsync(0);
+        await realDelay(0);
+      }
+      await work;
+    };
+    const accessedAt = (await lastUsedAt(skillVersionCachePaths(dueCompany, dueSkill.skillId, dueSkill.versionIds[0]!).access))!;
+    await completeIo(startRetention(5000, service, { shouldAdvanceTime: false, now: accessedAt }));
+    expect(await exists(expiring)).toBe(true);
+    // Keep virtual time fixed during real I/O; measure scheduler fairness,
+    // independent of filesystem latency and CI CPU scheduling.
+    while (coordinator!.nextWakeAt() !== null) {
+      await vi.advanceTimersByTimeAsync(Math.max(0, coordinator!.nextWakeAt()! - Date.now()));
+      await completeIo(Promise.all([...passes]));
+      await realDelay(0);
+      expect(idleWorkSnapshot().active).toBe(0);
+    }
     expect(pageTimes.length).toBeGreaterThan(5);
     expect(Math.max(...pageTimes.slice(1).map((at, index) => at - pageTimes[index]!))).toBeLessThan(3000);
     expect(await exists(expiring)).toBe(false); await quiet();

server/src/__tests__/status-card-evaluation-work.test.ts

@@ -112,6 +112,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     const created = await cards.create(companyId, { interestPrompt: "Blockers", titlePinned: false, agentId, refreshPolicy }, actor);
     const compile = await cards.requestCompile(created.id, actor);
     await writeback(created.id, compile.generatingIssue.id, true);
+    if (!coordinator && mode === "reactive") await cards.tickDueStatusCards();
     return cards.getById(created.id).then(row => row!);
   }
 
@@ -144,6 +145,104 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect(errors).toEqual([]);
   });
 
+  it.each([true, false])("rechecks source hints consumed during summary writeback without forcing unchanged generation (relevant=%s)", async relevant => {
+    const evaluation = statusCardEvaluationService(db, { wakeup });
+    const runs = vi.spyOn(evaluation, "run"), hints = vi.spyOn(evaluation, "changed");
+    await start(evaluation);
+    const created = await card();
+    await quiet();
+    const cards = statusCardService(db);
+    const generation = await cards.requestRefresh(created.id, { trigger: "manual", actor });
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning();
+    await db.update(issues).set({ checkoutRunId: run!.id }).where(eq(issues.id, generation.generatingIssue!.id));
+    const writer = { agentId, runId: run!.id };
+    await settled();
+    const entered = Promise.withResolvers<void>(), release = Promise.withResolvers<void>();
+    const factory = companySearch.companySearchService;
+    let hold = true;
+    const searchedCompanies: string[] = [];
+    vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => {
+      const search = factory(owner), actual = search.search;
+      search.search = async (...args) => {
+        searchedCompanies.push(args[0]);
+        const result = await actual(...args);
+        if (hold) { hold = false; entered.resolve(); await release.promise; }
+        return result;
+      };
+      return search;
+    });
+    const writing = cards.writeSummary(created.id, { markdown: "Current blockers", changeSummary: "Completed before source change", generationIssueId: generation.generatingIssue!.id }, writer);
+    await entered.promise;
+    const passes = runs.mock.calls.length;
+    hints.mockClear();
+    try {
+      await issueService(db).create(companyId, { title: "Source committed after summary search", status: relevant ? "blocked" : "backlog", priority: "medium" });
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(500);
+        await Promise.allSettled([...evaluations]);
+        expect(hints.mock.calls.some(([scope]) => scope?.has(`changes:${companyId}`))).toBe(true);
+        expect(runs.mock.calls.length).toBeGreaterThan(passes);
+      }, { timeout: 5000 });
+      expect((await cards.getById(created.id))!.generatingIssueId).toBe(generation.generatingIssue!.id);
+      expect(await evaluation.next()).toBeNull();
+      expect(wakeup).not.toHaveBeenCalled();
+      expect(searchedCompanies).toEqual([companyId]);
+    } finally { release.resolve(); }
+    await writing;
+    await settled();
+    if (relevant) {
+      const pending = (await cards.getById(created.id))!;
+      expect(pending.generatingIssueId).toBeNull();
+      expect(pending.pendingChangeCount).toBe(1);
+      expect(pending.nextEvalAt).toBeInstanceOf(Date);
+      expect(wakeup).not.toHaveBeenCalled();
+      await vi.advanceTimersByTimeAsync(Math.max(0, pending.nextEvalAt!.getTime() - Date.now()));
+      await settled();
+      const generated = (await cards.getById(created.id))!;
+      expect(generated.generatingIssueId).toBeTruthy();
+      expect(wakeup).toHaveBeenCalledOnce();
+      expect(await db.select().from(agentWakeupRequests)).toHaveLength(1);
+      await writeback(created.id, generated.generatingIssueId!);
+      expect((await db.select().from(statusCardUpdates).where(eq(statusCardUpdates.generationIssueId, generated.generatingIssueId!))))
+        .toMatchObject([{ status: "ok", finishedAt: expect.any(Date) }]);
+    }
+    await quiet();
+    expect(wakeup).toHaveBeenCalledTimes(relevant ? 1 : 0);
+    if (!relevant) expect(searchedCompanies).toEqual([companyId, companyId]);
+    expect(new Set(searchedCompanies)).toEqual(new Set([companyId]));
+    expect(errors).toEqual([]);
+  });
+
+  it("rolls back the summary and its post-commit recheck obligation together", async () => {
+    await start(); const created = await card(); await quiet();
+    const cards = statusCardService(db);
+    const generation = await cards.requestRefresh(created.id, { trigger: "manual", actor });
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning();
+    await db.update(issues).set({ checkoutRunId: run!.id }).where(eq(issues.id, generation.generatingIssue!.id));
+    await settled();
+    const before = (await cards.getById(created.id))!;
+    const revisions = await cards.listSummaryRevisions(created.id);
+    await expect(db.transaction(async tx => {
+      await statusCardService(tx as unknown as typeof db).writeSummary(created.id, { markdown: "Rolled back summary", changeSummary: "Rollback", generationIssueId: generation.generatingIssue!.id }, { agentId, runId: run!.id });
+      throw new Error("summary rollback");
+    })).rejects.toThrow("summary rollback");
+    await quiet();
+    expect((await cards.getById(created.id))!).toMatchObject({ generatingIssueId: before.generatingIssueId, nextEvalAt: before.nextEvalAt, nextEvalReason: before.nextEvalReason });
+    expect(await cards.listSummaryRevisions(created.id)).toEqual(revisions);
+    expect(wakeup).not.toHaveBeenCalled(); expect(errors).toEqual([]);
+  });
+
+  it.each(["manual", "interval"] as const)("preserves %s writeback cadence without a reactive recheck", async mode => {
+    const created = await card(mode), cards = statusCardService(db);
+    const generation = await cards.requestRefresh(created.id, { trigger: "manual", actor });
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning();
+    await db.update(issues).set({ checkoutRunId: run!.id }).where(eq(issues.id, generation.generatingIssue!.id));
+    const result = await cards.writeSummary(created.id, { markdown: "Current blockers", changeSummary: "Complete", generationIssueId: generation.generatingIssue!.id }, { agentId, runId: run!.id });
+    expect(result.card.generatingIssueId).toBeNull();
+    expect(result.card.nextEvalAt?.getTime() ?? null).toBe(mode === "interval" ? result.card.lastGeneratedAt!.getTime() + 15 * 60_000 : null);
+    expect(wakeup).not.toHaveBeenCalled();
+  });
+
   it("keeps card deletion scoped to its company and a missing deletion quiet", async () => {
     const removed = await card();
     await card();
@@ -448,6 +547,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     vi.setSystemTime(pending.nextEvalAt!); await restarted.run(new AbortController().signal);
     expect(wakeup).toHaveBeenCalledOnce();
     await writeback(created.id, (await statusCardService(db).getById(created.id))!.generatingIssueId!);
+    await restarted.run(new AbortController().signal);
     expect((await statusCardService(db).getById(created.id))!.nextEvalReason).toBe("membership");
   });
 

server/src/services/heartbeat/run-control.ts

@@ -1,3 +1,4 @@
+import { instanceSettingsService } from "../instance-settings.js";
 import { writeDeferredExecutionWait } from "../deferred-execution-wait.js";
 import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
 import { observeStoppedWorkspaceRecovery } from "../workspace-recovery.js";
@@ -777,7 +778,8 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
   async function resumeExecutionWaitComments(scope: DeferredReceiptRecoveryOptions = {}) {
     if ((await getSchedulingSuppression()).suppressed) return;
     await reconcileTaskChatRetryReceipts(db, scope);
-    const retries = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId })
+    const retries = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId,
+      conversationAgentId: issues.conversationAgentId })
       .from(agentWakeupRequests).innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId),
         sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`,
         eq(issues.assigneeAgentId, agentWakeupRequests.agentId)))
@@ -788,8 +790,13 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
         sql`${agentWakeupRequests.payload}->'taskRetry'->>'chatActionId' is null`,
         notInArray(issues.status, ["done", "cancelled"])))
       .orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
+    const chatEnabled = !retries.some(retry => retry.conversationAgentId)
+      || (await instanceSettingsService(db).getExperimental()).enableAgentChat;
     for (const retry of retries) {
+      // Accepted chat work waits for the flag's committed eligibility signal.
+      // Public Retry still rejects new requests while the feature is disabled.
       await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(eq(agentWakeupRequests.id, retry.id));
+      if (retry.conversationAgentId && !chatEnabled) continue;
       await resumeTaskRetry(retry.companyId, retry.id).catch(err => {
         scope.onError?.(err); logger.warn({ err, requestId: retry.id }, "Task retry dispatch deferred");
       });

server/src/services/instance-settings.ts

@@ -626,12 +626,16 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
           }
         }
       }
+      const chatBecameEnabled = !toExperimentalView(current.experimental).enableAgentChat
+        && toExperimentalView(nextExperimental).enableAgentChat;
       const now = new Date();
       const [updated] = await db.transaction(async tx => {
         if (patch.enableStatusCards !== undefined) await notifyStatusCardEvaluation(tx);
+        if (chatBecameEnabled) await notifyDeferredExecutionEligibility(tx);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*");
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
+        if (chatBecameEnabled) await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAssignments, "*");
         if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
         if (patch.enableWorktreeRunExecution !== undefined) {
           await notifyRunDispatch(tx);

server/src/services/status-cards.ts

@@ -969,12 +969,8 @@ export function statusCardService(
         pendingChangeCount: 0,
         pendingChangeHash: null,
         lastChangeAt: null,
-        nextEvalReason: current.refreshPolicy.mode === "reactive" &&
-          filterStatusCardChanges(diffStatusCardFingerprint(snapshot, watchedNow), current.refreshPolicy).length > 0
-          ? "refresh" : membershipAt ? "membership" : null,
-        nextEvalAt: current.refreshPolicy.mode === "reactive" &&
-          filterStatusCardChanges(diffStatusCardFingerprint(snapshot, watchedNow), current.refreshPolicy).length > 0
-          ? now : nextEvaluationAt(current, membershipAt, now),
+        nextEvalReason: current.refreshPolicy.mode === "reactive" ? "refresh" : membershipAt ? "membership" : null,
+        nextEvalAt: current.refreshPolicy.mode === "reactive" ? now : nextEvaluationAt(current, membershipAt, now),
         updatedAt: now,
       }).where(and(eq(statusCards.id, current.id), eq(statusCards.generatingIssueId, input.generationIssueId))).returning();
       if (!next) throw conflict("Status-card generation was superseded by a newer task");