Merge commit 'a0f5caa00241e9f2653992f47106bbe38fec130f' into codex/sleep-final-qualified-18401

* commit 'a0f5caa00241e9f2653992f47106bbe38fec130f':
  test: align deferred recovery clocks with real execution
  Keep recovery acceptance timers on their owning clock
  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

# Conflicts:
#	server/src/services/instance-settings.ts

doc/plans/2026-10-30-issue-recovery-sleep.md

@@ -23,6 +23,10 @@ Producer coverage uses the existing run-dispatch, execution-control, issue-revie
 
 The new owners replace only these four startup and periodic calls. Deferred comments, silent active runs and residual orphan/resource recovery retain their existing backstop until the next slice. Every previously migrated native-owner skip remains unchanged.
 
-A controlled process test prints an execution proof but cannot apply a task disposition. The test operator completes that task through issue service. If the existing continuation policy already saved a disposition-repair retry, the test preserves it until its stored deadline. Only after all provider work has joined does it advance the clock; dispatch then revalidates the completed task and cancels the retry. The final quiet proof requires no registered deadlines, no tracked in-flight work and no owner SQL over twenty former polling minutes. It does not claim a saved future retry is already drained.
+The ordinary dependency and watchdog process controls request completion through a loopback test-operator boundary before exiting. That boundary validates the actual run/company/agent and uses issue service; the watchdog control also disables its source through watchdog service. These cases prove normal completion without creating new disposition debt. The separate incomplete-process control prints an execution proof but cannot apply a task disposition, so the test operator completes that task through issue service after the existing policy saves a retry. If the existing continuation policy already saved a disposition-repair retry, the test preserves its actual timer and stored deadline on the real clock. Cleanup joins both competing coordinators before inspecting cancellation. A separate acceptance case deliberately waits for a production-created retry and verifies that exact row becomes cancelled, then all registered deadlines settle before the twenty-minute virtual idle measurement. Its120-second budget covers the existing60-second retry,30-second reconstruction and real I/O; the ordinary dependency test retains90seconds. Installing fake timers after an owner armed a real timer left that timer waiting on wall time despite virtual advances, so cleanup does not switch clocks while any owner obligation remains. The final quiet proof requires no registered deadlines, no tracked in-flight work and no owner SQL. It does not claim a saved future retry is already drained.
 
 Active action repair has no separate grace or lease countdown. Its delayed disposition repair first writes a scheduled-retry run, then mirrors the date into action wakePolicy.retryAt and timeoutAt. The dispatch owner owns that saved run deadline. Other skipped gates wait for their existing execution, wait, pause, budget, or finalizer producer; the company action worker does not add a polling fallback.
+
+Agent Chat is an instance-wide eligibility gate for saved conversation comments. An effective disabled-to-enabled transition publishes the existing stranded-assignment topic before the settings write in the same transaction. The worker reconstructs eligible conversation assignments after commit; other issue owners are not activated by this flag. Effective unchanged values, unrelated flags and disabling do not publish assignment hints. A rolled-back enable leaves the accepted comment parked.
+
+The real PostgreSQL control accepts a valid conversation comment while chat is enabled, disables chat before registering recovery, and proves the outbox stays quiet. Committing the ordinary settings enable alone launches one real process. That process calls a controlled loopback reply boundary which writes the agent comment through issue service; normal conversation completion consumes the message and returns to waiting. Rollback, unchanged/unrelated settings and disable-before-dispatch controls retain input without launching a run. Idle checks inspect actual PostgreSQL statements over twenty virtual minutes, including transactional queries.

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/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);