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.