import { notifyGitHubReviewProjection, repairSupersededGitHubReviews } from "./chat-channels/github/review-projection-work.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { notifyGitHubReviewChecks, registerGitHubReviewCheckWork } from "./chat-channels/github/review-check-work.js"; import { githubReviewCheckState } from "./chat-channels/github/review-check-state.js"; import { GitHubPublicationLeaseLost, withGitHubPublicationLease, } from "./chat-github-publication-lease.js"; import { and, eq, isNull, or, sql } from "drizzle-orm"; import { chatActions, chatEndpoints, chatGitHubConfigurations, chatGitHubReviews, type Db, } from "@paperclipai/db"; import type { GitHubReviewEventContext } from "@paperclipai/shared"; import { runtimePublicOrigin } from "./cloud-runtime-identity.js"; import { githubBotRepositoryToken, githubBotRequest, } from "./chat-github-client.js"; /** Check status follows the ordinary chat delivery/task/run. This outbox never * schedules an agent. Assessments are published by the governed review tool. */ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { async function enqueue( endpoint: typeof chatEndpoints.$inferSelect, event: GitHubReviewEventContext, admitted: boolean, reason: string, ) { const [config] = await db .select() .from(chatGitHubConfigurations) .where(eq(chatGitHubConfigurations.endpointId, endpoint.id)); if (!config?.configuration.toolsEnabled) return; const policy = { ...config.configuration.defaults, ...config.configuration.repositories[event.repositoryId], }; if (policy.ratingThreshold === null && !admitted) return; await db.transaction(async tx => { await notifyGitHubReviewChecks(tx, { companyId: endpoint.companyId, endpointId: endpoint.id, repositoryId: event.repositoryId, number: event.pullNumber }); await tx .insert(chatActions) .values({ companyId: endpoint.companyId, endpointId: endpoint.id, kind: "github_review_check", providerActionId: `github-check:${event.deliveryId}`, payload: { version: 1, event, admitted, reason, runtimeGeneration: (endpoint.setup as { runtimeGeneration?: number }) .runtimeGeneration ?? 0, }, status: "received", }) .onConflictDoNothing(); }); } async function dispatch(id: string, canContinue = () => true) { if (!canContinue()) return; const [action] = await db .select() .from(chatActions) .where( and( eq(chatActions.id, id), eq(chatActions.kind, "github_review_check"), ), ) .limit(1); if (!action || ["processed", "cancelled"].includes(action.status)) return; const event = action.payload.event as GitHubReviewEventContext; const guardedFetch: typeof fetch = (input, init) => { if (!canContinue()) throw new GitHubPublicationLeaseLost(); return fetchImpl(input, init); }; if (!canContinue()) return; await withGitHubPublicationLease( db, { companyId: action.companyId, endpointId: action.endpointId, repositoryId: event.repositoryId, number: event.pullNumber, }, guardedFetch, async (lease) => { const [fresh] = await db .select() .from(chatActions) .where(eq(chatActions.id, id)); if (!fresh || ["processed", "cancelled"].includes(fresh.status)) return; let attempts = Number(fresh.result?.attempts ?? 0); try { const currentState = await githubReviewCheckState(db, fresh); if (currentState.disposition !== "publish") { await lease.commit(async tx => { await tx.update(chatActions).set({ status: currentState.disposition === "connection_changed" ? "cancelled" : "processed", result: { code: currentState.disposition, ...(currentState.disposition === "assessment_owns_check" ? { reviewId: currentState.reviewId } : {}) }, updatedAt: new Date(), }).where(eq(chatActions.id, id)); }); return; } const { source, review, run, terminal, state, status } = currentState; const policy = review?.policySnapshot ?? { ...source.configuration.defaults, ...source.configuration.repositories[event.repositoryId], }; const conclusion = policy.ratingThreshold === null ? "neutral" : "action_required"; const title = state === "manual_required" ? "Authorized manual review required" : state === "incomplete" ? "Agent finished without a complete assessment" : state === "error" ? "Review execution did not complete" : state === "running" ? "Agent is reviewing this commit" : "Waiting for the assigned agent"; const token = await githubBotRepositoryToken( db, action.companyId, action.endpointId, event.repositoryId, lease.fetch, ); const prefix = `/repos/${event.repository.split("/").map(encodeURIComponent).join("/")}`; const request = ( path: string, options?: Parameters[3], ) => githubBotRequest( lease.fetch, token, `${prefix}${path}`, options, ); const current = await request<{ head: { sha: string } }>( `/pulls/${event.pullNumber}`, ); if (current.head.sha !== event.headSha) { if (review) await lease.commit(async (tx) => { await tx .update(chatGitHubReviews) .set({ state: "superseded", updatedAt: new Date() }) .where( and( eq(chatGitHubReviews.id, review.id), isNull(chatGitHubReviews.assessment), ), ); }); await lease.commit(async (tx) => { await notifyGitHubReviewProjection(tx, { companyId: action.companyId, endpointId: action.endpointId }); await tx .update(chatActions) .set({ status: "cancelled", result: { code: "stale_head" }, updatedAt: new Date(), }) .where(eq(chatActions.id, id)); }); return; } if (fresh.result?.state !== state) { const origin = runtimePublicOrigin(); const detailsPath = review ? `issues/${review.issueId}` : `apps/chat/${source.endpoint.id}/reviews`; const detailsUrl = origin ? `${origin}/${encodeURIComponent(source.companyPrefix)}/${detailsPath}` : null; const externalId = `${action.endpointId}:${event.pullNumber}:${event.headSha}`; const checks = await request<{ check_runs: Array<{ id: number; status?: string; external_id?: string; app?: { id?: number }; }>; }>( `/commits/${event.headSha}/check-runs?check_name=Paperclip%20Review&per_page=100`, ); const check = checks.check_runs.find( (check) => check.external_id === externalId && String(check.app?.id) === source.endpoint.botExternalId, ); // GitHub retains a completed check's conclusion when PATCHed back to // queued/in_progress. A fresh attempt needs a new check run with the // same stable name; otherwise a prior success still looks passing. const updateCheck = check && !(check.status === "completed" && status !== "completed"); const posted = await request<{ id: number; html_url: string }>( updateCheck ? `/check-runs/${check.id}` : "/check-runs", { method: updateCheck ? "PATCH" : "POST", body: { name: "Paperclip Review", head_sha: event.headSha, external_id: externalId, ...(detailsUrl ? { details_url: detailsUrl } : {}), status, ...(status === "completed" ? { conclusion } : {}), output: { title, summary: `${title}. ${state === "manual_required" ? "An authorized person can mention the bot to request a review of this head. Automatic review was not authorized by the bot configuration." : "This check follows the assigned Paperclip agent's task execution. Only a validated assessment can produce a passing score."}\n\nCommit: ${event.headSha}`, }, }, }, ); if (review) await lease.commit(async (tx) => { await notifyGitHubReviewProjection(tx, { companyId: action.companyId, endpointId: action.endpointId }); await tx .update(chatGitHubReviews) .set({ state, runId: run?.id ?? review.runId, checkId: String(posted.id), checkUrl: posted.html_url, ...(terminal ? { conclusion } : {}), updatedAt: new Date(), }) .where( and( eq(chatGitHubReviews.id, review.id), isNull(chatGitHubReviews.assessment), ), ); }); } await lease.commit(async (tx) => { await tx .update(chatActions) .set({ status: status === "completed" ? "processed" : "received", result: { state, attempts: 0, retryable: true, retryAt: new Date(Date.now() + 5000).toISOString(), }, updatedAt: new Date(), }) .where(eq(chatActions.id, id)); }); } catch (error) { if (error instanceof GitHubPublicationLeaseLost || !canContinue()) return; attempts++; await lease.commit(async (tx) => { await tx .update(chatActions) .set({ status: "failed", result: { attempts, retryable: attempts < 8, retryAt: new Date( Date.now() + Math.min(300000, 1000 * 2 ** attempts), ).toISOString(), code: "check_publication_failed", }, updatedAt: new Date(), }) .where(eq(chatActions.id, id)); }); } }, ); } const repairSupersededReviews = () => repairSupersededGitHubReviews(db); async function processPending(limit = 10) { await repairSupersededReviews(); const rows = await db .select({ id: chatActions.id }) .from(chatActions) .where( and( eq(chatActions.kind, "github_review_check"), or( and( eq(chatActions.status, "received"), sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= now())`, ), and( eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`, sql`(${chatActions.result}->>'retryAt')::timestamptz <= now()`, ), ), ), ) .orderBy(chatActions.updatedAt) .limit(limit); for (const row of rows) await dispatch(row.id); } return { enqueue, processPending, repairSupersededReviews, registerCheckWork: (coordinator: ReturnType, canRun: () => boolean) => registerGitHubReviewCheckWork(db, coordinator, dispatch, canRun), }; }