From 873e80f0ae1e80911590afa16a626afceed05562 Mon Sep 17 00:00:00 2001 From: Gandy2025 Date: Fri, 7 Aug 2026 18:00:40 +0800 Subject: [PATCH] fix(server): claim App task reply publication across runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five accepted webhook events on one pull request at one unchanged head dispatched five runs, and each published its own byte-identical Context Reviewer verdict. Publication idempotency was scoped to a single run's message row, so it could not see a sibling run, and the hidden `first-tree-github-task-reply-run` marker was written but never read back as a publish guard. Claim publication on the publisher's existing run-independent payload hash — repository, entity, body — before the run claims itself, so exactly one run per exact reply reaches GitHub. A losing run reads the marker back off the entity: finding its reply already published makes the loss terminal, and not finding it reports the owner as still in flight and leaves the run retryable. A definitive GitHub rejection releases the claim; an unknown write keeps it so nobody republishes behind a comment that may be live. Co-Authored-By: Claude Opus 5 --- .../github-webhook-routing-regression.md | 9 +- .../github-task-reply-publisher.test.ts | 206 ++++++++++++++++-- .../github-task-reply-publication-claim.ts | 137 ++++++++++++ .../services/github-task-reply-publisher.ts | 79 +++++++ .../shared/src/schemas/github-task-reply.ts | 6 + 5 files changed, 420 insertions(+), 17 deletions(-) create mode 100644 packages/server/src/services/github-task-reply-publication-claim.ts diff --git a/packages/qa/cases/cross-surface/github-webhook-routing-regression.md b/packages/qa/cases/cross-surface/github-webhook-routing-regression.md index 3507d9576d..d052b99ee2 100644 --- a/packages/qa/cases/cross-surface/github-webhook-routing-regression.md +++ b/packages/qa/cases/cross-surface/github-webhook-routing-regression.md @@ -85,6 +85,11 @@ App login or adding a GitHub-specific post-delivery branch. Simulate an unknown GitHub write, then confirm retry reconciles the App actor, hidden run marker, and exact body before returning rather than blindly creating another comment. Zero matches, duplicate exact matches, and list failure must all remain unknown with no second POST. +- Deliver several supported events on one pull request without moving its head, so more than one run is dispatched for + the same entity, and let each run reach its terminal reply. Confirm the entity ends with exactly one App-authored + comment for a given reply body: the first run publishes, and every later run reports + `GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION` without a second GitHub write. Push a new head, let the reviewer render a + verdict against it, and confirm that changed reply still publishes. - Attempt publication from a different Agent, client, runtime session, chat, repository-scoped role assignment, inactive membership, and spoofed user-authored `githubTask*` metadata. Confirm each fails before GitHub mutation. Confirm a reply that mentions the App is rejected. Historical boolean/no-run markers remain renderable but cannot publish. @@ -109,8 +114,8 @@ cards do not invent personnel context, and webhook content is never treated as a task capability and host GitHub permissions. `FAIL`: a reproducible regression in authentication, tenant resolution, followed-chat/card delivery, automatic task or -wake routing, recipient-bound publication, self-output suppression, whole-request deduplication, or Context Reviewer -claim coverage. +wake routing, recipient-bound publication, self-output suppression, whole-request deduplication, cross-run publication +claim coverage, or Context Reviewer claim coverage. `BLOCKED`: the isolated run cell cannot provision a disposable App/installation, webhook credential, bound entity, or connected runtime needed by the selected observations. diff --git a/packages/server/src/__tests__/github-task-reply-publisher.test.ts b/packages/server/src/__tests__/github-task-reply-publisher.test.ts index c0aea52554..7fdf653114 100644 --- a/packages/server/src/__tests__/github-task-reply-publisher.test.ts +++ b/packages/server/src/__tests__/github-task-reply-publisher.test.ts @@ -550,6 +550,106 @@ describe("GitHub App task reply publisher", () => { }), ).rejects.toThrow(/reserved for server-authored GitHub task reply runs/); }); + + // Regression: five webhook events on one pull request at one unchanged head + // dispatched five runs, and each published its own byte-identical Context + // Reviewer verdict. Run-scoped idempotency could not see a sibling run, so + // the guarantee has to be cross-run. + describe("cross-run publication claim", () => { + const VERDICT_AT_HEAD = "**Recommendation: request changes**\n\nReviewed exact head `32e5d2d4`."; + + it("lets one run publish the verdict and stops a sibling run repeating it on the same entity", async () => { + const fixture = await createRunFixture(getApp()); + const sibling = await createSiblingRun(fixture); + const github = entityGithubFetcher(); + + await expect( + submitGithubTaskReply(publishInput(fixture, github.fetcher, VERDICT_AT_HEAD)), + ).resolves.toMatchObject({ commentId: 901 }); + await expect(submitGithubTaskReply(publishInput(sibling, github.fetcher, VERDICT_AT_HEAD))).rejects.toMatchObject( + { statusCode: 409, code: "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION" }, + ); + + expect(github.comments()).toHaveLength(1); + expect(github.comments()[0]?.body).toContain(`first-tree-github-task-reply-run:${fixture.runId}`); + // The loser read its own reply back off the entity, so its loss is + // terminal rather than a blind retry. + const [message] = await fixture.app.db + .select({ metadata: messages.metadata }) + .from(messages) + .where(eq(messages.id, sibling.messageId)); + expect(message?.metadata.githubTaskReplySubmission).toMatchObject({ + state: "failed", + code: "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION", + }); + }); + + it("publishes again once the head moves and the verdict changes", async () => { + const fixture = await createRunFixture(getApp()); + const sibling = await createSiblingRun(fixture); + const github = entityGithubFetcher(); + + await expect( + submitGithubTaskReply(publishInput(fixture, github.fetcher, VERDICT_AT_HEAD)), + ).resolves.toMatchObject({ commentId: 901 }); + await expect( + submitGithubTaskReply( + publishInput(sibling, github.fetcher, "**Recommendation: approve**\n\nReviewed exact head `9f01ab77`."), + ), + ).resolves.toMatchObject({ commentId: 902 }); + + expect(github.comments()).toHaveLength(2); + }); + + it("claims once when two runs for the same exact head publish concurrently", async () => { + const fixture = await createRunFixture(getApp()); + const sibling = await createSiblingRun(fixture); + const github = entityGithubFetcher(); + + const results = await Promise.allSettled([ + submitGithubTaskReply(publishInput(fixture, github.fetcher, VERDICT_AT_HEAD)), + submitGithubTaskReply(publishInput(sibling, github.fetcher, VERDICT_AT_HEAD)), + ]); + + expect(results.filter((result) => result.status === "fulfilled")).toHaveLength(1); + const [rejected] = results.filter((result) => result.status === "rejected"); + expect(rejected?.reason).toMatchObject({ statusCode: 409, code: "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION" }); + expect(github.comments()).toHaveLength(1); + expect(githubCommentPosts(github.fetcher)).toHaveLength(1); + }); + + it("releases the claim when GitHub definitively rejects the write", async () => { + const fixture = await createRunFixture(getApp()); + const sibling = await createSiblingRun(fixture); + const rejecting = entityGithubFetcher({ post: "reject" }); + + await expect( + submitGithubTaskReply(publishInput(fixture, rejecting.fetcher, VERDICT_AT_HEAD)), + ).rejects.toMatchObject({ code: "GITHUB_TASK_REPLY_GITHUB_REJECTED" }); + + const accepting = entityGithubFetcher(); + await expect( + submitGithubTaskReply(publishInput(sibling, accepting.fetcher, VERDICT_AT_HEAD)), + ).resolves.toMatchObject({ commentId: 901 }); + expect(accepting.comments()).toHaveLength(1); + }); + + it("keeps the claim after an unknown write so no sibling run republishes behind it", async () => { + const fixture = await createRunFixture(getApp()); + const sibling = await createSiblingRun(fixture); + const unknown = entityGithubFetcher({ post: "unknown" }); + + await expect( + submitGithubTaskReply(publishInput(fixture, unknown.fetcher, VERDICT_AT_HEAD)), + ).rejects.toMatchObject({ code: "GITHUB_TASK_REPLY_GITHUB_UNKNOWN" }); + + const siblingGithub = entityGithubFetcher(); + await expect( + submitGithubTaskReply(publishInput(sibling, siblingGithub.fetcher, VERDICT_AT_HEAD)), + ).rejects.toMatchObject({ code: "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION" }); + expect(githubCommentPosts(siblingGithub.fetcher)).toHaveLength(0); + }); + }); }); async function createRunFixture( @@ -603,20 +703,56 @@ async function createRunFixture( type: "group", participantIds: [agent.uuid], }); - const runId = randomUUID(); const entityType = options.entityType ?? "issue"; - const entityUrl = `https://github.com/owner/repo/${entityType === "issue" ? "issues" : "pull"}/42`; + const run = await sendGithubTaskRun({ + app, + chatId: chat.id, + humanAgentUuid: admin.humanAgentUuid, + organizationId: admin.organizationId, + agentUuid: agent.uuid, + entityType, + }); + return { app, admin, agent, runtimeToken, chatId: chat.id, entityType, ...run }; +} + +/** + * A second automatically routed run for the SAME organization, chat, and + * entity — the shape webhook dispatch produces when several accepted events + * land on one pull request at one unchanged head. + */ +async function createSiblingRun(fixture: Awaited>) { + const run = await sendGithubTaskRun({ + app: fixture.app, + chatId: fixture.chatId, + humanAgentUuid: fixture.admin.humanAgentUuid, + organizationId: fixture.admin.organizationId, + agentUuid: fixture.agent.uuid, + entityType: fixture.entityType, + }); + return { ...fixture, ...run }; +} + +async function sendGithubTaskRun(input: { + app: ReturnType>; + chatId: string; + humanAgentUuid: string; + organizationId: string; + agentUuid: string; + entityType: "issue" | "pull_request"; +}) { + const runId = randomUUID(); + const entityUrl = `https://github.com/owner/repo/${input.entityType === "issue" ? "issues" : "pull"}/42`; const { message } = await sendMessage( - app.db, - chat.id, - admin.humanAgentUuid, + input.app.db, + input.chatId, + input.humanAgentUuid, { source: "github", format: "card", content: { type: "github_event", reason: "mentioned", - event: entityType === "issue" ? "issues" : "pull_request", + event: input.entityType === "issue" ? "issues" : "pull_request", action: "opened", kind: "commented", repository: "owner/repo", @@ -624,21 +760,21 @@ async function createRunFixture( title: "Task", body: "@test-app-slug do it", url: entityUrl, - entity: { type: entityType, key: "owner/repo#42", url: entityUrl }, - teamAgentTask: { agentUuid: agent.uuid, runId }, + entity: { type: input.entityType, key: "owner/repo#42", url: entityUrl }, + teamAgentTask: { agentUuid: input.agentUuid, runId }, }, metadata: { - mentions: [agent.uuid], + mentions: [input.agentUuid], source: "github", systemSender: "github", - teamAgentTask: { agentUuid: agent.uuid, runId }, + teamAgentTask: { agentUuid: input.agentUuid, runId }, githubTaskRun: true, githubTaskRunId: runId, - githubTaskOrganizationId: admin.organizationId, - githubTaskAgentUuid: agent.uuid, - githubTaskManagerHumanAgentId: admin.humanAgentUuid, + githubTaskOrganizationId: input.organizationId, + githubTaskAgentUuid: input.agentUuid, + githubTaskManagerHumanAgentId: input.humanAgentUuid, githubTaskRepository: "owner/repo", - githubTaskEntityType: entityType, + githubTaskEntityType: input.entityType, githubTaskEntityNumber: 42, githubTaskEntityUrl: entityUrl, githubTaskReplySubmission: { state: "pending" }, @@ -646,7 +782,7 @@ async function createRunFixture( }, { allowSystemSender: true, allowGithubTaskRun: true }, ); - return { app, admin, agent, runtimeToken, chatId: chat.id, messageId: message.id, runId }; + return { messageId: message.id, runId }; } function publishInput( @@ -700,6 +836,46 @@ function successfulGithubFetcher( }); } +/** + * A GitHub App fake that keeps the entity's comment thread, so a losing run's + * hidden-marker read-back sees exactly what an earlier run published. + */ +function entityGithubFetcher(options: { post?: "accept" | "reject" | "unknown" } = {}) { + const comments: Array<{ id: number; html_url: string; user: { login: string }; body: string }> = []; + let nextCommentId = 901; + const fetcher = vi.fn(async (url, init) => { + const target = String(url); + if (target.endsWith("/access_tokens")) { + return jsonResponse( + { + token: "installation-token", + expires_at: "2026-12-15T18:00:00Z", + permissions: { metadata: "read", issues: "write", pull_requests: "write" }, + repository_selection: "selected", + }, + 201, + ); + } + if (target.endsWith("/issues/42/comments") && init?.method === "POST") { + if (options.post === "reject") return new Response("rejected", { status: 422 }); + if (options.post === "unknown") throw new TypeError("socket closed after request dispatch"); + const payload = JSON.parse(String(init.body)) as { body: string }; + const id = nextCommentId++; + const comment = { + id, + html_url: `https://github.com/owner/repo/issues/42#issuecomment-${id}`, + user: { login: "test-app-slug[bot]" }, + body: payload.body, + }; + comments.push(comment); + return jsonResponse(comment, 201); + } + if (target.includes("/issues/42/comments?per_page=100")) return jsonResponse(comments); + return new Response("not found", { status: 404 }); + }); + return { fetcher, comments: () => comments }; +} + function failingGithubFetcher(phase: "token" | "comment", status: number) { return vi.fn(async (url, init) => { const target = String(url); diff --git a/packages/server/src/services/github-task-reply-publication-claim.ts b/packages/server/src/services/github-task-reply-publication-claim.ts new file mode 100644 index 0000000000..c5ea282dc3 --- /dev/null +++ b/packages/server/src/services/github-task-reply-publication-claim.ts @@ -0,0 +1,137 @@ +import { GITHUB_TASK_REPLY_RUN_MARKER_PREFIX } from "@first-tree/shared"; +import type { Database } from "../db/connection.js"; +import { claimEvent, unclaimEvent } from "./event-dedup.js"; +import { type GithubIssueComment, listIssueCommentsForRun } from "./github-app.js"; + +/** + * Cross-run publication guard for App-authored GitHub task replies. + * + * The submission state on a run's own message row makes ONE run idempotent. It + * says nothing about a sibling run, so duplicate or concurrent dispatch for the + * same entity published one identical comment per run — five byte-identical + * Context Reviewer verdicts on one unchanged pull request head is the case this + * module exists to make impossible. + * + * Two guards, in order: + * + * 1. An atomic claim keyed on the publisher's existing `payloadHash` + * (`sha256([repository, entityType, entityNumber, body])`). That hash is + * already run-independent by construction: it identifies the exact comment + * about to be written, not the run writing it. A verdict rendered against a + * moved head names a different head and therefore hashes differently, so a + * real re-review still publishes; a byte-identical repeat cannot. + * + * 2. For the run that loses the claim, a read-back of the hidden + * `first-tree-github-task-reply-run` marker on the entity's existing App + * comments. Until now that marker was written and never read as a publish + * guard, so a losing run had no way to learn its reply was already on + * GitHub. The read-back turns the loss into a definite answer — "your exact + * reply is already published at " — instead of a blind retry. + */ + +const PUBLICATION_CLAIM_PLATFORM = "github-task-reply-publication"; + +/** + * `processed_events` already provides exactly-once claim semantics through its + * `(event_id, platform)` unique index; the dedicated `platform` value keeps + * publication claims from ever colliding with webhook delivery ids. + */ +export function githubTaskReplyPublicationClaimKey(organizationId: string, payloadHash: string): string { + return `${organizationId}:${payloadHash}`; +} + +/** + * Take the publication claim for this exact reply. + * + * Call inside the transaction that also claims the run, and before that run + * claim: a lost publication claim must leave the run untouched so it stays + * retryable. `INSERT ... ON CONFLICT DO NOTHING` makes the race decision one + * atomic statement rather than a read-then-write, which is what makes two runs + * dispatched at the same instant resolve to exactly one publisher. + */ +export async function claimGithubTaskReplyPublication( + db: Database, + input: { organizationId: string; payloadHash: string }, +): Promise { + return claimEvent( + db, + githubTaskReplyPublicationClaimKey(input.organizationId, input.payloadHash), + PUBLICATION_CLAIM_PLATFORM, + ); +} + +/** + * Release a claim whose GitHub write was definitively rejected, so the reply + * becomes publishable again. + * + * Never call this for an *unknown* write outcome: the comment may exist, and + * re-opening the claim would authorize the duplicate this module prevents. The + * owning run's existing reconciliation path resolves those. + * + * A claim can therefore outlive its owner — an unresolved unknown write, or a + * crash in the narrow window between the claim transaction committing and + * GitHub answering. That strands one exact body on one entity, not the + * reviewer: the next run re-reads live state and publishes whatever it finds + * then, which is a different body and a different claim. + */ +export async function releaseGithubTaskReplyPublication( + db: Database, + input: { organizationId: string; payloadHash: string }, +): Promise { + await unclaimEvent( + db, + githubTaskReplyPublicationClaimKey(input.organizationId, input.payloadHash), + PUBLICATION_CLAIM_PLATFORM, + ); +} + +/** + * Find an App comment on this entity whose caller-authored body is exactly the + * reply we were about to publish, whichever run published it. + * + * Returns `null` when nothing matches and when GitHub cannot be read — a losing + * run must not publish either way, and an unreadable entity is reported as "the + * owner is still in flight" rather than as "nothing is published". + */ +export async function findPublishedGithubTaskReply(input: { + token: string; + appSlug: string; + owner: string; + repo: string; + entityNumber: number; + body: string; + fetcher?: typeof fetch; +}): Promise { + const comments = await listIssueCommentsForRun( + input.token, + { + owner: input.owner, + repo: input.repo, + issueNumber: input.entityNumber, + // Every run marker starts with this prefix, so the shared lister returns + // the App's task replies for *all* runs on the entity — which is exactly + // the cross-run view a publish guard needs. + marker: GITHUB_TASK_REPLY_RUN_MARKER_PREFIX, + appSlug: input.appSlug, + }, + { fetcher: input.fetcher }, + ).catch(() => null); + if (!comments) return null; + const expected = input.body.trimEnd(); + return comments.find((comment) => stripRunMarker(comment.body) === expected) ?? null; +} + +/** + * Recover the caller-authored body from a published reply. Mirrors the + * publisher's `replyBody` composition exactly — body, blank line, hidden run + * marker — and refuses anything that does not match that shape so an + * unrelated App comment can never be read as a duplicate. + */ +function stripRunMarker(commentBody: string): string | null { + const separator = `\n\n${GITHUB_TASK_REPLY_RUN_MARKER_PREFIX}`; + const markerStart = commentBody.lastIndexOf(separator); + if (markerStart < 0) return null; + const marker = commentBody.slice(markerStart + 2); + if (!/^$/.test(marker)) return null; + return commentBody.slice(0, markerStart); +} diff --git a/packages/server/src/services/github-task-reply-publisher.ts b/packages/server/src/services/github-task-reply-publisher.ts index 842e3fad58..17cd4878da 100644 --- a/packages/server/src/services/github-task-reply-publisher.ts +++ b/packages/server/src/services/github-task-reply-publisher.ts @@ -36,6 +36,11 @@ import { } from "./github-app.js"; import { isGithubAppTargetLogin } from "./github-audience.js"; import { extractMentions } from "./github-normalize.js"; +import { + claimGithubTaskReplyPublication, + findPublishedGithubTaskReply, + releaseGithubTaskReplyPublication, +} from "./github-task-reply-publication-claim.js"; import { getOrgContextReviewRuntime } from "./org-settings.js"; import { getTeamAgentUuid } from "./team-agent-settings.js"; @@ -199,6 +204,15 @@ export async function submitGithubTaskReply(input: { }; } + // Cross-run guard, taken before this run claims itself. Run-scoped + // idempotency cannot see a sibling run dispatched for the same entity, so + // without this every duplicate or concurrent dispatch published its own + // identical comment. Losing leaves the run `pending` — it is a retryable + // loss, not a terminal one — and the caller resolves it against GitHub. + if (!(await claimGithubTaskReplyPublication(db, { organizationId: run.organizationId, payloadHash }))) { + return { kind: "duplicate" as const, run }; + } + const attemptId = uuidv7(); const claimed = await setSubmissionIf(db, run.messageId, "pending", { state: "submitting", @@ -211,6 +225,16 @@ export async function submitGithubTaskReply(input: { return { kind: "claimed" as const, run, attemptId }; }); if (claim.kind === "submitted") return claim.response; + if (claim.kind === "duplicate") { + throw await duplicatePublication({ + db: input.db, + run: claim.run, + payloadHash, + body: request.body, + github, + fetcher: input.fetcher, + }); + } if (claim.kind === "reconcile") { return reconcileUnknownSubmission({ db: input.db, @@ -252,6 +276,14 @@ export async function submitGithubTaskReply(input: { ); } const mapped = mapGithubMutationError(error); + // GitHub definitively refused the write, so no comment exists and the + // reply must become publishable again. Released only on this branch — + // an unknown outcome keeps the claim so nobody republishes behind a + // comment that may already be live. + await releaseGithubTaskReplyPublication(input.db, { + organizationId: claim.run.organizationId, + payloadHash, + }); await setSubmissionForAttempt(input.db, claim.run.messageId, claim.attemptId, { state: "failed", payloadHash, @@ -651,6 +683,53 @@ async function reconcileUnknownSubmission(input: { return submittedResponse(submitted); } +/** + * Resolve a lost publication claim into a definite answer for the losing run. + * + * Reads the hidden run marker back off the entity's existing App comments — + * the guard the marker was always shaped for but never used as. Finding this + * exact reply already published makes the loss terminal, so the run stops + * instead of rephrasing and trying again. Not finding it means the owning run + * is still in flight, so the run stays `pending` and can take the claim over + * later if that owner ends up releasing it. + */ +async function duplicatePublication(input: { + db: Database; + run: RunFacts; + payloadHash: string; + body: string; + github: { token: string; appSlug: string; owner: string; repo: string }; + fetcher?: typeof fetch; +}): Promise { + const published = await findPublishedGithubTaskReply({ + token: input.github.token, + appSlug: input.github.appSlug, + owner: input.github.owner, + repo: input.github.repo, + entityNumber: input.run.entityNumber, + body: input.body, + fetcher: input.fetcher, + }); + if (!published) { + return new GithubTaskReplyPublisherError( + 409, + "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION", + `Another run already claimed publication of this exact reply on ${input.run.repository}#${input.run.entityNumber} and it is still in flight. Do not publish it again.`, + ); + } + await setSubmissionIf(input.db, input.run.messageId, "pending", { + state: "failed", + payloadHash: input.payloadHash, + code: "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION", + failedAt: new Date().toISOString(), + }); + return new GithubTaskReplyPublisherError( + 409, + "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION", + `This exact reply is already published on ${input.run.repository}#${input.run.entityNumber} at ${published.htmlUrl}. Do not publish it again.`, + ); +} + function submissionFromComment(input: { payloadHash: string; comment: GithubIssueComment; diff --git a/packages/shared/src/schemas/github-task-reply.ts b/packages/shared/src/schemas/github-task-reply.ts index 3d4d4287e9..e14c0fdb09 100644 --- a/packages/shared/src/schemas/github-task-reply.ts +++ b/packages/shared/src/schemas/github-task-reply.ts @@ -113,6 +113,12 @@ export const GITHUB_TASK_REPLY_ERROR_CODES = [ "GITHUB_TASK_REPLY_RUN_FORBIDDEN", "GITHUB_TASK_REPLY_RUN_ALREADY_SUBMITTED", "GITHUB_TASK_REPLY_RUN_PAYLOAD_MISMATCH", + /** + * A different run already owns the publication of this exact reply on this + * exact entity. Run-scoped idempotency cannot see that; the cross-run + * publication claim can. The losing run must exit without publishing. + */ + "GITHUB_TASK_REPLY_DUPLICATE_PUBLICATION", "GITHUB_TASK_REPLY_ENTITY_UNSUPPORTED", "GITHUB_TASK_REPLY_APP_NOT_INSTALLED", "GITHUB_TASK_REPLY_APP_PERMISSION_REQUIRED",