diff --git a/doc/connections/GITHUB-REVIEW-BOT.md b/doc/connections/GITHUB-REVIEW-BOT.md index c059e0208db..4a5c8dd50c9 100644 --- a/doc/connections/GITHUB-REVIEW-BOT.md +++ b/doc/connections/GITHUB-REVIEW-BOT.md @@ -188,11 +188,8 @@ old runs cannot publish over the latest head. One current summary is updated in place, with history and task/run links retained. Stable finding keys prevent duplicate inline comments on repeated reviews. -GitHub bot tools allow up to 60 seconds by default for their provider requests -and publication receipts. An explicit shorter tool deadline still applies. -The assessment and publication intent are saved before publication. A timeout -does not cancel that durable publication or prove that GitHub rejected it; -check the review's publication links before starting another review. +See [Publication recovery](#publication-recovery) for deadlines, retries, and +lost-response receipt recovery. The check's **Details** link opens its Paperclip task on the current instance hostname, or the connector's Reviews page when no task has been created yet. @@ -212,6 +209,42 @@ where supported. Paperclip does not change repository rules. GitHub account and repository plan restrictions may limit required-check enforcement. If automatic execution is disallowed, a gated head requests an authorized manual review. +## Publication recovery + +GitHub bot tools have a sixty-second default deadline. A caller can request a +shorter deadline. The assessment and publication intent are saved before provider +writes, so a caller timeout does not prove that publication failed. + +Temporary network errors, HTTP 408/429, and GitHub server errors use bounded +backoff with jitter. Reads and installation-token issuance can retry within the +call. Publication writes retry through the durable outbox, where App-owned +comment markers and check identities recover lost responses. An unknown write +is never repeated without checking its existing provider receipt. + +The caller waits for up to three publication attempts within a forty-five-second +retry window. Longer delays remain queued. A dedicated worker restores received +actions and future retry deadlines after restart without another inbound event. +The durable limit is eight failed attempts. Authorization, configuration, and +the current PR head are checked again before dispatch. Permanent permission or +validation failures, changed authority, and superseded reviews stop retries. + +Rate-limit retries respect GitHub's `Retry-After` and primary reset headers; +secondary limits without a deadline wait at least one minute. See +[GitHub's retry guidance](https://docs.github.com/en/rest/using-the-rest-api/best-practices-for-using-the-rest-api#handle-rate-limit-errors-appropriately). + +Repeat the exact `submit_review`, `comment`, or `update_comment` call to receive +the current durable receipt after a timeout. Preserve the assessment and +idempotency key; do not start another review to recover publication. A result +with `status: failed` and `receipt.retryable: true`, or `status: received`, is +still pending. Claim publication only after `status: processed`. Receipt recovery +does not itself mark a task Done; the agent must finish its task after confirming +the receipt. + +For a call that required human approval, the durable publication keeps that +verified approval for its exact operation. An exact caller retry only reads the +current receipt; the worker owns provider retries. A new policy denial or revoked +authority still stops recovery, and new content still requires its own approval. + ## Hosted ingress Dedicated Apps use Cloud as a sealed transport gateway. Cloud routes opaque diff --git a/maps/agent-access-map/chat-channels.md b/maps/agent-access-map/chat-channels.md index c79a9ef8477..df9e94958ea 100644 --- a/maps/agent-access-map/chat-channels.md +++ b/maps/agent-access-map/chat-channels.md @@ -87,6 +87,17 @@ remaining operations before qualifying the whole capability. Gaps: Normal agent replies work through server delivery. Explicit endpoint publication APIs are board management paths. +GitHub publication recovery (2026-10-14): assigned-agent `submit_review`, `comment`, +and `update_comment` calls reuse their run-, task-, endpoint-, and argument-bound +publication on an exact retry. Fresh gateway policy and provider authority remain +required. The durable worker resumes transient failures after restart and stops +at its retry cap, revoked authority, or a superseded head. A pending receipt is +not task completion. Source: +[publication service](../../server/src/services/chat-github-reviews.ts) and +[gateway](../../server/src/services/tool-gateway.ts). Verification uses +PostgreSQL-backed provider fixtures; this entry does not claim live provider +qualification for the new retry path. + Speko browser sessions use `POST /api/companies/{companyId}/voice-sessions`, with GET inspection/notification and POST `/{sessionId}/end` operations below that collection. These routes require board identity; ordinary agent credentials diff --git a/maps/feature-map/chat-channels.md b/maps/feature-map/chat-channels.md index 1b16924c035..1a5854af390 100644 --- a/maps/feature-map/chat-channels.md +++ b/maps/feature-map/chat-channels.md @@ -58,6 +58,13 @@ For GitHub, submit a disposable PR and verify the summary heading uses the bot's display name and the check remains **Paperclip Review**. For two separate bots on one PR, verify their App-scoped checks and comments remain independent. +For GitHub publication recovery, interrupt a response after GitHub accepts a +write and verify one summary, finding, and App-scoped check. Retry the same tool +call and verify it returns the saved receipt. Leave a transient failure queued, +restart the instance, and verify publication resumes at its saved deadline +without another inbound message. Revoked access and permanent rejections must +stop retries. These recipes are separate from live provider proof. + Configure optional project and workspace defaults at the connection and destination levels. Start a new conversation and inspect its task workspace. Verify that inheritance, an explicit selection, and clearing a default remain distinct. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index fea8b7f8f6a..24fa9b9ac7f 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -15,7 +15,7 @@ import { fastResponseService } from "../services/fast-responses.js"; import { fastResponseReceipt } from "../services/fast-response-provider.js"; import { fastResponseRequests } from "@paperclipai/db"; import { toolAccessService } from "../services/tool-access.js"; -import { toolActionRequests, toolInvocations } from "@paperclipai/db"; +import { toolActionRequests, toolInvocations, toolPolicies } from "@paperclipai/db"; import { GitHubPublicationLeaseLost, withGitHubPublicationLease } from "../services/chat-github-publication-lease.js"; import { secretService } from "../services/secrets.js"; import { githubChatWizardService } from "../services/chat-github-wizard.js"; @@ -2555,6 +2555,235 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const session = { companyId: f.companyId, agentId: f.assignedAgentId, issueId: conversation.issueId, runId: run.id }; return { ...f, conversation, run, session }; } + async function publicationRetryFixture() { + const f = await toolReplyFixture("github:paperclipai/paperclip:418", true); + const head = "b".repeat(40); + const inline: Array<{ id: number; body: string; html_url: string; user: { login: string } }> = []; + const checks: Array<{ id: number; external_id: string; app: { id: number }; html_url: string }> = []; + const writes: Array<{ path: string; method: string }> = []; + const provider = async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.endsWith("/pulls/418")) return Response.json({ number: 418, state: "open", draft: false, + head: { sha: head }, base: { sha: "a".repeat(40), ref: "master" }, user: { id: 42, login: "octocat", type: "User" }, labels: [] }); + if (url.includes("/files?")) return Response.json([{ filename: "src/math.ts", status: "modified", patch: "@@ -1 +1 @@\n-old\n+new" }]); + if (url.includes("/pulls/418/comments?")) return Response.json(inline); + if (url.includes("/check-runs?")) return Response.json({ check_runs: checks }); + if (init?.method === "POST" && url.endsWith("/pulls/418/comments")) { + writes.push({ path: url, method: init.method }); + const item = { id: 420 + inline.length, body: JSON.parse(String(init.body)).body, + html_url: "https://github.com/acme/repo/pull/418#discussion_r420", user: { login: f.endpoint.botUsername! } }; + inline.push(item); + return Response.json(item); + } + if (url.includes("/check-runs") && init?.method) { + writes.push({ path: url, method: init.method }); + const body = JSON.parse(String(init.body)); + const check = checks[0] ?? { id: 419, external_id: body.external_id, app: { id: Number(f.endpoint.botExternalId) }, html_url: "https://github.com/checks/419" }; + if (!checks.length) checks.push(check); + return Response.json(check); + } + return undefined; + }; + f.setSupplementalProviderFetch(provider); + const service = githubChatReviewService(db, f.providerFetch); + await service.execute(f.session, "begin_review", { reviewedCommit: head }); + const assessment = { reviewedCommit: head, score: 3, complete: true, summary: "Advisory review", rationale: "Reviewed the file", + coverage: { reviewedPaths: ["src/math.ts"], omittedPaths: [], limitations: [] }, findings: [] }; + const action = async () => (await db.select().from(chatActions).where(and( + eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication"), + )))[0]; + return { ...f, service, head, assessment, provider, action, inline, checks, writes }; + } + it.each([false, true])("recovers an approved queued GitHub review without another approval (revoked: %s)", async revoked => { + const f = await publicationRetryFixture(); + let unavailable = true; + let attempts = 0; + f.setSupplementalProviderFetch(async (input, init) => { + if (String(input).includes("/issues/comments/") && init?.method === "PATCH") { + attempts++; + if (unavailable) return new Response("Unavailable", { status: 503, headers: { "retry-after": "120" } }); + } + return f.provider(input, init); + }); + vi.stubGlobal("fetch", f.providerFetch); + onTestFinished(() => { vi.unstubAllGlobals(); }); + await initializeRunIdentity(db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, + responsibleUserId: null, cause: "company_default" }); + const { createToolGatewayService } = await import("../services/tool-gateway.js"); + const gateway = createToolGatewayService(db, { toolActionSigningSecret: "github-approval-retry-test-secret" }); + const session = await gateway.createSession({ companyId: f.companyId, agentId: f.assignedAgentId, runId: f.run.id }); + const tool = (await githubBotToolsForSession(db, f.session)).find(tool => tool.upstreamToolName === "submit_review")!; + await db.insert(toolPolicies).values({ companyId: f.companyId, name: "Review publication approval", policyType: "require_approval", + priority: 0, selectors: { catalogEntryId: tool.catalogEntryId! } }); + await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: f.assessment })) + .rejects.toMatchObject({ reasonCode: "approval_required" }); + const [approval] = await db.select().from(toolActionRequests).where(eq(toolActionRequests.companyId, f.companyId)); + expect(await gateway.approveActionRequest({ companyId: f.companyId, actionRequestId: approval.id, actor: { userId: "owner-user" } })) + .toMatchObject({ status: "executed" }); + const action = await f.action(); + expect(action).toMatchObject({ status: "failed", result: { retryable: true } }); + unavailable = false; + if (revoked) await db.insert(toolPolicies).values({ companyId: f.companyId, name: "Revoke approved publication", policyType: "block", + priority: -1, selectors: { catalogEntryId: tool.catalogEntryId! } }); + await db.update(chatActions).set({ result: { ...action.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, action.id)); + await f.service.processPending(); + expect((await f.action()).status).toBe(revoked ? "cancelled" : "processed"); + if (revoked) await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: f.assessment })).rejects.toThrow(); + else expect(await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: f.assessment })) + .toMatchObject({ status: "replayed", result: { data: { status: "processed", score: 3 } } }); + expect(attempts).toBe(revoked ? 1 : 2); + expect(await db.select().from(toolActionRequests).where(eq(toolActionRequests.companyId, f.companyId))).toHaveLength(1); + }, 25_000); + it("recovers lost inline and check responses without duplicate GitHub publications", async () => { + const f = await publicationRetryFixture(); + const lost = new Set(); + f.setSupplementalProviderFetch(async (input, init) => { + const response = await f.provider(input, init); + if (init?.method === "POST" && /\/(?:comments|check-runs)$/.test(String(input)) && !lost.has(String(input))) { + lost.add(String(input)); + throw new Error("Provider accepted the write but the response was lost"); + } + return response; + }); + const result = await f.service.execute(f.session, "submit_review", { ...f.assessment, + findings: [{ key: "note", path: "src/math.ts", line: 1, side: "RIGHT", severity: "error", category: "correctness", body: "Example finding" }], + }, "lost-receipts"); + expect(result).toMatchObject({ status: "processed", score: 3, conclusion: "neutral" }); + expect(f.inline).toHaveLength(1); + expect(f.checks).toHaveLength(1); + expect(f.writes.filter(write => write.method === "POST")).toHaveLength(2); + expect(f.responseComments.size).toBe(1); + expect((await f.action()).status).toBe("processed"); + }, 25_000); + it("resumes a delayed GitHub publication after coordinator restart without another inbound event", async () => { + const f = await publicationRetryFixture(); + let unavailable = true; + let attempts = 0; + f.setSupplementalProviderFetch(async (input, init) => { + if (String(input).includes("/issues/comments/") && init?.method === "PATCH") { + attempts++; + if (unavailable) return new Response("Unavailable", { status: 503, headers: { "retry-after": "120" } }); + } + return f.provider(input, init); + }); + const result = await f.service.execute(f.session, "submit_review", f.assessment, "scheduled-review"); + expect(result).toMatchObject({ status: "failed", receipt: { retryable: true, attempts: 1 } }); + const action = await f.action(); + expect(await f.service.nextPendingAt()).toBe(Date.parse(String(action.result?.retryAt))); + expect(await f.service.processPending()).toBe(0); + const due = new Date(Date.now() + 1500); + await db.update(chatActions).set({ result: { ...action.result, retryAt: due.toISOString() } }).where(eq(chatActions.id, action.id)); + unavailable = false; + const { createDeliveryWorkCoordinator } = await import("../services/delivery-work-coordinator.js"); + const { registerChatQueueWork } = await import("../services/chat-delivery-work.js"); + const { DELIVERY_QUEUES } = await import("../services/delivery-work-notifications.js"); + const restarted = githubChatReviewService(db, f.providerFetch); + const errors = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: errors }); + try { + await registerChatQueueWork(coordinator, { queue: DELIVERY_QUEUES.chatGitHubPublications, + run: restarted.processPending, next: restarted.nextPendingAt }, () => true).ready; + expect(coordinator.nextWakeAt()).toBe(due.getTime()); + expect(attempts).toBe(1); + await expect.poll(async () => (await f.action()).status, { timeout: 10_000 }).toBe("processed"); + expect(attempts).toBe(2); + expect(errors).not.toHaveBeenCalled(); + expect(await restarted.nextPendingAt()).toBeNull(); + expect(f.responseComments.size).toBe(1); + } finally { + await coordinator.stop(); + } + }, 25_000); + it.each([403, 422])("does not resubmit a permanent GitHub publication rejection (HTTP %s)", async status => { + const f = await publicationRetryFixture(); + let attempts = 0; + f.setSupplementalProviderFetch(async (input, init) => { + if (String(input).includes("/issues/comments/") && init?.method === "PATCH") { + attempts++; + return new Response("Rejected", { status }); + } + return f.provider(input, init); + }); + const result = await f.service.execute(f.session, "submit_review", f.assessment, "permanent-failure"); + expect(result).toMatchObject({ status: "failed", receipt: { retryable: false, attempts: 1, providerStatus: status } }); + await f.service.execute(f.session, "submit_review", f.assessment, "permanent-failure"); + expect(await f.service.processPending()).toBe(0); + expect(await f.service.nextPendingAt()).toBeNull(); + expect(attempts).toBe(1); + }); + it.each(["revoked", "exhausted"] as const)("stops a queued GitHub publication after %s", async mode => { + const f = await publicationRetryFixture(); + let attempts = 0; + f.setSupplementalProviderFetch(async (input, init) => { + if (String(input).includes("/issues/comments/") && init?.method === "PATCH") { + attempts++; + return new Response("Unavailable", { status: 503, headers: { "retry-after": "120" } }); + } + return f.provider(input, init); + }); + await f.service.execute(f.session, "submit_review", f.assessment, "bounded-retry"); + const action = await f.action(); + await db.update(chatActions).set({ result: { ...action.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, action.id)); + if (mode === "revoked") await db.update(chatIdentityLinks).set({ status: "revoked" }).where(eq(chatIdentityLinks.endpointId, f.endpoint.id)); + for (let attempt = 1; attempt < (mode === "exhausted" ? 8 : 2); attempt++) { + const current = await f.action(); + await db.update(chatActions).set({ result: { ...current.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, current.id)); + await f.service.processPending(); + } + await f.service.processPending(); + expect((await f.action()).result?.retryable).toBe(false); + expect((await f.action()).status).toBe(mode === "revoked" ? "cancelled" : "failed"); + expect(await f.service.nextPendingAt()).toBeNull(); + expect(attempts).toBe(mode === "exhausted" ? 8 : 1); + }); + it.each([503, 429])("automatically retries a transient GitHub review publication (HTTP %s)", async providerStatus => { + const f = await toolReplyFixture("github:paperclipai/paperclip:418", true); + const head = "b".repeat(40); + let summaryAttempts = 0; + const checks: Record[] = []; + f.setSupplementalProviderFetch(async (input, init) => { + const url = String(input); + if (url.endsWith("/pulls/418")) return Response.json({ number: 418, state: "open", draft: false, + head: { sha: head }, base: { sha: "a".repeat(40), ref: "master" }, user: { id: 42, login: "octocat", type: "User" }, labels: [] }); + if (url.includes("/files?")) return Response.json([{ filename: "src/math.ts", status: "modified", patch: "@@ -1 +1 @@\n-old\n+new" }]); + if (url.includes("/check-runs?")) return Response.json({ check_runs: [] }); + if (url.includes("/check-runs") && init?.method) { + checks.push(JSON.parse(String(init.body))); + return Response.json({ id: 419, html_url: "https://github.com/checks/419" }); + } + if (url.includes("/issues/comments/") && init?.method === "PATCH" && ++summaryAttempts === 1) + return new Response("Temporary provider failure", { status: providerStatus, headers: { "retry-after": "1" } }); + return undefined; + }); + vi.stubGlobal("fetch", f.providerFetch); + onTestFinished(() => { vi.unstubAllGlobals(); }); + const { createToolGatewayService } = await import("../services/tool-gateway.js"); + const gateway = createToolGatewayService(db); + const session = await gateway.createSession({ companyId: f.companyId, agentId: f.assignedAgentId, runId: f.run.id }); + const tool = (await githubBotToolsForSession(db, f.session)).find(tool => tool.upstreamToolName === "submit_review")!; + await githubChatReviewService(db, f.providerFetch).execute(f.session, "begin_review", { reviewedCommit: head }); + const assessment = { reviewedCommit: head, score: 3, complete: true, summary: "Advisory review", rationale: "Reviewed the file", + coverage: { reviewedPaths: ["src/math.ts"], omittedPaths: [], limitations: [] }, findings: [] }; + const idempotencyKey = randomUUID(); + const result = await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment, idempotencyKey }); + expect(result).toMatchObject({ status: "completed", result: { data: { status: "processed", score: 3, conclusion: "neutral" } } }); + expect(summaryAttempts).toBe(2); + expect(f.responseComments.size).toBe(1); + expect(checks).toEqual([expect.objectContaining({ conclusion: "neutral", status: "completed" })]); + const replay = await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment, idempotencyKey }); + expect(replay).toMatchObject({ result: { data: { status: "processed", score: 3, conclusion: "neutral" } } }); + expect(summaryAttempts).toBe(2); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication"))); + expect(action).toMatchObject({ status: "processed" }); + expect((await readChatGitHubReviewsByEndpointId(f.endpoint.id))).toHaveLength(1); + await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: { ...assessment, score: 5 }, idempotencyKey })) + .rejects.toMatchObject({ status: 409, reasonCode: "github_idempotency_conflict" }); + await db.insert(toolPolicies).values({ companyId: f.companyId, name: "Revoke review publication", policyType: "block", priority: 0, + selectors: { catalogEntryId: tool.catalogEntryId! }, + }); + await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment, idempotencyKey })).rejects.toThrow(); + expect(summaryAttempts).toBe(2); + }, 25_000); it("lists each GitHub bot tool once when an inline thread shares the task", async () => { const f = await toolReplyFixture("github:paperclipai/paperclip:418"); const { id: _id, ...conversation } = f.conversation; @@ -2572,7 +2801,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { it.each([ { label: "returns the confirmed GitHub assessment when publication takes longer than ten seconds", delayMs: 11_000, timeoutMs: undefined }, { label: "preserves an explicit shorter GitHub tool deadline and reconciles without duplicate publication", delayMs: 200, timeoutMs: 50 }, - ])("$label", async ({ delayMs, timeoutMs }) => { + ])("recovers a durable GitHub publication receipt ($label)", async ({ delayMs, timeoutMs }) => { const f = await toolReplyFixture("github:paperclipai/paperclip:418", true); const head = "b".repeat(40); const checks: Record[] = []; @@ -2616,7 +2845,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(result).toMatchObject({ error: { status: 504, reasonCode: "tool_timeout" } }); } const writes = f.responseWrites.length; - await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment }); + const replay = await gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters: assessment }); + expect(replay).toMatchObject({ result: { data: { status: "processed", score: 5, conclusion: "neutral" } } }); + const [invocation] = await db.select().from(toolInvocations).where(eq(toolInvocations.id, replay.invocationId)); + expect(invocation).toMatchObject({ status: "succeeded", errorCode: null, errorMessage: null }); expect(f.responseWrites).toHaveLength(writes); }, 40_000); it("does not post a GitHub working comment when the scheduler declines the request", async () => { diff --git a/server/src/__tests__/chat-delivery-work.test.ts b/server/src/__tests__/chat-delivery-work.test.ts index fe2c9bc68a0..6fbee3e2364 100644 --- a/server/src/__tests__/chat-delivery-work.test.ts +++ b/server/src/__tests__/chat-delivery-work.test.ts @@ -3,6 +3,7 @@ import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signal import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { registerChatDeliveryWork } from "../services/chat-delivery-work.js"; import { notifyChatDeliveryWork, notifyChatPublicationWork } from "../services/chat-work-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js"; const coordinators: ReturnType[] = []; afterEach(async () => { @@ -27,6 +28,8 @@ function setup() { nextInboundDeliveryAt: vi.fn(async (): Promise => null), nextPublicationAt: vi.fn(async (): Promise => null), nextSlackReceiptAt: vi.fn(async (): Promise => null), + processPendingGitHubReviewPublications: vi.fn(async () => {}), + nextGitHubReviewPublicationAt: vi.fn(async (): Promise => null), onPublicationsSettled: vi.fn(), }; const start = () => registerChatDeliveryWork(coordinator, service, enabled); @@ -38,7 +41,7 @@ describe("chat delivery work", () => { const s = setup(); await s.start().ready; await vi.advanceTimersByTimeAsync(3_600_000); - for (const run of [s.service.processQueuedDeliveries, s.service.scheduleQueuedPublications, s.service.processPendingSlackFileUploadReceipts]) expect(run).toHaveBeenCalledTimes(1); + for (const run of [s.service.processQueuedDeliveries, s.service.scheduleQueuedPublications, s.service.processPendingSlackFileUploadReceipts, s.service.processPendingGitHubReviewPublications]) expect(run).toHaveBeenCalledTimes(1); expect(s.coordinator.nextWakeAt()).toBeNull(); expect(vi.getTimerCount()).toBe(0); }); @@ -100,6 +103,23 @@ describe("chat delivery work", () => { expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1); expect(vi.getTimerCount()).toBe(0); }); + it("restores GitHub retry deadlines on startup and wakes after a publication commit", async () => { + const s = setup(); + const due = Date.now() + 60_000; + s.service.nextGitHubReviewPublicationAt.mockResolvedValue(due); + await s.start().ready; + expect(s.coordinator.nextWakeAt()).toBe(due); + await vi.advanceTimersByTimeAsync(59_999); + expect(s.service.processPendingGitHubReviewPublications).toHaveBeenCalledTimes(1); + s.service.nextGitHubReviewPublicationAt.mockResolvedValue(null); + await vi.advanceTimersByTimeAsync(1); + expect(s.service.processPendingGitHubReviewPublications).toHaveBeenCalledTimes(2); + await s.owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications)); + await vi.advanceTimersByTimeAsync(1); + expect(s.service.processPendingGitHubReviewPublications).toHaveBeenCalledTimes(3); + expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); it("retries failed selection without losing its startup recovery", async () => { const s = setup(); s.service.nextSlackReceiptAt.mockRejectedValueOnce(new Error("connection lost")); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 6b205ed94a9..b0e0eb6ba2c 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -55,6 +55,7 @@ export { createChatSdkStatePersistence } from "./chat-channels/sdk-state.js"; import { createChatInteractionCallbacks } from "./chat-channels/interactions/callbacks.js"; import { createChatInteractionContinuations } from "./chat-channels/interactions/continuations.js"; import { createGitHubInstallationLifecycle } from "./chat-channels/github/installation-lifecycle.js"; +import { githubChatReviewService } from "./chat-github-reviews.js"; import { createGitHubAppSetup } from "./chat-channels/github/app-setup.js"; import { createSlackTaskAuthorityRegistration } from "./chat-channels/slack/task-authority.js"; import { createGitHubPublicationPolicy } from "./chat-channels/github/publication-policy.js"; @@ -3370,6 +3371,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { return publicationQueue.processPendingPublications(...args); } + const githubReviews = githubChatReviewService(db, fetchImpl); const { scheduleGitHubMaintenance } = createGitHubMaintenance(db, fetchImpl, { scheduleMessageProcessing, isShuttingDown: () => shuttingDown, }); @@ -3592,6 +3594,8 @@ return { processPublicationMaintenance, nextInboundDeliveryAt, nextPublicationAt, + processPendingGitHubReviewPublications: githubReviews.processPending, + nextGitHubReviewPublicationAt: githubReviews.nextPendingAt, nextSlackReceiptAt, onPublicationsSettled: (wake: () => void) => { wakePublications = wake; }, prepareFailedChatRunRetry, diff --git a/server/src/services/chat-channels/github/maintenance.ts b/server/src/services/chat-channels/github/maintenance.ts index 3e45da8c350..8d09defd0f3 100644 --- a/server/src/services/chat-channels/github/maintenance.ts +++ b/server/src/services/chat-channels/github/maintenance.ts @@ -1,5 +1,4 @@ import type { Db } from "@paperclipai/db"; -import { githubChatReviewService } from "../../chat-github-reviews.js"; import { githubResponseCommentService } from "../../chat-github-response-comments.js"; import { githubReviewCheckService } from "../../chat-github-checks.js"; @@ -14,7 +13,6 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet options.scheduleMessageProcessing(async () => { try { // Provider I/O must not block Slack, Teams, or ordinary message receipts. - await githubChatReviewService(db, fetchImpl).processPending(); await githubResponseCommentService(db, fetchImpl).processPending(); await githubReviewCheckService(db, fetchImpl).processPending(); } finally { diff --git a/server/src/services/chat-channels/github/retry.test.ts b/server/src/services/chat-channels/github/retry.test.ts new file mode 100644 index 00000000000..37ee218e4c5 --- /dev/null +++ b/server/src/services/chat-channels/github/retry.test.ts @@ -0,0 +1,164 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { githubBotRequest } from "../../chat-github-client.js"; +import { forbidden, unprocessable } from "../../../errors.js"; +import { githubProviderRetry, githubPublicationRetry } from "./retry.js"; + +afterEach(() => vi.useRealTimers()); + +describe("GitHub publication retries", () => { + it.each([408, 500, 502, 503, 504])( + "retries a safe GitHub read after HTTP %s", + async (status) => { + vi.useFakeTimers(); + const provider = vi + .fn() + .mockResolvedValueOnce(new Response("Unavailable", { status })) + .mockResolvedValueOnce(Response.json({ head: "current" })); + const request = githubBotRequest( + provider, + "private-token", + "/repos/acme/repo/pulls/1", + ); + await vi.advanceTimersByTimeAsync(2500); + await expect(request).resolves.toEqual({ head: "current" }); + expect(provider).toHaveBeenCalledTimes(2); + }, + ); + it("retries installation token issuance without retrying arbitrary provider writes", async () => { + vi.useFakeTimers(); + const provider = vi + .fn() + .mockResolvedValueOnce(new Response("Unavailable", { status: 503 })) + .mockResolvedValueOnce(Response.json({ token: "issued" })); + const request = githubBotRequest( + provider, + null, + "/app/installations/42/access_tokens", + { method: "POST" }, + ); + await vi.advanceTimersByTimeAsync(2500); + await expect(request).resolves.toEqual({ token: "issued" }); + provider.mockResolvedValue(new Response("Unavailable", { status: 503 })); + await expect( + githubBotRequest(provider, null, "/repos/acme/repo/pulls/1/comments", { + method: "POST", + }), + ).rejects.toMatchObject({ + details: { providerStatus: 503, retryable: true }, + }); + expect(provider).toHaveBeenCalledTimes(3); + }); + it.each([401, 403, 404, 422])( + "does not retry permission or validation rejection HTTP %s", + async (status) => { + const provider = vi.fn(async () => + Response.json({ message: "Rejected private-token" }, { status }), + ); + const error = await githubBotRequest( + provider, + "private-token", + "/repos/acme/repo/pulls/1", + ).catch((error) => error); + expect(error).toMatchObject({ + details: { providerStatus: status, retryable: false }, + }); + expect(String(error)).not.toContain("private-token"); + expect(provider).toHaveBeenCalledTimes(1); + }, + ); + it("honors provider retry and primary rate-limit reset deadlines", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-10-14T12:00:00Z")); + expect( + githubProviderRetry( + new Response(null, { status: 429, headers: { "retry-after": "120" } }), + ), + ).toMatchObject({ retryable: true, retryAfterMs: 120_000 }); + expect( + githubProviderRetry( + new Response(null, { + status: 403, + headers: { + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": String((Date.now() + 300_000) / 1000), + }, + }), + ), + ).toMatchObject({ retryable: true, retryAfterMs: 300_000 }); + expect( + githubProviderRetry( + new Response(null, { + status: 503, + headers: { "retry-after": "Wed, 14 Oct 2026 12:02:00 GMT" }, + }), + ), + ).toMatchObject({ retryAfterMs: 120_000 }); + }); + it("defers a secondary rate limit beyond the inline window without sleeping or resending", async () => { + const provider = vi.fn(async () => + Response.json( + { message: "You have exceeded a secondary rate limit" }, + { status: 403 }, + ), + ); + await expect( + githubBotRequest(provider, null, "/repos/acme/repo/pulls/1"), + ).rejects.toMatchObject({ + details: { + retryable: true, + retryAfterMs: 60_000, + code: "github_bot_rate_limited", + }, + }); + expect(provider).toHaveBeenCalledTimes(1); + }); + it.each(["headers", "body", "json"])( + "bounds retries after an interrupted %s response", + async (phase) => { + vi.useFakeTimers(); + const provider = vi.fn(async () => { + if (phase === "headers") + throw new Error("private-token transport failed"); + if (phase === "json") return new Response("{truncated"); + return new Response( + new ReadableStream({ + pull(controller) { + controller.error(new Error("private-token body interrupted")); + }, + }), + ); + }); + const error = githubBotRequest( + provider, + "private-token", + "/repos/acme/repo/pulls/1", + ).catch((error) => error); + await vi.advanceTimersByTimeAsync(10_000); + expect(await error).toMatchObject({ + status: 422, + details: { retryable: true }, + }); + expect(String(await error)).not.toContain("private-token"); + expect(provider).toHaveBeenCalledTimes(3); + }, + ); + it("stops durable retries at the attempt cap and for revoked authority", () => { + const temporary = unprocessable("Unavailable", { + retryable: true, + providerStatus: 503, + retryAfterMs: 120_000, + }); + const before = Date.now(); + expect(githubPublicationRetry(temporary, 7).retryable).toBe(true); + expect( + Date.parse(githubPublicationRetry(temporary, 7).retryAt), + ).toBeGreaterThanOrEqual(before + 120_000); + expect(githubPublicationRetry(temporary, 8).retryable).toBe(false); + expect(githubPublicationRetry(forbidden("Revoked"), 1).retryable).toBe( + false, + ); + expect( + githubPublicationRetry(unprocessable("Invalid assessment"), 1).retryable, + ).toBe(false); + }); +}); diff --git a/server/src/services/chat-channels/github/retry.ts b/server/src/services/chat-channels/github/retry.ts new file mode 100644 index 00000000000..3d85c633998 --- /dev/null +++ b/server/src/services/chat-channels/github/retry.ts @@ -0,0 +1,79 @@ +import { HttpError } from "../../../errors.js"; + +export const GITHUB_PUBLICATION_MAX_ATTEMPTS = 8; +export const GITHUB_INLINE_RETRY_WINDOW_MS = 45_000; +export const GITHUB_INLINE_MAX_ATTEMPTS = 3; + +export function githubRetryDelay(attempt: number) { + return ( + Math.min(300_000, 1000 * 2 ** attempt) + Math.floor(Math.random() * 250) + ); +} + +export function githubProviderRetry( + response: Response, + secondaryRateLimit = false, +) { + const rateLimited = + response.status === 429 || + (response.status === 403 && + (response.headers.has("retry-after") || + response.headers.get("x-ratelimit-remaining") === "0" || + secondaryRateLimit)); + const retryable = + rateLimited || response.status === 408 || response.status >= 500; + const value = response.headers.get("retry-after"); + const seconds = value === null ? NaN : Number(value); + const retryAfter = value === null ? NaN : Date.parse(value); + const reset = Number(response.headers.get("x-ratelimit-reset")); + const retryAfterMs = + Number.isFinite(seconds) && seconds >= 0 + ? seconds * 1000 + : Number.isFinite(retryAfter) + ? Math.max(0, retryAfter - Date.now()) + : 0; + const resetMs = + response.headers.get("x-ratelimit-remaining") === "0" && + Number.isFinite(reset) + ? Math.max(0, reset * 1000 - Date.now()) + : 0; + return { + code: rateLimited + ? "github_bot_rate_limited" + : "github_bot_operation_failed", + providerStatus: response.status, + retryable, + retryAfterMs: Math.max( + retryAfterMs, + resetMs, + rateLimited && !retryAfterMs && !resetMs ? 60_000 : 0, + ), + }; +} + +export function githubPublicationRetry(error: unknown, attempts: number) { + const details = + error instanceof HttpError && + error.details && + typeof error.details === "object" + ? (error.details as { + retryable?: boolean; + retryAfterMs?: number; + providerStatus?: number; + }) + : null; + const retryable = + attempts < GITHUB_PUBLICATION_MAX_ATTEMPTS && + (details?.retryable ?? !(error instanceof HttpError && error.status < 500)); + return { + attempts, + retryable, + retryAt: new Date( + Date.now() + + Math.max(githubRetryDelay(attempts), details?.retryAfterMs ?? 0), + ).toISOString(), + ...(details?.providerStatus + ? { providerStatus: details.providerStatus } + : {}), + }; +} diff --git a/server/src/services/chat-delivery-work.ts b/server/src/services/chat-delivery-work.ts index ed790894d30..2b64d212441 100644 --- a/server/src/services/chat-delivery-work.ts +++ b/server/src/services/chat-delivery-work.ts @@ -8,6 +8,8 @@ type ChatDeliveryService = { nextInboundDeliveryAt(): Promise; nextPublicationAt(): Promise; nextSlackReceiptAt(): Promise; + processPendingGitHubReviewPublications(): Promise; + nextGitHubReviewPublicationAt(): Promise; onPublicationsSettled(wake: () => void): void; }; @@ -21,6 +23,7 @@ export function registerChatDeliveryWork( { queue: DELIVERY_QUEUES.chatInbound, run: service.processQueuedDeliveries, next: service.nextInboundDeliveryAt }, { queue: DELIVERY_QUEUES.chatPublications, run: service.scheduleQueuedPublications, next: service.nextPublicationAt }, { queue: DELIVERY_QUEUES.chatReceipts, run: service.processPendingSlackFileUploadReceipts, next: service.nextSlackReceiptAt }, + { queue: DELIVERY_QUEUES.chatGitHubPublications, run: service.processPendingGitHubReviewPublications, next: service.nextGitHubReviewPublicationAt }, ]; const workers = tasks.map(task => { const worker = registerChatQueueWork(coordinator, task, canRun); diff --git a/server/src/services/chat-github-client.ts b/server/src/services/chat-github-client.ts index e6ad3c8bd62..65d93445111 100644 --- a/server/src/services/chat-github-client.ts +++ b/server/src/services/chat-github-client.ts @@ -3,6 +3,7 @@ import { and, eq } from "drizzle-orm"; import { chatEndpoints, toolConnections, type Db } from "@paperclipai/db"; import { secretService } from "./secrets.js"; import { conflict, forbidden, unprocessable } from "../errors.js"; +import { githubPublicationRetry, githubProviderRetry, GITHUB_INLINE_RETRY_WINDOW_MS, GITHUB_INLINE_MAX_ATTEMPTS } from "./chat-channels/github/retry.js"; export function githubAppJwt( appId: string, @@ -25,6 +26,28 @@ export async function githubBotRequest( token: string | null, path: string, options: { method?: string; body?: unknown; accept?: string } = {}, +): Promise { + const deadline = Date.now() + GITHUB_INLINE_RETRY_WINDOW_MS; + const safeToRetry = (options.method ?? "GET") === "GET" || + (options.method === "POST" && /^\/app\/installations\/[1-9][0-9]*\/access_tokens$/.test(path)); + for (let attempt = 1; ; attempt++) { + try { + return await githubBotRequestOnce(fetchImpl, token, path, options, Math.min(25_000, Math.max(1, deadline - Date.now()))); + } catch (error) { + const retry = githubPublicationRetry(error, attempt); + const delay = Date.parse(retry.retryAt) - Date.now(); + if (!safeToRetry || !retry.retryable || attempt >= GITHUB_INLINE_MAX_ATTEMPTS || Date.now() + delay >= deadline) throw error; + await new Promise(resolve => setTimeout(resolve, delay)); + } + } +} + +async function githubBotRequestOnce( + fetchImpl: typeof fetch, + token: string | null, + path: string, + options: { method?: string; body?: unknown; accept?: string }, + timeoutMs: number, ): Promise { if ( !path.startsWith("/") || @@ -38,7 +61,7 @@ export async function githubBotRequest( response = await fetchImpl(`https://api.github.com${path}`, { method: options.method ?? "GET", redirect: "error", - signal: AbortSignal.timeout(25_000), + signal: AbortSignal.timeout(timeoutMs), headers: { accept: options.accept ?? "application/vnd.github+json", "x-github-api-version": "2022-11-28", @@ -54,18 +77,39 @@ export async function githubBotRequest( } catch { throw unprocessable( "GitHub is temporarily unavailable. Retry this operation.", + { code: "github_bot_temporarily_unavailable", retryable: true }, ); } if (!response.ok) { + let secondaryRateLimit = false; + if (response.status === 403) { + const reader = response.body?.getReader(); + try { + const chunks: Uint8Array[] = []; + let size = 0; + while (reader) { + const part = await reader.read(); + if (part.done) break; + size += part.value.byteLength; + if (size > 8192) break; + chunks.push(part.value); + } + secondaryRateLimit = /secondary rate limit|rate limit exceeded|abuse detection/i.test(Buffer.concat(chunks).toString("utf8")); + } catch { + } finally { + await reader?.cancel().catch(() => {}); + reader?.releaseLock(); + } + } await response.body?.cancel(); throw unprocessable( - `GitHub rejected this operation (HTTP ${response.status}). Check the App installation and repository permissions.`, - { code: "github_bot_operation_failed", providerStatus: response.status }, + `GitHub rejected this operation (HTTP ${response.status}).`, + githubProviderRetry(response, secondaryRateLimit), ); } if (response.status === 204) return undefined as T; const reader = response.body?.getReader(); - if (!reader) throw unprocessable("GitHub returned an empty response"); + if (!reader) throw unprocessable("GitHub returned an empty response", { code: "github_bot_invalid_response", retryable: true }); const chunks: Uint8Array[] = []; let size = 0; try { @@ -80,7 +124,14 @@ export async function githubBotRequest( chunks.push(part.value); } const body = Buffer.concat(chunks).toString("utf8"); - return (options.accept?.includes("diff") ? body : JSON.parse(body)) as T; + try { + return (options.accept?.includes("diff") ? body : JSON.parse(body)) as T; + } catch { + throw unprocessable("GitHub returned an invalid response", { code: "github_bot_invalid_response", retryable: true }); + } + } catch (error) { + if (error instanceof Error && "status" in error) throw error; + throw unprocessable("GitHub response was interrupted. Retry this operation.", { code: "github_bot_temporarily_unavailable", retryable: true }); } finally { await reader.cancel().catch(() => {}); reader.releaseLock(); diff --git a/server/src/services/chat-github-publication-lease.ts b/server/src/services/chat-github-publication-lease.ts index 4de90dac9d1..36cf8a9880d 100644 --- a/server/src/services/chat-github-publication-lease.ts +++ b/server/src/services/chat-github-publication-lease.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, gt, lte, sql } from "drizzle-orm"; import { chatEndpointLeases, type Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; type Transaction = Parameters[0]>[0]; const TTL_SECONDS = 90; @@ -113,13 +114,9 @@ export async function withGitHubPublicationLease( } finally { clearInterval(timer); await renewal; - await db - .delete(chatEndpointLeases) - .where( - and( - eq(chatEndpointLeases.id, claimed.id), - eq(chatEndpointLeases.token, token), - ), - ); + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); + await tx.delete(chatEndpointLeases).where(and(eq(chatEndpointLeases.id, claimed.id), eq(chatEndpointLeases.token, token))); + }); } } diff --git a/server/src/services/chat-github-reviews.ts b/server/src/services/chat-github-reviews.ts index 2e3e27c2c5a..9367e02f1b3 100644 --- a/server/src/services/chat-github-reviews.ts +++ b/server/src/services/chat-github-reviews.ts @@ -8,6 +8,7 @@ import { runtimePublicOrigin } from "./cloud-runtime-identity.js"; import { projectSafeChatPublicationText } from "./chat-publication-projection.js"; import { githubReviewCheckService } from "./chat-github-checks.js"; import { githubReviewHeading } from "./chat-channels/github/review-heading.js"; +import { githubPublicationRetry, GITHUB_INLINE_RETRY_WINDOW_MS, GITHUB_INLINE_MAX_ATTEMPTS } from "./chat-channels/github/retry.js"; import { createHash } from "node:crypto"; import { githubResponseCommentService, writeGitHubResponseComment } from "./chat-github-response-comments.js"; import { and, desc, eq, gt, isNull, or, sql } from "drizzle-orm"; @@ -59,6 +60,7 @@ import { githubPreviousAssessment, } from "./chat-github-events.js"; import { logActivity } from "./activity-log.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; type Session = { companyId: string; @@ -457,8 +459,10 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { toolName: string, parameters: unknown, writer: Db | Parameters[0]>[0] = db, - ) { + ): Promise { + if (writer === db) return db.transaction(tx => stage(source, operation, payload, key, invocationId, toolName, parameters, tx)); const actionKey = `github_publication:${source.issue.id}:${operation}:${hash(key)}`; + await notifyDeliveryWork(writer, DELIVERY_QUEUES.chatGitHubPublications); const values = { companyId: source.endpoint.companyId, endpointId: source.endpoint.id, @@ -524,13 +528,75 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { receipt: action?.result ?? null, }; } + function publicationIsDue(action: { status: string; result: Record | null }) { + return action.status === "received" || (action.status === "failed" && action.result?.retryable === true && + Date.parse(String(action.result.retryAt)) <= Date.now()); + } + async function publishAndWait(actionId: string, deadline: number) { + let failures = 0; + while (Date.now() < deadline) { + await processPublication(actionId); + const result = await receipt(actionId); + if (result.status !== "received" && !(result.status === "failed" && result.receipt?.retryable === true)) return result; + if (result.status === "failed") failures++; + const delay = result.status === "received" ? 1000 : Math.max(0, Date.parse(String(result.receipt?.retryAt)) - Date.now()); + if (!Number.isFinite(delay) || Date.now() + delay >= deadline || failures >= GITHUB_INLINE_MAX_ATTEMPTS) return result; + await new Promise(resolve => setTimeout(resolve, delay)); + } + return receipt(actionId); + } + async function readPublicationReceipt(session: Session, name: string, input: unknown, invocationId: string) { + const source = await scope(session); + const [action] = await db.select().from(chatActions).where(and( + eq(chatActions.companyId, session.companyId), eq(chatActions.endpointId, source.endpoint.id), + eq(chatActions.conversationId, source.conversation.id), eq(chatActions.kind, "github_review_publication"), + sql`${chatActions.payload}->>'invocationId' = ${invocationId}`, sql`${chatActions.payload}->>'toolName' = ${name}`, + )); + if (!action) return null; + const binding = action.payload.session as Session; + if (binding.companyId !== session.companyId || binding.agentId !== session.agentId || binding.runId !== session.runId || + binding.issueId !== session.issueId || hash(action.payload.parameters) !== hash(input)) + throw conflict("This GitHub invocation is bound to a different publication"); + await assertPublicationAuthority(source, action); + const result = await receipt(action.id); + if (name !== "submit_review") return result; + const [review] = await db.select().from(chatGitHubReviews).where(and( + eq(chatGitHubReviews.companyId, session.companyId), eq(chatGitHubReviews.id, String(action.payload.reviewId)), + eq(chatGitHubReviews.runId, session.runId!), eq(chatGitHubReviews.endpointId, source.endpoint.id), + )); + if (!review?.assessment) throw conflict("GitHub publication assessment is unavailable"); + return { reviewId: review.id, ...result, score: review.assessment.score, conclusion: review.conclusion }; + } async function execute( session: Session, name: string, input: unknown, invocationId?: string, ) { + const deadline = Date.now() + GITHUB_INLINE_RETRY_WINDOW_MS; const source = await scope(session); + if (invocationId && ["comment", "update_comment", "submit_review"].includes(name)) { + const [existing] = await db.select().from(chatActions).where(and( + eq(chatActions.companyId, session.companyId), eq(chatActions.endpointId, source.endpoint.id), + eq(chatActions.conversationId, source.conversation.id), eq(chatActions.kind, "github_review_publication"), + sql`${chatActions.payload}->>'invocationId' = ${invocationId}`, + sql`${chatActions.payload}->>'toolName' = ${name}`, + )); + if (existing) { + const binding = existing.payload.session as Session; + if (binding.companyId !== session.companyId || binding.agentId !== session.agentId || + binding.runId !== session.runId || binding.issueId !== session.issueId || hash(existing.payload.parameters) !== hash(input)) + throw conflict("This GitHub invocation is bound to a different publication"); + const result = await publishAndWait(existing.id, deadline); + if (name !== "submit_review") return result; + const [review] = await db.select().from(chatGitHubReviews).where(and( + eq(chatGitHubReviews.companyId, session.companyId), eq(chatGitHubReviews.id, String(existing.payload.reviewId)), + eq(chatGitHubReviews.runId, session.runId!), eq(chatGitHubReviews.endpointId, source.endpoint.id), + )); + if (!review?.assessment) throw conflict("GitHub publication assessment is unavailable"); + return { reviewId: review.id, ...result, score: review.assessment.score, conclusion: review.conclusion }; + } + } const api = await client(source); if (name === "comment" || name === "update_comment") { const parsed = commentSchema.parse(input); @@ -543,8 +609,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { name, input, ); - await processPublication(actionId); - return receipt(actionId); + return publishAndWait(actionId, deadline); } if (source.isIssue) throw conflict( @@ -807,10 +872,9 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { tx, ); }); - await processPublication(actionId); return { reviewId: review.id, - ...(await receipt(actionId)), + ...(await publishAndWait(actionId, deadline)), score: assessment.score, conclusion: githubReviewConclusion( assessment, @@ -846,8 +910,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { name, input, ); - await processPublication(actionId); - return receipt(actionId); + return publishAndWait(actionId, deadline); } throw notFound("Unknown GitHub bot tool"); } @@ -936,7 +999,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { ); if ( invocation?.approvalState === "approved" && - ["executing", "completed"].includes(invocation.status) + ["executing", "completed", "succeeded", "failed", "timed_out"].includes(invocation.status) ) return; } @@ -957,8 +1020,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { .limit(1); if ( !action || - action.status === "processed" || - action.status === "cancelled" + !publicationIsDue(action) ) return; const [binding] = await db @@ -981,7 +1043,14 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { const number = binding?.thread.match( /^github:[^:]+:(?:issue:)?([1-9][0-9]*)(?::rc:[1-9][0-9]*)?$/, )?.[1]; - if (!number || !binding?.repository.providerRepositoryId) return; + if (!number || !binding?.repository.providerRepositoryId) { + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); + await tx.update(chatActions).set({ status: "cancelled", result: { code: "publication_binding_unavailable", retryable: false }, updatedAt: new Date() }) + .where(and(eq(chatActions.id, actionId), eq(chatActions.companyId, action.companyId))); + }); + return; + } await withGitHubPublicationLease( db, { @@ -996,11 +1065,11 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { .select() .from(chatActions) .where(eq(chatActions.id, actionId)); - if (!fresh || ["processed", "cancelled"].includes(fresh.status)) return; + if (!fresh || !publicationIsDue(fresh)) return; const session = action.payload.session as Session; - if (!session || session.companyId !== action.companyId) - throw forbidden("Invalid GitHub publication binding"); try { + if (!session || session.companyId !== action.companyId) + throw forbidden("Invalid GitHub publication binding"); const source = await scope(session, true); if ( source.endpoint.id !== action.endpointId || @@ -1339,6 +1408,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { }; } else throw forbidden("Unknown GitHub publication"); await lease.commit(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); await tx .update(chatActions) .set({ @@ -1405,16 +1475,14 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { ); }); await lease.commit(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); await tx .update(chatActions) .set({ status: superseded || denied ? "cancelled" : "failed", result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify({ - attempts, - retryable: !superseded && !denied && attempts < 8, - retryAt: new Date( - Date.now() + Math.min(300000, 1000 * 2 ** attempts), - ).toISOString(), + ...githubPublicationRetry(error, attempts), + ...(superseded || denied ? { retryable: false } : {}), code: superseded ? "stale_head" : denied @@ -1440,6 +1508,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { .where( and( eq(chatActions.kind, "github_review_publication"), + sql`coalesce(${publicationLeaseExpiresAt()}, now()) <= now()`, or( eq(chatActions.status, "received"), and( @@ -1461,5 +1530,24 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { await processPublication(row.id); return rows.length; } - return { execute, processPending, scope }; + function publicationLeaseExpiresAt() { + return sql`(select max(l.expires_at) from chat_endpoint_leases l + inner join chat_conversations c on c.id = ${chatActions.conversationId} and c.company_id = ${chatActions.companyId} + inner join chat_endpoint_resources r on r.id = c.resource_id and r.company_id = ${chatActions.companyId} + where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId} + and l.lease_key = 'github-publication:' || (r.metadata->>'providerRepositoryId') || ':' || + substring(c.external_thread_id from '^github:[^:]+:(?:issue:)?([1-9][0-9]*)'))`; + } + async function nextPendingAt(): Promise { + const [row] = await db.select({ at: sql`min(greatest( + case when ${chatActions.status} = 'received' then now() + else (${chatActions.result}->>'retryAt')::timestamptz end, + ${publicationLeaseExpiresAt()} + ))` }).from(chatActions).where(and( + eq(chatActions.kind, "github_review_publication"), + or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)), + )); + return row?.at ? new Date(row.at).getTime() : null; + } + return { execute, readPublicationReceipt, processPending, nextPendingAt, scope }; } diff --git a/server/src/services/chat-github-tools.ts b/server/src/services/chat-github-tools.ts index e958c9ed83f..4c4c8f73583 100644 --- a/server/src/services/chat-github-tools.ts +++ b/server/src/services/chat-github-tools.ts @@ -100,7 +100,7 @@ export const GITHUB_BOT_TOOLS = [ name: "submit_review", title: "Submit a review assessment", description: - "Submit a structured assessment for this task's exact PR head. Use begin_review before starting an explicitly requested review. Paperclip validates coverage and score, publishes allowed summary/findings, and computes the Paperclip Review check. Coverage reviewedPaths and omittedPaths name only allowed changed files from read_pull_request(files); describe additional context in the rationale. Follow the schema length limits. Incomplete analysis cannot pass. This never formally approves a PR.", + "Submit a structured assessment for this task's exact PR head. Use begin_review before starting an explicitly requested review. Paperclip validates coverage and score, publishes allowed summary/findings, and computes the Paperclip Review check. Coverage reviewedPaths and omittedPaths name only allowed changed files from read_pull_request(files); describe additional context in the rationale. Follow the schema length limits. Incomplete analysis cannot pass. This never formally approves a PR. Transient publication failures retry automatically. A received result or failed result with receipt.retryable=true remains queued; do not claim publication yet. After a timeout, repeat the exact call to recover its durable receipt. Do not begin another review or revise the assessment just to retry publication.", risk: "write", // Share the input contract with server validation so discovery includes every // length/array bound; hidden limits caused real agents to abandon publication. diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 658c5880dab..88387c21709 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = { chatSessionStops: "chat-session-stops", chatProviderEffects: "chat-provider-effects", chatGitHubIngress: "chat-github-ingress", + chatGitHubPublications: "chat-github-publications", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", chatTelegramSubscriptions: "chat-telegram-subscriptions", diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 902d5e0d294..b107011c696 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -10150,12 +10150,21 @@ export function createToolGatewayService( argumentsHash: argumentValidation.summary.sha256 ?? "", }); if (replay?.matched) { + let replayResult: unknown = replay.result; + if (tool.providerType === "paperclip_github_chat" && ["comment", "update_comment", "submit_review"].includes(tool.upstreamToolName ?? "")) { + const access = await policyService.decide(policyInputForTool({ session, tool, parameters: effectiveParameters, consumeRateLimit: false })); + if (!access.allowed && access.decision !== "require_approval") + throw new ToolGatewayHttpError(policyErrorStatus(access), access.explanation, access.reasonCode); + const receipt = await githubChatReviewService(db).readPublicationReceipt(session, tool.upstreamToolName ?? "", effectiveParameters, replay.invocationId); + if (receipt) replayResult = validateToolContent({ value: { content: JSON.stringify(receipt), data: receipt }, + direction: "result", sensitiveMode: "redact", promptInjectionMode: "block" }).value; + } return { invocationId: replay.invocationId, status: "replayed" as const, tool: virtualToolName ?? tool.name, targetTool: virtualToolName ? tool.name : undefined, - result: replay.result, + result: replayResult, }; } } @@ -10542,7 +10551,23 @@ export function createToolGatewayService( const retryingSlackRateLimit = recorded.replayed && accessDecision.allowed && tool.providerType === "paperclip_slack_chat" && session.agentId && session.runId && session.issueId ? await claimSlackRateLimitRetry(db, { companyId: session.companyId, agentId: session.agentId, runId: session.runId, issueId: session.issueId, endpointId: String(asRecord(tool.providerMetadata)?.endpointId ?? ""), identityContextId: session.identityContextId }, invocationId) : false; - if (recorded.replayed && !retryingSlackRateLimit) { + const gitHubPublicationReplay = recorded.replayed && + tool.providerType === "paperclip_github_chat" && + ["comment", "update_comment", "submit_review"].includes(tool.upstreamToolName ?? "") && + recorded.invocation.approvalState === "not_required"; + const gitHubReplayPolicyAllowed = accessDecision.allowed; + const gitHubReplayBindingMatches = + recorded.invocation.agentId === session.agentId && recorded.invocation.runId === session.runId && + recorded.invocation.issueId === session.issueId && recorded.invocation.connectionId === tool.connectionId && + recorded.invocation.toolName === tool.name && recorded.invocation.argumentsHash === effectiveArgumentsSummary.sha256 && + ["authorized", "executing", "succeeded", "failed", "timed_out"].includes(recorded.invocation.status); + if (gitHubPublicationReplay && gitHubReplayPolicyAllowed && !gitHubReplayBindingMatches) + throw new ToolGatewayHttpError(409, "Idempotency key belongs to a different GitHub publication", "github_idempotency_conflict"); + const retryingGitHubPublication = gitHubPublicationReplay && accessDecision.allowed && + recorded.invocation.approvalState === "not_required" && gitHubReplayBindingMatches; + if (recorded.replayed && !retryingSlackRateLimit && !retryingGitHubPublication) { + if (gitHubPublicationReplay && !gitHubReplayPolicyAllowed) + throw new ToolGatewayHttpError(policyErrorStatus(accessDecision), accessDecision.explanation, accessDecision.reasonCode); await writeAudit({ session, companyId: session.companyId, @@ -10618,6 +10643,7 @@ export function createToolGatewayService( .update(toolInvocations) .set({ status: "executing", + ...(retryingGitHubPublication ? { errorCode: null, errorMessage: null, completedAt: null } : {}), startedAt: new Date(), updatedAt: new Date(), })