fix(server): wake interaction continuations from committed responses (#17653)

https://entire.com/gh/paperclipai/-/trails/204/changes/paperclip/204  ## Thinking Path > - Paperclip coordinates agent work and human approvals. > - Human decisions already persist the continuation action that resumes an agent. > - A response can commit before its callback or publication finishes. > - Periodic maintenance repeatedly discovers those rows, including when idle. > - Committed writes should wake the existing owner; saved retries and claims supply real deadlines. > - Startup reconstructs durable work, and settled history becomes quiet. ## Linked Issues or Issue Description Refs #17651, which is merged. This existing continuation slice targets master and removes its duty from periodic chat maintenance. **Problem:** A saved `interaction_wakeup` action can survive a committed human response whose callback crashes. Without producer coverage, removing maintenance strands the continuation. Repeated empty maintenance also keeps the database active. **Expected behavior:** The outer transaction's settlement wakes the continuation owner. It handles ready conversation publications, dispatches under existing authority, preserves interrupted/retry work, and becomes quiet after settlement. ## What Changed - Register continuation recovery with the shared scoped deadline coordinator. Reconstruct existing issued/processing actions and retain the existing retry and stale-claim deadlines. - Notify from both enqueue producers and owner-local claim/settlement writes. Preserve outer commit/rollback and avoid moving provider or heartbeat calls into database transactions. - Give the existing publication queue an optional company/endpoint/conversation scope. The continuation owner gives ready cards in that conversation a bounded delivery opportunity before claiming the action; FIFO, capacity, ambiguity and provider authority remain with the queue. - Carry admission/cancellation through publication preparation and check before every new send. Join in-flight requests and retain pending actions/publications through drain, standby, shutdown and resume. - Remove only recurring continuation recovery; keep the manual API. Add no table. A concurrent partial index on existing pending continuation actions and matching company/id keyset paging prevent retained history from being scanned. Parent #17651's indexed confirmation discovery and current master are retained. ## Verification Published head: `a1d16f84b3fae361d5d648670a5964486370c3a3`, on actual landed terminal-confirmation parent `ee67d2ece9bda157dfad9bdf25175b1d004ed341`. - **75 distinct current cases pass:** 32 selected real PostgreSQL chat cases (14 continuation lifecycle/producer/stop cases, one new continuation query-cost case, and 17 parent confirmation/query-cost cases; 1,671 skipped), and all 43 cases in complete interaction-publication and app/startup suites. Database build, server typecheck/build, module boundaries and fresh migrations through 0399 pass. The original landed-parent matrix passed 74 cases before the additional query-cost repair; those are not added again. - The new real-SQL regression seeds 20,000 completed/unrelated actions and one pending processing claim. Before repair it examines 20,345 rows and hits 20,421 buffers. The final startup and company-scoped queries must examine at most two action rows and fewer than 20 action buffers; empty scoped recovery must also stay below 20 buffers. Planner settings are unchanged. Migration `0399_ambitious_onslaught.sql` is concurrent/idempotent and uses the existing top-level migration runner; prior SQL/journal entries and the five newest snapshots are preserved. - The matrix covers callback recovery, concurrent responses, malformed prefixes, more than one page, saved retries/stale claims, outer rollback/commit, deferred/future/paused publication states, restart, drain and stop before each new provider request. Quiet assertions inspect real selects/transactions across virtual idle time. - Normal integration `db5a84c0f8b73a0ce863f4d2fbaee0a115cf044f` preserves the complete 12-path owned binary patch and all 12 unrelated current-master paths, including managed Cloud sandbox changes. A verified semantic base resolves squash-ancestry conflicts; both normal parents and the exact tree are checked. The subsequent index/keyset repair is a deliberate additional change, tested separately; the final patch is not claimed identical to the original 12-file implementation. No force push or rebase occurs. - The same 74 pre-index cases passed on the earlier prepared and actual landed parents. The new query-cost regression failed before its repair, then passes on the repaired source. Earlier 57-case and broad acceptance retain their original revision scopes; repeated cases are not added to the current count. The audit-only follow-up leaves all owned server source/test hashes unchanged. - [The owner audit](https://github.com/paperclipai/paperclip/blob/a1d16f84b3fae361d5d648670a5964486370c3a3/doc/plans/2026-10-25-interaction-continuation-sleep-audit.md#acceptance-on-the-actual-landed-terminal-confirmation-parent) retains the original missing-signal regression, the skipped standalone publication suite, and the review's stop-before-send failures with their corrections. Skips and earlier failing revisions are not relabelled as current passes. - **Fresh exact-head CI and Greptile 5/5 without unresolved actionable findings remain required.** The former remote head's reviews are stale. No full local repository suite, live provider or deployed sleep result is claimed. ## Risks - Process-local transaction hints cover supported application writers and shared pools. Independent SQL writers and other replicas do not gain a new wake guarantee. - The publication queue's existing FIFO, capacity, retry/ambiguity and authorization contracts still govern effects. In-flight provider requests can settle after Stop; the fence prevents a subsequent request from starting. - Missing a supported producer can defer recovery until another change or startup. Producer and rollback tests cover this slice's supported paths. - Worker quiet is separate from host/database suspension. External alarm coverage and Cloud sleep gates are unchanged. ## Model Used OpenAI Codex, using repository inspection, code execution and real PostgreSQL tests. The exact model identifier and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-11-10 20:46:28 +01:00 · 2026-10-28 15:45:18 -05:00
1 parent 287b04d72f
commit 063cfd4381
@@ -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,");
  });
@@ -45693,6 +45693,349 @@ 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("bounds interaction continuation recovery reads with completed history", async () => {
+    const context = await telegramConfirmationRecoveryFixture();
+    await expect(context.onAction(context.actionEvent))
+      .rejects.toThrow("injected post-interaction confirmation settlement crash");
+    const [action] = await db.select().from(chatActions).where(and(
+      eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"),
+    ));
+    expect(action).toBeDefined();
+    await db.update(chatActions).set({ status: "processing", updatedAt: new Date() }).where(eq(chatActions.id, action.id));
+    const prefix = `continuation-plan:${randomUUID()}:`;
+    await db.execute(sql`
+      INSERT INTO chat_actions (company_id, endpoint_id, kind, provider_action_id, status)
+      SELECT ${context.fixture.companyId}::uuid, ${context.endpoint.id}::uuid,
+        CASE WHEN n <= 10000 THEN 'interaction_wakeup' ELSE 'unrelated_action' END,
+        ${prefix} || n,
+        CASE WHEN n <= 10000 THEN 'processed' ELSE 'processing' END
+      FROM generate_series(1, 20000) AS n
+    `);
+    await db.execute(sql`VACUUM (ANALYZE) chat_actions`);
+    const statements: { sql: string; params: unknown[] }[] = [];
+    const select = db.select.bind(db);
+    const reads = vi.spyOn(db, "select").mockImplementation((...args) => {
+      const selection = select(...args), from = selection.from.bind(selection);
+      vi.spyOn(selection, "from").mockImplementation(table => {
+        const query = from(table);
+        if (table === chatActions) {
+          const execute = query.execute.bind(query);
+          vi.spyOn(query, "execute").mockImplementation(async () => {
+            const statement = query.toSQL();
+            if (statement.params.includes("interaction_wakeup") || statement.sql.includes("'interaction_wakeup'")) {
+              statements.push(statement);
+            }
+            return execute();
+          });
+        }
+        return query;
+      });
+      return selection;
+    });
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    try {
+      await context.service.registerInteractionContinuationWork(coordinator, () => true).ready;
+      expect(statements.length).toBeGreaterThan(0);
+      const startup = statements[0], beforeHint = statements.length;
+      await db.transaction(async tx => {
+        await signalDatabaseWork(tx, DELIVERY_QUEUES.chatInteractionContinuations,
+          JSON.stringify({ companyId: context.fixture.companyId }));
+      });
+      await vi.waitFor(() => expect(statements.length).toBeGreaterThan(beforeHint), { timeout: 5_000 });
+      const scoped = statements.at(-1)!;
+      expect(scoped.params).toContain(context.fixture.companyId);
+      await coordinator.stop(); reads.mockRestore();
+      for (const statement of [startup, scoped]) {
+        expect(await db.$client.unsafe(statement.sql, statement.params as Parameters<typeof db.$client.unsafe>[1]))
+          .toMatchObject([{ id: action.id }]);
+        const result = await db.$client.unsafe(
+          `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${statement.sql}`,
+          statement.params as Parameters<typeof db.$client.unsafe>[1],
+        );
+        type PlanNode = { Plans?: PlanNode[]; [key: string]: unknown };
+        const nodes = (node: PlanNode): PlanNode[] => [node, ...(node.Plans ?? []).flatMap(nodes)];
+        const actionReads = nodes(result[0]["QUERY PLAN"][0].Plan)
+          .filter(node => node["Relation Name"] === "chat_actions");
+        expect(actionReads).toHaveLength(1);
+        expect(Number(actionReads[0]["Actual Rows"]) + Number(actionReads[0]["Rows Removed by Filter"] ?? 0),
+          JSON.stringify({ params: statement.params, actionReads })).toBeLessThanOrEqual(2);
+        expect(Number(actionReads[0]["Shared Hit Blocks"]) + Number(actionReads[0]["Shared Read Blocks"]))
+          .toBeLessThan(20);
+      }
+      await db.update(chatActions).set({ status: "processed" }).where(eq(chatActions.id, action.id));
+      await db.execute(sql`ANALYZE chat_actions`);
+      expect(await db.$client.unsafe(scoped.sql, scoped.params as Parameters<typeof db.$client.unsafe>[1])).toHaveLength(0);
+      const empty = await db.$client.unsafe(
+        `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${scoped.sql}`, scoped.params as Parameters<typeof db.$client.unsafe>[1],
+      );
+      expect(empty[0]["QUERY PLAN"][0].Plan["Shared Hit Blocks"]
+        + empty[0]["QUERY PLAN"][0].Plan["Shared Read Blocks"]).toBeLessThan(20);
+      expect(errors).toEqual([]);
+    } finally {
+      await coordinator.stop(); reads.mockRestore();
+      await db.delete(chatActions).where(sql`${chatActions.providerActionId} LIKE ${`${prefix}%`}`);
+    }
+  });
+
+  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.each(["drain", "standby", "shutdown"] as const)("stops starting continuation publications during %s", async mode => {
+    const context = await telegramConfirmationRecoveryFixture();
+    await expect(context.onAction(context.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash");
+    const next = await issueThreadInteractionService(db).create(
+      { id: context.conversation.issueId, companyId: context.fixture.companyId },
+      { kind: "request_confirmation", continuationPolicy: "wake_assignee", title: "Next decision",
+        payload: { version: 1, prompt: "A second decision after the accepted response?" } },
+      { agentId: context.fixture.assignedAgentId },
+    );
+    const nextPublication = () => db.select({ state: chatPublications.state }).from(chatPublications).where(
+      eq(chatPublications.idempotencyKey, `interaction:${next.id}:${context.endpoint.id}`));
+    expect(await nextPublication()).toEqual([{ state: "pending" }]);
+    const provider = (context.service.runtime as unknown as FakeChatSdkRuntime).endpoints.get(context.endpoint.id)!;
+    const originalThread = provider.thread.bind(provider);
+    let entered = false;
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const threadSpy = vi.spyOn(provider, "thread").mockImplementation(threadId => {
+      const thread = originalThread(threadId), edit = thread.adapter.editMessage.bind(thread.adapter);
+      thread.adapter.editMessage = async (...args) => { entered = true; await gate; return edit(...args); };
+      return thread;
+    });
+    let canRun = true;
+    const errors: unknown[] = [];
+    let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
+      canReconcile: () => true, onError: error => errors.push(error) });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const ready = context.service.registerInteractionContinuationWork(coordinator, () => canRun).ready;
+    let stopping: Promise<void> | undefined;
+    try {
+      await vi.waitFor(() => expect(entered).toBe(true), { timeout: 10_000 });
+      if (mode === "shutdown") stopping = coordinator.stop(); else canRun = false;
+      release(); await ready; await stopping;
+      await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+      expect(provider.edits).toHaveLength(1);
+      expect(await nextPublication()).toEqual([{ state: "pending" }]);
+      expect(context.wakeup).not.toHaveBeenCalled();
+      if (mode === "shutdown") {
+        coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
+          canReconcile: () => true, onError: error => errors.push(error) });
+        await context.service.registerInteractionContinuationWork(coordinator, () => canRun).ready;
+      } else {
+        canRun = true;
+        await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now()));
+      }
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(10);
+        expect(context.wakeup).toHaveBeenCalledTimes(1);
+        expect(await nextPublication()).toEqual([{ state: "published" }]);
+        expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
+      }, { timeout: 10_000 });
+      expect(errors).toEqual([]);
+    } finally {
+      release(); await ready; await stopping; await coordinator.stop(); vi.useRealTimers();
+      threadSpy.mockRestore(); await context.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;
@@ -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());
@@ -3350,7 +3350,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.
@@ -3475,7 +3474,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
    withCredentialMutationLease, receiptReactionCompletionRunId,
  });
  const { processSelectedPublication } = publicationWorker;
-  const { processPendingInteractionWakeups } = createChatInteractionContinuations(db, {
+  const { processPendingInteractionWakeups, registerInteractionContinuationWork } = createChatInteractionContinuations(db, {
+    preparePublications: (scope, canContinue) => processPendingPublications(25, { maintenance: false, scope, canContinue }),
    redactError, CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS, get heartbeat() { return options.heartbeat; },
  });
  const { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions, registerTerminalConfirmationWork } = createChatInteractionCallbacks(db, {
@@ -3502,6 +3502,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,
@@ -1,3 +1,4 @@
+import { notifyChatContinuationWork } from "../../chat-continuation-notifications.js";
 import { connectionIntentService } from "../../connection-intents.js";
 import { queueIssueAssignmentWakeup, type IssueAssignmentWakeupDeps } from "../../issue-assignment-wakeup.js";
 import { chatInteractionPresentation } from "../../chat-interaction-presentation.js";
@@ -580,6 +581,7 @@ export function createChatConfirmationActions(db: Db, options: ChatConfirmationO
        })
        .onConflictDoNothing();
      if (shouldWake && issue.assigneeAgentId) {
+        await notifyChatContinuationWork(tx, { companyId: issue.companyId, issueId: issue.id });
        await tx
          .insert(chatActions)
          .values({
@@ -1,8 +1,14 @@
+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,
+  inArray,
+  sql,
  isNotNull,
  isNull,
  lte,
@@ -32,6 +38,7 @@ import { issueThreadInteractionService } from "../../issue-thread-interactions.j

 type InteractionContinuationsOptions = {
  heartbeat: IssueAssignmentWakeupDeps;
+  preparePublications: (scope: { companyId: string; endpointId: string; conversationId: string }, canContinue: () => boolean) => Promise<unknown>;
  redactError: (error: unknown) => string;
  CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS: number;
 };
@@ -42,6 +49,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 +63,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 +83,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 }, canContinue);
+      }
+      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 +114,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
            ),
          ),
        )
-        .returning();
+        .returning());
      if (!claimed) continue;
      const payload = claimed.payload;
      const requestedByActorType =
@@ -122,7 +143,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
        !requestedByActorType ||
        !requestedByActorId
      ) {
-        await db
+        await writeAction(tx => tx
          .update(chatActions)
          .set({
            status: "failed",
@@ -134,7 +155,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
              eq(chatActions.id, claimed.id),
              eq(chatActions.status, "processing"),
            ),
-          );
+          ));
        continue;
      }
      const currentIssue = await db
@@ -167,7 +188,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 +200,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
              eq(chatActions.id, claimed.id),
              eq(chatActions.status, "processing"),
            ),
-          );
+          ));
        processed += 1;
        continue;
      }
@@ -325,7 +346,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 +358,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 +379,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 +414,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 +433,7 @@ export function createChatInteractionContinuations(db: Db, options: InteractionC
              eq(chatActions.id, claimed.id),
              eq(chatActions.status, "processing"),
            ),
-          );
+          ));
        if (!durable) {
          logger.warn(
            {
@@ -428,5 +449,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: { companyId: string; id: string } | undefined;
+        while (!signal?.aborted) {
+          const rows = await db.select(selection).from(chatActions).where(and(
+            sql`${chatActions.kind} = 'interaction_wakeup' AND ${chatActions.status} IN ('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 ? sql`(${chatActions.companyId}, ${chatActions.id}) > (${cursor.companyId}, ${cursor.id})` : undefined,
+          )).orderBy(asc(chatActions.companyId), asc(chatActions.id)).limit(100);
+          entries.push(...rows.map(entry));
+          if (rows.length < 100) return entries;
+          cursor = rows[rows.length - 1];
+        }
+        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 };
 }
@@ -181,13 +181,22 @@ 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, canContinue = () => true }: {
+      waitForCompletion?: boolean; maintenance?: boolean; canContinue?: () => boolean;
+      scope?: { companyId: string; endpointId: string; conversationId: string };
+    } = {},
  ) {
-    if (isShuttingDown()) return 0;
+    if (isShuttingDown() || !canContinue()) return 0;
    if (maintenance) {
      await enqueueInboundWakeupPublications();
      await reconcileTerminalConfirmationActions(limit);
    }
+    if (isShuttingDown() || !canContinue()) return 0;
+    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 +205,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),
@@ -278,7 +288,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
    let failed = false;
    let firstError: unknown;
    try {
-      while (attemptedIds.length < limit && !failed && !isShuttingDown()) {
+      while (attemptedIds.length < limit && !failed && !isShuttingDown() && canContinue()) {
        if (publicationEndpointTasks.size >= PUBLICATION_ENDPOINT_CONCURRENCY) {
          if (!waitForCompletion) break;
          await Promise.race(publicationEndpointTasks.values());
@@ -295,7 +305,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 +318,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;
@@ -318,7 +329,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
            publicationEndpointTasks.size >= PUBLICATION_ENDPOINT_CONCURRENCY ||
            attemptedIds.length >= limit ||
            failed ||
-            isShuttingDown()
+            (isShuttingDown() || !canContinue())
          )
            break;
          if (publicationEndpointTasks.has(selectedPublication.endpointId))
@@ -330,6 +341,7 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
          // endpoint-exclusive credential fence instead of timing out behind it.
          const task = Promise.resolve()
            .then(async () => {
+              if (isShuttingDown() || !canContinue()) return;
              await processSelectedPublication(selectedPublication);
              await settleExternalConversation(db, selectedPublication.companyId, selectedPublication.issueId).catch((err) => {
                logger.warn({ err, publicationId: selectedPublication.id }, "Slack conversation settlement deferred to reconciliation");
@@ -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 }));
+}
@@ -1,3 +1,4 @@
+import { notifyChatContinuationWork } from "./chat-continuation-notifications.js";
 import { chatInteractionForm } from "./chat-interaction-forms.js";
 import { slackInteractionRequiresBoard } from "./chat-channels/slack/cards.js";
 import { answeredChatQuestionBody, chatInteractionPresentation } from "./chat-interaction-presentation.js";
@@ -710,51 +711,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();
+      });
    }
  }

@@ -59,6 +59,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",