From 338bf757617ca447518b472c31002474a1968dc9 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 26 Oct 2026 10:26:45 -0500 Subject: [PATCH] fix(chat): admit task Retry after GitHub reconnect (#17733) ## Thinking Path > - Paperclip manages agents and their tasks. > - Task Retry must preserve saved input and enforce current access. > - A GitHub reconnect invalidates old runtime callbacks. > - Retry reused that old callback authority and rejected valid new intent. > - Retry also replayed an older failed request when a follow-up was already saved. > - This change admits a deliberate Retry under current authority and transfers saved input once. ## Linked Issues or Issue Description Refs #16965, #17053 and #17317. These repairs preserve work and contain old executors. This change fixes the subsequent chat admission boundary. A search found no overlapping open reconnect/admission fix. **What happened?** Ordinary task Retry returned `chat_failed_run_retry_not_authorized` after the same GitHub App reconnected. The endpoint was healthy, but the saved request still carried its old runtime generation. A newer queued follow-up also made Retry reject the older failed request instead of continuing the follow-up. No successor was admitted. HTTP qualification found two related defects. Concurrent clicks could omit the canonical chat action from a response. Signed-in board dispatch also passed an incomplete task resource to the permission resolver, which rejected a permitted mutation. **Expected behavior** A deliberate Retry checks current authority for the same App, sender, repository, conversation and task. It continues saved input once. A later reconnect, changed App, revoked access or changed source prevents dispatch. Old runtime callbacks remain invalid. **Steps to reproduce** 1. Admit a GitHub request and record a failed native run. 2. Save a follow-up while execution is blocked. 3. Reconnect the same GitHub App. 4. Click ordinary task Retry. 5. Before this change, admission returns 409. Without reconnect, the newer input alone also causes 409. **Paperclip version or commit** Baseline source: `41c42895fa862c586ff841f8c50200d93ffc83a4`. **Deployment mode** Source checkout on macOS. Regressions use real PostgreSQL transactions and the assembled chat, HTTP route and heartbeat admission services. Provider boundaries are simulated. No hosted recovery is claimed. ## What Changed - Record the admitting App identity for new GitHub deliveries. Historical deliveries can use their original working-comment binding. Reauthorize deliberate Retry under a pinned current generation and credential-reference fingerprint. - Select the complete saved-input queue from the failed run's actual conversation before replaying the older failed batch. Keep exact-run Retry bounded while ordinary task Retry supports larger queues. Transfer its receipts with the dispatch transaction. Recover the same ownership after restart or a lost response. Release unadmitted input after a definitive rejection. - Preserve canonical chat action IDs across concurrent clicks. Supply current task status and assignment fields to board mutation checks. - Keep exact-run Retry, stale webhook, containment, source integrity, publication, pause and budget checks. Update recovery contracts and verification recipes. ## Verification - Red: all three new PostgreSQL regressions failed on unchanged production code. Reconnect failed at chat authorization. Newer input failed at the obsolete-source guard. The combined case also failed. - Green: 61 selected chat integration cases passed. These include 27 new cases and 34 existing cases for containment, retained-source recovery, exact batches and source revocation. - A further red/green PostgreSQL regression reproduces the 51-message overflow identified during review. All 51 messages reach one successor, including a later Retry of that successor. A separate 52-message consumed-batch regression also passes. - A competing conversation retains its own deferred message. Distinct automatic policy contexts are checked independently; revoking only the later policy blocks the whole unadmitted batch. Batch validation still rejects a later message with a changed App binding or deleted source. A real PostgreSQL query capture reduced the 51-message preparation from 1,103 calls to fewer than 180. - All 16 durable task Retry tests passed. - The real HTTP route returns 202 and admits one queued successor for concurrent deliveries. Fault injection covers rollback during input handoff, response loss after dispatch commit, and permission loss when resuming an existing durable wait. A wait only adopts input after current admission succeeds; a one-shot settlement marker keeps status projection read-only. A new deliberate Retry works after another reconnect or a failed successor. - Current-App, repository, sender, source edit/delete, reassignment, endpoint pause and board-permission controls passed. Historical deliveries remain unchanged. - Final-head CI: 55 checks passed, three expected skips. The unchanged minute-bucket rate-limit assertion passed locally against PostgreSQL and on its one CI rerun. Greptile reviewed the final commit at 5/5 with no unresolved threads. - Workspace typecheck and full build passed. Map alignment, module boundaries, UI token gates and `git diff --check` passed. - Hosted task recovery and a live provider turn remain unverified. The local tests hold agent capacity to inspect admission without starting a model or publishing a GitHub response. ## Risks - A deliberate GitHub Retry can admit old saved input under current connection authority. It requires the same App binding and current access. It pins that admission; it does not rewrite the historical delivery or make stale callbacks valid. - The durable chat queue has an optional admission callback. It transfers input only when the transaction contains a valid queued execution or saved wait. Other chat consumers have no callback and retain their behavior. - No migration, Cloud dependency, deployment, configuration change or workspace cleanup is included. ## Model Used OpenAI GPT-6 through Codex, with reasoning, source inspection, code execution and GitHub tools. The exact serving variant and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [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 --- doc/task-recovery.md | 22 +- maps/agent-access-map/recovery.md | 7 + maps/feature-map/recovery.md | 8 + .../chat-channels.integration.test.ts | 353 ++++++++++++++++++ server/src/services/chat-channels.ts | 18 +- .../chat-channels/credential-fingerprint.ts | 10 + .../chat-channels/failed-run-retry.ts | 206 ++++++---- .../chat-channels/github/task-preparation.ts | 2 +- .../github/task-retry-authorization.ts | 126 +++++++ .../inbound/wakeup-authorization.ts | 17 +- .../chat-channels/task-retry-input.ts | 95 +++++ server/src/services/durable-chat-wakeup.ts | 23 +- server/src/services/heartbeat/queue.ts | 28 +- server/src/services/task-retry.ts | 15 + 14 files changed, 808 insertions(+), 122 deletions(-) create mode 100644 server/src/services/chat-channels/credential-fingerprint.ts create mode 100644 server/src/services/chat-channels/github/task-retry-authorization.ts create mode 100644 server/src/services/chat-channels/task-retry-input.ts diff --git a/doc/task-recovery.md b/doc/task-recovery.md index 870b8299a34..714d921a97b 100644 --- a/doc/task-recovery.md +++ b/doc/task-recovery.md @@ -15,8 +15,26 @@ new generation; a repeated HTTP request cannot. The existing exact-run retry endpoint retains its source-bound contract. Connected conversations use their existing inbound authorization owner. A board -Retry changes the workspace strategy, not the external principal, source batch, -conversation access or publication authority. +Retry preserves external principal, conversation access and publication checks. +A GitHub Retry selects already-saved, undelivered input before replaying the +failed run's older request. The selected receipts transfer to the retry owner +under the dispatch transaction. They link to its one successor, including after +a restart or a lost dispatch response. Rejected, unadmitted retries release the +saved input for another deliberate Retry. + +A deliberate GitHub Retry can admit saved input after the same App reconnects. +It verifies the App binding in the accepted delivery or its historical working +comment receipt, then checks current sender, repository, conversation and task +access. The retry records the current runtime generation and credential-reference +fingerprint. Another reconnect invalidates that admission. Historical deliveries, +stale webhook callbacks and the exact-run retry contract retain their fences. + +Task Retry keeps the complete saved-input batch, including batches larger than +the exact-run endpoint's limit. Source rows, receipts and invalidations are read +in batches. Within one transaction, messages with the same principal and automatic +policy context share a current-access check; every message still proves its App, +destination, comment link and immutable source. Different automatic policy contexts +receive separate access checks. Input in another conversation remains queued. A connected-conversation Retry can save its exact request while the matching system recovery generation verifies the old executor. Saving that wait grants diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index e0c6003c9f2..e1973773a38 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -4,6 +4,13 @@ Product map: [Recovery](../feature-map/recovery.md). Feature ID: `recovery`. +Ordinary board Retry can reauthorize saved GitHub input for the same App after +reconnect. It pins current runtime authority and preserves sender, repository, +conversation and task checks. Pending input takes precedence over the older +failed request and transfers once to the retry owner. This grants no new agent +Retry API or stale webhook authority. Local PostgreSQL and HTTP-route regressions +cover these transitions; hosted execution and delivery remain separate acceptance. + Source revision: `3bbc2104b65caf9f7763938a01260a5673e12f7a`. Review depth and executed proof are separate. The operation records enumerate the named behavior and access boundaries. Their evidence states describe the reviewed sources; run results and environment limits belong in the PR or verification artifacts. A listed API or source test does not establish a live authorized journey. Follow the [map contract](../README.md). diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index b72eb30ea42..8c56541e248 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -89,6 +89,14 @@ The connected-chat database regression in exercises admission and queue continuation. It does not establish hosted provider recovery or successful delivery. +For GitHub, save a follow-up behind a failed run, then reconnect the same App. +Click ordinary Retry and verify that one successor receives the saved follow-up +instead of replaying the older request. Reload and repeat the same request key; +there must be one successor and no duplicate publication. A second reconnect +after admission, changed App, revoked sender or changed repository must prevent +dispatch. An unadmitted rejection keeps the follow-up available for another +deliberate Retry. Historical deliveries must remain unchanged. + ### `post-stop-message` After a stop, send a new instruction on the same task or inspect one already diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index f844a18bce3..095ca9aa105 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -131,6 +131,7 @@ import { deliverReconciledExecutions } from "../services/execution-recovery-reso import { unadmittedChatWakeupCondition, authorizeFailedChatRunRetryWake, + createDurableChatWakeupRequest, } from "../services/durable-chat-wakeup.js"; import { createChatWebhookDiagnostics, @@ -60661,6 +60662,358 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { ...context, issue, action, receipt, runId }; } + async function currentGitHubTaskRetryFixture(withPending = false, pendingCount = 1) { + const f = await failedChatRetryFixture("github", "42", true); + const companyId = f.fixture.companyId, issueId = f.issue.id; + await githubResponseCommentService(db, f.providerFetch).tryAcknowledge(f.action.deliveryId!); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, + nativePhase: "terminal_failure", runnerProfileJson: {} }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: f.runId, phase: "terminal_failure" }); + await db.insert(heartbeatRunEvents).values({ companyId, agentId: f.fixture.assignedAgentId, runId: f.runId, + seq: 1, eventType: "native.local_process_stopped", payload: { processPid: 999999999, processGroupId: null } }); + let pending: typeof chatActions.$inferSelect | undefined; + if (withPending) { + for (let index = 0; index < pendingCount; index++) { + await deliverMessage({ callbacks: f.runtime.configurations.get(f.endpoint.id)!.callbacks, + endpointId: f.endpoint.id, provider: "github", thread: f.thread.thread, + message: makeMessage({ id: String(990092 + index), text: `@maya continue saved follow-up ${index}`, userId: "42", mentioned: true }), + trigger: "mention" }); + const [{ source }] = await db.select({ source: chatActions }).from(chatActions) + .innerJoin(chatDeliveries, eq(chatDeliveries.id, chatActions.deliveryId)).where(and( + eq(chatActions.conversationId, f.conversation.id), eq(chatActions.kind, "inbound_wakeup"), + eq(chatDeliveries.providerEventId, `${f.thread.thread.id}:${990092 + index}`))); + pending ??= source; + await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution", runId: null }) + .where(eq(agentWakeupRequests.id, source.id)); + await githubResponseCommentService(db, f.providerFetch).tryAcknowledge(source.deliveryId!); + } + } + const [originalDelivery] = await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, f.action.deliveryId!)); + + await db.update(agents).set({ adapterType: "paperclip_runner", status: "running", + runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } }) + .where(eq(agents.id, f.fixture.assignedAgentId)); + await db.insert(heartbeatRuns).values({ companyId, agentId: f.fixture.assignedAgentId, + status: "running", startedAt: new Date() }); + f.wakeup.mockImplementation(heartbeatService(db).wakeup); + return { ...f, pending, originalDelivery }; + } + + it.each(["reconnected", "newer_input", "reconnected_newer_input"] as const)( + "admits current GitHub task Retry without replaying obsolete input: %s", async kind => { + const f = await currentGitHubTaskRetryFixture(kind.includes("newer_input")); + const { pending, originalDelivery } = f; + try { + const companyId = f.fixture.companyId, issueId = f.issue.id; + if (kind.includes("reconnected")) await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user"); + const input = { db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant", + observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry }; + const request = await requestTaskRetry(input); + expect(request.accepted).toBe(true); + expect((await requestTaskRetry(input)).requestId).toBe(request.requestId); + expect(await f.service.processFailedChatRunRetry(request.chatActionId!)).toMatchObject({ status: "queued" }); + const [wake] = await readAgentWakeupRequestsById(request.chatActionId!); + expect(wake.runId).toBeTruthy(); + const expectedComment = pending?.payload.commentId ?? f.action.payload.commentId; + const [run] = await readHeartbeatRunsById(wake.runId!); + expect(run).toMatchObject({ status: "queued", retryOfRunId: f.runId, + contextSnapshot: { wakeCommentIds: [expectedComment], taskRetryRequestId: request.requestId } }); + if (pending) expect((await readAgentWakeupRequestsById(pending.id))[0]) + .toMatchObject({ status: "coalesced", runId: wake.runId }); + await f.service.processFailedChatRunRetry(request.chatActionId!); + await heartbeatService(db).resumeQueuedRuns(); + expect((await readHeartbeatRunsByCompanyId(companyId)).filter(r => r.contextSnapshot?.taskRetryRequestId === request.requestId)) + .toHaveLength(1); + expect((await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, originalDelivery.id)))[0].normalizedEvent) + .toEqual(originalDelivery.normalizedEvent); + } finally { await f.service.shutdown(); } + }, + ); + + it.each(["legacy_receipt", "concurrent_clicks", "lost_dispatch_reply", "restart", "retry_failed_successor", "http", "handoff_rollback", "board_permission_wait", "peer_conversation", "backlog_51", "consumed_backlog_52"] as const)( + "preserves one current GitHub Retry across %s", async mode => { + const f = await currentGitHubTaskRetryFixture(true, mode.includes("backlog") ? 51 : mode === "peer_conversation" ? 2 : 1); + try { + const companyId = f.fixture.companyId, issueId = f.issue.id; + if (mode === "consumed_backlog_52") { + const sources = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id), + eq(chatActions.kind, "inbound_wakeup"))).orderBy(asc(chatActions.createdAt), asc(chatActions.id)); + expect(sources).toHaveLength(52); + const commentIds = sources.map(source => source.payload.commentId); + await db.update(heartbeatRuns).set({ finishedAt: new Date(), contextSnapshot: { + ...(await readHeartbeatRunsById(f.runId))[0].contextSnapshot, + wakeCommentIds: commentIds, wakeCommentId: commentIds.at(-1), + } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(agentWakeupRequests).set({ runId: f.runId, status: "coalesced", + payload: sql`${agentWakeupRequests.payload} || ${JSON.stringify({ coalescedIntoWakeupRequestId: f.action.id })}::jsonb`, + }).where(and(inArray(agentWakeupRequests.id, sources.map(source => source.id)), + sql`${agentWakeupRequests.id} <> ${f.action.id}::uuid`)); + } + let peerInput: typeof chatActions.$inferSelect | undefined; + if (mode === "peer_conversation") { + const originId = randomUUID(); + await db.insert(chatConversations).values({ ...f.conversation, id: originId, sessionGeneration: 2 }); + await db.update(chatActions).set({ conversationId: originId, + payload: sql`jsonb_set(${chatActions.payload}, '{sessionGeneration}', '2'::jsonb)`, + }).where(inArray(chatActions.id, [f.action.id, f.pending!.id])); + await db.update(chatDeliveries).set({ conversationId: originId }) + .where(inArray(chatDeliveries.id, [f.action.deliveryId!, f.pending!.deliveryId!])); + await db.update(chatMessageLinks).set({ conversationId: originId }) + .where(inArray(chatMessageLinks.deliveryId, [f.action.deliveryId!, f.pending!.deliveryId!])); + [peerInput] = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id), + eq(chatActions.kind, "inbound_wakeup"))); + expect(peerInput).toBeDefined(); + } + if (mode === "legacy_receipt") { + const [originalEndpoint] = await readChatEndpointsById(f.endpoint.id); + await db.insert(chatActions).values({ companyId, endpointId: f.endpoint.id, + conversationId: f.conversation.id, deliveryId: f.pending!.deliveryId, + kind: "github_response_comment", providerActionId: `github_response:${f.pending!.deliveryId}`, + payload: { botExternalId: originalEndpoint.botExternalId, agentId: f.fixture.assignedAgentId }, + status: "processed" }).onConflictDoNothing(); + await db.update(chatDeliveries).set({ + normalizedEvent: sql`jsonb_set(${chatDeliveries.normalizedEvent}, '{githubAuthority}', + (${chatDeliveries.normalizedEvent}->'githubAuthority') - 'botExternalId')`, + }).where(eq(chatDeliveries.id, f.pending!.deliveryId!)); + } + await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user"); + const input = { db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant", + observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry }; + if (mode === "board_permission_wait") { + input.actorId = `retry-operator-${companyId}`; + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: input.actorId, + status: "active", membershipRole: "operator" }); + } + const queryEvents: Record[] = []; + const capture = mode === "backlog_51" ? createDatabaseResponseCapture({ start: Date.now() - 1, + end: Date.now() + 59_000 }, event => queryEvents.push(event)) : undefined; + if (capture) installDatabaseResponseCapture(db, { chatActions, chatDeliveries, chatMessageLinks, issueComments, + agentWakeupRequests, chatEndpoints, chatConversations }, capture); + let one: Awaited>, two: typeof one; + if (mode === "http") { + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { req.actor = boardActor(companyId); next(); }); + app.use("/api", issueRoutes(db, createStorageService().storage, { chatRunRetries: f.service })); + app.use(errorHandler); + const send = () => request(app).post(`/api/issues/${issueId}/retry`).send({ + idempotencyKey: input.idempotencyKey, observedRunId: input.observedRunId, + }).expect(202); + const replies = await Promise.all([send(), send()]); + [one, two] = replies.map(response => response.body); + } else [one, two] = await Promise.all([requestTaskRetry(input), requestTaskRetry({ ...input, + idempotencyKey: mode === "concurrent_clicks" ? randomUUID() : input.idempotencyKey })]); + if (capture) { + capture.flush(); + expect(Number(queryEvents.find(event => event.event === "paperclip.db_response_capture.summary.v1")?.startedQueries)) + .toBeLessThan(180); + } + expect(two.requestId).toBe(one.requestId); + expect(one).toMatchObject({ accepted: true, chatActionId: expect.any(String) }); + let service = f.service; + if (mode === "restart") { + await service.shutdown(); + service = createService(f.runtime, f.providerFetch, { wakeup: heartbeatService(db).wakeup }).service; + } + if (mode === "lost_dispatch_reply") f.wakeup.mockImplementationOnce(async (...args) => { + await heartbeatService(db).wakeup(...args); + throw new Error("Simulated response loss after dispatch commit"); + }); + if (mode === "handoff_rollback") f.wakeup.mockImplementationOnce(async (agentId, options) => { + const original = options!.durableChatRequest!; + return heartbeatService(db).wakeup(agentId, { ...options, durableChatRequest: createDurableChatWakeupRequest({ + ...original, onAdmitted: async (tx, receipt) => { + await original.onAdmitted!(tx, receipt); + throw new Error("Simulated interruption before admission commit"); + }, + }) }); + }); + if (mode === "board_permission_wait") await db.update(companyMemberships).set({ membershipRole: "viewer" }) + .where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, input.actorId))); + await Promise.all([service.processFailedChatRunRetry(one.chatActionId!), + service.processFailedChatRunRetry(one.chatActionId!)]); + if (mode === "handoff_rollback" || mode === "board_permission_wait") { + expect(await readAgentWakeupRequestsById(one.chatActionId!)).toHaveLength(0); + expect((await readAgentWakeupRequestsById(f.pending!.id))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === one.requestId)).toHaveLength(0); + } + if (mode === "board_permission_wait") await db.update(companyMemberships).set({ membershipRole: "operator" }) + .where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, input.actorId))); + await vi.waitFor(async () => { + await service.processFailedChatRunRetry(one.chatActionId!); + expect((await readAgentWakeupRequestsById(one.chatActionId!))[0]?.runId, + JSON.stringify((await db.select().from(chatActions).where(eq(chatActions.id, one.chatActionId!)))[0]?.result)).toBeTruthy(); + }, { timeout: 10_000, interval: 100 }); + const [parent] = await readAgentWakeupRequestsById(one.chatActionId!); + expect(parent.runId).toBeTruthy(); + expect((await readAgentWakeupRequestsById(f.pending!.id))[0]).toMatchObject({ + status: "coalesced", runId: mode === "consumed_backlog_52" ? f.runId : parent.runId, + }); + if (peerInput) expect((await readAgentWakeupRequestsById(peerInput.id))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + if (mode === "backlog_51") { + const saved = await db.select({ id: chatActions.id, payload: chatActions.payload, + status: agentWakeupRequests.status, runId: agentWakeupRequests.runId }).from(chatActions) + .innerJoin(agentWakeupRequests, eq(agentWakeupRequests.id, chatActions.id)).where(and( + eq(chatActions.conversationId, f.conversation.id), eq(chatActions.kind, "inbound_wakeup"), + sql`${chatActions.id} <> ${f.action.id}::uuid`, + )).orderBy(asc(chatActions.createdAt), asc(chatActions.id)); + expect(saved).toHaveLength(51); + expect(saved.every(source => source.status === "coalesced" && source.runId === parent.runId)).toBe(true); + expect((await readHeartbeatRunsById(parent.runId!))[0].contextSnapshot?.wakeCommentIds) + .toEqual(saved.map(source => source.payload.commentId)); + } + await service.processFailedChatRunRetry(one.chatActionId!); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === one.requestId)).toHaveLength(1); + expect(await db.select().from(chatActions).where(and(eq(chatActions.companyId, companyId), + eq(chatActions.kind, "failed_run_retry")))).toHaveLength(1); + if (mode === "retry_failed_successor" || mode === "backlog_51") { + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), + errorCode: "adapter_failed", runtimeMode: "native", nativeIssueId: issueId, nativePhase: "terminal_failure", + }).where(eq(heartbeatRuns.id, parent.runId!)); + await db.update(agentWakeupRequests).set({ status: "failed" }).where(eq(agentWakeupRequests.id, one.chatActionId!)); + await db.update(issues).set({ executionRunId: null, status: "blocked" }).where(eq(issues.id, issueId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: parent.runId!, phase: "terminal_failure" }); + await db.insert(heartbeatRunEvents).values({ companyId, agentId: f.fixture.assignedAgentId, runId: parent.runId!, + seq: 1, eventType: "native.local_process_stopped", payload: { processPid: 999999998, processGroupId: null } }); + await service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user"); + const next = await requestTaskRetry({ ...input, observedRunId: parent.runId!, idempotencyKey: randomUUID(), + prepareChatRetry: service.prepareFailedChatRunRetry }); + expect(next.requestId).not.toBe(one.requestId); + expect(await service.processFailedChatRunRetry(next.chatActionId!)).toMatchObject({ status: "queued" }); + const [second] = await readAgentWakeupRequestsById(next.chatActionId!); + expect((await readHeartbeatRunsById(second.runId!))[0]).toMatchObject({ + retryOfRunId: parent.runId, contextSnapshot: { + wakeCommentIds: (await readHeartbeatRunsById(parent.runId!))[0].contextSnapshot!.wakeCommentIds, + }, + }); + } + } finally { await f.service.shutdown(); } + }, + ); + + it("leaves saved GitHub input pending when a persisted Retry wait loses board authority", async () => { + const f = await currentGitHubTaskRetryFixture(true); + try { + const companyId = f.fixture.companyId, issueId = f.issue.id, actorId = `retry-operator-${companyId}`; + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: actorId, + status: "active", membershipRole: "operator" }); + const intent = await requestTaskRetry({ db, companyId, issueId, actorId, actorSource: "cloud_tenant", + observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry }); + let savedOptions: Parameters[1]; + f.wakeup.mockImplementationOnce(async (_agentId, options) => { + savedOptions = options; + const request = options!.durableChatRequest!; + // Persist the pre-handoff wait as a server interrupted before queue adoption would. + await db.insert(agentWakeupRequests).values({ id: request.id, companyId, agentId: request.agentId, + source: options!.source, triggerDetail: options!.triggerDetail, reason: options!.reason, + requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId, + requestedAt: request.requestedAt, idempotencyKey: request.idempotencyKey, + status: "deferred_issue_execution", payload: { ...options!.payload, + taskRetryRequestId: intent.requestId, _paperclipWakeContext: options!.contextSnapshot } }); + return null; + }); + expect(await f.service.processFailedChatRunRetry(intent.chatActionId!)).toMatchObject({ status: "deferred" }); + await db.update(companyMemberships).set({ membershipRole: "viewer" }).where(and( + eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, actorId))); + expect(await heartbeatService(db).wakeup(f.fixture.assignedAgentId, savedOptions!)).toBeNull(); + expect((await readAgentWakeupRequestsById(f.pending!.id))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === intent.requestId)).toHaveLength(0); + await db.update(companyMemberships).set({ membershipRole: "operator" }).where(and( + eq(companyMemberships.companyId, companyId), eq(companyMemberships.principalId, actorId))); + await heartbeatService(db).wakeup(f.fixture.assignedAgentId, savedOptions!); + const [parent] = await readAgentWakeupRequestsById(intent.chatActionId!); + expect(parent.runId).toBeTruthy(); + expect((await readAgentWakeupRequestsById(f.pending!.id))[0]).toMatchObject({ status: "coalesced", runId: parent.runId }); + } finally { await f.service.shutdown(); } + }); + + it.each(["changed_bot", "disabled_repository", "retargeted_repository", "revoked_sender", "paused_endpoint", + "edited_source", "deleted_source", "reassigned_task", "second_reconnect", "later_message_wrong_bot", "later_message_deleted", "later_message_revoked_policy"] as const)( + "keeps current GitHub task Retry authority gated: %s", async gate => { + const f = await currentGitHubTaskRetryFixture(true, gate.startsWith("later_message") ? 2 : 1); + try { + const companyId = f.fixture.companyId, issueId = f.issue.id; + await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user"); + const input = { db, companyId, issueId, actorId: "owner-user", actorSource: "cloud_tenant", + observedRunId: f.runId, idempotencyKey: randomUUID(), prepareChatRetry: f.service.prepareFailedChatRunRetry }; + if (gate === "later_message_revoked_policy") { + const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.id, f.conversation.resourceId!)); + await db.update(chatEndpointResources).set({ availability: "available", + metadata: { ...resource.metadata, providerRepositoryId: "418" }, + }).where(eq(chatEndpointResources.id, resource.id)); + await db.insert(chatGitHubConfigurations).values({ companyId, endpointId: f.endpoint.id, + updatedByUserId: "owner-user", configuration: { version: 1, toolsEnabled: true, responsibleUserId: "owner-user", + memberAccess: "all_linked", people: [], repositories: {}, + defaults: { ...defaultGitHubReviewPolicy(), invocation: "linked_authors", events: ["opened", "synchronize"] }, + } }); + const sources = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id), + eq(chatActions.kind, "inbound_wakeup"), sql`${chatActions.id} <> ${f.action.id}::uuid`)) + .orderBy(asc(chatActions.createdAt), asc(chatActions.id)); + for (const [index, source] of sources.entries()) { + const context = { event: index === 0 ? "opened" : "synchronize", deliveryId: source.deliveryId, + repositoryId: "418", repository: resource.providerResourceId, + pullNumber: 7, title: "Saved review request", body: "", baseSha: "a".repeat(40), headSha: "b".repeat(40), + baseBranch: "master", author: { id: "42", login: "octocat", isBot: false }, + sender: { id: "42", login: "octocat" }, draft: false, labels: [] }; + await db.update(chatDeliveries).set({ + normalizedEvent: sql`${chatDeliveries.normalizedEvent} || ${JSON.stringify({ githubAutomatic: { context } })}::jsonb`, + }).where(eq(chatDeliveries.id, source.deliveryId!)); + } + } + const accepted = await requestTaskRetry(input); + if (gate.startsWith("later_message")) { + const [later] = await db.select().from(chatActions).where(and(eq(chatActions.conversationId, f.conversation.id), + eq(chatActions.kind, "inbound_wakeup"), sql`${chatActions.id} not in (${f.action.id}::uuid, ${f.pending!.id}::uuid)`)); + if (gate === "later_message_wrong_bot") await db.update(chatDeliveries).set({ + normalizedEvent: sql`jsonb_set(${chatDeliveries.normalizedEvent}, '{githubAuthority,botExternalId}', ${JSON.stringify(randomUUID())}::jsonb)`, + }).where(eq(chatDeliveries.id, later.deliveryId!)); + else if (gate === "later_message_deleted") await db.update(issueComments).set({ deletedAt: new Date() }) + .where(eq(issueComments.id, String(later.payload.commentId))); + else await db.update(chatGitHubConfigurations).set({ revision: sql`${chatGitHubConfigurations.revision} + 1`, + configuration: sql`jsonb_set(${chatGitHubConfigurations.configuration}, '{defaults,events}', '["opened"]'::jsonb)`, + }).where(eq(chatGitHubConfigurations.endpointId, f.endpoint.id)); + } + if (gate === "changed_bot") await db.update(chatEndpoints).set({ botExternalId: randomUUID() }) + .where(eq(chatEndpoints.id, f.endpoint.id)); + if (gate === "disabled_repository") await db.update(chatEndpointResources).set({ enabled: false }) + .where(eq(chatEndpointResources.id, f.conversation.resourceId!)); + if (gate === "retargeted_repository") await db.update(chatEndpointResources).set({ providerResourceId: "another/repository" }) + .where(eq(chatEndpointResources.id, f.conversation.resourceId!)); + if (gate === "revoked_sender") await db.update(chatIdentityLinks).set({ status: "revoked" }) + .where(and(eq(chatIdentityLinks.endpointId, f.endpoint.id), eq(chatIdentityLinks.principalId, f.pending!.principalId!))); + if (gate === "paused_endpoint") await db.update(chatEndpoints).set({ status: "paused" }) + .where(eq(chatEndpoints.id, f.endpoint.id)); + if (gate === "edited_source") await db.update(issueComments).set({ body: "Changed after Retry", updatedAt: new Date() }) + .where(eq(issueComments.id, String(f.pending!.payload.commentId))); + if (gate === "deleted_source") await db.update(issueComments).set({ deletedAt: new Date() }) + .where(eq(issueComments.id, String(f.pending!.payload.commentId))); + if (gate === "reassigned_task") await db.update(issues).set({ assigneeAgentId: f.fixture.replacementAgentId }) + .where(eq(issues.id, issueId)); + if (gate === "second_reconnect") await f.service.configure(f.endpoint.id, { action: "reconnect" }, "owner-user"); + expect(await f.service.processFailedChatRunRetry(accepted.chatActionId!)).toMatchObject({ status: "failed" }); + expect((await readAgentWakeupRequestsById(f.pending!.id))[0]).toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect((await readHeartbeatRunsByCompanyId(companyId)) + .filter(run => run.contextSnapshot?.taskRetryRequestId === accepted.requestId)).toHaveLength(0); + const [settled] = await db.select().from(chatActions).where(eq(chatActions.id, accepted.chatActionId!)); + expect(settled.result?.taskRetryInputReconciled).toBe(true); + const [intentReceipt] = await readAgentWakeupRequestsById(accepted.requestId!); + await f.service.processFailedChatRunRetry(accepted.chatActionId!); + expect((await readAgentWakeupRequestsById(accepted.requestId!))[0].updatedAt).toEqual(intentReceipt.updatedAt); + if (gate === "second_reconnect") { + const next = await requestTaskRetry({ ...input, idempotencyKey: randomUUID() }); + expect(next.requestId).not.toBe(accepted.requestId); + expect(await f.service.processFailedChatRunRetry(next.chatActionId!)).toMatchObject({ status: "queued" }); + } + } finally { await f.service.shutdown(); } + }, + ); + it.each([ "settled", "distinct_account", diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index e53608b49de..fc41debb711 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,4 @@ +import { credentialFingerprint } from "./chat-channels/credential-fingerprint.js"; import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; import { resolveInboundConversationBinding } from "./chat-channels/inbound/conversation-binding.js"; @@ -312,20 +313,6 @@ function runtimeGeneration(setup: ChatEndpointSetupState): number { return Number.isSafeInteger(value) && value >= 0 ? value : 0; } -function credentialFingerprint(refs: ToolCredentialSecretRef[]): string { - const stable = refs - .map((ref) => ({ - configPath: ref.configPath, - secretId: ref.secretId, - versionSelector: ref.versionSelector ?? "latest", - })) - .sort((left, right) => - `${left.configPath}:${left.secretId}:${left.versionSelector}`.localeCompare( - `${right.configPath}:${right.secretId}:${right.versionSelector}`, - ), - ); - return createHash("sha256").update(JSON.stringify(stable)).digest("hex"); -} type ConversationRow = typeof chatConversations.$inferSelect; type DeliveryRow = typeof chatDeliveries.$inferSelect; @@ -847,7 +834,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { lockTelegramCallbackNoticeAuthority } = createTelegramCallbackNoticeAuthority({ nonDirectDestinationAllowed, lockCurrentPrincipalAuthorization }); const { stageProviderEffect, processProviderEffect, processPendingProviderEffects } = createProviderEffects(db, { endpointRecord, runtimeCallbackEndpoint, withCredentialMutationLease, runtimeFor, invalidateRuntime, lockCurrentPrincipalAuthorization, redactError, MAX_ERROR_TEXT, nonDirectDestinationAllowed, lockTelegramCallbackNoticeAuthority }); const { publicationFiles } = createOutboundAttachments(db, { get storage() { return options.storage; } }); - const { authorizeInboundWakeup, authorizeInboundWorkStart, authorizeFastResponse } = createInboundWakeAuthority(db, { runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization }); + const { authorizeInboundWakeup, authorizeGitHubTaskRetryBatch, authorizeInboundWorkStart, authorizeFastResponse } = createInboundWakeAuthority(db, { runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization }); const { ingestAttachments, attachmentOmissionDetail } = createInboundAttachments(db, { authorizeInboundWakeup, lifecycleRuntimeFence, sanitizeFilename, issuesSvc, isExternalActionAuthorizationChange, redactError, get storage() { return options.storage; } }); const { stageInboundWakeup, acceptInboundWakeup, settleRejectedInboundWakeups, processInboundWakeup } = createInboundWakeups(db, { authorizeInboundWakeup, authorizeInboundWorkStart, attachmentOmissionDetail, isExternalActionAuthorizationChange, redactError, fetchImpl, get heartbeat() { return options.heartbeat; } }); const { settleTelegramAttachmentOnlyFailure } = createTelegramAttachmentFailure(db, { @@ -1203,6 +1190,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { pendingFailedRetries, } = createChatFailedRunRetry(db, { authorizeInboundWakeup, + authorizeGitHubTaskRetryBatch, isExternalActionAuthorizationChange, inboundWakeScope, lifecycleRuntimeFence, diff --git a/server/src/services/chat-channels/credential-fingerprint.ts b/server/src/services/chat-channels/credential-fingerprint.ts new file mode 100644 index 00000000000..57f472140c7 --- /dev/null +++ b/server/src/services/chat-channels/credential-fingerprint.ts @@ -0,0 +1,10 @@ +import { createHash } from "node:crypto"; +import type { ToolCredentialSecretRef } from "@paperclipai/shared"; + +export function credentialFingerprint(refs: ToolCredentialSecretRef[]): string { + const stable = refs.map(ref => ({ configPath: ref.configPath, secretId: ref.secretId, + versionSelector: ref.versionSelector ?? "latest" })).sort((left, right) => + `${left.configPath}:${left.secretId}:${left.versionSelector}`.localeCompare( + `${right.configPath}:${right.secretId}:${right.versionSelector}`)); + return createHash("sha256").update(JSON.stringify(stable)).digest("hex"); +} diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index 6310d448a66..c18a15bdcb8 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -3,6 +3,8 @@ import { readTaskChatRetryIntent, taskChatRetryCanAwaitRecovery, taskChatRetryEx import { notifyChatActionWork, notifyChatPublicationWork } from "../chat-work-notifications.js"; import { FAILED_RUN_RETRY_STALE_MS, dueChatAction } from "../chat-action-work.js"; import { readChatAuthorizationRun } from "./authorization-run-read.js"; +import { selectTaskRetryInput, claimTaskRetryInput, settleTaskRetryInput, releaseUnadmittedTaskRetryInput, taskRetryInputSchema, type TaskRetryInput } from "./task-retry-input.js"; +import type { TaskRetryWakeAuthority } from "./github/task-retry-authorization.js"; import { createHash, @@ -93,14 +95,16 @@ type RetryRuntimeFence = { credentialFingerprint: string; generation: number }; export type ChatFailedRetrySource = Awaited["failedChatRetrySource"]>>; type ChatFailedRunRetryOptions = { + authorizeGitHubTaskRetryBatch: ReturnType["authorizeGitHubTaskRetryBatch"]; authorizeInboundWakeup: ( database: DbOrTransaction, action: typeof chatActions.$inferSelect, - notice?: { nonblocking: true; terminal: boolean }, + notice?: { nonblocking: true; terminal: boolean; taskRetry?: TaskRetryWakeAuthority }, ) => Promise<{ conversation: Pick; endpoint: Pick; delivery: DeliveryRow; + retryRuntimeFence?: RetryRuntimeFence; }>; isExternalActionAuthorizationChange: (error: unknown) => boolean; inboundWakeScope: (action: typeof chatActions.$inferSelect) => string; @@ -112,6 +116,7 @@ type ChatFailedRunRetryOptions = { export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOptions) { const { authorizeInboundWakeup, + authorizeGitHubTaskRetryBatch, isExternalActionAuthorizationChange, inboundWakeScope, lifecycleRuntimeFence, @@ -131,14 +136,14 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti } } - function retryCommentIds(context: Record): string[] { + function retryCommentIds(context: Record, taskRetryInput?: TaskRetryInput): string[] { const ids = context.wakeCommentIds ?? (context.wakeCommentId ? [context.wakeCommentId] : []); if ( !Array.isArray(ids) || ids.length === 0 || - ids.length > 50 || + (!taskRetryInput && ids.length > 50) || ids.some((id) => typeof id !== "string" || !isUuidLike(id)) || new Set(ids).size !== ids.length || (context.wakeCommentId !== undefined && @@ -168,6 +173,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti failedRunId: string; sourceScopeSha256: string; }>; + taskRetryInput?: TaskRetryInput; sourceScopeSha256: string; }; function hashFailedRetrySource( @@ -651,6 +657,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti runId?: string; publication?: boolean; taskRetryIntentId?: string; + taskRetryInput?: TaskRetryInput; + taskRetryReauthorization?: TaskRetryWakeAuthority; /** Internal selector for already committed output, never retry admission. */ committedResponse?: { resultId: string; @@ -734,7 +742,22 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti (context.taskKey !== issue.identifier && context.taskKey !== issue.id) ) throw failedChatRetryDenied(); - const commentIds = retryCommentIds(context); + const selection = input.taskRetryInput && taskRetryInputSchema.safeParse(input.taskRetryInput); + if (selection && !selection.success) throw failedChatRetryDenied(); + const selectedInput = selection?.success ? selection.data : undefined; + if (selectedInput && (!input.taskRetryIntentId || provider !== "github")) throw failedChatRetryDenied(); + const reauthorization = input.taskRetryReauthorization ?? (selectedInput ? { + intentId: input.taskRetryIntentId!, sourceRunId: failedRun.id, runtimeFence: selectedInput.runtimeFence, + } : undefined); + const pendingSources = selectedInput?.sourceActionIds.length ? await tx.select().from(chatActions).where(and( + eq(chatActions.companyId, input.companyId), eq(chatActions.kind, "inbound_wakeup"), + sql`${chatActions.payload}->>'issueId' = ${issue.id}`, + inArray(chatActions.id, selectedInput.sourceActionIds), + )) : null; + if (pendingSources && pendingSources.length !== selectedInput!.sourceActionIds.length) throw failedChatRetryDenied(); + const pendingById = new Map(pendingSources?.map(source => [source.id, source])); + const commentIds = pendingSources ? selectedInput!.sourceActionIds.map(id => + String(pendingById.get(id)!.payload.commentId)) : retryCommentIds(context, selectedInput); const [accepted] = await tx .select() .from(nativeRunResults) @@ -868,7 +891,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti answerDelivery ) throw failedChatRetryDenied(); - if (context.chatFailedRunRetry) { + if (context.chatFailedRunRetry && !pendingSources) { const hint = context.chatFailedRunRetry as Record; const [priorIntent] = await tx .select() @@ -914,6 +937,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti committedResponse: undefined, taskRetryIntentId: typeof priorIntent.payload.taskRetryIntentId === "string" ? priorIntent.payload.taskRetryIntentId : undefined, + taskRetryInput: priorIntent.payload.taskRetryInput as TaskRetryInput | undefined, + retryActionId: priorIntent.id, + taskRetryReauthorization: reauthorization, }, ancestors, ); @@ -945,6 +971,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti return hashFailedRetrySource({ ...parentScope, failedRunId: failedRun.id, + ...(selectedInput ? { taskRetryInput: { ...selectedInput, runtimeFence: reauthorization?.runtimeFence } } : {}), retryAncestors: [ ...parent.retryAncestors, { @@ -958,20 +985,21 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti // Include every receipt that fed this run, including a deferred owner's // coalesced children. A mutable latest-comment hint cannot drop a sibling. const admittedBatch = - await tx.execute(sql`select source.id, source.payload->>'commentId' as comment_id + pendingSources ? [] : await tx.execute(sql`select source.id, source.payload->>'commentId' as comment_id from agent_wakeup_requests receipt join chat_actions source on source.id = receipt.id where receipt.company_id = ${input.companyId}::uuid and receipt.agent_id = ${input.agentId}::uuid and (receipt.run_id = ${failedRun.id}::uuid or receipt.id = ${failedRun.wakeupRequestId}::uuid or receipt.payload->>'coalescedIntoWakeupRequestId' = ${failedRun.wakeupRequestId}) - and source.company_id = ${input.companyId}::uuid and source.kind = 'inbound_wakeup' limit 51`); + and source.company_id = ${input.companyId}::uuid and source.kind = 'inbound_wakeup' + ${selectedInput ? sql`` : sql`limit 51`}`); if ( - admittedBatch.length !== commentIds.length || + !pendingSources && (admittedBatch.length !== commentIds.length || admittedBatch.some( (entry) => !commentIds.includes(String(entry.comment_id)), - ) + )) ) throw failedChatRetryDenied(); - const sources = await tx + const sources = pendingSources ?? await tx .select() .from(chatActions) .where( @@ -986,8 +1014,9 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti ), ); if (sources.length !== commentIds.length) throw failedChatRetryDenied(); + const sourcesByCommentId = new Map(sources.map(source => [source.payload.commentId, source])); const ordered = commentIds.map( - (id) => sources.find((action) => action.payload.commentId === id)!, + (id) => sourcesByCommentId.get(id)!, ); const first = ordered[0]!; if ( @@ -1011,39 +1040,53 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti ); if (!identity[0]?.locked) throw new NativeChatReviewPresentationContentionError(); + const actionIds = ordered.map(action => action.id); + const currentActions = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, input.companyId), + inArray(chatActions.id, actionIds))).orderBy(asc(chatActions.id)).for("update", { noWait: true }); + const deliveries = await tx.select().from(chatDeliveries).where(and(eq(chatDeliveries.companyId, input.companyId), + inArray(chatDeliveries.id, ordered.map(action => action.deliveryId!)))) + .orderBy(asc(chatDeliveries.id)).for("update", { noWait: true }); + const comments = await tx.select().from(issueComments).where(and(eq(issueComments.companyId, input.companyId), + eq(issueComments.issueId, issue.id), inArray(issueComments.id, commentIds))) + .orderBy(asc(issueComments.id)).for("share", { noWait: true }); + const receipts = await tx.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, input.companyId), + inArray(agentWakeupRequests.id, [...actionIds, failedRun.wakeupRequestId!, ...(input.retryActionId ? [input.retryActionId] : [])]))) + .orderBy(asc(agentWakeupRequests.id)).for("share", { noWait: true }); + const actionsById = new Map(currentActions.map(action => [action.id, action])); + const deliveriesById = new Map(deliveries.map(delivery => [delivery.id, delivery])); + const commentsById = new Map(comments.map(comment => [comment.id, comment])); + const receiptsById = new Map(receipts.map(receipt => [receipt.id, receipt])); + const lifecycleRows = await tx.select({ + target: sql`${chatDeliveries.normalizedEvent}->'message'->>'targetProviderEventId'`, + context: sql>`${chatDeliveries.normalizedEvent}->'runtimeContext'`, + }).from(chatDeliveries).where(and(eq(chatDeliveries.companyId, input.companyId), + eq(chatDeliveries.endpointId, first.endpointId), + or(isNull(chatDeliveries.conversationId), eq(chatDeliveries.conversationId, first.conversationId)), + ne(chatDeliveries.state, "filtered"), inArray(chatDeliveries.eventKind, ["message_updated", "message_deleted", "message_restored"]), + inArray(sql`${chatDeliveries.normalizedEvent}->'message'->>'targetProviderEventId'`, deliveries.map(delivery => delivery.providerEventId)))); + const lifecycleKey = (target: string, context: Record | null) => + JSON.stringify([target, Object.entries(context ?? {}).sort(([left], [right]) => left.localeCompare(right))]); + const invalidated = new Set(lifecycleRows.map(row => lifecycleKey(row.target, row.context))); + const batch = reauthorization ? await authorizeGitHubTaskRetryBatch(tx, ordered, { + nonblocking: true, terminal: input.publication === true, taskRetry: reauthorization, + }).catch(error => { if (isExternalActionAuthorizationChange(error)) throw failedChatRetryDenied(); throw error; }) : null; const canonicalSources: Array> = []; let destination: Awaited> | null = null; for (const action of ordered) { - destination = await authorizeInboundWakeup(tx, action, { + destination = batch?.get(action.id) ?? await authorizeInboundWakeup(tx, action, { nonblocking: true, terminal: input.publication === true, + ...(reauthorization ? { taskRetry: reauthorization } : {}), }).catch((error) => { if (isExternalActionAuthorizationChange(error)) throw failedChatRetryDenied(); throw error; }); - const [delivery] = await tx - .select() - .from(chatDeliveries) - .where(eq(chatDeliveries.id, action.deliveryId!)) - .for("update", { noWait: true }); - const [current] = await tx - .select() - .from(chatActions) - .where(eq(chatActions.id, action.id)) - .for("update", { noWait: true }); - const [comment] = await tx - .select() - .from(issueComments) - .where( - and( - eq(issueComments.id, String(action.payload.commentId)), - eq(issueComments.companyId, input.companyId), - eq(issueComments.issueId, issue.id), - ), - ) - .for("share", { noWait: true }); + if (reauthorization && !reauthorization.runtimeFence) reauthorization.runtimeFence = destination.retryRuntimeFence; + const delivery = deliveriesById.get(action.deliveryId!); + const current = actionsById.get(action.id); + const comment = commentsById.get(String(action.payload.commentId)); if ( !current || current.status !== "processed" || @@ -1053,43 +1096,17 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti !comment || comment.deletedAt || comment.updatedAt.getTime() !== comment.createdAt.getTime() || - comment.updatedAt > + (!pendingSources && comment.updatedAt > (input.committedResponse ? accepted!.createdAt - : failedRun.finishedAt!) || + : failedRun.finishedAt!)) || destination.endpoint.provider !== provider || JSON.stringify(lifecycleRuntimeFence(delivery)) !== JSON.stringify(lifecycleRuntimeFence(destination.delivery)) ) throw failedChatRetryDenied(); - const [lifecycle] = await tx - .select({ id: chatDeliveries.id }) - .from(chatDeliveries) - .where( - and( - eq(chatDeliveries.companyId, input.companyId), - eq(chatDeliveries.endpointId, first.endpointId), - or( - isNull(chatDeliveries.conversationId), - eq(chatDeliveries.conversationId, first.conversationId), - ), - ne(chatDeliveries.state, "filtered"), - sql`${chatDeliveries.normalizedEvent}->'runtimeContext' = ${JSON.stringify(lifecycleRuntimeFence(delivery))}::jsonb`, - inArray(chatDeliveries.eventKind, [ - "message_updated", - "message_deleted", - "message_restored", - ]), - sql`${chatDeliveries.normalizedEvent}->'message'->>'targetProviderEventId' = ${delivery.providerEventId}`, - ), - ) - .limit(1); - if (lifecycle) throw failedChatRetryDenied(); - const [receipt] = await tx - .select() - .from(agentWakeupRequests) - .where(eq(agentWakeupRequests.id, action.id)) - .for("share", { noWait: true }); + if (invalidated.has(lifecycleKey(delivery.providerEventId, lifecycleRuntimeFence(delivery)))) throw failedChatRetryDenied(); + const receipt = receiptsById.get(action.id); if (!receipt) throw failedChatRetryDenied(); assertFailedRetryReceipt( createDurableChatWakeupRequest({ @@ -1107,6 +1124,17 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti }), receipt, ); + if (pendingSources) { + const direct = receipt.status === "deferred_issue_execution" && !receipt.runId + && receipt.payload?.coalescedIntoWakeupRequestId === undefined; + const owned = input.retryActionId && receipt.status === "coalesced" + && receipt.payload?.coalescedIntoWakeupRequestId === input.retryActionId; + if (!direct && !owned) throw failedChatRetryDenied(); + if (owned && receipt.runId) { + const owner = receiptsById.get(input.retryActionId!); + if (owner?.runId !== receipt.runId) throw failedChatRetryDenied(); + } + } let owner = receipt; const seen = new Set(); for ( @@ -1126,25 +1154,16 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti ) throw failedChatRetryDenied(); seen.add(ownerId); - const [next] = await tx - .select() - .from(agentWakeupRequests) - .where( - and( - eq(agentWakeupRequests.id, ownerId), - eq(agentWakeupRequests.companyId, input.companyId), - ), - ) - .for("share", { noWait: true }); + const next = receiptsById.get(ownerId); if (!next) throw failedChatRetryDenied(); owner = next; } if ( - owner.id !== failedRun.wakeupRequestId || + (!pendingSources && (owner.id !== failedRun.wakeupRequestId || owner.runId !== failedRun.id || owner.agentId !== input.agentId || owner.requestedByActorType !== receipt.requestedByActorType || - owner.requestedByActorId !== receipt.requestedByActorId || + owner.requestedByActorId !== receipt.requestedByActorId)) || ["skipped", "cancelled"].includes(receipt.status) ) throw failedChatRetryDenied(); @@ -1158,7 +1177,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti bodySha256: createHash("sha256").update(comment.body).digest("hex"), omissionReasons: action.payload.attachmentOmissionReasons ?? {}, receiptId: receipt.id, - ownerId: owner.id, + ownerId: pendingSources ? action.id : owner.id, }); } if (!destination) throw failedChatRetryDenied(); @@ -1204,6 +1223,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti | "system", requestedByActorId: String(first.payload.requestedByActorId), retryAncestors: [], + ...(selectedInput ? { taskRetryInput: { ...selectedInput, + runtimeFence: selectedInput.runtimeFence ?? destination.retryRuntimeFence } } : {}), }; return hashFailedRetrySource(scope); } @@ -1240,7 +1261,8 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti retryIntentId?: string; }, ): Promise<{ actionId: string; issueId: string }> { - const source = await failedChatRetrySource(tx, { ...input, taskRetryIntentId: input.retryIntentId }); + const taskRetryInput = input.retryIntentId ? await selectTaskRetryInput(tx, input) : undefined; + const source = await failedChatRetrySource(tx, { ...input, taskRetryIntentId: input.retryIntentId, taskRetryInput }); const providerActionId = input.retryIntentId ? `task-retry:${input.retryIntentId}` : `failed-run-retry:${source.failedRunId}`; const [existing] = await tx @@ -1351,6 +1373,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti failedRunId: snapshot.payload.failedRunId, taskRetryIntentId: typeof snapshot.payload.taskRetryIntentId === "string" ? snapshot.payload.taskRetryIntentId : undefined, + taskRetryInput: snapshot.payload.taskRetryInput as TaskRetryInput | undefined, }); if (typeof snapshot.payload.taskRetryIntentId === "string" && !await readTaskChatRetryIntent(tx as unknown as Db, { companyId: input.companyId, issueId: input.issueId, agentId: input.agentId, @@ -1444,7 +1467,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti context.source !== `chat:${source.provider}` || context.taskKey !== source.taskKey || context.retryOfRunId !== source.failedRunId || - JSON.stringify(retryCommentIds(context)) !== + JSON.stringify(retryCommentIds(context, source.taskRetryInput)) !== JSON.stringify(source.commentIds) || hint?.version !== 1 || hint.actionId !== source.intent.id || @@ -1518,8 +1541,19 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti }; } + async function reconcileFailedTaskRetry(action: typeof chatActions.$inferSelect) { + if (!["failed", "cancelled"].includes(action.status) || !action.payload.taskRetryIntentId || + action.result?.taskRetryInputReconciled === true) return action; + await releaseUnadmittedTaskRetryInput(db, { companyId: action.companyId, retryActionId: action.id }); + const [settled] = await db.update(chatActions).set({ + result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || '{"taskRetryInputReconciled":true}'::jsonb`, + }).where(and(eq(chatActions.companyId, action.companyId), eq(chatActions.id, action.id), + inArray(chatActions.status, ["failed", "cancelled"]))).returning(); + return settled ?? action; + } + async function claimFailedChatRunRetry(actionId: string) { - const [candidate] = await db + const [snapshot] = await db .select() .from(chatActions) .where( @@ -1528,7 +1562,10 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti eq(chatActions.kind, "failed_run_retry"), ), ); - if (!candidate) throw notFound("Chat retry request not found"); + if (!snapshot) throw notFound("Chat retry request not found"); + // Reconcile once at settlement; a crash before the marker is safely + // retried here. Status projection itself performs no writes or row locks. + const candidate = await reconcileFailedTaskRetry(snapshot); const now = new Date(); if ( isShuttingDown() || @@ -1571,7 +1608,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti ) .returning(); }); - return failedChatRetryReceipt(current ?? claimed); + return failedChatRetryReceipt(await reconcileFailedTaskRetry(current ?? claimed)); }; try { const scope = { @@ -1644,6 +1681,12 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti true, // The durable wake callback validates admission; later phases recheck containment. ); }, + onAdmitted: async (tx, receipt) => { + if (!source.taskRetryInput) return; + await claimTaskRetryInput(tx, { companyId: claimed.companyId, + retryActionId: claimed.id, sourceActionIds: source.taskRetryInput.sourceActionIds }); + await settleTaskRetryInput(tx, { companyId: claimed.companyId, retryActionId: claimed.id, runId: receipt.runId }); + }, }), }); } @@ -1652,6 +1695,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti .from(agentWakeupRequests) .where(eq(agentWakeupRequests.id, claimed.id)); if (!receipt) throw new Error("chat_retry_receipt_not_committed"); + await settleTaskRetryInput(db, { companyId: claimed.companyId, retryActionId: claimed.id, runId: receipt.runId }); return settle( ["skipped", "failed", "cancelled"].includes(receipt.status) ? "failed" diff --git a/server/src/services/chat-channels/github/task-preparation.ts b/server/src/services/chat-channels/github/task-preparation.ts index 403b45c537e..d44ce61de02 100644 --- a/server/src/services/chat-channels/github/task-preparation.ts +++ b/server/src/services/chat-channels/github/task-preparation.ts @@ -37,7 +37,7 @@ export async function retainGitHubDeliveryAuthority( linkedUserId: string | null, taskUserId: string | null, ): Promise { const guest = !linkedUserId; - await taskTx.update(chatDeliveries).set({ normalizedEvent: sql`${chatDeliveries.normalizedEvent} || ${JSON.stringify({ githubAuthority: { guest, responsibleUserId: taskUserId, sponsorUserId: guest ? taskEndpoint.sponsorUserId : null } })}::jsonb` }).where(eq(chatDeliveries.id, deliveryId)); + await taskTx.update(chatDeliveries).set({ normalizedEvent: sql`${chatDeliveries.normalizedEvent} || ${JSON.stringify({ githubAuthority: { guest, botExternalId: taskEndpoint.botExternalId, responsibleUserId: taskUserId, sponsorUserId: guest ? taskEndpoint.sponsorUserId : null } })}::jsonb` }).where(eq(chatDeliveries.id, deliveryId)); } export async function retainGitHubReviewReceipt(taskTx: DbOrTransaction, endpoint: EndpointRow, issueId: string, deliveryId: string, githubAutomatic: AutomaticRequest): Promise { diff --git a/server/src/services/chat-channels/github/task-retry-authorization.ts b/server/src/services/chat-channels/github/task-retry-authorization.ts new file mode 100644 index 00000000000..52e51f38ebf --- /dev/null +++ b/server/src/services/chat-channels/github/task-retry-authorization.ts @@ -0,0 +1,126 @@ +import { and, eq, inArray } from "drizzle-orm"; +import { chatActions, chatConversations, chatDeliveries, chatEndpointResources, chatMessageLinks, issueComments, toolConnections } from "@paperclipai/db"; +import { forbidden } from "../../../errors.js"; +import type { DbOrTransaction, EndpointRow, LifecycleRuntimeFence } from "../runtime-types.js"; +import { credentialFingerprint } from "../credential-fingerprint.js"; +import { readTaskChatRetryIntent } from "../task-retry-authority.js"; + +export type TaskRetryWakeAuthority = { + intentId: string; + sourceRunId: string; + runtimeFence?: LifecycleRuntimeFence; +}; + +export function githubTaskRetryAppMatches(endpoint: EndpointRow, delivery: typeof chatDeliveries.$inferSelect, + response: { payload: Record } | undefined) { + const original = delivery.normalizedEvent.githubAuthority as { botExternalId?: unknown } | undefined; + return original?.botExternalId !== undefined ? original.botExternalId === endpoint.botExternalId + : response?.payload.botExternalId === endpoint.botExternalId && response.payload.agentId === endpoint.assignedAgentId; +} + +export function githubTaskRetryDestinationMatches(delivery: typeof chatDeliveries.$inferSelect, + conversation: typeof chatConversations.$inferSelect, resource: typeof chatEndpointResources.$inferSelect | null) { + const originalResource = delivery.normalizedEvent.resource as { providerResourceId?: unknown } | undefined; + const originalThread = delivery.normalizedEvent.conversation as { externalThreadId?: unknown } | undefined; + const automatic = (delivery.normalizedEvent.githubAutomatic ?? delivery.normalizedEvent.githubIssue) as + { context?: { repositoryId?: unknown } } | undefined; + return resource && originalResource?.providerResourceId === resource.providerResourceId + && originalThread?.externalThreadId === conversation.externalThreadId + && (automatic?.context?.repositoryId === undefined + || automatic.context.repositoryId === resource.metadata.providerRepositoryId); +} + +export async function githubTaskRetryRuntimeFence(tx: DbOrTransaction, endpoint: EndpointRow, + action: typeof chatActions.$inferSelect, delivery: typeof chatDeliveries.$inferSelect, + authority: TaskRetryWakeAuthority): Promise { + const deny = () => forbidden("This accepted chat message is no longer authorized to start work", + { code: "chat_action_authorization_changed" }); + const retry = await readTaskChatRetryIntent(tx as Parameters[0], { + companyId: action.companyId, issueId: String(action.payload.issueId), + agentId: String(action.payload.agentId), sourceRunId: authority.sourceRunId, intentId: authority.intentId, + }); + if (!retry || endpoint.provider !== "github" || !endpoint.botExternalId || action.kind !== "inbound_wakeup" + || action.status !== "processed" || delivery.state !== "processed") throw deny(); + const [response] = await tx.select({ payload: chatActions.payload }).from(chatActions).where(and( + eq(chatActions.companyId, action.companyId), eq(chatActions.endpointId, endpoint.id), + eq(chatActions.conversationId, action.conversationId!), eq(chatActions.deliveryId, delivery.id), + eq(chatActions.kind, "github_response_comment"), + )); + if (!githubTaskRetryAppMatches(endpoint, delivery, response)) throw deny(); + const [connection] = await tx.select({ refs: toolConnections.credentialSecretRefs, + enabled: toolConnections.enabled, status: toolConnections.status }).from(toolConnections).where(and( + eq(toolConnections.companyId, endpoint.companyId), eq(toolConnections.id, endpoint.connectionId), + )); + if (!connection?.enabled || connection.status !== "active") throw deny(); + const generation = Number((endpoint.setup as { runtimeGeneration?: unknown }).runtimeGeneration ?? 0); + if (!Number.isSafeInteger(generation) || generation < 0) throw deny(); + return authority.runtimeFence ?? { generation, credentialFingerprint: credentialFingerprint(connection.refs ?? []) }; +} + +type GitHubRetryBinding = { + endpoint: EndpointRow; + conversation: typeof chatConversations.$inferSelect; + resource: typeof chatEndpointResources.$inferSelect | null; + delivery: typeof chatDeliveries.$inferSelect; + retryRuntimeFence?: LifecycleRuntimeFence; +}; + +export function createGitHubTaskRetryBatchAuthority(authorize: ( + tx: DbOrTransaction, action: typeof chatActions.$inferSelect, + notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority }, +) => Promise) { + async function authorizeGitHubTaskRetryBatch(tx: DbOrTransaction, actions: Array, + notice: { nonblocking: true; terminal: boolean; taskRetry: TaskRetryWakeAuthority }) { + const deny = () => forbidden("This accepted chat message is no longer authorized to start work", + { code: "chat_action_authorization_changed" }); + const first = actions[0]; + if (!first) throw deny(); + const commonScope = (action: typeof chatActions.$inferSelect) => JSON.stringify([ + action.companyId, action.endpointId, action.conversationId, action.principalId, action.kind, action.status, + ...["version", "issueId", "agentId", "sessionGeneration", "requestedByActorType", "requestedByActorId"] + .map(key => action.payload[key]), + ]); + if (actions.some(action => commonScope(action) !== commonScope(first))) throw deny(); + const deliveries = await tx.select().from(chatDeliveries).where(and(eq(chatDeliveries.companyId, first.companyId), + inArray(chatDeliveries.id, actions.map(action => action.deliveryId!)))); + const byDelivery = new Map(deliveries.map(delivery => [delivery.id, delivery])); + const legacy = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, first.companyId), + eq(chatActions.endpointId, first.endpointId), eq(chatActions.conversationId, first.conversationId!), + eq(chatActions.kind, "github_response_comment"), inArray(chatActions.deliveryId, deliveries.map(delivery => delivery.id)))); + const legacyByDelivery = new Map(legacy.map(action => [action.deliveryId, action])); + const links = await tx.select({ deliveryId: chatMessageLinks.deliveryId, commentId: chatMessageLinks.commentId, + authorUserId: issueComments.authorUserId, deletedAt: issueComments.deletedAt }).from(chatMessageLinks) + .innerJoin(issueComments, and(eq(issueComments.id, chatMessageLinks.commentId), + eq(issueComments.companyId, first.companyId), eq(issueComments.issueId, String(first.payload.issueId)))) + .where(and(eq(chatMessageLinks.companyId, first.companyId), eq(chatMessageLinks.endpointId, first.endpointId), + eq(chatMessageLinks.conversationId, first.conversationId!), eq(chatMessageLinks.direction, "inbound"), + inArray(chatMessageLinks.deliveryId, deliveries.map(delivery => delivery.id)))); + const expectedUserId = first.payload.requestedByActorType === "user" ? first.payload.requestedByActorId : null; + const linked = new Set(links.filter(link => (!link.deletedAt || notice.terminal) && link.authorUserId === expectedUserId) + .map(link => `${link.deliveryId}:${link.commentId}`)); + const groups = new Map(); + const authorized = new Map(); + for (const action of actions) { + const delivery = byDelivery.get(action.deliveryId!); + if (!delivery || !linked.has(`${delivery.id}:${action.payload.commentId}`)) throw deny(); + // Reuse only the same live grant within this transaction. Automatic review + // policies with different contexts still receive their own full check. + const grant = JSON.stringify(delivery.normalizedEvent.githubAutomatic ?? delivery.normalizedEvent.githubIssue ?? null); + let current = groups.get(grant); + if (!current) { + current = await authorize(tx, action, notice); + if (current.endpoint.provider !== "github") throw deny(); + notice.taskRetry.runtimeFence ??= current.retryRuntimeFence; + groups.set(grant, current); + } + if (delivery.state !== "processed" || delivery.endpointId !== current.endpoint.id || + delivery.conversationId !== current.conversation.id || delivery.principalId !== action.principalId || + !githubTaskRetryAppMatches(current.endpoint, delivery, legacyByDelivery.get(delivery.id)) || + !githubTaskRetryDestinationMatches(delivery, current.conversation, current.resource)) throw deny(); + authorized.set(action.id, { ...current, delivery }); + } + return authorized; + } + + return authorizeGitHubTaskRetryBatch; +} diff --git a/server/src/services/chat-channels/inbound/wakeup-authorization.ts b/server/src/services/chat-channels/inbound/wakeup-authorization.ts index ecdca242e9b..f4da4fa7cce 100644 --- a/server/src/services/chat-channels/inbound/wakeup-authorization.ts +++ b/server/src/services/chat-channels/inbound/wakeup-authorization.ts @@ -11,6 +11,7 @@ import { agentWakeupRequests, agents, chatActions, chatConversations, chatDelive import { forbidden } from "../../../errors.js"; import { resolveGitHubWakeAuthority, isGitHubAutomaticDelivery } from "../github/wake-authority.js"; import { assertSlackWakeMention } from "../slack/wake-authority.js"; +import { githubTaskRetryRuntimeFence, githubTaskRetryDestinationMatches, createGitHubTaskRetryBatchAuthority, type TaskRetryWakeAuthority } from "../github/task-retry-authorization.js"; import type { createChatInboundAuthority } from "./authorization.js"; import type { createChatMessageLifecycle } from "./lifecycle.js"; type WakeAuthorityOptions = { @@ -41,7 +42,7 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions async function authorizeInboundWakeup( tx: DbOrTransaction, action: typeof chatActions.$inferSelect, - notice?: { nonblocking: true; terminal: boolean }, + notice?: { nonblocking: true; terminal: boolean; taskRetry?: TaskRetryWakeAuthority }, ) { const payload = action.payload; const deny = () => @@ -106,6 +107,8 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions .limit(1) .then((rows) => rows[0] ?? null); const fence = delivery ? lifecycleRuntimeFence(delivery) : null; + const currentFence = endpoint && delivery && notice?.taskRetry + ? await githubTaskRetryRuntimeFence(tx, endpoint, action, delivery, notice.taskRetry) : fence; if ( !endpoint || endpoint.assignedAgentId !== payload.agentId || @@ -114,8 +117,8 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions delivery.endpointId !== endpoint.id || delivery.conversationId !== action.conversationId || delivery.principalId !== action.principalId || - !fence || - !(await runtimeCallbackEndpoint(tx as unknown as DbTransaction, endpoint.id, fence, [ + !currentFence || + !(await runtimeCallbackEndpoint(tx as unknown as DbTransaction, endpoint.id, currentFence, [ "verifying", "active", ])) @@ -160,6 +163,7 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions endpoint, action.principalId, ); + if (notice?.taskRetry && !githubTaskRetryDestinationMatches(delivery, conversation, resource)) throw deny(); if (endpoint.provider === "github") { const automatic = await resolveGitHubWakeAuthority(tx, endpoint, delivery); if (automatic.kind === "denied") throw deny(); @@ -214,9 +218,12 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions source.authorUserId !== expectedUserId ) throw deny(); - return { issue, delivery, endpoint, conversation }; + return { issue, delivery, endpoint, conversation, resource, + ...(notice?.taskRetry && currentFence ? { retryRuntimeFence: currentFence } : {}) }; } + const authorizeGitHubTaskRetryBatch = createGitHubTaskRetryBatchAuthority(authorizeInboundWakeup); + async function authorizeInboundWorkStart( tx: DbOrTransaction, action: typeof chatActions.$inferSelect, @@ -241,5 +248,5 @@ export function createInboundWakeAuthority(db: Db, options: WakeAuthorityOptions attachments: normalized.message?.attachments?.map(a => `${a.name ?? "Attachment"} (${a.mimeType ?? "unknown type"})`) }; } - return { authorizeInboundWakeup, authorizeInboundWorkStart, authorizeFastResponse }; + return { authorizeInboundWakeup, authorizeGitHubTaskRetryBatch, authorizeInboundWorkStart, authorizeFastResponse }; } diff --git a/server/src/services/chat-channels/task-retry-input.ts b/server/src/services/chat-channels/task-retry-input.ts new file mode 100644 index 00000000000..b5fb8a30233 --- /dev/null +++ b/server/src/services/chat-channels/task-retry-input.ts @@ -0,0 +1,95 @@ +import { z } from "zod"; +import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm"; +import { agentWakeupRequests, chatActions, chatConversations, chatEndpoints, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import type { DbOrTransaction } from "./runtime-types.js"; + +export const taskRetryInputSchema = z.object({ + // This is a server-selected snapshot of the durable queue, not client input. + // Truncating it strands older requests or makes the newer-input guard reject + // every Retry. Dispatch already carries the full queue with bounded inline text. + sourceActionIds: z.array(z.string().uuid()).refine(ids => new Set(ids).size === ids.length), + runtimeFence: z.object({ generation: z.number().int().nonnegative(), + credentialFingerprint: z.string().regex(/^[a-f0-9]{64}$/) }).optional(), +}); +export type TaskRetryInput = z.infer; + +export async function selectTaskRetryInput(tx: DbOrTransaction, input: { + companyId: string; issueId: string; agentId: string; failedRunId: string; +}) { + const [conversation] = await tx.select({ id: chatConversations.id }).from(chatConversations) + .innerJoin(chatEndpoints, eq(chatEndpoints.id, chatConversations.endpointId)) + .innerJoin(chatActions, and(eq(chatActions.companyId, chatConversations.companyId), + eq(chatActions.conversationId, chatConversations.id))) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, chatConversations.companyId), + eq(heartbeatRuns.wakeupRequestId, chatActions.id))).where(and( + eq(chatConversations.companyId, input.companyId), eq(chatConversations.issueId, input.issueId), + eq(heartbeatRuns.id, input.failedRunId), eq(heartbeatRuns.agentId, input.agentId), + eq(chatEndpoints.provider, "github"), eq(chatEndpoints.assignedAgentId, input.agentId), + eq(chatEndpoints.externalExecutionPolicy, "restricted"), + )).limit(1); + if (!conversation) return undefined; + const pending = await tx.select({ id: chatActions.id }).from(chatActions) + .innerJoin(agentWakeupRequests, and(eq(agentWakeupRequests.companyId, chatActions.companyId), + eq(agentWakeupRequests.id, chatActions.id))).where(and( + eq(chatActions.companyId, input.companyId), eq(chatActions.conversationId, conversation.id), + eq(chatActions.kind, "inbound_wakeup"), eq(chatActions.status, "processed"), + eq(agentWakeupRequests.agentId, input.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"), + isNull(agentWakeupRequests.runId), + )).orderBy(asc(chatActions.createdAt), asc(chatActions.id)).for("update", { of: agentWakeupRequests }); + return taskRetryInputSchema.parse({ sourceActionIds: pending.map(row => row.id) }); +} + +export async function claimTaskRetryInput(tx: DbOrTransaction, input: { + companyId: string; retryActionId: string; sourceActionIds: string[]; +}) { + if (!input.sourceActionIds.length) return; + await tx.update(agentWakeupRequests).set({ status: "coalesced", updatedAt: new Date(), + payload: sql`coalesce(${agentWakeupRequests.payload}, '{}'::jsonb) || ${JSON.stringify({ + coalescedIntoWakeupRequestId: input.retryActionId })}::jsonb`, + }).where(and(eq(agentWakeupRequests.companyId, input.companyId), + inArray(agentWakeupRequests.id, input.sourceActionIds), eq(agentWakeupRequests.status, "deferred_issue_execution"), + isNull(agentWakeupRequests.runId))); +} + +export async function settleTaskRetryInput(tx: DbOrTransaction, input: { companyId: string; retryActionId: string; runId: string | null }) { + if (!input.runId) return; + await tx.update(agentWakeupRequests).set({ runId: input.runId, updatedAt: new Date() }).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.status, "coalesced"), + isNull(agentWakeupRequests.runId), + sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId' = ${input.retryActionId}`, + )); +} + +export async function releaseUnadmittedTaskRetryInput(db: Db, input: { companyId: string; retryActionId: string }) { + await db.transaction(async tx => { + const [snapshot] = await tx.select({ payload: chatActions.payload }).from(chatActions).where(and( + eq(chatActions.companyId, input.companyId), eq(chatActions.id, input.retryActionId), + )); + const issueId = z.string().uuid().safeParse(snapshot?.payload.issueId); + if (!issueId.success) return; + const [issue] = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, input.companyId), eq(issues.id, issueId.data), + )).for("update"); + if (!issue) return; + const [action] = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, input.companyId), + eq(chatActions.id, input.retryActionId), eq(chatActions.kind, "failed_run_retry"), + sql`${chatActions.payload}->>'issueId' = ${issue.id}`, + inArray(chatActions.status, ["failed", "cancelled"]))).for("update"); + if (!action || typeof action.payload.taskRetryIntentId !== "string") return; + const [receipt] = await tx.select({ runId: agentWakeupRequests.runId }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.retryActionId), + )).for("update"); + if (receipt?.runId) return; + await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", finishedAt: null, + payload: sql`${agentWakeupRequests.payload} - 'coalescedIntoWakeupRequestId'`, updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.status, "coalesced"), + isNull(agentWakeupRequests.runId), + sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId' = ${input.retryActionId}`)); + await tx.update(agentWakeupRequests).set({ status: "failed", finishedAt: new Date(), updatedAt: new Date() }).where(and( + eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, action.payload.taskRetryIntentId), + eq(agentWakeupRequests.reason, "task_retry"), isNull(agentWakeupRequests.runId), + inArray(agentWakeupRequests.status, ["deferred_issue_execution", "queued", "claimed"]), + sql`${agentWakeupRequests.payload}#>>'{taskRetry,chatActionId}' = ${action.id}`, + )); + }); +} diff --git a/server/src/services/durable-chat-wakeup.ts b/server/src/services/durable-chat-wakeup.ts index 62324a15ece..b9d9f721a28 100644 --- a/server/src/services/durable-chat-wakeup.ts +++ b/server/src/services/durable-chat-wakeup.ts @@ -1,6 +1,6 @@ import { createHash } from "node:crypto"; import { agentWakeupRequests, chatActions, chatDeliveries, chatEndpoints, type Db } from "@paperclipai/db"; -import { and, eq, or, sql, type SQLWrapper } from "drizzle-orm"; +import { and, eq, inArray, or, sql, type SQLWrapper } from "drizzle-orm"; import { HttpError } from "../errors.js"; export interface CommittedChatResponseAuthorizationInput { @@ -211,6 +211,7 @@ export interface DurableChatWakeupRequest { readonly idempotencyKey: string; readonly failedRunRetry?: { readonly failedRunId: string }; readonly authorize: (tx: Db) => Promise; + readonly onAdmitted?: (tx: Db, receipt: { runId: string | null }) => Promise; } const trustedRequests = new WeakSet(); @@ -290,6 +291,26 @@ export function assertDurableChatWakeupReceipt( } } +export function withDurableChatAdmission(request: DurableChatWakeupRequest | undefined, + admit: (tx: Parameters[0]>[0]) => Promise) { + return async (tx: Parameters[0]>[0]) => { + const outcome = await admit(tx); + // An existing deferred receipt is not proof this transaction passed its + // gates. Only dispatch or an explicitly persisted wait admits the input. + if (request?.onAdmitted && (outcome.admitted === true || ["queued", "durable", "coalesced"].includes(outcome.kind))) { + const [receipt] = await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, request.companyId), eq(agentWakeupRequests.id, request.id), + inArray(agentWakeupRequests.status, ["queued", "claimed", "deferred_issue_execution", "coalesced"]), + )); + if (receipt) { + assertDurableChatWakeupReceipt(request, receipt); + await request.onAdmitted(tx as unknown as Db, { runId: receipt.runId }); + } + } + return outcome; + }; +} + /** Work-start policy only: ongoing tool and publication authorization must not * revoke an already-started run when the operator enables mention gating. */ export async function slackMentionAllowsRunStart( diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 35ff18e6e88..2ac1a3f0960 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,5 +1,4 @@ -import { authorizationService } from "../authorization.js"; -import { assertTaskRetryConversationAllowed, readTaskRetryIntent, retireSupersededTaskRetry } from "../task-retry.js"; +import { assertTaskRetryConversationAllowed, readTaskRetryIntent, retireSupersededTaskRetry, taskRetryMutationAllowed } from "../task-retry.js"; import { insertIssueComment } from "../issue-comment-writes.js"; import { withExternalAdmissionGuard, assertNoExternalOverlap } from "../../modules/external-agents/index.js"; import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js"; @@ -50,6 +49,7 @@ import { slackMentionAllowsRunStart, assertDurableChatWakeupReceipt, assertDurableChatWakeupRequest, + withDurableChatAdmission, authorizeFailedChatRunRetryWake, FailedChatRunRetryAuthorizationError, unadmittedChatWakeupCondition, @@ -2809,7 +2809,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const cancelledRunsToEmit: (typeof heartbeatRuns.$inferSelect)[] = []; const outcome = await db - .transaction(async (tx) => { + .transaction(withDurableChatAdmission(durableRequest, async (tx) => { await tx.execute( sql`select id from issues where id = ${issueId} and company_id = ${agent.companyId} for update`, ); @@ -2847,11 +2847,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (pending.id !== opts.taskRetryRequestId || retry?.generationId !== pending.id || retry.chatActionId || retry.actorId !== opts.requestedByActorId || retry.sourceRunId !== enrichedContextSnapshot.previousRunId) return { kind: "deferred" as const }; - const permission = await authorizationService(tx as unknown as Db).decide({ - actor: { type: "board", userId: retry.actorId, source: retry.actorSource }, action: "issue:mutate", - resource: { type: "issue", companyId: agent.companyId, issueId }, - }); - if (!permission.allowed) return { kind: "deferred" as const }; + if (!await taskRetryMutationAllowed(tx as unknown as Db, { companyId: agent.companyId, issueId, + actorId: retry.actorId, actorSource: retry.actorSource })) return { kind: "deferred" as const }; } if (opts.queuedCommentRequestId) { const ids = await undeliveredLegacyUserCommentIds(tx as unknown as Db, @@ -3004,11 +3001,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende if (!failedChatRetry || retry?.chatActionId !== durableRequest.id || retry.actorId !== taskRetryActorId || retry.sourceRunId !== durableRequest.failedRunRetry?.failedRunId) throw new FailedChatRunRetryAuthorizationError(); - const permission = await authorizationService(tx as unknown as Db).decide({ - actor: { type: "board", userId: retry.actorId, source: retry.actorSource }, action: "issue:mutate", - resource: { type: "issue", companyId: agent.companyId, issueId }, - }); - if (!permission.allowed) return { kind: "deferred" as const }; + if (!await taskRetryMutationAllowed(tx as unknown as Db, { companyId: agent.companyId, issueId, + actorId: retry.actorId, actorSource: retry.actorSource })) return { kind: "deferred" as const }; } } else { if ( @@ -3311,7 +3305,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${JSON.stringify(condition)}::jsonb)`, updatedAt: new Date(), }).where(eq(agentWakeupRequests.id, executionWaitRequestId)); - return { kind: "deferred" as const }; + return { kind: "deferred" as const, admitted: true }; } if (durableRequest || wakeCommentId || hasInteractionContinuationWakeContext(enrichedContextSnapshot) || @@ -3348,7 +3342,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }, }); } - return { kind: "deferred" as const }; + return { kind: "deferred" as const, admitted: true }; }; const explicitContinuationRunId = randomUUID(); const executionBlocker = await getExecutionBlocker( @@ -3974,7 +3968,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }; } if (admission.kind === "deferred") { - return { kind: "deferred" as const }; + return { kind: "deferred" as const, admitted: true }; } // admission.kind === "proceed": no active run absorbed this wake, // so fall through to the ordinary queue path below. @@ -4357,7 +4351,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // transitions to "running" — Fix A (lazy locking). return { kind: "queued" as const, run: newRun }; - }) + })) .catch((error) => { if (isExternalChatWaitAuthorizationContention(error)) return { kind: "deferred" as const }; diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 1937712ac23..9300b07d661 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -8,6 +8,20 @@ import { createSystemRecovery, readSystemRecovery } from "./system-recovery-stat import { updateSystemRecovery } from "./system-recovery.js"; import { instanceSettingsService } from "./instance-settings.js"; import { notifyChatActionWork } from "./chat-work-notifications.js"; +import { authorizationService } from "./authorization.js"; + +export async function taskRetryMutationAllowed(db: Db, input: { + companyId: string; issueId: string; actorId: string; actorSource: TaskRetryIntent["actorSource"]; +}) { + const [issue] = await db.select({ status: issues.status, assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId }).from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))); + if (!issue) return false; + const permission = await authorizationService(db).decide({ + actor: { type: "board", userId: input.actorId, source: input.actorSource }, action: "issue:mutate", + resource: { type: "issue", companyId: input.companyId, issueId: input.issueId, ...issue }, + }); + return permission.allowed; +} export async function assertTaskRetryConversationAllowed( db: Db, @@ -115,6 +129,7 @@ export async function requestTaskRetry(input: { const intent: TaskRetryIntent = { schema: "paperclip.task-retry.v1", generationId: active?.id ?? id, sourceRunId: latest.id, actorId: input.actorId, actorSource: taskRetryIntentSchema.shape.actorSource.parse(input.actorSource ?? "session"), strategy: "evaluate", + ...(active && activeIntent?.chatActionId ? { chatActionId: activeIntent.chatActionId } : {}), ...(active ? { canonicalIntentId: active.id } : {}) }; // Preserve external chat authority through its existing admission owner. // Generic board Retry must never manufacture a chat-inbound request.