From 5a72061ccc5ce6acf5524bb5c3b1cb8d10de5147 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 12 Oct 2026 09:49:34 -0500 Subject: [PATCH] refactor(chat): extract publication replacement decisions (#15844) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip publishes agent progress and results back to chat. > - Some updates replace an earlier message while others need a new message. > - The current outbound link and task generation determine that authority. > - These decisions are spread through the large chat service. > - This PR groups publication replacement and supersession decisions in one module. > - Reviewers can follow those decisions without reading provider transport code. ## Linked Issues or Issue Description Refs #15836, #15834 and #15823. Stack slice 4 of 5, based on `codex/chat-identity-links`. Merge the predecessor first, then retarget this PR. Do not merge the stack out of order. Stack order: #15823 (provider messages) → #15834 (resource reconciliation) → #15836 (identity links) → #15844 (publication replacement) → #15845 (failed-run retries). GitHub native stack merging preserves this order and automatically restacks the remaining branches. ## What Changed - Extract shared publication replacement and supersession queries from the large service. - Keep exact outbound ownership, run/wake scope, consumed progress, interaction replacement and retained-source authorization together. - Put the iMessage typing-versus-progress policy in `imessage-photon/publications.ts`. - Preserve endpoint lookup timing and all ownership query bodies. The multi-provider close allowlist remains shared dispatch metadata. - Restack onto the revised #15836 and record why this boundary remains mostly shared. ## Verification - The same 49 existing publication and Photon integration cases passed before and after the provider-policy move. Other cases were excluded by the filter. - Cases cover close-owned progress, ambiguous delivery, interleaved Slack control, Telegram confirmation, FIFO replacement and Photon channel behavior. - Server typecheck, module-boundary and whitespace checks passed. Only the iMessage predicate/reason moved; ownership queries are unchanged. - Historical server shard 2 hit a cleanup deadlock; its unchanged [isolated rerun passed](https://github.com/paperclipai/paperclip/actions/runs/38134431077/job/114457999409). [Original failure](https://github.com/paperclipai/paperclip/actions/runs/38134431077/job/114455332022). - Tests use disposable databases and fixture provider transports, not live third-party execution. - Fresh Greptile review is 5/5 on `8e2da97b6dc11eb446128b61c789edb529c50d5a`, with a successful check and no open review threads. - Final-stack `pnpm -r typecheck` and `pnpm build` passed at `57ec5ba94aecd860471f9b14db46f99120c7c790`. - Historical pre-restack CI failed on `ci / General tests (server (11/12))` and its aggregate `ci / verify`. All 1,255 tests passed, but worker-exit cleanup in unchanged `server/src/__tests__/setup-supertest.ts:32` failed with EACCES removing a temporary `paperclip-vitest-home-*` directory. The [original job](https://github.com/paperclipai/paperclip/actions/runs/38256549828/job/114836065382) and the [single unchanged rerun](https://github.com/paperclipai/paperclip/actions/runs/38256549828/job/114839735273) failed the same way. No further reruns were requested. After GitHub automatically restacked the branches, the full cumulative run on #15845 passed this server shard. This middle PR uses the repository’s policy-only stack CI scope; the top PR runs all suites. - The final full-local-suite outcome for this cumulative stack is recorded in #15845; do not infer that gate from the focused tests above. - Merge verification on the restacked tree: 118 focused identity, publication, Photon and retry integration tests passed; server typecheck and module-boundary checks passed. GitHub’s rebased trees exactly match the previously reviewed file contents. - All current-head required checks passed on `8e2da97b6dc11eb446128b61c789edb529c50d5a`: policy CI (middle-stack scope; full cumulative suites passed on #15845); Greptile 5/5 with no unresolved threads. ## Risks - Splitting shared ownership queries per channel would duplicate authorization rules. They remain in the shared module. - Preserve the issue-to-run lock order and retained-source callback. The factory creates no worker or mutable state. - This PR depends on #15836. Merge and retarget in stack order. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and repository tools. The exact runtime model ID and context-window size are not exposed in this session. No separate agent was used. ## 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 - [ ] 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 - [x] All Paperclip CI gates are green - [x] 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 --- .../2026-10-10-chat-channels-refactor.md | 30 + server/src/services/chat-channels.ts | 906 +--------------- .../imessage-photon/publications.ts | 9 + .../chat-channels/publication-replacement.ts | 965 ++++++++++++++++++ 4 files changed, 1006 insertions(+), 904 deletions(-) create mode 100644 server/src/services/chat-channels/imessage-photon/publications.ts create mode 100644 server/src/services/chat-channels/publication-replacement.ts diff --git a/doc/plans/2026-10-10-chat-channels-refactor.md b/doc/plans/2026-10-10-chat-channels-refactor.md index ce51837f239..bba089b0c2a 100644 --- a/doc/plans/2026-10-10-chat-channels-refactor.md +++ b/doc/plans/2026-10-10-chat-channels-refactor.md @@ -203,3 +203,33 @@ Slack connect journey: discovery, preview, outsider access request, duplicate request, member confirmation, provider notice, consumed-token rejection and endpoint isolation. Receipt queries, access-request transaction and notice payload retain their statements with explicit parameter substitutions. + +### Slice 4: publication replacement and supersession + +Move 13 functions into `publication-replacement.ts`. The module selects the exact +provider message a publication can replace and suppresses obsolete progress. It +uses the database, endpoint lookup and retained-source authorization callback. +Transport execution, publication claims and receipt reactions remain coordinated +by the service. No state or registration moves in this slice. + +- Source: 37,295 lines / 1,436,867 bytes → 36,393 lines / 1,400,636 bytes. +- Destination: 963 lines / 38,724 bytes. +- Baseline: all 51 checks and 5/5 review passed on slice 3 (`581dd656c1`, #15836). +- All 13 bodies and every remaining statement match the predecessor exactly. + Server typecheck, module boundaries and whitespace checks pass. +- Existing CI fixtures cover exact outbound ownership, interleaved control + messages, consumed progress, interactions, committed failure replacement and + issue-to-run lock order. Local database capacity remains blocked. + +### 2026-10-12 channel-folder reassessment: slice 4 + +Publication replacement is shared ownership logic: exact outbound links, run and +wake scope, consumed progress messages, interaction prompts and retained-source +authorization. Keep those queries together. The close-message provider allowlist +is shared dispatch metadata, not four separate implementations. + +Move the iMessage queued/working suppression policy and explanation to +`imessage-photon/publications.ts`. Preserve when endpoint lookup occurs and leave +all ownership queries unchanged. Baseline: 49 public-service publication and +Photon cases passed, including close-owned progress, interleaved Slack control, +Telegram confirmation and FIFO replacement. The same 49 cases passed after. Server typecheck and module boundaries pass. diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 5930982dd0b..2e88f05b99c 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,4 @@ +import { createChatPublicationReplacement } from "./chat-channels/publication-replacement.js"; import { createChatIdentityLinks } from "./chat-channels/identity-links.js"; import { createChatResourceReconciliation } from "./chat-channels/resource-reconciliation.js"; import { githubRepositoryInventoryItemFromPayload } from "./chat-channels/github/resources.js"; @@ -2427,6 +2428,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }), }); const issuesSvc = issueService(db); + const { runIdFromMilestonePublication, runOwnershipMilestoneSupersessionReason, runPublicationToReplace, receiptReactionCompletionRunId, inboundWakePublicationToReplace, interactionResolutionPublicationToReplace, interactionPromptPublicationToReplace, progressSupersededByFastResponse, runProgressSupersededByPublishedInteraction, taskStatusPublicationToReplace, closeProgressToReplace } = createChatPublicationReplacement(db, { endpointRecord, authorizeRetainedChatSourcePublication }); const secrets = secretService(db); const questionResponses = questionResponseDeliveryService(db, { heartbeat: @@ -32677,541 +32679,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); } - function runIdFromMilestonePublication( - publication: typeof chatPublications.$inferSelect, - ): string | null { - if (!publication.payload.progressState) return null; - const match = - /^run:([^:]+):(?:queued|working|waiting_for_input|completed|failed):/.exec( - publication.idempotencyKey, - ); - const runId = match?.[1] ?? null; - // This identifier is also compared against a UUID column below. Old or - // manually repaired rows must not be allowed to turn the global - // publication sweep into a PostgreSQL cast error. - return isUuidLike(runId) ? runId : null; - } - - async function runOwnershipMilestoneSupersessionReason( - publication: typeof chatPublications.$inferSelect, - ): Promise { - const progress = publication.payload.progressState; - if (progress && ["queued", "working"].includes(progress)) { - const endpoint = await endpointRecord(publication.endpointId); - if (endpoint?.endpoint.provider === "imessage-photon") return "iMessage uses typing instead of progress bubbles"; - } - if ( - !progress || - !["queued", "working", "waiting_for_input"].includes(progress) - ) { - return null; - } - const runId = runIdFromMilestonePublication(publication); - if (!runId) return null; - const run = await db.transaction(async (tx) => { - // Use native finalization's issue -> run lock order. If terminalization - // won, the current status suppresses this obsolete provider update. If - // this short authorization snapshot wins, the progress was still true - // at its durable send boundary and the later final remains authoritative. - const currentIssue = await tx - .select({ id: issues.id }) - .from(issues) - .where( - and( - eq(issues.id, publication.issueId), - eq(issues.companyId, publication.companyId), - ), - ) - .for("update") - .then((rows) => rows[0] ?? null); - if (!currentIssue) return null; - return tx - .select({ - status: heartbeatRuns.status, - errorCode: heartbeatRuns.errorCode, - }) - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.id, runId), - eq(heartbeatRuns.companyId, publication.companyId), - sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${publication.issueId}`, - ), - ) - .for("update") - .then((rows) => rows[0] ?? null); - }); - const baseProgressKey = `run:${runId}:${progress}:${publication.endpointId}`; - const nativeProgressPrefix = `run:${runId}:working:${publication.endpointId}:native:`; - const isTerminalizableProgress = - (progress === "queued" || progress === "working") && - (publication.idempotencyKey === baseProgressKey || - publication.idempotencyKey.startsWith(nativeProgressPrefix)); - if ( - isTerminalizableProgress && - run && - ["succeeded", "interrupted", "failed", "cancelled", "timed_out"].includes( - run.status, - ) - ) { - return "Run reached a terminal state before progress delivery"; - } - // Match only the closed ownership-attention projection. Historical progress - // without an extant run retains its existing delivery semantics. - const ownershipBlocked = - run?.status === "running" && - [ - "native_execution_ownership_unverified", - "native_adopted_runner_authentication_timeout", - ].includes(run.errorCode ?? ""); - if (progress === "waiting_for_input") { - return ownershipBlocked - ? null - : "Run ownership attention no longer applies to the bound task"; - } - return ownershipBlocked - ? "Run progress was superseded by ownership recovery attention" - : null; - } - - async function providerProgressLaneConsumed( - publication: typeof chatPublications.$inferSelect, - providerMessageId: string, - ): Promise { - // Published progress rows retain their old provider ID after an edit. - // Consult that exact message's current outbound link, not the conversation - // tail: an authored answer or failure has consumed the lane, while an - // interleaved task-status update may still be replaced normally. - const [consumed] = await db - .select({ id: chatPublications.id }) - .from(chatMessageLinks) - .innerJoin( - chatPublications, - and( - eq(chatPublications.id, chatMessageLinks.publicationId), - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - eq(chatPublications.providerMessageId, providerMessageId), - eq(chatPublications.state, "published"), - ), - ) - .where( - and( - eq(chatMessageLinks.companyId, publication.companyId), - eq(chatMessageLinks.endpointId, publication.endpointId), - eq(chatMessageLinks.conversationId, publication.conversationId), - eq(chatMessageLinks.direction, "outbound"), - eq(chatMessageLinks.providerMessageId, providerMessageId), - or( - and( - isNotNull(chatPublications.commentId), - sql`${chatPublications.payload}->>'progressState' is null`, - ), - sql`${chatPublications.payload}->>'progressState' = 'failed'`, - ), - ), - ) - .limit(1); - return Boolean(consumed); - } - - async function committedResponseMayReplaceFailure( - publication: typeof chatPublications.$inferSelect, - runId: string, - providerMessageId: string, - ): Promise { - if ( - !publication.commentId || - publication.payload.progressState !== undefined - ) - return false; - return db - .transaction(async (tx) => { - const [run] = await tx - .select() - .from(heartbeatRuns) - .where( - and( - eq(heartbeatRuns.id, runId), - eq(heartbeatRuns.companyId, publication.companyId), - eq(heartbeatRuns.nativeIssueId, publication.issueId), - eq(heartbeatRuns.runtimeMode, "native"), - eq(heartbeatRuns.status, "succeeded"), - isNotNull(heartbeatRuns.finishedAt), - ), - ) - .for("share", { noWait: true }); - const presentation = run?.resultJson?.presentationDecision as - Record | undefined; - if ( - !run?.resultJson?.nativeCommittedChatResponse || - presentation?.commentId !== publication.commentId - ) - return false; - const [comment] = await tx - .select({ id: issueComments.id }) - .from(issueComments) - .where( - and( - eq(issueComments.id, publication.commentId!), - eq(issueComments.companyId, publication.companyId), - eq(issueComments.issueId, publication.issueId), - eq(issueComments.createdByRunId, runId), - eq(issueComments.authorAgentId, run.agentId), - isNull(issueComments.deletedAt), - ), - ) - .for("share", { noWait: true }); - if (!comment) return false; - const [priorAnswer] = await tx - .select({ id: chatPublications.id }) - .from(chatPublications) - .innerJoin( - issueComments, - and( - eq(issueComments.id, chatPublications.commentId), - eq(issueComments.companyId, publication.companyId), - eq(issueComments.createdByRunId, runId), - ), - ) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - ne(chatPublications.id, publication.id), - inArray(chatPublications.state, [ - "published", - "streaming", - "delivery_unknown", - ]), - sql`${chatPublications.payload}->>'progressState' is null`, - ), - ) - .limit(1); - if (priorAnswer) return false; - // A historical progress row can retain this ID after a later edit. - // Only its CURRENT outbound link may grant this narrow exception; an - // authored answer or another run's failure must remain consumed. - const [failure] = await tx - .select({ id: chatPublications.id }) - .from(chatMessageLinks) - .innerJoin( - chatPublications, - and( - eq(chatPublications.id, chatMessageLinks.publicationId), - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - eq(chatPublications.providerMessageId, providerMessageId), - eq( - chatPublications.idempotencyKey, - `run:${runId}:failed:${publication.endpointId}`, - ), - eq(chatPublications.state, "published"), - isNull(chatPublications.commentId), - sql`${chatPublications.payload}->>'progressState' = 'failed'`, - ), - ) - .where( - and( - eq(chatMessageLinks.companyId, publication.companyId), - eq(chatMessageLinks.endpointId, publication.endpointId), - eq(chatMessageLinks.conversationId, publication.conversationId), - eq(chatMessageLinks.providerMessageId, providerMessageId), - eq(chatMessageLinks.direction, "outbound"), - ), - ); - if (!failure) return false; - // The marker is read from the authoritative run, never the publication - // payload. Recheck accepted result/decision plus the complete current - // source batch, actor, access and epoch before borrowing its failure lane. - // The transport claim repeats those checks at the I/O boundary. - return authorizeRetainedChatSourcePublication(tx, publication); - }) - .catch((error: unknown) => { - if (isExternalChatWaitAuthorizationContention(error)) - throw new NativeChatReviewPresentationContentionError(); - throw error; - }); - } - - async function runPublicationToReplace( - publication: typeof chatPublications.$inferSelect, - payload: SafeChatPublicationPayload, - ): Promise { - if (payload.attachmentIds?.length) return null; - const currentRunId = - runIdFromMilestonePublication(publication) ?? - (publication.commentId - ? await db - .select({ runId: issueComments.createdByRunId }) - .from(issueComments) - .where(eq(issueComments.id, publication.commentId)) - .then((rows) => rows[0]?.runId ?? null) - : null); - if (!currentRunId) return null; - return db - .select({ - id: chatPublications.id, - commentId: chatPublications.commentId, - providerMessageId: chatPublications.providerMessageId, - payload: chatPublications.payload, - }) - .from(chatPublications) - .leftJoin(issueComments, eq(issueComments.id, chatPublications.commentId)) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.state, "published"), - isNotNull(chatPublications.providerMessageId), - or( - like(chatPublications.idempotencyKey, `run:${currentRunId}:%`), - eq(issueComments.createdByRunId, currentRunId), - ), - ), - ) - .orderBy(desc(chatPublications.createdAt), desc(chatPublications.id)) - .then(async (rows) => { - // Progress updates are one replaceable provider-message lane per run. - // The first durable agent comment may turn that placeholder into the - // terminal response, but later comments from the same run are distinct - // user-visible outputs and must be posted separately. Re-editing the - // placeholder for each comment silently erases the earlier replies. - if ( - publication.commentId && - rows.some( - (row) => - row.commentId !== null && row.payload.progressState === undefined, - ) - ) - return null; - const replacement = rows.find( - (row) => - Boolean(row.providerMessageId) && - row.payload.progressState !== undefined, - ); - if (!replacement?.providerMessageId) return null; - if ( - (await providerProgressLaneConsumed( - publication, - replacement.providerMessageId, - )) && - !(await committedResponseMayReplaceFailure( - publication, - currentRunId, - replacement.providerMessageId, - )) - ) - return null; - // Replacement identity belongs to the run, not to the provider-visible - // tail. A status/control reply may legitimately interleave while the run - // is active; making the tail the edit candidate would strand this run's - // working placeholder forever. The query is bounded by endpoint, - // conversation (the task generation), and run id, so an interleaved - // control or another run can never donate its provider message here. - return replacement.providerMessageId; - }); - } - - async function receiptReactionCompletionRunId( - tx: Db, - publication: typeof chatPublications.$inferSelect, - payload: SafeChatPublicationPayload, - ): Promise { - if ( - payload.progressState === "completed" || - payload.progressState === "failed" - ) { - return runIdFromMilestonePublication(publication); - } - if (publication.commentId) { - const runId = await tx - .select({ runId: issueComments.createdByRunId }) - .from(issueComments) - .where( - and( - eq(issueComments.id, publication.commentId), - eq(issueComments.companyId, publication.companyId), - ), - ) - .then((rows) => rows[0]?.runId ?? null); - return isUuidLike(runId) ? runId : null; - } - if ( - payload.interactionId && - publication.idempotencyKey === - `interaction:${payload.interactionId}:${publication.endpointId}` - ) { - const runId = await tx - .select({ runId: issueThreadInteractions.sourceRunId }) - .from(issueThreadInteractions) - .where( - and( - eq(issueThreadInteractions.id, payload.interactionId), - eq(issueThreadInteractions.companyId, publication.companyId), - eq(issueThreadInteractions.issueId, publication.issueId), - ), - ) - .then((rows) => rows[0]?.runId ?? null); - return isUuidLike(runId) ? runId : null; - } - return null; - } - - async function inboundWakePublicationToReplace( - publication: typeof chatPublications.$inferSelect, - payload: SafeChatPublicationPayload, - ): Promise { - if (payload.attachmentIds?.length) return null; - const notice = parseInboundWakePublicationKey(publication.idempotencyKey); - let wakeId: string | null = - notice && notice.state !== "queued" ? notice.wakeId : null; - let runId = runIdFromMilestonePublication(publication); - if ( - !notice && - !runId && - payload.interactionId && - publication.idempotencyKey === - `interaction:${payload.interactionId}:${publication.endpointId}` - ) { - const [interaction] = await db - .select({ runId: issueThreadInteractions.sourceRunId }) - .from(issueThreadInteractions) - .where( - and( - eq(issueThreadInteractions.id, payload.interactionId), - eq(issueThreadInteractions.companyId, publication.companyId), - eq(issueThreadInteractions.issueId, publication.issueId), - inArray(issueThreadInteractions.kind, [ - "ask_user_questions", - "request_confirmation", - ]), - ), - ); - runId = interaction?.runId ?? null; - } - if (!notice && !runId && publication.commentId) { - const [comment] = await db - .select({ runId: issueComments.createdByRunId }) - .from(issueComments) - .where( - and( - eq(issueComments.id, publication.commentId), - eq(issueComments.companyId, publication.companyId), - eq(issueComments.issueId, publication.issueId), - ), - ); - runId = comment?.runId ?? null; - } - let runContext: Record | null = null; - if (runId) { - if (publication.commentId) { - const [priorAnswer] = await db - .select({ id: chatPublications.id }) - .from(chatPublications) - .innerJoin( - issueComments, - eq(issueComments.id, chatPublications.commentId), - ) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.state, "published"), - eq(issueComments.createdByRunId, runId), - sql`${chatPublications.payload}->>'progressState' is null`, - ), - ) - .limit(1); - if (priorAnswer) return null; - } - const [run] = await db - .select({ - wakeId: heartbeatRuns.wakeupRequestId, - context: heartbeatRuns.contextSnapshot, - }) - .from(heartbeatRuns) - .innerJoin( - agentWakeupRequests, - and( - eq(agentWakeupRequests.id, heartbeatRuns.wakeupRequestId), - eq(agentWakeupRequests.runId, heartbeatRuns.id), - eq(agentWakeupRequests.companyId, heartbeatRuns.companyId), - eq(agentWakeupRequests.agentId, heartbeatRuns.agentId), - // Failed/cancelled is the admitted run's outcome, not loss of - // its original admission. Its terminal milestone still owns - // this exact lane even if no working update was sent first. - ne(agentWakeupRequests.status, "skipped"), - ), - ) - .where( - and( - eq(heartbeatRuns.id, runId), - eq(heartbeatRuns.companyId, publication.companyId), - sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${publication.issueId}`, - ), - ); - wakeId = run?.wakeId ?? null; - runContext = run?.context ?? null; - } - if (!wakeId) return null; - const [queued] = await db - .select({ - commentId: chatPublications.commentId, - providerMessageId: chatPublications.providerMessageId, - }) - .from(chatPublications) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - eq(chatPublications.state, "published"), - isNotNull(chatPublications.providerMessageId), - eq( - chatPublications.idempotencyKey, - inboundWakePublicationKey( - wakeId, - "queued", - publication.endpointId, - publication.conversationId, - ), - ), - ), - ); - if (!queued?.commentId || !queued.providerMessageId) return null; - // The run helper may have rejected this same provider ID because its - // queued/working placeholder is already an answer or terminal failure. - // Falling back to the older wake row must not erase that newer output. - if ( - await providerProgressLaneConsumed(publication, queued.providerMessageId) - ) - return null; - if (notice && notice.state !== "queued") - return queued.commentId === publication.commentId - ? queued.providerMessageId - : null; - // A deferred owner can be a Board wake into which the exact external - // comment coalesced. The run must actually contain that source comment; - // wake ID alone cannot authorize a different successor's edit lane. - return runContext && - (runContext.wakeCommentId === queued.commentId || - runContext.commentId === queued.commentId || - (Array.isArray(runContext.wakeCommentIds) && - runContext.wakeCommentIds.includes(queued.commentId))) - ? queued.providerMessageId - : null; - } - async function stageTerminalReceiptReactionRemovals( tx: Db, input: { @@ -33236,375 +32703,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); } - async function interactionResolutionPublicationToReplace( - publication: typeof chatPublications.$inferSelect, - payload: SafeChatPublicationPayload, - ): Promise { - if ( - !publication.idempotencyKey.startsWith("interaction-resolution:") || - !payload.interactionId - ) - return null; - return db - .select({ providerMessageId: chatPublications.providerMessageId }) - .from(chatPublications) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - eq( - chatPublications.idempotencyKey, - `interaction:${payload.interactionId}:${publication.endpointId}`, - ), - eq(chatPublications.state, "published"), - isNotNull(chatPublications.providerMessageId), - ), - ) - .then((rows) => rows[0]?.providerMessageId ?? null); - } - - async function interactionPromptPublicationToReplace( - publication: typeof chatPublications.$inferSelect, - payload: SafeChatPublicationPayload, - ): Promise { - if ( - !payload.interactionId || - publication.idempotencyKey !== - `interaction:${payload.interactionId}:${publication.endpointId}` - ) { - return null; - } - const interaction = await db - .select({ - kind: issueThreadInteractions.kind, - sourceRunId: issueThreadInteractions.sourceRunId, - }) - .from(issueThreadInteractions) - .where( - and( - eq(issueThreadInteractions.id, payload.interactionId), - eq(issueThreadInteractions.companyId, publication.companyId), - eq(issueThreadInteractions.issueId, publication.issueId), - inArray(issueThreadInteractions.kind, [ - "ask_user_questions", - "request_confirmation", - ]), - ), - ) - .then((rows) => rows[0] ?? null); - if (!interaction?.sourceRunId) return null; - - // A provider-visible interaction is this run's response, not an additional - // message beside its progress indicator. Move the durable interaction - // publication onto the exact source run's provider-message lane. This - // retires both the normal working placeholder and the queued placeholder - // when a very fast run asks before the working update is published. - for (const progressState of ["working", "queued"] as const) { - const providerMessageId = await db - .select({ providerMessageId: chatPublications.providerMessageId }) - .from(chatPublications) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - eq( - chatPublications.idempotencyKey, - `run:${interaction.sourceRunId}:${progressState}:${publication.endpointId}`, - ), - eq(chatPublications.state, "published"), - isNotNull(chatPublications.providerMessageId), - ), - ) - .then((rows) => rows[0]?.providerMessageId ?? null); - if (providerMessageId) return providerMessageId; - } - return null; - } - - async function progressSupersededByFastResponse(publication: typeof chatPublications.$inferSelect): Promise { - if (!["queued", "working"].includes(publication.payload.progressState ?? "")) return false; - const runId = runIdFromMilestonePublication(publication); - const [run] = runId ? await db.select({ context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, publication.companyId), eq(heartbeatRuns.id, runId))) : []; - const sourceCommentId = typeof run?.context?.wakeCommentId === "string" ? run.context.wakeCommentId : typeof run?.context?.commentId === "string" ? run.context.commentId : publication.commentId; - if (!sourceCommentId) return false; - const [receipt] = await db.select({ id: fastResponseRequests.id }).from(fastResponseRequests).innerJoin(chatPublications, - and(eq(chatPublications.companyId, publication.companyId), eq(chatPublications.commentId, fastResponseRequests.commentId), eq(chatPublications.state, "published"))) - .where(and(eq(fastResponseRequests.companyId, publication.companyId), eq(fastResponseRequests.sourceCommentId, sourceCommentId), eq(fastResponseRequests.endpointId, publication.endpointId), eq(fastResponseRequests.conversationId, publication.conversationId))).limit(1); - return Boolean(receipt); - } - - async function runProgressSupersededByPublishedInteraction( - publication: typeof chatPublications.$inferSelect, - ): Promise { - if ( - !["queued", "working"].includes(publication.payload.progressState ?? "") - ) { - return false; - } - const sourceRunId = runIdFromMilestonePublication(publication); - if (!sourceRunId) return false; - const interactionIds = await db - .select({ id: issueThreadInteractions.id }) - .from(issueThreadInteractions) - .where( - and( - eq(issueThreadInteractions.companyId, publication.companyId), - eq(issueThreadInteractions.issueId, publication.issueId), - eq(issueThreadInteractions.sourceRunId, sourceRunId), - inArray(issueThreadInteractions.kind, [ - "ask_user_questions", - "request_confirmation", - ]), - ), - ) - .then((rows) => rows.map((row) => row.id)); - if (interactionIds.length === 0) return false; - return db - .select({ id: chatPublications.id }) - .from(chatPublications) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - eq(chatPublications.state, "published"), - inArray( - chatPublications.idempotencyKey, - interactionIds.map( - (interactionId) => - `interaction:${interactionId}:${publication.endpointId}`, - ), - ), - ), - ) - .limit(1) - .then((rows) => rows.length > 0); - } - - async function taskStatusPublicationToReplace( - publication: typeof chatPublications.$inferSelect, - payload: SafeChatPublicationPayload, - ): Promise { - if ( - !publication.idempotencyKey.startsWith("control:status:") || - payload.attachmentIds?.length - ) - return null; - const rows = await db - .select({ - commentId: chatPublications.commentId, - commentRunId: issueComments.createdByRunId, - idempotencyKey: chatPublications.idempotencyKey, - payload: chatPublications.payload, - providerMessageId: chatPublications.providerMessageId, - }) - .from(chatPublications) - .leftJoin(issueComments, eq(issueComments.id, chatPublications.commentId)) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.state, "published"), - isNotNull(chatPublications.providerMessageId), - ), - ) - .orderBy(desc(chatPublications.createdAt), desc(chatPublications.id)); - const rowRunId = (row: (typeof rows)[number]) => { - const milestoneMatch = - /^run:([^:]+):(?:queued|working|waiting_for_input|completed|failed):/.exec( - row.idempotencyKey, - ); - return milestoneMatch?.[1] ?? row.commentRunId ?? null; - }; - for (const candidate of rows) { - if ( - !candidate.providerMessageId || - !["queued", "working"].includes(candidate.payload.progressState ?? "") - ) { - continue; - } - const runId = rowRunId(candidate); - if (!runId) continue; - const laneClosed = rows.some((row) => { - if (rowRunId(row) !== runId) return false; - return ( - ["waiting_for_input", "completed", "failed"].includes( - row.payload.progressState ?? "", - ) || - (row.commentId !== null && row.payload.progressState === undefined) - ); - }); - if (!laneClosed) return candidate.providerMessageId; - } - return null; - } - - async function closeProgressToReplace( - publication: typeof chatPublications.$inferSelect, - endpoint: EndpointRow, - ): Promise<{ providerMessageId: string; runId: string } | null> { - if ( - !["telegram", "slack", "discord", "microsoft-teams"].includes( - endpoint.provider, - ) || - !publication.idempotencyKey.startsWith("control:close:") - ) - return null; - const current = alias(chatPublications, "close_current_progress"); - const plainText = (table: typeof chatPublications | typeof current) => - and( - sql`jsonb_typeof(${table.payload}->'text') = 'string'`, - sql`${table.payload}->'card' is null`, - sql`${table.payload}->'interactionId' is null`, - sql`${table.payload}->'transportPart' is null`, - sql`(${table.payload}->'attachmentIds' is null or ${table.payload}->'attachmentIds' = '[]'::jsonb)`, - ); - const ownedMilestone = (table: typeof chatPublications | typeof current) => - or( - sql`${table.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':queued:' || ${endpoint.id}`, - sql`${table.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${endpoint.id}`, - sql`${table.idempotencyKey} like 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${endpoint.id} || ':native:%'`, - ); - // A historical progress row alone is not edit authority: its exact current - // outbound link must still belong to that run's progress (or an interleaved - // status control). Read under the same credential lease as final delivery. - const lanes = await db - .selectDistinct({ - providerMessageId: chatMessageLinks.providerMessageId, - runId: heartbeatRuns.id, - }) - .from(chatPublications) - .innerJoin( - heartbeatRuns, - and( - eq(heartbeatRuns.companyId, publication.companyId), - eq(heartbeatRuns.agentId, endpoint.assignedAgentId), - sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${publication.issueId}`, - sql`split_part(${chatPublications.idempotencyKey}, ':', 2) = ${heartbeatRuns.id}::text`, - ), - ) - .innerJoin( - chatMessageLinks, - and( - eq(chatMessageLinks.companyId, publication.companyId), - eq(chatMessageLinks.endpointId, publication.endpointId), - eq(chatMessageLinks.conversationId, publication.conversationId), - eq(chatMessageLinks.direction, "outbound"), - eq( - chatMessageLinks.providerMessageId, - chatPublications.providerMessageId, - ), - ), - ) - .innerJoin( - current, - and( - eq(current.id, chatMessageLinks.publicationId), - eq(current.companyId, publication.companyId), - eq(current.endpointId, publication.endpointId), - eq(current.conversationId, publication.conversationId), - eq(current.issueId, publication.issueId), - eq(current.state, "published"), - eq(current.providerMessageId, chatMessageLinks.providerMessageId), - isNull(current.commentId), - plainText(current), - or( - and( - sql`${current.payload}->>'progressState' in ('queued', 'working')`, - ownedMilestone(current), - ), - like(current.idempotencyKey, "control:status:%"), - ), - ), - ) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - eq(chatPublications.state, "published"), - isNull(chatPublications.commentId), - plainText(chatPublications), - sql`${chatPublications.payload}->>'progressState' in ('queued', 'working')`, - ownedMilestone(chatPublications), - ), - ) - .limit(2); - // One close receipt owns one edit. Multiple independent lanes require a - // separate durable multi-effect protocol, never an arbitrary latest guess. - if (lanes.length !== 1) return null; - const lane = lanes[0]!; - const [possiblyConsumed] = await db - .select({ id: chatPublications.id }) - .from(chatPublications) - .leftJoin( - issueComments, - and( - eq(issueComments.id, chatPublications.commentId), - eq(issueComments.companyId, publication.companyId), - eq(issueComments.issueId, publication.issueId), - ), - ) - .where( - and( - eq(chatPublications.companyId, publication.companyId), - eq(chatPublications.endpointId, publication.endpointId), - eq(chatPublications.conversationId, publication.conversationId), - eq(chatPublications.issueId, publication.issueId), - or( - inArray(chatPublications.state, [ - "pending", - "retry", - "streaming", - "delivery_unknown", - "published", - ]), - // Stopping retries (including an operator's unknown-delivery - // cancellation) does not prove an attempted edit never arrived. - and( - inArray(chatPublications.state, ["cancelled", "failed"]), - gt(chatPublications.attempts, 0), - ), - ), - or( - eq(issueComments.createdByRunId, lane.runId), - // Question/confirmation prompts consume this same run's progress - // ID through interactionPromptPublicationToReplace, but have no - // commentId. A stale working link is not proof that their edit - // never reached the provider (including an unconfirmed receipt). - sql`exists ( - select 1 from issue_thread_interactions consuming_interaction - where consuming_interaction.company_id = ${publication.companyId}::uuid - and consuming_interaction.issue_id = ${publication.issueId}::uuid - and consuming_interaction.source_run_id = ${lane.runId}::uuid - and consuming_interaction.kind in ('ask_user_questions', 'request_confirmation') - and ${chatPublications.payload}->>'interactionId' = consuming_interaction.id::text - and ${chatPublications.idempotencyKey} = 'interaction:' || consuming_interaction.id::text || ':' || ${publication.endpointId} - )`, - eq( - chatPublications.idempotencyKey, - `run:${lane.runId}:completed:${endpoint.id}`, - ), - eq( - chatPublications.idempotencyKey, - `run:${lane.runId}:failed:${endpoint.id}`, - ), - ), - ), - ) - .limit(1); - // A final may have reached the provider before its database receipt. Even an - // unchanged working link cannot disprove that ambiguous provider effect. - return possiblyConsumed ? null : lane; - } - async function currentTaskControlPayload( publication: typeof chatPublications.$inferSelect, ): Promise { diff --git a/server/src/services/chat-channels/imessage-photon/publications.ts b/server/src/services/chat-channels/imessage-photon/publications.ts new file mode 100644 index 00000000000..d85b4410cb5 --- /dev/null +++ b/server/src/services/chat-channels/imessage-photon/publications.ts @@ -0,0 +1,9 @@ +import type { SafeChatPublicationPayload } from "@paperclipai/shared"; + +export function imessageProgressSupersessionReason( + progress: SafeChatPublicationPayload["progressState"], +): string | null { + return progress && ["queued", "working"].includes(progress) + ? "iMessage uses typing instead of progress bubbles" + : null; +} diff --git a/server/src/services/chat-channels/publication-replacement.ts b/server/src/services/chat-channels/publication-replacement.ts new file mode 100644 index 00000000000..626161201a2 --- /dev/null +++ b/server/src/services/chat-channels/publication-replacement.ts @@ -0,0 +1,965 @@ +import { imessageProgressSupersessionReason } from "./imessage-photon/publications.js"; +import { fastResponseRequests } from "@paperclipai/db"; + +import { + and, + desc, + eq, + gt, + inArray, + isNotNull, + isNull, + like, + ne, + or, + sql, +} from "drizzle-orm"; +import { alias } from "drizzle-orm/pg-core"; + +import type { Db } from "@paperclipai/db"; + +import { + agentWakeupRequests, + chatEndpoints, + chatMessageLinks, + chatPublications, + heartbeatRuns, + issueComments, + issueThreadInteractions, + issues, +} from "@paperclipai/db"; +import type { + SafeChatPublicationPayload, +} from "@paperclipai/shared"; +import { + isUuidLike, +} from "@paperclipai/shared"; + +import { + NativeChatReviewPresentationContentionError, +} from "../native-runtime/native-chat-review-presentation.js"; +import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js"; + +import { + inboundWakePublicationKey, + parseInboundWakePublicationKey, +} from "../chat-inbound-wakeup-publications.js"; + +type EndpointRow = typeof chatEndpoints.$inferSelect; +type DbOrTransaction = Db | Parameters[0]>[0]; +type ChatPublicationReplacementOptions = { + endpointRecord: (endpointId: string) => Promise<{ endpoint: EndpointRow } | null>; + authorizeRetainedChatSourcePublication: (database: DbOrTransaction, publication: typeof chatPublications.$inferSelect) => Promise; +}; + +export function createChatPublicationReplacement(db: Db, options: ChatPublicationReplacementOptions) { + const { endpointRecord, authorizeRetainedChatSourcePublication } = options; + + + function runIdFromMilestonePublication( + publication: typeof chatPublications.$inferSelect, + ): string | null { + if (!publication.payload.progressState) return null; + const match = + /^run:([^:]+):(?:queued|working|waiting_for_input|completed|failed):/.exec( + publication.idempotencyKey, + ); + const runId = match?.[1] ?? null; + // This identifier is also compared against a UUID column below. Old or + // manually repaired rows must not be allowed to turn the global + // publication sweep into a PostgreSQL cast error. + return isUuidLike(runId) ? runId : null; + } + + async function runOwnershipMilestoneSupersessionReason( + publication: typeof chatPublications.$inferSelect, + ): Promise { + const progress = publication.payload.progressState; + const typingReason = imessageProgressSupersessionReason(progress); + if (typingReason) { + const endpoint = await endpointRecord(publication.endpointId); + if (endpoint?.endpoint.provider === "imessage-photon") return typingReason; + } + if ( + !progress || + !["queued", "working", "waiting_for_input"].includes(progress) + ) { + return null; + } + const runId = runIdFromMilestonePublication(publication); + if (!runId) return null; + const run = await db.transaction(async (tx) => { + // Use native finalization's issue -> run lock order. If terminalization + // won, the current status suppresses this obsolete provider update. If + // this short authorization snapshot wins, the progress was still true + // at its durable send boundary and the later final remains authoritative. + const currentIssue = await tx + .select({ id: issues.id }) + .from(issues) + .where( + and( + eq(issues.id, publication.issueId), + eq(issues.companyId, publication.companyId), + ), + ) + .for("update") + .then((rows) => rows[0] ?? null); + if (!currentIssue) return null; + return tx + .select({ + status: heartbeatRuns.status, + errorCode: heartbeatRuns.errorCode, + }) + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.id, runId), + eq(heartbeatRuns.companyId, publication.companyId), + sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${publication.issueId}`, + ), + ) + .for("update") + .then((rows) => rows[0] ?? null); + }); + const baseProgressKey = `run:${runId}:${progress}:${publication.endpointId}`; + const nativeProgressPrefix = `run:${runId}:working:${publication.endpointId}:native:`; + const isTerminalizableProgress = + (progress === "queued" || progress === "working") && + (publication.idempotencyKey === baseProgressKey || + publication.idempotencyKey.startsWith(nativeProgressPrefix)); + if ( + isTerminalizableProgress && + run && + ["succeeded", "interrupted", "failed", "cancelled", "timed_out"].includes( + run.status, + ) + ) { + return "Run reached a terminal state before progress delivery"; + } + // Match only the closed ownership-attention projection. Historical progress + // without an extant run retains its existing delivery semantics. + const ownershipBlocked = + run?.status === "running" && + [ + "native_execution_ownership_unverified", + "native_adopted_runner_authentication_timeout", + ].includes(run.errorCode ?? ""); + if (progress === "waiting_for_input") { + return ownershipBlocked + ? null + : "Run ownership attention no longer applies to the bound task"; + } + return ownershipBlocked + ? "Run progress was superseded by ownership recovery attention" + : null; + } + + async function providerProgressLaneConsumed( + publication: typeof chatPublications.$inferSelect, + providerMessageId: string, + ): Promise { + // Published progress rows retain their old provider ID after an edit. + // Consult that exact message's current outbound link, not the conversation + // tail: an authored answer or failure has consumed the lane, while an + // interleaved task-status update may still be replaced normally. + const [consumed] = await db + .select({ id: chatPublications.id }) + .from(chatMessageLinks) + .innerJoin( + chatPublications, + and( + eq(chatPublications.id, chatMessageLinks.publicationId), + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + eq(chatPublications.providerMessageId, providerMessageId), + eq(chatPublications.state, "published"), + ), + ) + .where( + and( + eq(chatMessageLinks.companyId, publication.companyId), + eq(chatMessageLinks.endpointId, publication.endpointId), + eq(chatMessageLinks.conversationId, publication.conversationId), + eq(chatMessageLinks.direction, "outbound"), + eq(chatMessageLinks.providerMessageId, providerMessageId), + or( + and( + isNotNull(chatPublications.commentId), + sql`${chatPublications.payload}->>'progressState' is null`, + ), + sql`${chatPublications.payload}->>'progressState' = 'failed'`, + ), + ), + ) + .limit(1); + return Boolean(consumed); + } + + async function committedResponseMayReplaceFailure( + publication: typeof chatPublications.$inferSelect, + runId: string, + providerMessageId: string, + ): Promise { + if ( + !publication.commentId || + publication.payload.progressState !== undefined + ) + return false; + return db + .transaction(async (tx) => { + const [run] = await tx + .select() + .from(heartbeatRuns) + .where( + and( + eq(heartbeatRuns.id, runId), + eq(heartbeatRuns.companyId, publication.companyId), + eq(heartbeatRuns.nativeIssueId, publication.issueId), + eq(heartbeatRuns.runtimeMode, "native"), + eq(heartbeatRuns.status, "succeeded"), + isNotNull(heartbeatRuns.finishedAt), + ), + ) + .for("share", { noWait: true }); + const presentation = run?.resultJson?.presentationDecision as + Record | undefined; + if ( + !run?.resultJson?.nativeCommittedChatResponse || + presentation?.commentId !== publication.commentId + ) + return false; + const [comment] = await tx + .select({ id: issueComments.id }) + .from(issueComments) + .where( + and( + eq(issueComments.id, publication.commentId!), + eq(issueComments.companyId, publication.companyId), + eq(issueComments.issueId, publication.issueId), + eq(issueComments.createdByRunId, runId), + eq(issueComments.authorAgentId, run.agentId), + isNull(issueComments.deletedAt), + ), + ) + .for("share", { noWait: true }); + if (!comment) return false; + const [priorAnswer] = await tx + .select({ id: chatPublications.id }) + .from(chatPublications) + .innerJoin( + issueComments, + and( + eq(issueComments.id, chatPublications.commentId), + eq(issueComments.companyId, publication.companyId), + eq(issueComments.createdByRunId, runId), + ), + ) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + ne(chatPublications.id, publication.id), + inArray(chatPublications.state, [ + "published", + "streaming", + "delivery_unknown", + ]), + sql`${chatPublications.payload}->>'progressState' is null`, + ), + ) + .limit(1); + if (priorAnswer) return false; + // A historical progress row can retain this ID after a later edit. + // Only its CURRENT outbound link may grant this narrow exception; an + // authored answer or another run's failure must remain consumed. + const [failure] = await tx + .select({ id: chatPublications.id }) + .from(chatMessageLinks) + .innerJoin( + chatPublications, + and( + eq(chatPublications.id, chatMessageLinks.publicationId), + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + eq(chatPublications.providerMessageId, providerMessageId), + eq( + chatPublications.idempotencyKey, + `run:${runId}:failed:${publication.endpointId}`, + ), + eq(chatPublications.state, "published"), + isNull(chatPublications.commentId), + sql`${chatPublications.payload}->>'progressState' = 'failed'`, + ), + ) + .where( + and( + eq(chatMessageLinks.companyId, publication.companyId), + eq(chatMessageLinks.endpointId, publication.endpointId), + eq(chatMessageLinks.conversationId, publication.conversationId), + eq(chatMessageLinks.providerMessageId, providerMessageId), + eq(chatMessageLinks.direction, "outbound"), + ), + ); + if (!failure) return false; + // The marker is read from the authoritative run, never the publication + // payload. Recheck accepted result/decision plus the complete current + // source batch, actor, access and epoch before borrowing its failure lane. + // The transport claim repeats those checks at the I/O boundary. + return authorizeRetainedChatSourcePublication(tx, publication); + }) + .catch((error: unknown) => { + if (isExternalChatWaitAuthorizationContention(error)) + throw new NativeChatReviewPresentationContentionError(); + throw error; + }); + } + + async function runPublicationToReplace( + publication: typeof chatPublications.$inferSelect, + payload: SafeChatPublicationPayload, + ): Promise { + if (payload.attachmentIds?.length) return null; + const currentRunId = + runIdFromMilestonePublication(publication) ?? + (publication.commentId + ? await db + .select({ runId: issueComments.createdByRunId }) + .from(issueComments) + .where(eq(issueComments.id, publication.commentId)) + .then((rows) => rows[0]?.runId ?? null) + : null); + if (!currentRunId) return null; + return db + .select({ + id: chatPublications.id, + commentId: chatPublications.commentId, + providerMessageId: chatPublications.providerMessageId, + payload: chatPublications.payload, + }) + .from(chatPublications) + .leftJoin(issueComments, eq(issueComments.id, chatPublications.commentId)) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.state, "published"), + isNotNull(chatPublications.providerMessageId), + or( + like(chatPublications.idempotencyKey, `run:${currentRunId}:%`), + eq(issueComments.createdByRunId, currentRunId), + ), + ), + ) + .orderBy(desc(chatPublications.createdAt), desc(chatPublications.id)) + .then(async (rows) => { + // Progress updates are one replaceable provider-message lane per run. + // The first durable agent comment may turn that placeholder into the + // terminal response, but later comments from the same run are distinct + // user-visible outputs and must be posted separately. Re-editing the + // placeholder for each comment silently erases the earlier replies. + if ( + publication.commentId && + rows.some( + (row) => + row.commentId !== null && row.payload.progressState === undefined, + ) + ) + return null; + const replacement = rows.find( + (row) => + Boolean(row.providerMessageId) && + row.payload.progressState !== undefined, + ); + if (!replacement?.providerMessageId) return null; + if ( + (await providerProgressLaneConsumed( + publication, + replacement.providerMessageId, + )) && + !(await committedResponseMayReplaceFailure( + publication, + currentRunId, + replacement.providerMessageId, + )) + ) + return null; + // Replacement identity belongs to the run, not to the provider-visible + // tail. A status/control reply may legitimately interleave while the run + // is active; making the tail the edit candidate would strand this run's + // working placeholder forever. The query is bounded by endpoint, + // conversation (the task generation), and run id, so an interleaved + // control or another run can never donate its provider message here. + return replacement.providerMessageId; + }); + } + + async function receiptReactionCompletionRunId( + tx: Db, + publication: typeof chatPublications.$inferSelect, + payload: SafeChatPublicationPayload, + ): Promise { + if ( + payload.progressState === "completed" || + payload.progressState === "failed" + ) { + return runIdFromMilestonePublication(publication); + } + if (publication.commentId) { + const runId = await tx + .select({ runId: issueComments.createdByRunId }) + .from(issueComments) + .where( + and( + eq(issueComments.id, publication.commentId), + eq(issueComments.companyId, publication.companyId), + ), + ) + .then((rows) => rows[0]?.runId ?? null); + return isUuidLike(runId) ? runId : null; + } + if ( + payload.interactionId && + publication.idempotencyKey === + `interaction:${payload.interactionId}:${publication.endpointId}` + ) { + const runId = await tx + .select({ runId: issueThreadInteractions.sourceRunId }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.id, payload.interactionId), + eq(issueThreadInteractions.companyId, publication.companyId), + eq(issueThreadInteractions.issueId, publication.issueId), + ), + ) + .then((rows) => rows[0]?.runId ?? null); + return isUuidLike(runId) ? runId : null; + } + return null; + } + + async function inboundWakePublicationToReplace( + publication: typeof chatPublications.$inferSelect, + payload: SafeChatPublicationPayload, + ): Promise { + if (payload.attachmentIds?.length) return null; + const notice = parseInboundWakePublicationKey(publication.idempotencyKey); + let wakeId: string | null = + notice && notice.state !== "queued" ? notice.wakeId : null; + let runId = runIdFromMilestonePublication(publication); + if ( + !notice && + !runId && + payload.interactionId && + publication.idempotencyKey === + `interaction:${payload.interactionId}:${publication.endpointId}` + ) { + const [interaction] = await db + .select({ runId: issueThreadInteractions.sourceRunId }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.id, payload.interactionId), + eq(issueThreadInteractions.companyId, publication.companyId), + eq(issueThreadInteractions.issueId, publication.issueId), + inArray(issueThreadInteractions.kind, [ + "ask_user_questions", + "request_confirmation", + ]), + ), + ); + runId = interaction?.runId ?? null; + } + if (!notice && !runId && publication.commentId) { + const [comment] = await db + .select({ runId: issueComments.createdByRunId }) + .from(issueComments) + .where( + and( + eq(issueComments.id, publication.commentId), + eq(issueComments.companyId, publication.companyId), + eq(issueComments.issueId, publication.issueId), + ), + ); + runId = comment?.runId ?? null; + } + let runContext: Record | null = null; + if (runId) { + if (publication.commentId) { + const [priorAnswer] = await db + .select({ id: chatPublications.id }) + .from(chatPublications) + .innerJoin( + issueComments, + eq(issueComments.id, chatPublications.commentId), + ) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.state, "published"), + eq(issueComments.createdByRunId, runId), + sql`${chatPublications.payload}->>'progressState' is null`, + ), + ) + .limit(1); + if (priorAnswer) return null; + } + const [run] = await db + .select({ + wakeId: heartbeatRuns.wakeupRequestId, + context: heartbeatRuns.contextSnapshot, + }) + .from(heartbeatRuns) + .innerJoin( + agentWakeupRequests, + and( + eq(agentWakeupRequests.id, heartbeatRuns.wakeupRequestId), + eq(agentWakeupRequests.runId, heartbeatRuns.id), + eq(agentWakeupRequests.companyId, heartbeatRuns.companyId), + eq(agentWakeupRequests.agentId, heartbeatRuns.agentId), + // Failed/cancelled is the admitted run's outcome, not loss of + // its original admission. Its terminal milestone still owns + // this exact lane even if no working update was sent first. + ne(agentWakeupRequests.status, "skipped"), + ), + ) + .where( + and( + eq(heartbeatRuns.id, runId), + eq(heartbeatRuns.companyId, publication.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${publication.issueId}`, + ), + ); + wakeId = run?.wakeId ?? null; + runContext = run?.context ?? null; + } + if (!wakeId) return null; + const [queued] = await db + .select({ + commentId: chatPublications.commentId, + providerMessageId: chatPublications.providerMessageId, + }) + .from(chatPublications) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + eq(chatPublications.state, "published"), + isNotNull(chatPublications.providerMessageId), + eq( + chatPublications.idempotencyKey, + inboundWakePublicationKey( + wakeId, + "queued", + publication.endpointId, + publication.conversationId, + ), + ), + ), + ); + if (!queued?.commentId || !queued.providerMessageId) return null; + // The run helper may have rejected this same provider ID because its + // queued/working placeholder is already an answer or terminal failure. + // Falling back to the older wake row must not erase that newer output. + if ( + await providerProgressLaneConsumed(publication, queued.providerMessageId) + ) + return null; + if (notice && notice.state !== "queued") + return queued.commentId === publication.commentId + ? queued.providerMessageId + : null; + // A deferred owner can be a Board wake into which the exact external + // comment coalesced. The run must actually contain that source comment; + // wake ID alone cannot authorize a different successor's edit lane. + return runContext && + (runContext.wakeCommentId === queued.commentId || + runContext.commentId === queued.commentId || + (Array.isArray(runContext.wakeCommentIds) && + runContext.wakeCommentIds.includes(queued.commentId))) + ? queued.providerMessageId + : null; + } + + async function interactionResolutionPublicationToReplace( + publication: typeof chatPublications.$inferSelect, + payload: SafeChatPublicationPayload, + ): Promise { + if ( + !publication.idempotencyKey.startsWith("interaction-resolution:") || + !payload.interactionId + ) + return null; + return db + .select({ providerMessageId: chatPublications.providerMessageId }) + .from(chatPublications) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + eq( + chatPublications.idempotencyKey, + `interaction:${payload.interactionId}:${publication.endpointId}`, + ), + eq(chatPublications.state, "published"), + isNotNull(chatPublications.providerMessageId), + ), + ) + .then((rows) => rows[0]?.providerMessageId ?? null); + } + + async function interactionPromptPublicationToReplace( + publication: typeof chatPublications.$inferSelect, + payload: SafeChatPublicationPayload, + ): Promise { + if ( + !payload.interactionId || + publication.idempotencyKey !== + `interaction:${payload.interactionId}:${publication.endpointId}` + ) { + return null; + } + const interaction = await db + .select({ + kind: issueThreadInteractions.kind, + sourceRunId: issueThreadInteractions.sourceRunId, + }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.id, payload.interactionId), + eq(issueThreadInteractions.companyId, publication.companyId), + eq(issueThreadInteractions.issueId, publication.issueId), + inArray(issueThreadInteractions.kind, [ + "ask_user_questions", + "request_confirmation", + ]), + ), + ) + .then((rows) => rows[0] ?? null); + if (!interaction?.sourceRunId) return null; + + // A provider-visible interaction is this run's response, not an additional + // message beside its progress indicator. Move the durable interaction + // publication onto the exact source run's provider-message lane. This + // retires both the normal working placeholder and the queued placeholder + // when a very fast run asks before the working update is published. + for (const progressState of ["working", "queued"] as const) { + const providerMessageId = await db + .select({ providerMessageId: chatPublications.providerMessageId }) + .from(chatPublications) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + eq( + chatPublications.idempotencyKey, + `run:${interaction.sourceRunId}:${progressState}:${publication.endpointId}`, + ), + eq(chatPublications.state, "published"), + isNotNull(chatPublications.providerMessageId), + ), + ) + .then((rows) => rows[0]?.providerMessageId ?? null); + if (providerMessageId) return providerMessageId; + } + return null; + } + + async function progressSupersededByFastResponse(publication: typeof chatPublications.$inferSelect): Promise { + if (!["queued", "working"].includes(publication.payload.progressState ?? "")) return false; + const runId = runIdFromMilestonePublication(publication); + const [run] = runId ? await db.select({ context: heartbeatRuns.contextSnapshot }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, publication.companyId), eq(heartbeatRuns.id, runId))) : []; + const sourceCommentId = typeof run?.context?.wakeCommentId === "string" ? run.context.wakeCommentId : typeof run?.context?.commentId === "string" ? run.context.commentId : publication.commentId; + if (!sourceCommentId) return false; + const [receipt] = await db.select({ id: fastResponseRequests.id }).from(fastResponseRequests).innerJoin(chatPublications, + and(eq(chatPublications.companyId, publication.companyId), eq(chatPublications.commentId, fastResponseRequests.commentId), eq(chatPublications.state, "published"))) + .where(and(eq(fastResponseRequests.companyId, publication.companyId), eq(fastResponseRequests.sourceCommentId, sourceCommentId), eq(fastResponseRequests.endpointId, publication.endpointId), eq(fastResponseRequests.conversationId, publication.conversationId))).limit(1); + return Boolean(receipt); + } + + async function runProgressSupersededByPublishedInteraction( + publication: typeof chatPublications.$inferSelect, + ): Promise { + if ( + !["queued", "working"].includes(publication.payload.progressState ?? "") + ) { + return false; + } + const sourceRunId = runIdFromMilestonePublication(publication); + if (!sourceRunId) return false; + const interactionIds = await db + .select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.companyId, publication.companyId), + eq(issueThreadInteractions.issueId, publication.issueId), + eq(issueThreadInteractions.sourceRunId, sourceRunId), + inArray(issueThreadInteractions.kind, [ + "ask_user_questions", + "request_confirmation", + ]), + ), + ) + .then((rows) => rows.map((row) => row.id)); + if (interactionIds.length === 0) return false; + return db + .select({ id: chatPublications.id }) + .from(chatPublications) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + eq(chatPublications.state, "published"), + inArray( + chatPublications.idempotencyKey, + interactionIds.map( + (interactionId) => + `interaction:${interactionId}:${publication.endpointId}`, + ), + ), + ), + ) + .limit(1) + .then((rows) => rows.length > 0); + } + + async function taskStatusPublicationToReplace( + publication: typeof chatPublications.$inferSelect, + payload: SafeChatPublicationPayload, + ): Promise { + if ( + !publication.idempotencyKey.startsWith("control:status:") || + payload.attachmentIds?.length + ) + return null; + const rows = await db + .select({ + commentId: chatPublications.commentId, + commentRunId: issueComments.createdByRunId, + idempotencyKey: chatPublications.idempotencyKey, + payload: chatPublications.payload, + providerMessageId: chatPublications.providerMessageId, + }) + .from(chatPublications) + .leftJoin(issueComments, eq(issueComments.id, chatPublications.commentId)) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.state, "published"), + isNotNull(chatPublications.providerMessageId), + ), + ) + .orderBy(desc(chatPublications.createdAt), desc(chatPublications.id)); + const rowRunId = (row: (typeof rows)[number]) => { + const milestoneMatch = + /^run:([^:]+):(?:queued|working|waiting_for_input|completed|failed):/.exec( + row.idempotencyKey, + ); + return milestoneMatch?.[1] ?? row.commentRunId ?? null; + }; + for (const candidate of rows) { + if ( + !candidate.providerMessageId || + !["queued", "working"].includes(candidate.payload.progressState ?? "") + ) { + continue; + } + const runId = rowRunId(candidate); + if (!runId) continue; + const laneClosed = rows.some((row) => { + if (rowRunId(row) !== runId) return false; + return ( + ["waiting_for_input", "completed", "failed"].includes( + row.payload.progressState ?? "", + ) || + (row.commentId !== null && row.payload.progressState === undefined) + ); + }); + if (!laneClosed) return candidate.providerMessageId; + } + return null; + } + + async function closeProgressToReplace( + publication: typeof chatPublications.$inferSelect, + endpoint: EndpointRow, + ): Promise<{ providerMessageId: string; runId: string } | null> { + if ( + !["telegram", "slack", "discord", "microsoft-teams"].includes( + endpoint.provider, + ) || + !publication.idempotencyKey.startsWith("control:close:") + ) + return null; + const current = alias(chatPublications, "close_current_progress"); + const plainText = (table: typeof chatPublications | typeof current) => + and( + sql`jsonb_typeof(${table.payload}->'text') = 'string'`, + sql`${table.payload}->'card' is null`, + sql`${table.payload}->'interactionId' is null`, + sql`${table.payload}->'transportPart' is null`, + sql`(${table.payload}->'attachmentIds' is null or ${table.payload}->'attachmentIds' = '[]'::jsonb)`, + ); + const ownedMilestone = (table: typeof chatPublications | typeof current) => + or( + sql`${table.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':queued:' || ${endpoint.id}`, + sql`${table.idempotencyKey} = 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${endpoint.id}`, + sql`${table.idempotencyKey} like 'run:' || ${heartbeatRuns.id}::text || ':working:' || ${endpoint.id} || ':native:%'`, + ); + // A historical progress row alone is not edit authority: its exact current + // outbound link must still belong to that run's progress (or an interleaved + // status control). Read under the same credential lease as final delivery. + const lanes = await db + .selectDistinct({ + providerMessageId: chatMessageLinks.providerMessageId, + runId: heartbeatRuns.id, + }) + .from(chatPublications) + .innerJoin( + heartbeatRuns, + and( + eq(heartbeatRuns.companyId, publication.companyId), + eq(heartbeatRuns.agentId, endpoint.assignedAgentId), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${publication.issueId}`, + sql`split_part(${chatPublications.idempotencyKey}, ':', 2) = ${heartbeatRuns.id}::text`, + ), + ) + .innerJoin( + chatMessageLinks, + and( + eq(chatMessageLinks.companyId, publication.companyId), + eq(chatMessageLinks.endpointId, publication.endpointId), + eq(chatMessageLinks.conversationId, publication.conversationId), + eq(chatMessageLinks.direction, "outbound"), + eq( + chatMessageLinks.providerMessageId, + chatPublications.providerMessageId, + ), + ), + ) + .innerJoin( + current, + and( + eq(current.id, chatMessageLinks.publicationId), + eq(current.companyId, publication.companyId), + eq(current.endpointId, publication.endpointId), + eq(current.conversationId, publication.conversationId), + eq(current.issueId, publication.issueId), + eq(current.state, "published"), + eq(current.providerMessageId, chatMessageLinks.providerMessageId), + isNull(current.commentId), + plainText(current), + or( + and( + sql`${current.payload}->>'progressState' in ('queued', 'working')`, + ownedMilestone(current), + ), + like(current.idempotencyKey, "control:status:%"), + ), + ), + ) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + eq(chatPublications.state, "published"), + isNull(chatPublications.commentId), + plainText(chatPublications), + sql`${chatPublications.payload}->>'progressState' in ('queued', 'working')`, + ownedMilestone(chatPublications), + ), + ) + .limit(2); + // One close receipt owns one edit. Multiple independent lanes require a + // separate durable multi-effect protocol, never an arbitrary latest guess. + if (lanes.length !== 1) return null; + const lane = lanes[0]!; + const [possiblyConsumed] = await db + .select({ id: chatPublications.id }) + .from(chatPublications) + .leftJoin( + issueComments, + and( + eq(issueComments.id, chatPublications.commentId), + eq(issueComments.companyId, publication.companyId), + eq(issueComments.issueId, publication.issueId), + ), + ) + .where( + and( + eq(chatPublications.companyId, publication.companyId), + eq(chatPublications.endpointId, publication.endpointId), + eq(chatPublications.conversationId, publication.conversationId), + eq(chatPublications.issueId, publication.issueId), + or( + inArray(chatPublications.state, [ + "pending", + "retry", + "streaming", + "delivery_unknown", + "published", + ]), + // Stopping retries (including an operator's unknown-delivery + // cancellation) does not prove an attempted edit never arrived. + and( + inArray(chatPublications.state, ["cancelled", "failed"]), + gt(chatPublications.attempts, 0), + ), + ), + or( + eq(issueComments.createdByRunId, lane.runId), + // Question/confirmation prompts consume this same run's progress + // ID through interactionPromptPublicationToReplace, but have no + // commentId. A stale working link is not proof that their edit + // never reached the provider (including an unconfirmed receipt). + sql`exists ( + select 1 from issue_thread_interactions consuming_interaction + where consuming_interaction.company_id = ${publication.companyId}::uuid + and consuming_interaction.issue_id = ${publication.issueId}::uuid + and consuming_interaction.source_run_id = ${lane.runId}::uuid + and consuming_interaction.kind in ('ask_user_questions', 'request_confirmation') + and ${chatPublications.payload}->>'interactionId' = consuming_interaction.id::text + and ${chatPublications.idempotencyKey} = 'interaction:' || consuming_interaction.id::text || ':' || ${publication.endpointId} + )`, + eq( + chatPublications.idempotencyKey, + `run:${lane.runId}:completed:${endpoint.id}`, + ), + eq( + chatPublications.idempotencyKey, + `run:${lane.runId}:failed:${endpoint.id}`, + ), + ), + ), + ) + .limit(1); + // A final may have reached the provider before its database receipt. Even an + // unchanged working link cannot disprove that ambiguous provider effect. + return possiblyConsumed ? null : lane; + } + + return { runIdFromMilestonePublication, runOwnershipMilestoneSupersessionReason, runPublicationToReplace, receiptReactionCompletionRunId, inboundWakePublicationToReplace, interactionResolutionPublicationToReplace, interactionPromptPublicationToReplace, progressSupersededByFastResponse, runProgressSupersededByPublishedInteraction, taskStatusPublicationToReplace, closeProgressToReplace }; +}