fix(heartbeat): resume saved chat work on feature enablement

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

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

@@ -379,6 +379,99 @@ 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" });
+      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 vi.waitFor(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 vi.waitFor(async () => {
+        const saved = await read(); expect(saved.runId).toBeTruthy(); runId = saved.runId;
+      }, { timeout: 5_000 });
+      await vi.waitFor(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 { vi.useRealTimers(); 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 vi.waitFor(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 vi.waitFor(async () => {
+        const saved = await read(); expect(saved.runId).toBeTruthy(); runId = saved.runId;
+      }, { timeout: 5_000 });
+      await vi.waitFor(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 { vi.useRealTimers(); 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;

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,7 +790,12 @@ 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.
+      if (retry.conversationAgentId && !chatEnabled) continue;
       await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(eq(agentWakeupRequests.id, retry.id));
       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

@@ -625,8 +625,11 @@ 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 (chatBecameEnabled) await notifyDeferredExecutionEligibility(tx);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*");
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);