diff --git a/server/src/__tests__/github-review-wait.test.ts b/server/src/__tests__/github-review-wait.test.ts index cad6571fe42..f4874dff9a1 100644 --- a/server/src/__tests__/github-review-wait.test.ts +++ b/server/src/__tests__/github-review-wait.test.ts @@ -6,7 +6,41 @@ import { agents, chatActions, chatConversations, chatEndpointResources, chatEndp import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { captureReviewWaitQueries, seedGitHubReviewWait, setGitHubReviewAssessment, type CapturedReviewWaitQuery } from "./helpers/github-review-wait.js"; -import { readInheritedGitHubReviewWait } from "../services/chat-channels/github/review-wait.js"; +import { readInheritedGitHubReviewWait, readInheritedGitHubReviewWaits } from "../services/chat-channels/github/review-wait.js"; + +const reviewWaitChanges = ["disabled", "inactive", "review_state", "review_agent", "repository", "repository_id", + "thread", "publication", "identity", "endpoint_agent", "company"] as const; + +async function applyReviewWaitChange(db: Db, f: Awaited>, + change: typeof reviewWaitChanges[number]) { + if (change === "disabled") await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, f.connectionId)); + if (change === "inactive") await db.update(toolConnections).set({ status: "disabled" }).where(eq(toolConnections.id, f.connectionId)); + if (change === "review_state") await db.update(chatGitHubReviews).set({ state: "queued" }).where(eq(chatGitHubReviews.id, f.reviewId)); + if (change === "review_agent" || change === "endpoint_agent") { + const [other] = await db.insert(agents).values({ companyId: f.companyId, name: "Other", adapterType: "process" }).returning(); + if (change === "review_agent") await db.update(heartbeatRuns).set({ agentId: other!.id }).where(eq(heartbeatRuns.id, f.reviewRunId)); + else await db.update(chatEndpoints).set({ assignedAgentId: other!.id }).where(eq(chatEndpoints.id, f.endpointId)); + } + if (change === "repository") await db.update(chatEndpointResources).set({ providerResourceId: "other/repository" }).where(eq(chatEndpointResources.id, f.resourceId)); + if (change === "repository_id") await db.update(chatEndpointResources).set({ metadata: { providerRepositoryId: "43" } }).where(eq(chatEndpointResources.id, f.resourceId)); + if (change === "thread") await db.update(chatConversations).set({ externalThreadId: "github:synthetic/repository:8" }).where(eq(chatConversations.id, f.conversationId)); + if (change === "publication") await db.update(chatActions).set({ status: "received" }).where(eq(chatActions.id, f.publicationId)); + if (change === "identity") await db.update(runIdentityContexts).set({ status: "rejected" }).where(eq(runIdentityContexts.id, f.activeIdentityContextId!)); + if (change === "company") { + const otherCompanyId = randomUUID(); + await db.insert(companies).values({ id: otherCompanyId, name: "Other", issuePrefix: otherCompanyId }); + f.run = { ...f.run, companyId: otherCompanyId }; + } +} + +/** Record the SQL text of every statement that the client sends. */ +async function withStatements(db: Db, read: () => Promise) { + const client = db.$client as unknown as { unsafe: (query: string, ...rest: unknown[]) => unknown }; + const unsafe = client.unsafe; + const sql: string[] = []; + client.unsafe = (query: string, ...rest: unknown[]) => { sql.push(query); return unsafe.call(client, query, ...rest); }; + try { return { result: await read(), sql }; } finally { client.unsafe = unsafe; } +} describe("inherited GitHub review wait projection", () => { let temporary: Awaited>; @@ -54,32 +88,54 @@ describe("inherited GitHub review wait projection", () => { expect(await readInheritedGitHubReviewWait(db, f.run)).toEqual({ kind: "revoked" }); }); - it.each(["disabled", "inactive", "review_state", "review_agent", "repository", "repository_id", "thread", "publication", "identity", "endpoint_agent", "company"] as const)( + it.each(reviewWaitChanges)( "continues to reject a changed %s binding", async change => { const f = await seedGitHubReviewWait(db); expect((await readInheritedGitHubReviewWait(db, f.run))?.kind).toBe("published"); - if (change === "disabled") await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, f.connectionId)); - if (change === "inactive") await db.update(toolConnections).set({ status: "disabled" }).where(eq(toolConnections.id, f.connectionId)); - if (change === "review_state") await db.update(chatGitHubReviews).set({ state: "queued" }).where(eq(chatGitHubReviews.id, f.reviewId)); - if (change === "review_agent" || change === "endpoint_agent") { - const [other] = await db.insert(agents).values({ companyId: f.companyId, name: "Other", adapterType: "process" }).returning(); - if (change === "review_agent") await db.update(heartbeatRuns).set({ agentId: other!.id }).where(eq(heartbeatRuns.id, f.reviewRunId)); - else await db.update(chatEndpoints).set({ assignedAgentId: other!.id }).where(eq(chatEndpoints.id, f.endpointId)); - } - if (change === "repository") await db.update(chatEndpointResources).set({ providerResourceId: "other/repository" }).where(eq(chatEndpointResources.id, f.resourceId)); - if (change === "repository_id") await db.update(chatEndpointResources).set({ metadata: { providerRepositoryId: "43" } }).where(eq(chatEndpointResources.id, f.resourceId)); - if (change === "thread") await db.update(chatConversations).set({ externalThreadId: "github:synthetic/repository:8" }).where(eq(chatConversations.id, f.conversationId)); - if (change === "publication") await db.update(chatActions).set({ status: "received" }).where(eq(chatActions.id, f.publicationId)); - if (change === "identity") await db.update(runIdentityContexts).set({ status: "rejected" }).where(eq(runIdentityContexts.id, f.activeIdentityContextId!)); - if (change === "company") { - const otherCompanyId = randomUUID(); - await db.insert(companies).values({ id: otherCompanyId, name: "Other", issuePrefix: otherCompanyId }); - f.run = { ...f.run, companyId: otherCompanyId }; - } + await applyReviewWaitChange(db, f, change); expect((await readInheritedGitHubReviewWait(db, f.run))?.kind).not.toBe("published"); + expect((await readInheritedGitHubReviewWaits(db, [f.run]))[0]?.kind).not.toBe("published"); }, ); + it("reads the waits of many runs with one review read and the same results", async () => { + const [companyId, agentId] = [randomUUID(), randomUUID()]; + await db.insert(companies).values({ id: companyId, name: "Batch", issuePrefix: companyId }); + await db.insert(agents).values({ id: agentId, companyId, name: "Reviewer", adapterType: "process" }); + const fixtures: Array>> = []; + for (const change of [null, ...reviewWaitChanges, "newer_head", "no_identity", "no_issue", null] as const) { + const f = await seedGitHubReviewWait(db, change === "company" ? undefined : { companyId, agentId }); + if (change === "newer_head") { + const [review] = await db.select().from(chatGitHubReviews).where(eq(chatGitHubReviews.id, f.reviewId)); + await db.insert(chatGitHubReviews).values({ ...review!, id: randomUUID(), deliveryId: randomUUID(), + headSha: "2".repeat(40), state: "queued", createdAt: new Date(review!.createdAt.getTime() + 1_000) }); + } else if (change === "no_identity") f.run = { ...f.run, activeIdentityContextId: null }; + else if (change === "no_issue") f.run = { ...f.run, nativeIssueId: null }; + else if (change) await applyReviewWaitChange(db, f, change); + fixtures.push(f); + } + // A run can be read twice in one pass. + const [first] = fixtures; + // The "company" wait moved its run to a third company; its own company + // also has one published wait. + const otherCompany = await seedGitHubReviewWait(db); + const runs = [...fixtures.map(f => f.run), first!.run, otherCompany.run]; + const single = []; + for (const run of runs) single.push(await readInheritedGitHubReviewWait(db, run)); + expect(single.filter(wait => wait?.kind === "published")).toHaveLength(4); + // No identity, no task, and an endpoint of another agent have no inherited wait. + expect(single.filter(wait => wait === null)).toHaveLength(3); + expect(single.filter(wait => wait?.kind === "revoked")).toHaveLength(runs.length - 7); + + const { result, sql } = await withStatements(db, () => readInheritedGitHubReviewWaits(db, runs)); + expect(result).toEqual(single); + // One review read and one publication read for each company with a source + // comment, not one of each for every run. + expect(sql.filter(text => /join "chat_github_reviews"/i.test(text))).toHaveLength(2); + expect(sql.filter(text => /from "chat_actions"/i.test(text))).toHaveLength(2); + expect(await readInheritedGitHubReviewWaits(db, [])).toEqual([]); + }); + it("binds each call afresh while reusing the compiled query", async () => { const first = await seedGitHubReviewWait(db), second = await seedGitHubReviewWait(db); const queries: CapturedReviewWaitQuery[] = []; diff --git a/server/src/__tests__/helpers/github-review-wait.ts b/server/src/__tests__/helpers/github-review-wait.ts index ecdcbe0075f..789c4fe9fae 100644 --- a/server/src/__tests__/helpers/github-review-wait.ts +++ b/server/src/__tests__/helpers/github-review-wait.ts @@ -5,14 +5,19 @@ import { agents, chatActions, chatConversations, chatDeliveries, chatEndpointRes runIdentityContexts, toolApplications, toolConnections, type Db } from "@paperclipai/db"; import { defaultGitHubReviewPolicy } from "@paperclipai/shared"; -export async function seedGitHubReviewWait(db: Db) { - const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); +/** Seed one published inherited review wait. With `into`, the wait is added to + * an existing company and agent. */ +export async function seedGitHubReviewWait(db: Db, into?: { companyId: string; agentId: string }) { + const companyId = into?.companyId ?? randomUUID(), agentId = into?.agentId ?? randomUUID(); + const issueId = randomUUID(), runId = randomUUID(); const reviewRunId = randomUUID(), reviewId = randomUUID(), endpointId = randomUUID(), resourceId = randomUUID(); const conversationId = randomUUID(), commentId = randomUUID(), activeIdentityContextId = randomUUID(); const applicationId = randomUUID(), connectionId = randomUUID(), publicationId = randomUUID(); const headSha = "1".repeat(40), repository = "synthetic/repository", repositoryId = "42", pullNumber = 7; - await db.insert(companies).values({ id: companyId, name: "Synthetic review", issuePrefix: companyId }); - await db.insert(agents).values({ id: agentId, companyId, name: "Reviewer", adapterType: "process" }); + if (!into) { + await db.insert(companies).values({ id: companyId, name: "Synthetic review", issuePrefix: companyId }); + await db.insert(agents).values({ id: agentId, companyId, name: "Reviewer", adapterType: "process" }); + } await db.insert(issues).values({ id: issueId, companyId, title: "Review", assigneeAgentId: agentId }); await db.insert(heartbeatRuns).values([ { id: runId, companyId, agentId, nativeIssueId: issueId, responsibleUserId: "operator", activeIdentityContextId }, @@ -21,7 +26,7 @@ export async function seedGitHubReviewWait(db: Db) { await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorUserId: "operator", body: "Review" }); await db.insert(runIdentityContexts).values({ id: activeIdentityContextId, companyId, runId, revision: 1, responsibleUserId: "operator", messageId: commentId, cause: "test", correlationId: "test", status: "accepted" }); - await db.insert(toolApplications).values({ id: applicationId, companyId, name: "GitHub", type: "chat" }); + await db.insert(toolApplications).values({ id: applicationId, companyId, name: into ? `GitHub ${applicationId}` : "GitHub", type: "chat" }); await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "GitHub", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel", enabled: true, status: "active" }); await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "github", diff --git a/server/src/services/chat-channels/github/review-wait.ts b/server/src/services/chat-channels/github/review-wait.ts index c3481257780..7f490c327a4 100644 --- a/server/src/services/chat-channels/github/review-wait.ts +++ b/server/src/services/chat-channels/github/review-wait.ts @@ -22,7 +22,7 @@ export async function hasPublishedGitHubReview(db: Db, binding: { eq(chatActions.companyId, binding.companyId), or(...reviews.map(review => and( eq(chatActions.endpointId, review.endpointId), - eq(chatActions.providerActionId, `github_publication:${binding.issueId}:assessment:${createHash("sha256").update(JSON.stringify(review.id)).digest("hex")}`), + eq(chatActions.providerActionId, publicationActionId(binding.issueId, review.id)), ))), eq(chatActions.kind, "github_review_publication"), eq(chatActions.status, "processed"), @@ -33,20 +33,24 @@ export async function hasPublishedGitHubReview(db: Db, binding: { type WaitRun = Pick; +/** The columns of one inherited review. The single read and the batch read + * select the same columns. */ +const inheritedReviewFields = { + id: chatGitHubReviews.id, endpointId: chatGitHubReviews.endpointId, + state: chatGitHubReviews.state, repository: chatGitHubReviews.repository, + repositoryId: chatGitHubReviews.repositoryId, headSha: chatGitHubReviews.headSha, + reviewedCommit: sql`${chatGitHubReviews.assessment}->'reviewedCommit'`, reviewRunId: chatGitHubReviews.runId, + connectionEnabled: toolConnections.enabled, connectionStatus: toolConnections.status, + reviewAgentId: heartbeatRuns.agentId, reviewCompanyId: heartbeatRuns.companyId, + resourceRepository: chatEndpointResources.providerResourceId, + resourceRepositoryId: sql`${chatEndpointResources.metadata}->>'providerRepositoryId'`, + pullNumber: chatGitHubReviews.pullNumber, thread: chatConversations.externalThreadId, +}; + const inheritedReviewQueries = new WeakMap>(); function prepareInheritedReviewQuery(db: Db) { - return db.select({ - id: chatGitHubReviews.id, endpointId: chatGitHubReviews.endpointId, - state: chatGitHubReviews.state, repository: chatGitHubReviews.repository, - repositoryId: chatGitHubReviews.repositoryId, headSha: chatGitHubReviews.headSha, - reviewedCommit: sql`${chatGitHubReviews.assessment}->'reviewedCommit'`, reviewRunId: chatGitHubReviews.runId, - connectionEnabled: toolConnections.enabled, connectionStatus: toolConnections.status, - reviewAgentId: heartbeatRuns.agentId, reviewCompanyId: heartbeatRuns.companyId, - resourceRepository: chatEndpointResources.providerResourceId, - resourceRepositoryId: sql`${chatEndpointResources.metadata}->>'providerRepositoryId'`, - pullNumber: chatGitHubReviews.pullNumber, thread: chatConversations.externalThreadId, - }).from(chatMessageLinks) + return db.select(inheritedReviewFields).from(chatMessageLinks) .innerJoin(chatConversations, and(eq(chatConversations.id, chatMessageLinks.conversationId), eq(chatConversations.companyId, sql.placeholder("companyId")), eq(chatConversations.issueId, sql.placeholder("issueId")))) .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatConversations.endpointId), @@ -65,6 +69,8 @@ function prepareInheritedReviewQuery(db: Db) { .prepare("inherited_github_review_wait"); } +type InheritedReviewRow = Awaited["execute"]>>[number]; + function inheritedReviewQuery(db: Db) { let query = inheritedReviewQueries.get(db); if (!query) { @@ -74,35 +80,77 @@ function inheritedReviewQuery(db: Db) { return query; } -/** Passive review waits inherit the same accepted requester lineage as GitHub - * tools. This is not a grant to publish, run a monitor, or reuse attachments. */ -export async function readInheritedGitHubReviewWait(db: Db, run: WaitRun) { +type InheritedReviewKey = { commentId: string; issueId: string; agentId: string }; + +function inheritedReviewKey(key: InheritedReviewKey) { + return `${key.commentId}:${key.issueId}:${key.agentId}`; +} + +/** The latest review for each (source comment, task, agent) key of one + * company. For each key this is the row of the single prepared read: the same + * joins and filters, and DISTINCT ON keeps the first row in the same order. */ +async function readInheritedReviewsForKeys(db: Db, companyId: string, keys: InheritedReviewKey[]) { + const unique = (values: string[]) => [...new Set(values)]; + const rows = await db.selectDistinctOn([chatMessageLinks.commentId, chatConversations.issueId, chatEndpoints.assignedAgentId], { + ...inheritedReviewFields, + keyCommentId: chatMessageLinks.commentId, keyIssueId: chatConversations.issueId, keyAgentId: chatEndpoints.assignedAgentId, + }).from(chatMessageLinks) + .innerJoin(chatConversations, and(eq(chatConversations.id, chatMessageLinks.conversationId), + eq(chatConversations.companyId, companyId), + sql`${chatConversations.issueId} = any(${sql.param(unique(keys.map(key => key.issueId)))}::uuid[])`)) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatConversations.endpointId), + eq(chatEndpoints.companyId, companyId), eq(chatEndpoints.provider, "github"), + sql`${chatEndpoints.assignedAgentId} = any(${sql.param(unique(keys.map(key => key.agentId)))}::uuid[])`)) + .leftJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), + eq(toolConnections.companyId, companyId))) + .innerJoin(chatEndpointResources, and(eq(chatEndpointResources.id, chatConversations.resourceId), + eq(chatEndpointResources.companyId, companyId), eq(chatEndpointResources.endpointId, chatEndpoints.id))) + .innerJoin(chatGitHubReviews, and(eq(chatGitHubReviews.endpointId, chatEndpoints.id), + eq(chatGitHubReviews.companyId, companyId), eq(chatGitHubReviews.issueId, chatConversations.issueId))) + .leftJoin(heartbeatRuns, eq(heartbeatRuns.id, chatGitHubReviews.runId)) + .where(and(eq(chatMessageLinks.companyId, companyId), + sql`${chatMessageLinks.commentId} = any(${sql.param(unique(keys.map(key => key.commentId)))}::uuid[])`, + eq(chatMessageLinks.direction, "inbound"))) + .orderBy(chatMessageLinks.commentId, chatConversations.issueId, chatEndpoints.assignedAgentId, + desc(chatGitHubReviews.createdAt), desc(chatGitHubReviews.id)); + // The array filters can also match a mix of two keys. Only exact keys are read. + const byKey = new Map(); + for (const { keyCommentId, keyIssueId, keyAgentId, ...review } of rows) { + if (keyCommentId && keyIssueId && keyAgentId) { + byKey.set(inheritedReviewKey({ commentId: keyCommentId, issueId: keyIssueId, agentId: keyAgentId }), review); + } + } + return byKey; +} + +function publicationActionId(issueId: string, reviewId: string) { + return `github_publication:${issueId}:assessment:${createHash("sha256").update(JSON.stringify(reviewId)).digest("hex")}`; +} + +/** The accepted GitHub source comment of the run. A changed requester lineage + * revokes the inherited wait. */ +async function inheritedSourceComment(db: Db, run: WaitRun): Promise<{ kind: "revoked" } | { commentId: string } | null> { if (!run.activeIdentityContextId || !run.nativeIssueId) return null; let commentId: string | null; try { commentId = await chatRunSourceMessageId(db, run, { provider: "github" }); } catch (error) { - if (error instanceof HttpError && error.status === 403) return { kind: "revoked" as const }; + if (error instanceof HttpError && error.status === 403) return { kind: "revoked" }; throw error; } - if (!commentId) return null; - const [review] = await inheritedReviewQuery(db).execute({ - companyId: run.companyId, issueId: run.nativeIssueId, agentId: run.agentId, commentId, - }); - // A newer unreviewed head cannot inherit the prior head's publication. - if (!review) return null; - if (!review.connectionEnabled || review.connectionStatus !== "active" || review.state !== "completed" || - review.reviewCompanyId !== run.companyId || review.reviewAgentId !== run.agentId || - review.reviewedCommit !== review.headSha || - review.resourceRepository.toLowerCase() !== review.repository.toLowerCase() || - review.resourceRepositoryId !== review.repositoryId || - review.thread.toLowerCase() !== `github:${review.repository}:${review.pullNumber}`.toLowerCase()) return { kind: "revoked" as const }; - const [publication] = await db.select({ id: chatActions.id }).from(chatActions).where(and( - eq(chatActions.companyId, run.companyId), eq(chatActions.endpointId, review.endpointId), - eq(chatActions.kind, "github_review_publication"), eq(chatActions.status, "processed"), - eq(chatActions.providerActionId, `github_publication:${run.nativeIssueId}:assessment:${createHash("sha256").update(JSON.stringify(review.id)).digest("hex")}`), - )).limit(1); - if (!publication) return { kind: "revoked" as const }; + return commentId ? { commentId } : null; +} + +function inheritedReviewRevoked(run: WaitRun, review: InheritedReviewRow) { + return !review.connectionEnabled || review.connectionStatus !== "active" || review.state !== "completed" || + review.reviewCompanyId !== run.companyId || review.reviewAgentId !== run.agentId || + review.reviewedCommit !== review.headSha || + review.resourceRepository.toLowerCase() !== review.repository.toLowerCase() || + review.resourceRepositoryId !== review.repositoryId || + review.thread.toLowerCase() !== `github:${review.repository}:${review.pullNumber}`.toLowerCase(); +} + +function publishedInheritedWait(run: WaitRun, commentId: string, review: InheritedReviewRow) { // The caller must still verify live task ownership, destination and principal // policy. This normalized source is used only by the passive-wait boundary. return { kind: "published" as const, endpointId: review.endpointId, reviewId: review.id, reviewRunId: review.reviewRunId, @@ -113,3 +161,74 @@ export async function readInheritedGitHubReviewWait(db: Db, run: WaitRun) { issue: { id: run.nativeIssueId }, commentIds: [commentId] }, } }; } + +export type InheritedGitHubReviewWait = { kind: "revoked" } | ReturnType | null; + +/** Passive review waits inherit the same accepted requester lineage as GitHub + * tools. This is not a grant to publish, run a monitor, or reuse attachments. */ +export async function readInheritedGitHubReviewWait(db: Db, run: WaitRun): Promise { + const source = await inheritedSourceComment(db, run); + if (!source || !("commentId" in source)) return source; + const { commentId } = source; + const [review] = await inheritedReviewQuery(db).execute({ + companyId: run.companyId, issueId: run.nativeIssueId, agentId: run.agentId, commentId, + }); + // A newer unreviewed head cannot inherit the prior head's publication. + if (!review) return null; + if (inheritedReviewRevoked(run, review)) return { kind: "revoked" }; + const [publication] = await db.select({ id: chatActions.id }).from(chatActions).where(and( + eq(chatActions.companyId, run.companyId), eq(chatActions.endpointId, review.endpointId), + eq(chatActions.kind, "github_review_publication"), eq(chatActions.status, "processed"), + eq(chatActions.providerActionId, publicationActionId(run.nativeIssueId!, review.id)), + )).limit(1); + if (!publication) return { kind: "revoked" }; + return publishedInheritedWait(run, commentId, review); +} + +/** `readInheritedGitHubReviewWait` for the runs of one read-only pass, such + * as the waits of an issue page. Each run gets the same result as a separate + * call. The source lineage is still checked for each run. The reviews and the + * publications of all the runs of one company are read with one statement + * each, not with two statements for each run. */ +export async function readInheritedGitHubReviewWaits(db: Db, runs: WaitRun[]): Promise { + const sources = await Promise.all(runs.map(run => inheritedSourceComment(db, run))); + const results: InheritedGitHubReviewWait[] = sources.map(source => source && !("commentId" in source) ? source : null); + const byCompany = new Map>(); + runs.forEach((run, index) => { + const source = sources[index]; + if (!source || !("commentId" in source)) return; + const entries = byCompany.get(run.companyId) ?? []; + entries.push({ index, run, commentId: source.commentId }); + byCompany.set(run.companyId, entries); + }); + for (const [companyId, entries] of byCompany) { + const keyOf = ({ run, commentId }: { run: WaitRun; commentId: string }) => + ({ commentId, issueId: run.nativeIssueId!, agentId: run.agentId }); + const reviews = await readInheritedReviewsForKeys(db, companyId, entries.map(keyOf)); + const reviewed: Array<{ index: number; run: WaitRun; commentId: string; review: InheritedReviewRow; actionId: string }> = []; + for (const entry of entries) { + const review = reviews.get(inheritedReviewKey(keyOf(entry))); + // A newer unreviewed head cannot inherit the prior head's publication. + if (!review) continue; + if (inheritedReviewRevoked(entry.run, review)) { + results[entry.index] = { kind: "revoked" }; + continue; + } + reviewed.push({ ...entry, review, actionId: publicationActionId(entry.run.nativeIssueId!, review.id) }); + } + if (!reviewed.length) continue; + const publications = await db.select({ endpointId: chatActions.endpointId, providerActionId: chatActions.providerActionId }) + .from(chatActions).where(and( + eq(chatActions.companyId, companyId), + sql`${chatActions.endpointId} = any(${sql.param([...new Set(reviewed.map(entry => entry.review.endpointId))])}::uuid[])`, + eq(chatActions.kind, "github_review_publication"), eq(chatActions.status, "processed"), + sql`${chatActions.providerActionId} = any(${sql.param([...new Set(reviewed.map(entry => entry.actionId))])}::text[])`, + )); + const processed = new Set(publications.map(publication => `${publication.endpointId}:${publication.providerActionId}`)); + for (const { index, run, commentId, review, actionId } of reviewed) { + results[index] = processed.has(`${review.endpointId}:${actionId}`) + ? publishedInheritedWait(run, commentId, review) : { kind: "revoked" }; + } + } + return results; +} diff --git a/server/src/services/native-runtime/external-review-wait.ts b/server/src/services/native-runtime/external-review-wait.ts index 5d05c779cb1..2b3e05cd2ba 100644 --- a/server/src/services/native-runtime/external-review-wait.ts +++ b/server/src/services/native-runtime/external-review-wait.ts @@ -1,6 +1,6 @@ import { and, eq, inArray, sql } from "drizzle-orm"; import { chatConversations, chatEndpoints, heartbeatRuns, issues, statusDecisions, type Db } from "@paperclipai/db"; -import { readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js"; +import { readInheritedGitHubReviewWaits } from "../chat-channels/github/review-wait.js"; import { authorizeChatConversationsForBoundRuns } from "./chat-attachment-reuse.js"; import { EXTERNAL_CHAT_QUESTION_RESPONSE_KEY } from "./external-chat-question-response.js"; @@ -61,15 +61,19 @@ export async function listExternalReviewWaits(db: Db, companyId: string, issueId eq(chatEndpoints.status, "active"), )) .where(and(eq(issues.companyId, companyId), inArray(issues.id, issueIds), eq(issues.status, "in_review"))); - const authorizable = (await Promise.all(candidates.map(async ({ runId, agentId, contextSnapshot, responsibleUserId, activeIdentityContextId, ...wait }) => { - const inheritedWait = await readInheritedGitHubReviewWait(db, { - id: runId, companyId, agentId, responsibleUserId, activeIdentityContextId, - nativeIssueId: wait.issueId, issueId: wait.issueId, contextSnapshot, - }); - if (inheritedWait?.kind === "revoked") return null; - return { wait, binding: { companyId, issueId: wait.issueId, runId, agentId }, - contextSnapshot: inheritedWait?.context ?? contextSnapshot }; - }))).filter((candidate): candidate is NonNullable => candidate !== null); + // One review read and one publication read for every wait of the page, not + // two reads per wait. Each wait gets the same inherited result as before. + const inheritedWaits = await readInheritedGitHubReviewWaits(db, candidates.map(candidate => ({ + id: candidate.runId, companyId, agentId: candidate.agentId, responsibleUserId: candidate.responsibleUserId, + activeIdentityContextId: candidate.activeIdentityContextId, nativeIssueId: candidate.issueId, + issueId: candidate.issueId, contextSnapshot: candidate.contextSnapshot, + }))); + const authorizable = candidates.flatMap(({ runId, agentId, contextSnapshot, responsibleUserId: _user, activeIdentityContextId: _context, ...wait }, index) => { + const inheritedWait = inheritedWaits[index]; + if (inheritedWait?.kind === "revoked") return []; + return [{ wait, binding: { companyId, issueId: wait.issueId, runId, agentId }, + contextSnapshot: inheritedWait?.context ?? contextSnapshot }]; + }); // One authorization read for every wait of the page, not one per wait: the // per-wait reads each held a pool connection at the same moment. const authorized = await authorizeChatConversationsForBoundRuns(db, companyId, authorizable);