fix(server): wake interaction continuations from committed responses

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

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -93,6 +93,7 @@ describe("createChatReconciliationCoordinator", () => {
     expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications(");
     expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications(");
     expect(publicationMaintenance).not.toContain("reconcileTerminalConfirmationActions(");
+    expect(source).toContain("chatChannels.registerInteractionContinuationWork(deliveryWork,");
     expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,");
     expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,");
   });

server/src/__tests__/chat-channels.integration.test.ts

@@ -45012,6 +45012,205 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     return { fixture, callbacks, onAction: callbacks.onAction, endpoint, service, wakeup, settlementBarrier, dm, conversation, linkedUserId, principal, interaction, publication, acceptAction, actionEvent };
   }
 
+  it.each(["accept", "reject", "drain", "restart", "deferred", "failure", "paused endpoint", "future publication", "stale claim"] as const)("wakes interaction continuations from a committed response through %s", async mode => {
+    const context = await telegramConfirmationRecoveryFixture(mode === "reject" ? "reject" : "accept");
+    const errors: unknown[] = [];
+    let canRun = true;
+    let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const settle = async () => vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(10);
+      expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
+    }, { timeout: 10_000 });
+    const quiet = async () => {
+      const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
+      } finally { select.mockRestore(); transaction.mockRestore(); }
+    };
+    const originalWake = context.wakeup.getMockImplementation()!;
+    context.wakeup.mockImplementation(async (...args) => {
+      expect((context.service.runtime as unknown as FakeChatSdkRuntime).endpoints.get(context.endpoint.id)?.edits).toHaveLength(["paused endpoint", "future publication"].includes(mode) ? 0 : 1);
+      return originalWake(...args);
+    });
+    if (mode === "deferred") context.wakeup.mockResolvedValueOnce(null);
+    if (mode === "failure") context.wakeup.mockRejectedValueOnce(new Error("temporary continuation admission failure"));
+    try {
+      await context.service.registerInteractionContinuationWork(coordinator, () => canRun).ready;
+      await settle(); await quiet();
+      if (["drain", "paused endpoint", "future publication"].includes(mode)) canRun = false;
+      if (mode === "restart" || mode === "stale claim") await coordinator.stop();
+      await expect(context.onAction(context.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash");
+      if (["drain", "paused endpoint", "future publication"].includes(mode) || mode === "restart" || mode === "stale claim") {
+        await vi.advanceTimersByTimeAsync(1000);
+        await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+        expect(context.wakeup).not.toHaveBeenCalled();
+        await quiet();
+        if (mode === "paused endpoint") await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, context.endpoint.id));
+        if (mode === "future publication") await db.update(chatPublications).set({ state: "retry", nextAttemptAt: new Date(Date.now() + 60_000) })
+          .where(eq(chatPublications.idempotencyKey, `interaction-resolution:${context.interaction.id}:${context.endpoint.id}`));
+        if (mode === "stale claim") await db.update(chatActions).set({ status: "processing", updatedAt: new Date() }).where(and(
+          eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+        ));
+        canRun = true;
+        if (mode === "restart" || mode === "stale claim") {
+          coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
+            canReconcile: () => true, onError: error => errors.push(error) });
+          await context.service.registerInteractionContinuationWork(coordinator, () => canRun).ready;
+        } else if (coordinator.nextWakeAt() !== null) {
+          await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now()));
+        }
+      }
+      if (mode === "stale claim") {
+        const [action] = await db.select().from(chatActions).where(and(
+          eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+        ));
+        expect(coordinator.nextWakeAt()).toBe(action.updatedAt.getTime() + 30_000);
+        await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now() - 1));
+        expect(context.wakeup).not.toHaveBeenCalled();
+        await vi.advanceTimersByTimeAsync(1);
+      }
+      if (mode === "deferred" || mode === "failure") {
+        let retryAt = 0;
+        await vi.waitFor(async () => {
+          await vi.advanceTimersByTimeAsync(10);
+          const [action] = await db.select().from(chatActions).where(and(
+            eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+          ));
+          expect(action).toMatchObject({ status: "issued", result: { code: mode === "deferred" ? "interaction_wakeup_deferred" : "interaction_wakeup_failed" } });
+          retryAt = action.updatedAt.getTime() + 30_000;
+          expect(idleWorkSnapshot().active).toBe(0);
+          expect(coordinator.nextWakeAt()).toBe(retryAt);
+        }, { timeout: 10_000 });
+        expect(context.wakeup).toHaveBeenCalledTimes(1);
+        await vi.advanceTimersByTimeAsync(Math.max(0, retryAt - Date.now() - 1));
+        expect(context.wakeup).toHaveBeenCalledTimes(1);
+        await vi.advanceTimersByTimeAsync(1);
+      }
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(10);
+        expect(await db.select({ status: chatActions.status }).from(chatActions).where(and(
+          eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+        ))).toEqual([{ status: "processed" }]);
+      }, { timeout: 10_000 });
+      expect(context.wakeup).toHaveBeenCalledTimes(mode === "deferred" || mode === "failure" ? 2 : 1);
+      expect(context.wakeup).toHaveBeenCalledWith(context.fixture.assignedAgentId, expect.objectContaining({
+        idempotencyKey: `interaction:${context.interaction.id}:${mode === "reject" ? "rejected" : "accepted"}`, allowRunCoalescing: false,
+      }));
+      await settle(); await quiet();
+      expect(errors).toEqual([]);
+    } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
+  it("passes historical continuation prefixes and retains a response committed during dispatch", async () => {
+    const first = await telegramConfirmationRecoveryFixture();
+    const second = await telegramConfirmationRecoveryFixture();
+    await expect(first.onAction(first.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash");
+    const invalidIds = Array.from({ length: 101 }, (_, index) => `00000000-0000-4000-8002-${String(index + 1).padStart(12, "0")}`);
+    await db.insert(chatActions).values(invalidIds.map(id => ({
+      id, companyId: first.fixture.companyId, endpointId: first.endpoint.id,
+      kind: "interaction_wakeup", providerActionId: `invalid-continuation:${id}`, status: "issued", payload: {},
+    })));
+    let release!: () => void;
+    const barrier = new Promise<void>(resolve => { release = resolve; });
+    const originalWake = first.wakeup.getMockImplementation()!;
+    first.wakeup.mockImplementationOnce(async (...args) => { await barrier; return originalWake(...args); });
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const ready = first.service.registerInteractionContinuationWork(coordinator, () => true).ready;
+    try {
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(10);
+        expect(first.wakeup).toHaveBeenCalledTimes(1);
+      }, { timeout: 15_000 });
+      await expect(second.onAction(second.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash");
+      expect(await db.select({ status: chatActions.status }).from(chatActions).where(and(
+        eq(chatActions.companyId, second.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+      ))).toEqual([{ status: "issued" }]);
+      release(); await ready;
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(10);
+        expect(first.wakeup).toHaveBeenCalledTimes(2);
+        expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
+      }, { timeout: 15_000 });
+      expect(await db.select({ id: chatActions.id }).from(chatActions).where(and(
+        inArray(chatActions.id, invalidIds), eq(chatActions.status, "failed"),
+        eq(sql<string>`${chatActions.result}->>'code'`, "interaction_wakeup_payload_invalid"),
+      ))).toHaveLength(101);
+      expect(await db.select({ id: chatActions.id }).from(chatActions).where(and(
+        inArray(chatActions.companyId, [first.fixture.companyId, second.fixture.companyId]),
+        eq(chatActions.kind, "interaction_wakeup"), eq(chatActions.status, "processed"),
+      ))).toHaveLength(2);
+      const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
+      } finally { select.mockRestore(); transaction.mockRestore(); }
+      expect(errors).toEqual([]);
+    } finally {
+      release(); await ready; await coordinator.stop(); vi.useRealTimers();
+      await db.delete(chatActions).where(inArray(chatActions.id, invalidIds));
+      await first.service.shutdown(); await second.service.shutdown();
+    }
+  });
+
+  it("defers continuation dispatch until the outer response transaction commits", async () => {
+    const context = await telegramConfirmationRecoveryFixture();
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const settle = async () => vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(10);
+      expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
+    }, { timeout: 10_000 });
+    const originalWake = context.wakeup.getMockImplementation()!;
+    context.wakeup.mockImplementation(async (...args) => {
+      expect((context.service.runtime as unknown as FakeChatSdkRuntime).endpoints.get(context.endpoint.id)?.edits).toHaveLength(1);
+      return originalWake(...args);
+    });
+    try {
+      await context.service.registerInteractionContinuationWork(coordinator, () => true).ready;
+      await settle();
+      const [issue] = await db.select().from(issues).where(eq(issues.id, context.conversation.issueId));
+      const accept = async (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => {
+        await issueThreadInteractionService(tx as unknown as Db).acceptInteraction(issue, context.interaction.id, {}, { userId: context.linkedUserId });
+        await vi.advanceTimersByTimeAsync(100);
+        expect(context.wakeup).not.toHaveBeenCalled();
+        expect(await db.select({ status: issueThreadInteractions.status }).from(issueThreadInteractions)
+          .where(eq(issueThreadInteractions.id, context.interaction.id))).toEqual([{ status: "pending" }]);
+        expect(await db.select({ id: chatActions.id }).from(chatActions).where(and(
+          eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+        ))).toEqual([]);
+      };
+      await expect(db.transaction(async tx => {
+        await accept(tx);
+        throw new Error("rollback outer continuation decision");
+      })).rejects.toThrow("rollback outer continuation decision");
+      await settle();
+      expect(context.wakeup).not.toHaveBeenCalled();
+      const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
+      try {
+        await vi.advanceTimersByTimeAsync(120_000);
+        expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
+      } finally { select.mockRestore(); transaction.mockRestore(); }
+      await db.transaction(accept);
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(10);
+        expect(context.wakeup).toHaveBeenCalledTimes(1);
+      });
+      await settle();
+      expect(await db.select({ status: chatActions.status }).from(chatActions).where(and(
+        eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+      ))).toEqual([{ status: "processed" }]);
+      expect(errors).toEqual([]);
+    } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); }
+  });
+
   it.each(["accept", "reject", "drain", "restart", "failed selection"] as const)("recovers committed terminal confirmations without polling through %s", async mode => {
     const context = await telegramConfirmationRecoveryFixture(mode === "reject" ? "reject" : "accept");
     let canRun = true;

server/src/app.ts

@@ -1288,6 +1288,7 @@ export async function createApp(
   chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
+  chatChannels.registerInteractionContinuationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());

server/src/services/chat-channels.ts

@@ -3354,7 +3354,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
 
   async function processPublicationMaintenance(limit = 25) {
     scheduleGitHubMaintenance();
-    await processPendingInteractionWakeups(limit);
   }
 
   // The app refills free endpoint slots on commits and durable deadlines.
@@ -3479,7 +3478,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     withCredentialMutationLease, receiptReactionCompletionRunId,
   });
   const { processSelectedPublication } = publicationWorker;
-  const { processPendingInteractionWakeups } = createChatInteractionContinuations(db, {
+  const { processPendingInteractionWakeups, registerInteractionContinuationWork } = createChatInteractionContinuations(db, {
+    preparePublications: scope => processPendingPublications(25, { maintenance: false, scope }),
     redactError, CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS, get heartbeat() { return options.heartbeat; },
   });
   const { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions, registerTerminalConfirmationWork } = createChatInteractionCallbacks(db, {
@@ -3505,6 +3505,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
   const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor });
   const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
 return {
+    registerInteractionContinuationWork,
     registerTerminalConfirmationWork,
     loadFailedRetryNoticeScopes,
     reconcileFailedRetryNotices,

server/src/services/chat-channels/interactions/continuations.ts

@@ -1,8 +1,15 @@
+import { notifyChatContinuationWork, type ChatContinuationScope } from "../../chat-continuation-notifications.js";
+import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js";
+import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js";
+import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js";
 import {
   and,
   asc,
   desc,
   eq,
+  gt,
+  inArray,
+  sql,
   isNotNull,
   isNull,
   lte,
@@ -32,6 +39,7 @@ import { issueThreadInteractionService } from "../../issue-thread-interactions.j
 
 type InteractionContinuationsOptions = {
   heartbeat: IssueAssignmentWakeupDeps;
+  preparePublications: (scope: { companyId: string; endpointId: string; conversationId: string }) => Promise<unknown>;
   redactError: (error: unknown) => string;
   CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS: number;
 };
@@ -42,6 +50,9 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
 
   async function processPendingInteractionWakeups(
     limit = 100,
+    actionId?: string,
+    preparePublications = false,
+    canContinue: () => boolean = () => true,
   ): Promise<number> {
     const staleBefore = new Date(Date.now() - 30_000);
     const retryBefore = new Date(
@@ -53,6 +64,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
       .where(
         and(
           eq(chatActions.kind, "interaction_wakeup"),
+          actionId ? eq(chatActions.id, actionId) : undefined,
           or(
             and(
               eq(chatActions.status, "issued"),
@@ -72,7 +84,17 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
       .limit(limit);
     let processed = 0;
     for (const candidate of candidates) {
-      const [claimed] = await db
+      if (preparePublications && candidate.conversationId) {
+        await options.preparePublications({ companyId: candidate.companyId,
+          endpointId: candidate.endpointId, conversationId: candidate.conversationId });
+      }
+      if (!canContinue()) break;
+      const writeAction = <T>(write: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => PromiseLike<T>) =>
+        db.transaction(async tx => {
+          await notifyChatContinuationWork(tx, { companyId: candidate.companyId, actionId: candidate.id });
+          return write(tx);
+        });
+      const [claimed] = await writeAction(tx => tx
         .update(chatActions)
         .set({ status: "processing", updatedAt: new Date() })
         .where(
@@ -93,7 +115,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
             ),
           ),
         )
-        .returning();
+        .returning());
       if (!claimed) continue;
       const payload = claimed.payload;
       const requestedByActorType =
@@ -122,7 +144,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
         !requestedByActorType ||
         !requestedByActorId
       ) {
-        await db
+        await writeAction(tx => tx
           .update(chatActions)
           .set({
             status: "failed",
@@ -134,7 +156,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
               eq(chatActions.id, claimed.id),
               eq(chatActions.status, "processing"),
             ),
-          );
+          ));
         continue;
       }
       const currentIssue = await db
@@ -167,7 +189,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
         currentIssue.status === "done" ||
         currentIssue.status === "cancelled"
       ) {
-        await db
+        await writeAction(tx => tx
           .update(chatActions)
           .set({
             status: "processed",
@@ -179,7 +201,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
               eq(chatActions.id, claimed.id),
               eq(chatActions.status, "processing"),
             ),
-          );
+          ));
         processed += 1;
         continue;
       }
@@ -325,7 +347,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
             .limit(1)
             .then((rows) => rows[0] ?? null));
         if (!queuedRun && !durable) {
-          await db
+          await writeAction(tx => tx
             .update(chatActions)
             .set({
               status: "issued",
@@ -337,10 +359,10 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
                 eq(chatActions.id, claimed.id),
                 eq(chatActions.status, "processing"),
               ),
-            );
+            ));
           continue;
         }
-        await db
+        await writeAction(tx => tx
           .update(chatActions)
           .set({
             status: "processed",
@@ -358,7 +380,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
               eq(chatActions.id, claimed.id),
               eq(chatActions.status, "processing"),
             ),
-          );
+          ));
         processed += 1;
       } catch (error) {
         const durable = isUniqueViolation(error)
@@ -393,7 +415,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
             ? previousResult.attemptCount
             : 0;
         const attemptCount = previousAttempts + 1;
-        await db
+        await writeAction(tx => tx
           .update(chatActions)
           .set({
             status: durable ? "processed" : "issued",
@@ -412,7 +434,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
               eq(chatActions.id, claimed.id),
               eq(chatActions.status, "processing"),
             ),
-          );
+          ));
         if (!durable) {
           logger.warn(
             {
@@ -428,5 +450,53 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
     }
     return processed;
   }
-  return { processPendingInteractionWakeups };
+  function registerInteractionContinuationWork(
+    coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+    canRun: () => boolean,
+  ) {
+    type Entry = { id: string; companyId: string; issueId: string | null; dueAt: number };
+    const selection = {
+      id: chatActions.id, companyId: chatActions.companyId,
+      issueId: sql<string | null>`${chatActions.payload}->>'issueId'`,
+      status: chatActions.status, result: chatActions.result, updatedAt: chatActions.updatedAt,
+    };
+    const entry = (row: { id: string; companyId: string; issueId: string | null; status: string;
+      result: unknown; updatedAt: Date }): Entry => ({
+      id: row.id, companyId: row.companyId, issueId: row.issueId,
+      dueAt: row.status === "processing" ? row.updatedAt.getTime() + 30_000
+        : row.result === null ? 0 : row.updatedAt.getTime() + CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS,
+    });
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.chatInteractionContinuations,
+      decodeScope: (hint: string): ChatContinuationScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
+      matches: (value: Entry, scope?: ChatContinuationScope) => !scope ||
+        (value.companyId === scope.companyId && (!scope.actionId || value.id === scope.actionId) &&
+          (!scope.issueId || value.issueId === scope.issueId)),
+      load: async (scope?: ChatContinuationScope, signal?: AbortSignal): Promise<Entry[]> => {
+        const entries: Entry[] = [];
+        let cursor: string | undefined;
+        while (!signal?.aborted) {
+          const rows = await db.select(selection).from(chatActions).where(and(
+            eq(chatActions.kind, "interaction_wakeup"), inArray(chatActions.status, ["issued", "processing"]),
+            scope ? eq(chatActions.companyId, scope.companyId) : undefined,
+            scope?.actionId ? eq(chatActions.id, scope.actionId) : undefined,
+            scope?.issueId ? eq(sql<string>`${chatActions.payload}->>'issueId'`, scope.issueId) : undefined,
+            cursor ? gt(chatActions.id, cursor) : undefined,
+          )).orderBy(asc(chatActions.id)).limit(100);
+          entries.push(...rows.map(entry));
+          if (rows.length < 100) return entries;
+          cursor = rows[rows.length - 1].id;
+        }
+        throw signal?.reason ?? new Error("Interaction continuation recovery stopped");
+      },
+      check: async (value: Entry, signal?: AbortSignal) => {
+        await processPendingInteractionWakeups(1, value.id, true, () => canRun() && !signal?.aborted);
+        const [current] = await db.select(selection).from(chatActions).where(and(
+          eq(chatActions.id, value.id), inArray(chatActions.status, ["issued", "processing"]),
+        ));
+        return current ? entry(current) : null;
+      },
+    }, canRun);
+  }
+  return { processPendingInteractionWakeups, registerInteractionContinuationWork };
 }

server/src/services/chat-channels/publications/queue.ts

@@ -181,13 +181,21 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
 
   async function processPendingPublications(
     limit = 25,
-    { waitForCompletion = true, maintenance = true }: { waitForCompletion?: boolean; maintenance?: boolean } = {},
+    { waitForCompletion = true, maintenance = true, scope }: {
+      waitForCompletion?: boolean; maintenance?: boolean;
+      scope?: { companyId: string; endpointId: string; conversationId: string };
+    } = {},
   ) {
     if (isShuttingDown()) return 0;
     if (maintenance) {
       await enqueueInboundWakeupPublications();
       await reconcileTerminalConfirmationActions(limit);
     }
+    const scopeCondition = scope ? and(eq(chatPublications.companyId, scope.companyId),
+      eq(chatPublications.endpointId, scope.endpointId), eq(chatPublications.conversationId, scope.conversationId)) : undefined;
+    const pendingTasks = () => scope
+      ? [publicationEndpointTasks.get(scope.endpointId)].filter((task): task is Promise<void> => Boolean(task))
+      : [...publicationEndpointTasks.values()];
     const now = new Date();
     const staleBefore = new Date(now.getTime() - 60_000);
     // A process that disappears after the provider accepted a post but before
@@ -196,6 +204,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
     // checking the provider conversation.
     const quarantinedPublications = await db.transaction(async tx => {
       const condition = and(
+        scopeCondition,
         sql`not exists (select 1 from chat_endpoints e where e.id = ${chatPublications.endpointId} and e.publication_mode = 'explicit')`,
         eq(chatPublications.state, "streaming"),
         lte(chatPublications.updatedAt, staleBefore),
@@ -295,7 +304,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
           .select()
           .from(chatPublications)
           .where(
-            publicationEligibility({ now, busyEndpointIds, coolingTeamsPublicationIds, attemptedIds }),
+            and(scopeCondition, publicationEligibility({ now, busyEndpointIds, coolingTeamsPublicationIds, attemptedIds })),
           )
           .orderBy(
             asc(chatPublications.createdAt),
@@ -308,8 +317,9 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
           // exposing a new head that must be queried before declaring idle.
           if (busyEndpointIds.some((id) => !publicationEndpointTasks.has(id)))
             continue;
-          if (!waitForCompletion || publicationEndpointTasks.size === 0) break;
-          await Promise.race(publicationEndpointTasks.values());
+          const pending = pendingTasks();
+          if (!waitForCompletion || pending.length === 0) break;
+          await Promise.race(pending);
           continue;
         }
         let started = false;

server/src/services/chat-channels/telegram/confirmation-actions.ts

@@ -1,3 +1,4 @@
+import { notifyChatContinuationWork } from "../../chat-continuation-notifications.js";
 import type { DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js";
 
 import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js";
@@ -586,6 +587,7 @@ export function createTelegramConfirmationActions(db: Db, options: TelegramConfi
         })
         .onConflictDoNothing();
       if (shouldWake && issue.assigneeAgentId) {
+        await notifyChatContinuationWork(tx, { companyId: issue.companyId, issueId: issue.id });
         await tx
           .insert(chatActions)
           .values({

server/src/services/chat-continuation-notifications.ts

@@ -0,0 +1,8 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export type ChatContinuationScope = { companyId: string; issueId?: string; actionId?: string };
+
+export async function notifyChatContinuationWork(tx: object, scope: ChatContinuationScope) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInteractionContinuations,
+    JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, actionId: scope.actionId }));
+}

server/src/services/chat-interaction-publications.ts

@@ -1,3 +1,4 @@
+import { notifyChatContinuationWork } from "./chat-continuation-notifications.js";
 import { notifyChatPublicationWork } from "./chat-work-notifications.js";
 import { nativePhotonInteraction } from "./photon/interactions.js";
 import { randomBytes } from "node:crypto";
@@ -669,51 +670,54 @@ export async function enqueueTerminalIssueInteractionChatPublications(
         interaction.resolvedByAgentId === issue.assigneeAgentId
       )
     ) {
-      await db
-        .insert(chatActions)
-        .values({
-          companyId: interaction.companyId,
-          endpointId: wakeBinding.endpointId,
-          conversationId: wakeBinding.conversationId,
-          kind: "interaction_wakeup",
-          providerActionId: `interaction_wakeup:${interaction.id}`,
-          payload: {
-            version: 1,
-            interactionId: interaction.id,
-            interactionKind: interaction.kind,
-            interactionStatus: interaction.status,
-            issueId: interaction.issueId,
-            agentId: issue.assigneeAgentId,
-            sourceCommentId: interaction.sourceCommentId ?? null,
-            sourceRunId: interaction.sourceRunId ?? null,
-            requestedByActorType,
-            requestedByActorId,
-            ...(planTarget
-              ? {
-                  planReviewInteraction: {
-                    id: interaction.id,
-                    kind: interaction.kind,
-                    status: interaction.status,
-                    target: planTarget,
-                    acceptedTargetRevision:
-                      interaction.status === "accepted" ? planTarget : null,
-                    result: interaction.result,
-                  },
-                }
-              : {}),
-            ...(interaction.status === "accepted" && planTarget
-              ? {
-                  forceFreshSession: true,
-                  workspaceRefreshReason: "accepted_plan_confirmation",
-                }
-              : {}),
-            ...(interaction.resolvedByUserId
-              ? { requestedByUserId: interaction.resolvedByUserId }
-              : {}),
-          },
-          status: "issued",
-        })
-        .onConflictDoNothing();
+      await db.transaction(async tx => {
+        await notifyChatContinuationWork(tx, { companyId: interaction.companyId, issueId: interaction.issueId });
+        await tx
+          .insert(chatActions)
+          .values({
+            companyId: interaction.companyId,
+            endpointId: wakeBinding.endpointId,
+            conversationId: wakeBinding.conversationId,
+            kind: "interaction_wakeup",
+            providerActionId: `interaction_wakeup:${interaction.id}`,
+            payload: {
+              version: 1,
+              interactionId: interaction.id,
+              interactionKind: interaction.kind,
+              interactionStatus: interaction.status,
+              issueId: interaction.issueId,
+              agentId: issue.assigneeAgentId,
+              sourceCommentId: interaction.sourceCommentId ?? null,
+              sourceRunId: interaction.sourceRunId ?? null,
+              requestedByActorType,
+              requestedByActorId,
+              ...(planTarget
+                ? {
+                    planReviewInteraction: {
+                      id: interaction.id,
+                      kind: interaction.kind,
+                      status: interaction.status,
+                      target: planTarget,
+                      acceptedTargetRevision:
+                        interaction.status === "accepted" ? planTarget : null,
+                      result: interaction.result,
+                    },
+                  }
+                : {}),
+              ...(interaction.status === "accepted" && planTarget
+                ? {
+                    forceFreshSession: true,
+                    workspaceRefreshReason: "accepted_plan_confirmation",
+                  }
+                : {}),
+              ...(interaction.resolvedByUserId
+                ? { requestedByUserId: interaction.resolvedByUserId }
+                : {}),
+            },
+            status: "issued",
+          })
+          .onConflictDoNothing();
+      });
     }
   }
 

server/src/services/delivery-work-notifications.ts

@@ -56,6 +56,7 @@ export const DELIVERY_QUEUES = {
   chatPublications: "chat-publications",
   chatNativeProgress: "chat-native-progress",
   chatInboundWakeNotices: "chat-inbound-wake-notices",
+  chatInteractionContinuations: "chat-interaction-continuations",
   chatTerminalConfirmations: "chat-terminal-confirmations",
   chatFailedRetryNotices: "chat-failed-retry-notices",
   chatRunMilestones: "chat-run-milestones",