refactor(server): isolate bounded failed retry notice projection

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

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -1753,3 +1753,24 @@ wake notice audit for the producer inventory, regressions and fixture correction
 This update does not clear the older Grok/runtime-discovery CI failures, voice
 session recovery review, or inbound voice cleanup ownership findings, and it does
 not shrink the remaining original checklist.
+
+
+## Slice 30 gate and slice 31 projection checkpoint (2026-10-25)
+
+[#17647](https://github.com/paperclipai/paperclip/pull/17647) at
+`fee07f3fa8cbc489cb9bbf657a6065c6e00748c3` has all 53 checks passing,
+a fresh Greptile 5/5 review, and its fairness finding resolved. The pre-repair
+full chat/startup acceptance passed 1,486 cases; the bounded-candidate repair
+passes 44 selected integration cases and server typecheck/build. No merge occurred.
+
+The next separate branch, `codex/sleep-31-failed-retry-notices`, fixes a reproduced
+manual-cursor starvation defect and adds scoped, bounded retry-notice projection.
+Its normal periodic pass remains until producer/lifecycle migration is complete.
+See [the retry notice audit](2026-10-25-failed-retry-notices-sleep-audit.md).
+
+Fresh older-parent audit still shows #17490 at `31bcd19888` and #17572 at
+`8815dfad38` failing E2E shard 3 and its aggregate. #17576 still has unresolved
+voice inventory amplification findings 4371283330/4371307607, and #17580 still
+has unresolved company-offboarding/cleanup findings 4370459800/4370543984. These
+are required repairs, not exemptions. The full original migration scope,
+master reconciliation, and all latest-head gates remain required.

doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md

@@ -0,0 +1,141 @@
+# Failed-run retry notice migration
+
+Date: 2026-10-25. Branch: `codex/sleep-31-failed-retry-notices`.
+Parent: [#17647](https://github.com/paperclipai/paperclip/pull/17647),
+`fee07f3fa8cbc489cb9bbf657a6065c6e00748c3`.
+This is slice 31 of the recurring-server-work stack. No production worker is
+registered yet, and the periodic retry-notice pass remains until producer and
+lifecycle acceptance is complete. No PR has been opened for this slice yet.
+
+## Obligation and boundary
+
+A failed-run retry may have a durable wake receipt but no admitted run. Its
+existing `chat_actions` row has kind `failed_run_retry`; the same-ID,
+same-company `agent_wakeup_requests` row is authoritative for whether the retry
+waits or was declined. The notice is eligible while `runId` is null and the
+receipt is deferred, cancelled, failed, or skipped. A matching existing
+publication discharges that state. Queued and not-started are separate saved
+states; a later cancellation can owe a second notice.
+
+The retry service owns current source/batch authorization, task Retry generation,
+original principal, retained source digest, endpoint reach, provider binding,
+execution containment, and retry admission. The new projection module receives
+a narrow authorization callback returning the destination, task and source
+comment IDs. It does not reproduce that policy or add provider branches.
+Existing publication transport still authorizes immediately before I/O.
+
+The new projection module is `chat-channels/failed-retry-notices.ts`. It owns
+candidate selection, receipt locking, deduplicated insertion, exact cursor
+precision and scoped reconciliation. The service retains its existing public
+combined manual notice API. Nothing adds a database table or a framework adapter.
+
+## Current implementation and evidence
+
+- The manual cursor previously converted `createdAt` to JavaScript Date. A full
+  denied page with PostgreSQL microseconds was selected repeatedly, starving
+  a later valid retry. The regression reproduces through the public combined
+  API in `/tmp/sleep31-cursor-before-public-api.log`: the second call inserts
+  zero instead of the expected one notice.
+- `/tmp/sleep31-cursor-before.log` is an invalid first reproduction that attempted
+  to call a private service method. It is retained as failure evidence, not
+  evidence of the cursor defect.
+- Cursors now use PostgreSQL `timestamp::text` plus the UUID tiebreaker. Manual
+  selection retains its existing maximum batch of 50; scoped reconciliation
+  bounds candidate authorization work, not successful publications.
+- Permanent authority rejection settles the pass. Actual PostgreSQL/authority
+  contention is returned separately for the future worker to retain. Unknown
+  errors propagate. Run-bound and already projected states leave no candidates.
+- Five integration cases pass in `/tmp/sleep31-scoped-notices.log`: manual/scoped
+  denied-prefix pagination, scoped isolation and receipt contention/cancellation,
+  and the existing Slack/Telegram queued/working/selected-final identities.
+  The scoped cursor test observes exactly 50 then one candidate transaction.
+- Server typecheck passes in `/tmp/sleep31-scoped-typecheck.log`.
+  This establishes the projection contract, not idle worker behavior or host
+  suspension. Full acceptance, worker registration, CI and review remain open.
+
+## Producer audit to complete
+
+- [ ] Receipt admission/defer: trace actual heartbeat and wake-queue writers,
+  including the ordinary task Retry intent and the chat action receipt. A
+  callback returning a truthy result is not sufficient persistence evidence.
+- [ ] Receipt cancellation/failure: task tree, agent, budget, company, Slack Stop,
+  deferred queue writers, and retry-specific authorization denial. Reuse the
+  existing transaction signals where possible; test outer rollback and commit.
+- [ ] Admission/promotion and settlement: run-bound receipts stop owing notices;
+  retain notification ordering through transaction commit and failure.
+- [ ] Retry action lifecycle: staged, claimed, processed, failed, cancelled,
+  recovered, and superseded generations; determine which transitions can make
+  current source authorization succeed or fail.
+- [ ] Current authority restoration: endpoint/provider runtime, resources,
+  linked identity, membership, task/project privacy and explicit grants, and
+  original source/run lineage. Use existing eligibility signals rather than
+  scattering new wrappers through these owners.
+- [ ] Source and generation changes: inspect task Retry intent rewrites,
+  recovery completion, preserved source metadata, and repaired delivery records.
+- [ ] Mixed ordinary/retry receipt ownership: `findExistingDeferredWake` does
+  not exclude a failed-retry owner by kind. A retry's own `allowRunCoalescing:
+  false` does not establish what a later ordinary wake can select. The ordinary
+  retirement path requires an exact visible queued publication; retry-owned
+  publication authorization additionally requires its comment ID to equal the
+  final original retry source comment. Establish the reachable combinations with
+  real admission before exempting the retry denial writer from ordinary signals.
+- [ ] Independent clients/replicas: current process-local signals cover only the
+  supported owning application client and child pools. Do not claim external
+  SQL or another replica is observed.
+
+## Worker and acceptance still required
+
+- [ ] Register a scoped deadline worker using the shared coordinator, retaining
+  exact page cursors and contention across pages. Committed invalidation must
+  restart a cursor so restored earlier candidates are revisited.
+- [ ] Empty/configured-idle/retained-history quiet, real first admission, actual
+  producer cancellation, denied history, lock contention, retry after failure,
+  rollback/outer commit, restart, drain/resume, shutdown, and writes arriving
+  during selection and pagination.
+- [ ] Preserve all existing provider identity and authorization acceptance.
+- [ ] Remove only retry notices from periodic maintenance after the above passes.
+- [ ] Full required verification, template PR, latest-head checks and fresh 5/5
+  review with no unresolved findings. Do not merge the stack.
+
+No change to Cloud sleep-safety gates or external wake coverage is proposed.
+Terminal confirmations, interaction continuations and GitHub maintenance remain
+separate duties. The rest of the original migration ledger remains in scope.
+
+
+## Broader acceptance and retained failure evidence
+
+Server build passed in `/tmp/sleep31-scoped-build.log`. Broader selection
+`retry|Retry|pre-run FIFO` first finished with 180 passing and one failing case
+in `/tmp/sleep31-retry-acceptance.log`. The failing test injects the first
+transactional agent-status read error while publishing a selected final. It
+instead reached retry-notice maintenance because the reused test database still
+contained two deferred retry candidates from the deliberate failing reproductions.
+Direct SQL identified those candidates at 13:33:57 and 13:34:35, matching the
+invalid first reproduction and the valid before-fix cursor reproduction. Neither
+had a publication. This is fixture contamination, not an inferred flake from a
+later success.
+
+The cursor fixture now removes its own original staged action/receipt as well
+as its denied clones in `finally`, including on a failed assertion. The original
+database is retained as evidence. After applying migrations to a fresh isolated
+`paperclip_sleep31_retry_acceptance_20261025_test`, all 181 selected cases pass
+in `/tmp/sleep31-retry-acceptance-fresh.log`. No production error handling was
+relaxed to bypass the failure. The local timestamp/fixture evidence is test-only.
+
+Additional producer audit findings to resolve before wiring:
+
+- The admission reader can select a previously deferred retry as a later
+  ordinary wake's owner; the current ordinary publication path requires a
+  matching visible original queue notice, while retry-owned transport requires
+  its original final source comment. Reachability must be established, not
+  assumed from the retry's own no-coalescing option.
+- Wake-queue `cancelDeferredWake` and `failDeferredWake` signal the ordinary
+  inbound-notice helper rather than the broader milestone helper. A retry
+  notification added only to milestones would miss those real writers.
+- `normalizeDeferredWakeCommentIds` changes the saved payload without a notice
+  signal. Check whether removing merged/deleted input restores a retry's exact
+  source eligibility before classifying it as harmless bookkeeping.
+- `retireSupersededTaskRetry`, `requestTaskRetry`, and retry reconciliation write
+  the separate task-intent receipt. The notice candidate uses the chat-action
+  receipt; examine generation/source authorization before adding blanket wakes
+  to all intent timestamp writes.

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

@@ -62607,6 +62607,107 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     }
   });
 
+  it.each(["manual", "scoped"] as const)("advances failed retry notices past denied history with microsecond timestamps (%s)", async mode => {
+    const context = await failedChatRetryFixture("telegram");
+    const rejectedIds = Array.from({ length: 50 }, () => randomUUID());
+    let stagedActionId: string | undefined;
+    try {
+      const staged = await db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, {
+        companyId: context.fixture.companyId, issueId: context.issue.id,
+        agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user",
+      }));
+      stagedActionId = staged.actionId;
+      await context.service.processFailedChatRunRetry(staged.actionId);
+      const [receipt] = await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" })
+        .where(eq(agentWakeupRequests.id, staged.actionId)).returning();
+      const [action] = await db.select().from(chatActions).where(eq(chatActions.id, staged.actionId));
+      const before = new Date(action.createdAt.getTime() - 1000).toISOString();
+      await db.insert(chatActions).values(rejectedIds.map(id => ({
+        ...action, id, providerActionId: `failed-run-retry:${id}`,
+        payload: { ...action.payload, failedRunId: randomUUID() },
+        createdAt: sql`${before}::timestamptz + interval '0.000123 seconds'`,
+      })));
+      await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({
+        ...receipt, id, idempotencyKey: `retry-notice-cursor:${id}`,
+      })));
+      if (mode === "manual") {
+        expect(await context.service.enqueueInboundWakeupPublications(50)).toBe(0);
+        expect(await context.service.enqueueInboundWakeupPublications(50)).toBe(1);
+      } else {
+        const scope = { companyId: context.fixture.companyId, issueId: context.issue.id };
+        const transaction = vi.spyOn(db, "transaction");
+        try {
+          const first = await context.service.reconcileFailedRetryNotices({ scope, limit: 50 });
+          expect(first).toMatchObject({ inserted: 0, contended: false, nextCursor: expect.any(Object) });
+          expect(transaction).toHaveBeenCalledTimes(50); transaction.mockClear();
+          expect(await context.service.reconcileFailedRetryNotices({ scope, limit: 50, cursor: first.nextCursor }))
+            .toEqual({ inserted: 1, contended: false, nextCursor: null });
+          expect(transaction).toHaveBeenCalledTimes(1);
+        } finally { transaction.mockRestore(); }
+      }
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({
+        idempotencyKey: `wake:${staged.actionId}:queued:${context.endpoint.id}:${context.conversation.id}`,
+        payload: { text: "Your retry is queued.", progressState: "queued" },
+      })]);
+    } finally {
+      await context.service.shutdown();
+      const actionIds = stagedActionId ? [...rejectedIds, stagedActionId] : rejectedIds;
+      await db.delete(chatActions).where(inArray(chatActions.id, actionIds));
+      await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, actionIds));
+    }
+  });
+
+  it("scopes failed retry notices and retains receipt contention without retrying settled history", async () => {
+    const context = await failedChatRetryFixture("telegram");
+    const other = await failedChatRetryFixture("telegram");
+    let release = () => {};
+    let holder: Promise<unknown> | undefined;
+    try {
+      const staged = [];
+      for (const target of [context, other]) {
+        const action = await db.transaction(tx => target.service.prepareFailedChatRunRetry(tx, {
+          companyId: target.fixture.companyId, issueId: target.issue.id,
+          agentId: target.fixture.assignedAgentId, failedRunId: target.runId, initiatedByUserId: "owner-user",
+        }));
+        await target.service.processFailedChatRunRetry(action.actionId);
+        await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" })
+          .where(eq(agentWakeupRequests.id, action.actionId));
+        staged.push(action);
+      }
+      const scope = { companyId: context.fixture.companyId, issueId: context.issue.id };
+      expect(await context.service.loadFailedRetryNoticeScopes({ companyId: scope.companyId })).toEqual([scope]);
+      let locked!: () => void;
+      const ready = new Promise<void>(resolve => { locked = resolve; });
+      const hold = new Promise<void>(resolve => { release = resolve; });
+      holder = db.transaction(async tx => {
+        await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, staged[0].actionId)).for("update");
+        locked(); await hold;
+      });
+      await ready;
+      expect(await context.service.reconcileFailedRetryNotices({ scope }))
+        .toEqual({ inserted: 0, contended: true, nextCursor: null });
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]);
+      release(); await holder;
+      expect(await context.service.reconcileFailedRetryNotices({ scope }))
+        .toEqual({ inserted: 1, contended: false, nextCursor: null });
+      expect(await context.service.loadFailedRetryNoticeScopes(scope)).toEqual([]);
+      expect(await readChatPublicationsByEndpointId(other.endpoint.id)).toEqual([]);
+      await db.update(agentWakeupRequests).set({ status: "cancelled" })
+        .where(eq(agentWakeupRequests.id, staged[0].actionId));
+      expect(await context.service.reconcileFailedRetryNotices({ scope }))
+        .toEqual({ inserted: 1, contended: false, nextCursor: null });
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({
+        payload: { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" },
+      })]));
+      expect(await context.service.reconcileFailedRetryNotices({ scope }))
+        .toEqual({ inserted: 0, contended: false, nextCursor: null });
+      expect(await context.service.loadFailedRetryNoticeScopes(scope)).toEqual([]);
+      await db.update(agentWakeupRequests).set({ runId: other.runId })
+        .where(eq(agentWakeupRequests.id, staged[1].actionId));
+      expect(await other.service.loadFailedRetryNoticeScopes({ companyId: other.fixture.companyId })).toEqual([]);
+    } finally { release(); await holder; await context.service.shutdown(); await other.service.shutdown(); }
+  });
+
   it.each(["slack", "telegram"] as const)(
     "keeps a %s retry's queue, working, and selected final on its own provider identity",
     async (provider) => {

server/src/services/chat-channels.ts

@@ -1241,6 +1241,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
     processFailedChatRunRetry,
     processFailedChatRunRetries,
     enqueueFailedChatRetryPublications,
+    loadFailedRetryNoticeScopes,
+    reconcileFailedRetryNotices,
     pendingFailedRetries,
   } = createChatFailedRunRetry(db, {
     authorizeInboundWakeup,
@@ -3507,6 +3509,8 @@ 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 {
+    loadFailedRetryNoticeScopes,
+    reconcileFailedRetryNotices,
     registerTeamsAttemptRecoveryWork,
     registerTeamsConsentExpiryWork,
     registerCommandWork,

server/src/services/chat-channels/failed-retry-notices.ts

@@ -0,0 +1,111 @@
+import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
+import { agentWakeupRequests, chatActions, chatPublications, type Db } from "@paperclipai/db";
+import type { DbOrTransaction } from "./runtime-types.js";
+import { notifyChatPublicationWork } from "../chat-work-notifications.js";
+import { projectSafeChatPublication } from "../chat-publication-projection.js";
+import { inboundWakePublicationKey } from "../chat-inbound-wakeup-publications.js";
+import { NativeChatReviewPresentationContentionError } from "../native-runtime/native-chat-review-presentation.js";
+import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js";
+
+export type FailedRetryNoticeScope = { companyId: string; issueId?: string };
+export type FailedRetryNoticeCursor = { createdAt: string; id: string };
+type Options = {
+  authorize: (tx: DbOrTransaction, action: typeof chatActions.$inferSelect) => Promise<{
+    endpointId: string; conversationId: string; issueId: string; commentIds: string[];
+  }>;
+  isExternalActionAuthorizationChange: (error: unknown) => boolean;
+  isShuttingDown: () => boolean;
+};
+
+/** Retry authorization belongs to the retry service. This projection owns only
+ * the queued/not-started notice, selected from its existing durable receipt. */
+export function createFailedRetryNotices(db: Db, options: Options) {
+  let manualCursor: FailedRetryNoticeCursor | null = null;
+  async function candidates(limit: number, cursor: FailedRetryNoticeCursor | null, scope?: FailedRetryNoticeScope) {
+    return db.select({ action: chatActions, receipt: agentWakeupRequests,
+      cursorCreatedAt: sql<string>`${chatActions.createdAt}::text`,
+    }).from(chatActions).innerJoin(agentWakeupRequests, and(
+      eq(agentWakeupRequests.id, chatActions.id), eq(agentWakeupRequests.companyId, chatActions.companyId),
+    )).where(and(
+      eq(chatActions.kind, "failed_run_retry"), isNull(agentWakeupRequests.runId),
+      scope ? eq(chatActions.companyId, scope.companyId) : undefined,
+      scope?.issueId ? sql`${chatActions.payload}->>'issueId' = ${scope.issueId}` : undefined,
+      sql`${chatActions.payload}->>'issueId' ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'`,
+      cursor ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt}::timestamptz, ${cursor.id}::uuid)` : undefined,
+      inArray(agentWakeupRequests.status, ["deferred_issue_execution", "cancelled", "failed", "skipped"]),
+      sql`not exists (select 1 from chat_publications retry_notice where retry_notice.company_id = ${chatActions.companyId}
+        and retry_notice.idempotency_key = 'wake:' || ${chatActions.id}::text || ':' ||
+        case when ${agentWakeupRequests.status} = 'deferred_issue_execution' then 'queued' else 'not_started' end || ':' ||
+        ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`,
+    )).orderBy(asc(chatActions.createdAt), asc(chatActions.id)).limit(limit);
+  }
+  type Candidates = Awaited<ReturnType<typeof candidates>>;
+  function continuation(rows: Candidates, limit: number): FailedRetryNoticeCursor | null {
+    const last = rows.at(-1);
+    return rows.length === limit && last ? { createdAt: last.cursorCreatedAt, id: last.action.id } : null;
+  }
+  async function project(rows: Candidates, signal?: AbortSignal) {
+    let inserted = 0, contended = false;
+    for (const { action, receipt } of rows) {
+      signal?.throwIfAborted();
+      if (options.isShuttingDown()) throw new Error("Failed retry notice reconciliation stopped");
+      try {
+        inserted += await db.transaction(async tx => {
+          const source = await options.authorize(tx, action);
+          const [current] = await tx.select().from(agentWakeupRequests)
+            .where(eq(agentWakeupRequests.id, receipt.id)).for("update", { noWait: true });
+          if (!current || current.runId || current.status !== receipt.status) return 0;
+          const state = current.status === "deferred_issue_execution" ? "queued" : "not_started";
+          await notifyChatPublicationWork(tx);
+          return (await tx.insert(chatPublications).values({
+            companyId: action.companyId, endpointId: source.endpointId, conversationId: source.conversationId,
+            issueId: source.issueId, commentId: source.commentIds.at(-1)!,
+            idempotencyKey: inboundWakePublicationKey(action.id, state, source.endpointId, source.conversationId),
+            payload: projectSafeChatPublication({ classification: "external", source: "safe_milestone",
+              text: state === "queued" ? "Your retry is queued."
+                : "This retry was not started. Open the task in Paperclip for details.",
+              progressState: state === "queued" ? "queued" : "failed",
+            }), state: "pending",
+          }).onConflictDoNothing().returning({ id: chatPublications.id })).length;
+        });
+      } catch (error) {
+        if (isExternalChatWaitAuthorizationContention(error)
+          || error instanceof NativeChatReviewPresentationContentionError) contended = true;
+        else if (options.isExternalActionAuthorizationChange(error)
+          || (error && typeof error === "object" && "status" in error && error.status === 409)) continue;
+        else throw error;
+      }
+    }
+    return { inserted, contended };
+  }
+  async function enqueueFailedChatRetryPublications(limit: number) {
+    if (options.isShuttingDown()) return 0;
+    const size = Math.max(1, Math.min(50, limit));
+    const rows = await candidates(size, manualCursor);
+    manualCursor = continuation(rows, size);
+    return (await project(rows)).inserted;
+  }
+  async function loadFailedRetryNoticeScopes(scope?: FailedRetryNoticeScope, signal?: AbortSignal) {
+    const scopes = new Map<string, { companyId: string; issueId: string }>();
+    let cursor: FailedRetryNoticeCursor | null = null;
+    do {
+      signal?.throwIfAborted();
+      const rows = await candidates(200, cursor, scope);
+      for (const { action } of rows) {
+        const issueId = String(action.payload.issueId);
+        scopes.set(`${action.companyId}:${issueId}`, { companyId: action.companyId, issueId });
+      }
+      cursor = continuation(rows, 200);
+    } while (cursor);
+    return [...scopes.values()];
+  }
+  async function reconcileFailedRetryNotices(input: {
+    scope: FailedRetryNoticeScope; limit?: number; cursor?: FailedRetryNoticeCursor | null; signal?: AbortSignal;
+  }) {
+    input.signal?.throwIfAborted();
+    const limit = Math.max(1, Math.min(input.limit ?? 100, 200));
+    const rows = await candidates(limit, input.cursor ?? null, input.scope);
+    return { ...await project(rows, input.signal), nextCursor: continuation(rows, limit) };
+  }
+  return { enqueueFailedChatRetryPublications, loadFailedRetryNoticeScopes, reconcileFailedRetryNotices };
+}

server/src/services/chat-channels/failed-run-retry.ts

@@ -1,6 +1,7 @@
+import { createFailedRetryNotices } from "./failed-retry-notices.js";
 import { continuingTeamsFilePublication, type ContinuingTeamsFileTransfer } from "./microsoft-teams/file-transfer-source.js";
 import { readTaskChatRetryIntent, taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./task-retry-authority.js";
-import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-notifications.js";
+import { notifyChatActionWork } from "../chat-work-notifications.js";
 import { FAILED_RUN_RETRY_STALE_MS, dueChatAction } from "../chat-action-work.js";
 
 import {
@@ -78,13 +79,8 @@ import {
 import {
   NativeChatReviewPresentationContentionError,
 } from "../native-runtime/native-chat-review-presentation.js";
-import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js";
 import { CHAT_CONVERSATION_ID_KEY } from "../../modules/run-dispatch/index.js";
-import { projectSafeChatPublication } from "../chat-publication-projection.js";
 
-import {
-  inboundWakePublicationKey,
-} from "../chat-inbound-wakeup-publications.js";
 
 type DbOrTransaction = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
 type DeliveryRow = typeof chatDeliveries.$inferSelect;
@@ -1746,117 +1742,13 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
     return candidates.length;
   }
 
-  let failedRetryNoticeCursor: { createdAt: Date; id: string } | null = null;
-  async function enqueueFailedChatRetryPublications(
-    limit: number,
-  ): Promise<number> {
-    const cursor = failedRetryNoticeCursor;
-    const batchSize = Math.max(1, Math.min(50, limit));
-    const candidates = await db
-      .select({ action: chatActions, receipt: agentWakeupRequests })
-      .from(chatActions)
-      .innerJoin(
-        agentWakeupRequests,
-        and(
-          eq(agentWakeupRequests.id, chatActions.id),
-          eq(agentWakeupRequests.companyId, chatActions.companyId),
-        ),
-      )
-      .where(
-        and(
-          eq(chatActions.kind, "failed_run_retry"),
-          isNull(agentWakeupRequests.runId),
-          cursor
-            ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt.toISOString()}::timestamptz, ${cursor.id}::uuid)`
-            : undefined,
-          inArray(agentWakeupRequests.status, [
-            "deferred_issue_execution",
-            "cancelled",
-            "failed",
-            "skipped",
-          ]),
-          sql`not exists (select 1 from chat_publications retry_notice where retry_notice.company_id = ${chatActions.companyId}
-          and retry_notice.idempotency_key = 'wake:' || ${chatActions.id}::text || ':' ||
-          case when ${agentWakeupRequests.status} = 'deferred_issue_execution' then 'queued' else 'not_started' end || ':' ||
-          ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`,
-        ),
-      )
-      .orderBy(asc(chatActions.createdAt), asc(chatActions.id))
-      .limit(batchSize);
-    const last = candidates.at(-1)?.action;
-    failedRetryNoticeCursor =
-      candidates.length >= batchSize && last
-        ? { createdAt: last.createdAt, id: last.id }
-        : null;
-    let inserted = 0;
-    for (const { action, receipt } of candidates) {
-      try {
-        inserted += await db.transaction(async (tx) => {
-          const source = await authorizeFailedChatRunRetry(tx, {
-            companyId: action.companyId,
-            issueId: String(action.payload.issueId),
-            agentId: String(action.payload.agentId),
-            retryActionId: action.id,
-            publication: true,
-          });
-          const [current] = await tx
-            .select()
-            .from(agentWakeupRequests)
-            .where(eq(agentWakeupRequests.id, receipt.id))
-            .for("update", { noWait: true });
-          if (!current || current.runId || current.status !== receipt.status)
-            return 0;
-          const state =
-            current.status === "deferred_issue_execution"
-              ? "queued"
-              : "not_started";
-          await notifyChatPublicationWork(tx);
-          return (
-            await tx
-              .insert(chatPublications)
-              .values({
-                companyId: action.companyId,
-                endpointId: source.endpointId,
-                conversationId: source.conversationId,
-                issueId: source.issueId,
-                commentId: source.commentIds.at(-1)!,
-                idempotencyKey: inboundWakePublicationKey(
-                  action.id,
-                  state,
-                  source.endpointId,
-                  source.conversationId,
-                ),
-                payload: projectSafeChatPublication({
-                  classification: "external",
-                  source: "safe_milestone",
-                  text:
-                    state === "queued"
-                      ? "Your retry is queued."
-                      : "This retry was not started. Open the task in Paperclip for details.",
-                  progressState: state === "queued" ? "queued" : "failed",
-                }),
-                state: "pending",
-              })
-              .onConflictDoNothing()
-              .returning({ id: chatPublications.id })
-          ).length;
-        });
-      } catch (error) {
-        if (
-          isExternalChatWaitAuthorizationContention(error) ||
-          error instanceof NativeChatReviewPresentationContentionError ||
-          isExternalActionAuthorizationChange(error) ||
-          (error &&
-            typeof error === "object" &&
-            "status" in error &&
-            error.status === 409)
-        )
-          continue;
-        throw error;
-      }
-    }
-    return inserted;
-  }
+  const retryNotices = createFailedRetryNotices(db, {
+    authorize: (tx, action) => authorizeFailedChatRunRetry(tx, {
+      companyId: action.companyId, issueId: String(action.payload.issueId),
+      agentId: String(action.payload.agentId), retryActionId: action.id, publication: true,
+    }),
+    isExternalActionAuthorizationChange, isShuttingDown,
+  });
 
   return {
     retryCommentIds,
@@ -1869,7 +1761,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
     unregisterFailedRetryAuthority,
     processFailedChatRunRetry,
     processFailedChatRunRetries,
-    enqueueFailedChatRetryPublications,
+    ...retryNotices,
     pendingFailedRetries: () => [...failedRetryTasks.values()],
   };
 }