diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 015a2bde75a..37fe3dff804 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/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. diff --git a/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md b/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md new file mode 100644 index 00000000000..6be4dc53085 --- /dev/null +++ b/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. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index cc3801b12c1..9e747646f0e 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/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 | 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(resolve => { locked = resolve; }); + const hold = new Promise(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) => { diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 1b808df6164..8ed25ed01bd 100644 --- a/server/src/services/chat-channels.ts +++ b/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, diff --git a/server/src/services/chat-channels/failed-retry-notices.ts b/server/src/services/chat-channels/failed-retry-notices.ts new file mode 100644 index 00000000000..493c3e016e3 --- /dev/null +++ b/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`${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>; + 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(); + 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 }; +} diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index 2b5d2c9b026..aa434b1893e 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/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[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 { - 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()], }; }