mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
perf(github): read the inherited review waits of an issue list in one pass (#20673)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - An agent can review a GitHub pull request. After the agent publishes the review, the task can wait in review for the next GitHub reply. A later run of the same task can inherit this wait. > - Issue lists and inbox attention show these waits. For each waiting task, the server reads the inherited GitHub review wait: the source comment, the latest review, and the review publication. > - The server did these reads one time for each waiting task. A list request with N waiting tasks sent N copies of the prepared review statement and N copies of the publication read. > - On a hosted database, the prepared review statement ran about 270,000 times in about 2.5 hours (up to 87 calls per second). The run identity read before it ran the same number of times. The list read that calls it ran only about 1,400 times, so each list read started about 190 review reads on average. No other caller reads the waits of many runs in a loop. > - This pull request reads the reviews and the publications of all waits of one list request with one statement each. Each wait gets the same result as before. > - The benefit is about 200 times fewer review statements and fewer pool connections for each issue list. ## Linked Issues or Issue Description Refs #20400 (shares review-wait reads between the callers of one request). Refs #20551 (authorizes the waits of one issue page with one read; this change does the same for the inherited GitHub review reads of that page). **What happened** `listExternalReviewWaits` called `readInheritedGitHubReviewWait` one time for each candidate wait, inside `Promise.all`. Each call sent a run identity read, a source-message read, the prepared `inherited_github_review_wait` statement, and a publication read. A list with N waits sent each statement N times. **Expected behavior** One list request sends one review statement and one publication read for each company. The result of each wait does not change. **Steps to reproduce** 1. Put many tasks of one company in review with an inherited GitHub review wait. 2. Open the issue list or the inbox. 3. Count the statements of the database. The review statement runs one time for each waiting task. ## What Changed - `server/src/services/chat-channels/github/review-wait.ts`: - Split `readInheritedGitHubReviewWait` into shared steps: the source comment read, the review checks, the publication key, and the published result. The single-run read uses the same prepared statement and the same checks as before. - Add `readInheritedGitHubReviewWaits(db, runs)`. It reads the source comment of each run with the same lineage check as before. Then, for each company, one `DISTINCT ON` statement reads the latest review for each (source comment, task, agent) key, with the same joins, filters and order as the prepared statement. One read finds the processed publications. Each run gets the same result as a separate call: `null`, `revoked`, or `published`. - `server/src/services/native-runtime/external-review-wait.ts`: `listExternalReviewWaits` uses the batch read. The rest of the list and its authorization do not change. - `server/src/__tests__/github-review-wait.test.ts`: a new test reads 18 runs in one batch and compares each result with a separate single-run read. The runs include each revocation case (connection disabled or inactive, review state, review agent, repository, repository id, thread, publication, identity, company), a newer unreviewed head, no identity, no task, an endpoint of another agent, a run in another company and a duplicate run. The test also makes sure that the batch sends one review statement and one publication read for each company. The existing revocation tests also check the batch read. - `server/src/__tests__/helpers/github-review-wait.ts`: the seed helper can add a wait to an existing company and agent. ## Verification - `pnpm vitest run server/src/__tests__/github-review-wait.test.ts server/src/__tests__/github-review-check-state.test.ts server/src/__tests__/issue-list-review-wait-authorization.integration.test.ts server/src/__tests__/blocked-inbox-review-wait-scope.test.ts` - Focused local run of every test that imports the changed modules or the seed helper (7 files, 335 tests passed), including `external-chat-wait.integration.test.ts`, `external-review-wait.test.ts` and `legacy-execution-recovery.test.ts`. The change to the shared seed helper makes `pnpm verify` select the full server suite, so CI supplies that coverage. - Local measurement on real PostgreSQL (embedded PostgreSQL, `ANALYZE` after seeding). One company and agent, each wait has 5 reviews and 10 outbound messages. "Before" is the previous list code: one `readInheritedGitHubReviewWait` call for each run in `Promise.all`. "After" is `readInheritedGitHubReviewWaits`. The runs alternate order in each pair. Both return equal results in every sample. | Waits | Version | Statements | Review statements | Median ms | Min–max ms | Samples | | --- | --- | --- | --- | --- | --- | --- | | 200 | before | 800 | 200 | 106.8 | 98.0–117.5 | 10 | | 200 | after | 402 | 1 | 44.3 | 41.6–50.9 | 10 | | 20 | before | 80 | 20 | 9.8 | 8.6–14.7 | 12 | | 20 | after | 42 | 1 | 6.3 | 5.3–7.0 | 12 | - `EXPLAIN (ANALYZE, BUFFERS)` of the batch review statement for 200 waits: index scans on each join, 6.7 ms execution. The single statement: 0.05 ms execution and about 2.7 ms planning for each call. - The run identity read and the source-message read still run one time for each wait. They are cheap on the hosted database (about 0 ms each), and they keep the lineage check for each run. ## Risks - Low risk. The batch statement uses the same joins, filters and order as the prepared statement, and `DISTINCT ON` keeps the first row for each key. The array filters can also match a mix of two keys. The code uses only rows of exact requested keys. - The single-run read and its prepared statement do not change. The finalizer, the status committer, and the attachment reuse paths still use it. - Errors do not change: a 403 from the lineage check gives `revoked`, and other errors are thrown. - No migration. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (code reading, editing, shell, tests). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
c444da7d05
commit
e9db5e0d03
4 files changed
+253
-69
No files matched your search
@@ -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<ReturnType<typeof seedGitHubReviewWait>>,
|
||||
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<T>(db: Db, read: () => Promise<T>) {
|
||||
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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -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<Awaited<ReturnType<typeof seedGitHubReviewWait>>> = [];
|
||||
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[] = [];
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "companyId" | "agentId" | "responsibleUserId" | "activeIdentityContextId" | "nativeIssueId" | "issueId" | "contextSnapshot">;
|
||||
|
||||
/** 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<unknown>`${chatGitHubReviews.assessment}->'reviewedCommit'`, reviewRunId: chatGitHubReviews.runId,
|
||||
connectionEnabled: toolConnections.enabled, connectionStatus: toolConnections.status,
|
||||
reviewAgentId: heartbeatRuns.agentId, reviewCompanyId: heartbeatRuns.companyId,
|
||||
resourceRepository: chatEndpointResources.providerResourceId,
|
||||
resourceRepositoryId: sql<string | null>`${chatEndpointResources.metadata}->>'providerRepositoryId'`,
|
||||
pullNumber: chatGitHubReviews.pullNumber, thread: chatConversations.externalThreadId,
|
||||
};
|
||||
|
||||
const inheritedReviewQueries = new WeakMap<Db, ReturnType<typeof prepareInheritedReviewQuery>>();
|
||||
|
||||
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<unknown>`${chatGitHubReviews.assessment}->'reviewedCommit'`, reviewRunId: chatGitHubReviews.runId,
|
||||
connectionEnabled: toolConnections.enabled, connectionStatus: toolConnections.status,
|
||||
reviewAgentId: heartbeatRuns.agentId, reviewCompanyId: heartbeatRuns.companyId,
|
||||
resourceRepository: chatEndpointResources.providerResourceId,
|
||||
resourceRepositoryId: sql<string | null>`${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<ReturnType<ReturnType<typeof prepareInheritedReviewQuery>["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<string, InheritedReviewRow>();
|
||||
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<typeof publishedInheritedWait> | 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<InheritedGitHubReviewWait> {
|
||||
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<InheritedGitHubReviewWait[]> {
|
||||
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<string, Array<{ index: number; run: WaitRun; commentId: string }>>();
|
||||
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;
|
||||
}
|
||||
@@ -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<typeof candidate> => 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);
|
||||
|
||||
Reference in new issue
Block a user