diff --git a/apps/api/src/storage/task-board.ts b/apps/api/src/storage/task-board.ts index 5a1979e162..53a730122f 100644 --- a/apps/api/src/storage/task-board.ts +++ b/apps/api/src/storage/task-board.ts @@ -1185,7 +1185,8 @@ export class TaskBoardStorage { * the Super Agent, returning the updated item to the single winner and null * to everyone else. The auto-resolve reaction fires from two triggers that * can coincide — a reviewer's approval (`review-decision`) and the PR - * modal's poll (`prs-get`) — so, like `claimReviewer`, this fence must be + * modal's poll (`prs-get`) — so, like the reviewer dispatch's derived thread + * id, this fence must be * atomic: a read-then-write would let both dispatch a Super Agent run on the * same PR. The assignee re-check closes a second race: the caller's * assignee check runs against a read taken before this write, so a human @@ -1466,67 +1467,12 @@ export class TaskBoardStorage { .execute(); } - /** - * Atomically claim the (task, reviewer, cycle) slot, minting a token. - * `claimed` is false when the slot was already taken (a concurrent enqueue - * won the race) — the caller then skips enqueueing that reviewer. Either way - * returns the winning claim's `token`, which the reviewer run carries and - * echoes back to prove its identity when it records a decision. - */ - async claimReviewer( - taskBoardItemId: string, - reviewer: string, - cycleAt: Date, - ): Promise<{ claimed: boolean; token: string }> { - const token = `rtok_${crypto.randomUUID()}`; - const inserted = await this.db - .insertInto("task_board_review_claims") - .values({ - task_board_item_id: taskBoardItemId, - reviewer, - cycle_at: cycleAt, - token, - }) - .onConflict((oc) => - oc.columns(["task_board_item_id", "reviewer", "cycle_at"]).doNothing(), - ) - .returning("token") - .executeTakeFirst(); - if (inserted) return { claimed: true, token: inserted.token }; - const existing = await this.db - .selectFrom("task_board_review_claims") - .select("token") - .where("task_board_item_id", "=", taskBoardItemId) - .where("reviewer", "=", reviewer) - .where("cycle_at", "=", cycleAt) - .executeTakeFirst(); - return { claimed: false, token: existing?.token ?? token }; - } - - /** - * Release a reviewer's claim on a cycle — the counterpart to `claimReviewer` - * for when the dispatch it was minted for never actually ran (e.g. the - * enqueue itself threw). Without this, a transient dispatch failure leaves - * the slot permanently claimed with no thread behind it: `claimReviewer`'s - * unique (task, reviewer, cycle) key would refuse every retry for the rest - * of that review cycle, so that reviewer would simply never run. - */ - async releaseReviewerClaim( - taskBoardItemId: string, - reviewer: string, - cycleAt: Date, - ): Promise { - await this.db - .deleteFrom("task_board_review_claims") - .where("task_board_item_id", "=", taskBoardItemId) - .where("reviewer", "=", reviewer) - .where("cycle_at", "=", cycleAt) - .execute(); - } - - /** Resolve a review token to its claim (which reviewer, which cycle) for a - * task. Null when the token doesn't belong to this task — used to verify - * that a decision's caller really is the reviewer it claims to be. */ + /** Resolve a legacy review token to its claim for a task. Null when the token + * doesn't belong to this task. + * + * LEGACY: claims are no longer written — reviewer identity is an HMAC now + * (`tools/task-board/review-token.ts`). This only serves runs dispatched + * before that deploy; delete it with the table once they've drained. */ async resolveReviewClaimByToken( taskBoardItemId: string, token: string, diff --git a/apps/api/src/tools/task-board/enqueue-reviewer.test.ts b/apps/api/src/tools/task-board/enqueue-reviewer.test.ts index a1fc226326..4209453bc9 100644 --- a/apps/api/src/tools/task-board/enqueue-reviewer.test.ts +++ b/apps/api/src/tools/task-board/enqueue-reviewer.test.ts @@ -7,11 +7,11 @@ import { describe, expect, it, test } from "bun:test"; import type { TaskBoardItem } from "@/storage/types"; import { - hasSpentAttemptThisCycle, MAX_REVIEWER_ATTEMPTS, REVIEWER_DISALLOWED_TOOLS, reviewerAttemptsExhausted, reviewerHandledThisCycle, + spentAttemptsThisCycle, } from "./enqueue-reviewer"; import { REVIEW_RUN_TOOL_NAMES } from "./task-run-context"; @@ -101,8 +101,7 @@ describe("reviewerHandledThisCycle", () => { // The bug: a FAILED reviewer thread satisfied "created this cycle", so when // both reviewers died on a database-connection timeout the card sat In Review - // with two dead reviewer threads, a claim row nothing released, and no - // verdicts. A failure is not a review. + // with two dead reviewer threads and no verdicts. A failure is not a review. it("is FALSE for a reviewer thread that FAILED this cycle — it gets retried", () => { const task = taskWith([ thread({ @@ -112,15 +111,15 @@ describe("reviewerHandledThisCycle", () => { }), ]); expect(reviewerHandledThisCycle(task, "qa", CYCLE_START, NOW)).toBe(false); - expect(hasSpentAttemptThisCycle(task, "qa", CYCLE_START, NOW)).toBe(true); + expect(spentAttemptsThisCycle(task, "qa", CYCLE_START, NOW)).toBe(1); }); /** * The deadlock. A reviewer whose pod dies mid-run keeps `in_progress` * forever: the per-pod idle reaper can't see it and `failNeverStartedThreads` * only covers runs that never started. Taking the status at face value made - * the thread own the cycle permanently — claim spent, nothing re-dispatched, - * and a merge gate waiting on a verdict that was never coming. One card sat + * the thread own the cycle permanently — nothing re-dispatched, and a merge + * gate waiting on a verdict that was never coming. One card sat * that way while its co-reviewer had approved in 68 seconds. */ it("is FALSE for a non-terminal thread whose heartbeat went cold", () => { @@ -133,8 +132,9 @@ describe("reviewerHandledThisCycle", () => { }), ]); expect(reviewerHandledThisCycle(task, "qa", CYCLE_START, NOW)).toBe(false); - // …and its claim row is released first, or the retry loses to the corpse. - expect(hasSpentAttemptThisCycle(task, "qa", CYCLE_START, NOW)).toBe(true); + // …and it counts as an attempt, so the retry fences on a fresh id + // instead of colliding with the corpse. + expect(spentAttemptsThisCycle(task, "qa", CYCLE_START, NOW)).toBe(1); }); it("a warm heartbeat still owns the cycle — a slow reviewer is not a hung one", () => { @@ -147,7 +147,7 @@ describe("reviewerHandledThisCycle", () => { }), ]); expect(reviewerHandledThisCycle(task, "qa", CYCLE_START, NOW)).toBe(true); - expect(hasSpentAttemptThisCycle(task, "qa", CYCLE_START, NOW)).toBe(false); + expect(spentAttemptsThisCycle(task, "qa", CYCLE_START, NOW)).toBe(0); }); // The opposite mistake to the deadlock: a reviewer whose pod keeps dying must @@ -223,7 +223,7 @@ describe("reviewerHandledThisCycle", () => { createdAt: "2026-01-01T09:30:00Z", }), ]); - expect(hasSpentAttemptThisCycle(task, "qa", CYCLE_START, NOW)).toBe(false); + expect(spentAttemptsThisCycle(task, "qa", CYCLE_START, NOW)).toBe(0); }); it("scopes to the given reviewer — the other reviewer's thread and the Super Agent's don't count", () => { diff --git a/apps/api/src/tools/task-board/enqueue-reviewer.ts b/apps/api/src/tools/task-board/enqueue-reviewer.ts index 7a7b460db7..d5d1ff7279 100644 --- a/apps/api/src/tools/task-board/enqueue-reviewer.ts +++ b/apps/api/src/tools/task-board/enqueue-reviewer.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import type { StudioContext } from "@/core/studio-context"; import type { TaskBoardItem } from "@/storage/types"; import { @@ -12,6 +13,7 @@ import { emitTaskBoardUpdated, handTaskToHuman } from "./run-reactions"; import { enqueueAgentRunForTask } from "./enqueue-task-run"; import { resolveTaskRepoChoice } from "./claude-code-task-run"; import { isThreadRunStale } from "@/tools/thread/helpers"; +import { mintReviewToken } from "./review-token"; /** Thread statuses past which a reviewer run is done — a live run has a * non-terminal status. Mirrors the storage-layer set. */ @@ -120,10 +122,10 @@ export async function enqueueEnabledReviewers( const lastInReviewAt = await lastInReviewTime(ctx, task); const cycleAt = new Date(lastInReviewAt); - // Each reviewer's claim + enqueue is independent (a separate DB row keyed by - // its own kind), so run them CONCURRENTLY — this is on TASK_BOARD_ITEM_PRS_GET's - // synchronous poll path, and serial awaits doubled its latency once both QA - // and Code Reviewer are enabled. + // Each reviewer's enqueue is independent (its own fence id), so run them + // CONCURRENTLY — this is on TASK_BOARD_ITEM_PRS_GET's synchronous poll path, + // and serial awaits doubled its latency once both QA and Code Reviewer are + // enabled. await Promise.all( enabled.map(async (kind) => { // A dead end, not a wait — see `reviewerAttemptsExhausted`. @@ -138,52 +140,13 @@ export async function enqueueEnabledReviewers( } if (reviewerHandledThisCycle(task, kind, lastInReviewAt)) return; // Getting here with a dead reviewer thread from THIS cycle means the last - // attempt failed (see `reviewerHandledThisCycle`). Its claim row is still - // there, and the claim key is (task, reviewer, cycle) — so without - // releasing it first, `claimReviewer` below would lose to the corpse and - // the retry would be a no-op. - if (hasSpentAttemptThisCycle(task, kind, lastInReviewAt)) { - await ctx.storage.taskBoard - .releaseReviewerClaim(task.id, kind, cycleAt) - .catch((err) => - console.error( - `[task-board] ${kind} stale claim release failed`, - err, - ), - ); - } - // Atomically claim the reviewer's slot for this cycle. The claim dedups - // the two triggers (projector run-finish + the modal poll) that can fire - // at the same instant — the loser's `claimed` is false, so it skips - // instead of spawning a duplicate run. The claim's token binds the - // reviewer's later decision back to this dispatch. - let claimed: boolean; - let token: string; - try { - ({ claimed, token } = await ctx.storage.taskBoard.claimReviewer( - task.id, - kind, - cycleAt, - )); - } catch (err) { - console.error(`[task-board] ${kind} reviewer claim failed`, err); - return; - } - if (!claimed) return; - await enqueueReviewerForTask(ctx, task, kind, token).catch( - async (err) => { - console.error(`[task-board] ${kind} reviewer enqueue failed`, err); - // Nothing was dispatched — release the slot so the next poll/trigger - // can retry this reviewer instead of finding it permanently claimed. - await ctx.storage.taskBoard - .releaseReviewerClaim(task.id, kind, cycleAt) - .catch((releaseErr) => - console.error( - `[task-board] ${kind} reviewer claim release failed`, - releaseErr, - ), - ); - }, + // attempt failed (see `reviewerHandledThisCycle`), so this dispatch is a + // RETRY and needs a fence of its own — the previous attempt's thread id + // is taken, and reusing it would collapse the retry onto the corpse. + const attempt = spentAttemptsThisCycle(task, kind, lastInReviewAt); + await enqueueReviewerForTask(ctx, task, kind, cycleAt, attempt).catch( + (err) => + console.error(`[task-board] ${kind} reviewer enqueue failed`, err), ); }), ); @@ -197,22 +160,23 @@ export async function enqueueEnabledReviewers( */ export const MAX_REVIEWER_ATTEMPTS = 2; -/** Does this cycle already have a SPENT reviewer attempt of `kind` — one that - * failed, or one stuck non-terminal with a cold heartbeat? Decides whether the - * dispatch below is a retry (and so has a stale claim row to clear first). +/** How many SPENT reviewer attempts of `kind` this cycle already has — ones + * that failed, or that are stuck non-terminal with a cold heartbeat. It is the + * dispatch's attempt ordinal, and so part of its fence id: a retry must not + * derive the same thread id as the corpse it is replacing. * Pure; exported for the unit test. */ -export function hasSpentAttemptThisCycle( +export function spentAttemptsThisCycle( task: TaskBoardItem, kind: ReviewerKind, lastInReviewAt: number, now: number = Date.now(), -): boolean { - return task.threads.some( +): number { + return task.threads.filter( (thr) => isReviewerThreadTitle(thr.title, kind) && isSpentAttempt(thr, now) && new Date(thr.createdAt).getTime() >= lastInReviewAt, - ); + ).length; } /** A reviewer attempt that produced no verdict and never will: it failed, or it @@ -250,10 +214,10 @@ function reviewerThreadsThisCycle( /** * True when this reviewer has spent every attempt of the cycle on a FAILED run. * - * This is a dead end, not a wait: the claim key is (task, reviewer, cycle), so - * `claimReviewer` refuses every further dispatch until the card leaves and - * re-enters In Review — which only a reviewer verdict or a human can cause. The - * verdict is therefore never coming and the all-approved gate can never close, + * This is a dead end, not a wait: the budget is per (task, reviewer, cycle), so + * nothing dispatches this reviewer again until the card leaves and re-enters In + * Review — which only a reviewer verdict or a human can cause. The verdict is + * therefore never coming and the all-approved gate can never close, * so the caller hands the card to a person instead of letting the sweeper visit * it forever. Two cards sat In Review for six days on exactly this: one * approval each, and a QA Agent that had died twice. @@ -286,9 +250,8 @@ export function reviewerAttemptsExhausted( * created since the cycle started satisfied this, so when both reviewers died on * an infrastructure error (a database-connection timeout, in the burst that * prompted this) the card sat In Review with two dead reviewer threads and - * nothing to re-dispatch them — the claim row stayed, `claimReviewer` refused - * every retry for the rest of the cycle, and the verdicts never came. A failure - * is not a review. + * nothing to re-dispatch them for the rest of the cycle, and the verdicts never + * came. A failure is not a review. * * Bounded by `MAX_REVIEWER_ATTEMPTS` so a reviewer that cannot run doesn't loop: * once this cycle has that many failed attempts, the card is left alone for a @@ -311,9 +274,9 @@ export function reviewerHandledThisCycle( // A live run owns the cycle; never dispatch alongside it. "Live" is the // heartbeat, not the status: a reviewer whose pod died mid-run keeps // `in_progress` forever, and taking that at face value deadlocked the card — - // the claim stays spent so nothing re-dispatches, and the merge gate waits on - // a verdict that will never come. One sat that way while its co-reviewer had - // approved in 68 seconds. + // nothing re-dispatches, and the merge gate waits on a verdict that will + // never come. One sat that way while its co-reviewer had approved in 68 + // seconds. if (thisCycle.some((thr) => isReviewerThreadLive(thr, now))) return true; const spent = thisCycle.filter((thr) => isSpentAttempt(thr, now)); // Every attempt spent and the budget is gone — stop, a human owns it now. @@ -337,18 +300,54 @@ async function lastInReviewTime( return reviewCycleStart(activity); } +/** + * The one string that identifies a reviewer dispatch: (task, reviewer, cycle, + * attempt). Both fences below are derived from it, so they can only agree. + * `toISOString()` must be the ONLY serialization of the cycle — a formatting + * difference between the trigger paths silently breaks the fence. + */ +function reviewFenceKey( + taskId: string, + kind: ReviewerKind, + cycleAt: Date, + attempt: number, +): string { + return `review:${taskId}:${kind}:${cycleAt.toISOString()}:${attempt}`; +} + +/** + * The reviewer run's thread id, derived from the fence key so the `threads` PK + * IS the dispatch fence: the two triggers (60s sweeper, the task dialog's 10s + * poll) can race and the loser's insert conflicts instead of spawning a second + * reviewer run. + */ +function reviewerThreadId(fenceKey: string): string { + const digest = createHash("sha256") + .update(fenceKey) + .digest("hex") + .slice(0, 32); + return `thrd_${digest}`; +} + /** * Enqueue a single reviewer run: a fresh thread (titled `: `), * a "delegated to " timeline entry, and the review prompt dispatched * on the org's agent. The reviewer ends by calling `TASK_BOARD_REVIEW_DECISION`. + * + * `attempt` is the cycle's spent-attempt count — it only moves the fence, so a + * retry after a dead attempt gets ids of its own instead of colliding with the + * corpse. */ async function enqueueReviewerForTask( ctx: StudioContext, task: TaskBoardItem, kind: ReviewerKind, - reviewToken: string, + cycleAt: Date, + attempt: number, ): Promise { const organizationId = task.organizationId; + // Proves to TASK_BOARD_REVIEW_DECISION that the caller is this reviewer. + const reviewToken = mintReviewToken(task.id, kind, cycleAt); // Same harness the Super Agent runs on, for the same reason: a review needs // real `git`/`gh` on a checkout, and — the blocking one — only a @@ -435,7 +434,9 @@ async function enqueueReviewerForTask( ].join("\n"); // Create + link the reviewer thread and dispatch its run (shared plumbing). - await enqueueAgentRunForTask(ctx, task, { + const fenceKey = reviewFenceKey(task.id, kind, cycleAt, attempt); + const fenceThreadId = reviewerThreadId(fenceKey); + const { isNew } = await enqueueAgentRunForTask(ctx, task, { // A verdict is the last thing between this card and Done — it outranks // starting a new task for the next slot. runClass: "reviewer", @@ -452,7 +453,23 @@ async function enqueueReviewerForTask( } : {}), ...(repo ? { repo } : {}), + fence: { threadId: fenceThreadId, workflowID: fenceKey }, + }).catch(async (err) => { + // Nothing was dispatched, but the fence thread may already exist — and + // `reviewerHandledThisCycle` would then read it as this cycle's reviewer + // forever. Drop it so the next trigger retries. + await ctx.storage.threads + .delete(fenceThreadId) + .catch((delErr) => + console.error( + `[task-board] ${kind} fence thread cleanup failed`, + delErr, + ), + ); + throw err; }); + // A concurrent trigger got there first — it owns this reviewer's dispatch. + if (!isNew) return; // Timeline: "Super Agent delegated to " (machine actor → null), and // broadcast the now-linked thread so the card shows the reviewer session live diff --git a/apps/api/src/tools/task-board/enqueue-task-run.ts b/apps/api/src/tools/task-board/enqueue-task-run.ts index f056df163b..c817f3456c 100644 --- a/apps/api/src/tools/task-board/enqueue-task-run.ts +++ b/apps/api/src/tools/task-board/enqueue-task-run.ts @@ -53,8 +53,15 @@ export async function enqueueAgentRunForTask( pinnedRef?: string | null; /** Admission class for this run. See `dispatch-queue/run-priority.ts`. */ runClass?: RunClass; + /** + * Deterministic thread id + run workflow id, for a caller whose triggers can + * race (the reviewer enqueues). Both are `INSERT … ON CONFLICT DO NOTHING` + * in effect: the losing racer gets `isNew: false` and nothing is dispatched + * twice. Omit for a caller with a single trigger. + */ + fence?: { threadId: string; workflowID: string }; }, -): Promise<{ threadId: string }> { +): Promise<{ threadId: string; isNew: boolean }> { const organizationId = task.organizationId; const userId = task.assignedBy ?? task.createdBy; const harnessId = opts.harnessId ?? "decopilot"; @@ -63,6 +70,7 @@ export async function enqueueAgentRunForTask( const agentId = getDecopilotId(organizationId); const thread = await ctx.storage.threads.create({ + ...(opts.fence ? { id: opts.fence.threadId } : {}), organization_id: organizationId, title: opts.title, status: "in_progress", @@ -73,6 +81,8 @@ export async function enqueueAgentRunForTask( sandbox_provider_kind: "agent-sandbox", created_by: userId, }); + // Another trigger already created this run — it owns the dispatch. + if (!thread.isNew) return { threadId: thread.id, isNew: false }; // Bind the repo to the thread the way `load_repo` does — it's the only place a // repo persists for the synthetic Super Agent, and it's what makes @@ -144,41 +154,44 @@ export async function enqueueAgentRunForTask( runId: thread.id, }).emitRequestMessage(requestMessage); - await enqueueThreadRun({ - threadId: thread.id, - source: "background-tool", - request: { - messages: [requestMessage], - models: { - credentialId: model.credentialId, - thinking: { id: model.modelId, title: model.modelMeta.title }, - }, - agent: { id: agentId, ...(opts.agent ?? {}) }, - temperature: opts.temperature, - toolApprovalLevel: "auto", - mode: "default", - organizationId, - userId, - harnessId, - sandboxProviderKind: "agent-sandbox", - // Only meaningful for the repo-less sandbox run above: `resolveSandboxBranch` - // derives the key from the thread's repo when there is one, and needs the - // explicit bare key when there isn't. Carried in the durable snapshot, so a - // recovered re-dispatch resolves the same pod. - ...(opts.repo ? {} : sandboxBranch ? { branch: sandboxBranch } : {}), - taskId: thread.id, - // Reports tasks carry the subscription-billing stamp: their AI usage - // is included in the org subscription (billing/subsidized-runs.ts). - // `runClass` orders admission when the pod is at its cap — a reviewer or - // a retry outranks a brand-new task (see dispatch-queue/run-priority.ts). - // A free-form metadata string, so it changes no schema and no DBOS step - // I/O. Defaults to a new task: the class that nothing is waiting on. - runMetadata: { - ...taskRunMetadata(task), - [RUN_CLASS_METADATA_KEY]: opts.runClass ?? "new_task", + await enqueueThreadRun( + { + threadId: thread.id, + source: "background-tool", + request: { + messages: [requestMessage], + models: { + credentialId: model.credentialId, + thinking: { id: model.modelId, title: model.modelMeta.title }, + }, + agent: { id: agentId, ...(opts.agent ?? {}) }, + temperature: opts.temperature, + toolApprovalLevel: "auto", + mode: "default", + organizationId, + userId, + harnessId, + sandboxProviderKind: "agent-sandbox", + // Only meaningful for the repo-less sandbox run above: `resolveSandboxBranch` + // derives the key from the thread's repo when there is one, and needs the + // explicit bare key when there isn't. Carried in the durable snapshot, so a + // recovered re-dispatch resolves the same pod. + ...(opts.repo ? {} : sandboxBranch ? { branch: sandboxBranch } : {}), + taskId: thread.id, + // Reports tasks carry the subscription-billing stamp: their AI usage + // is included in the org subscription (billing/subsidized-runs.ts). + // `runClass` orders admission when the pod is at its cap — a reviewer or + // a retry outranks a brand-new task (see dispatch-queue/run-priority.ts). + // A free-form metadata string, so it changes no schema and no DBOS step + // I/O. Defaults to a new task: the class that nothing is waiting on. + runMetadata: { + ...taskRunMetadata(task), + [RUN_CLASS_METADATA_KEY]: opts.runClass ?? "new_task", + }, }, }, - }); + opts.fence ? { workflowID: opts.fence.workflowID } : undefined, + ); - return { threadId: thread.id }; + return { threadId: thread.id, isNew: true }; } diff --git a/apps/api/src/tools/task-board/review-decision.ts b/apps/api/src/tools/task-board/review-decision.ts index 197fe514eb..b3e13803b9 100644 --- a/apps/api/src/tools/task-board/review-decision.ts +++ b/apps/api/src/tools/task-board/review-decision.ts @@ -13,15 +13,17 @@ import { emitTaskBoardUpdated, handTaskToHuman } from "./run-reactions"; import { enqueueSuperAgentForTask } from "./enqueue-super-agent"; import { allEnabledReviewersVerifiedApproved, mergeLinkedPr } from "./merge-pr"; import { fetchPrConflict, pickActivePr } from "./prs-get"; +import { verifyReviewToken } from "./review-token"; import { reactToApprovedPrConflict } from "./conflict-reaction"; import { TaskQuotaError } from "@/billing/task-quota"; /** - * True when a resolved reviewToken claim actually belongs to THIS reviewer's - * claim on the CURRENT review cycle. + * True when a resolved LEGACY reviewToken claim actually belongs to THIS + * reviewer's claim on the CURRENT review cycle. (Current tokens are HMACs — + * see `review-token.ts`; this only serves runs dispatched before that deploy.) * - * `claimReviewer` mints a fresh row (and token) for every review cycle but - * never deletes the old one, so comparing only the reviewer field — as this + * The claims table held a fresh row (and token) per review cycle but never + * deleted the old one, so comparing only the reviewer field — as this * used to — let a token minted for an EARLIER cycle still verify: a reviewer * that kept its token from a prior bounce (visible in that run's own prompt, * see `enqueueReviewerForTask`) could replay it after being bounced back and @@ -113,14 +115,19 @@ export const TASK_BOARD_REVIEW_DECISION = defineTool({ throw new Error(`Task board item not found: ${taskBoardItemId}`); } - // Verify the caller's reviewToken against THIS task's CURRENT cycle. - const claim = reviewToken - ? await ctx.storage.taskBoard.resolveReviewClaimByToken( - taskBoardItemId, - reviewToken, - ) - : null; - const currentCycleAt = claim + // Verify the caller is the reviewer it claims to be, against THIS task's + // CURRENT cycle: the reviewToken must be the HMAC over (task, reviewer, + // cycle). An unverified decision is still recorded (so a dropped token + // never stalls the flow) but won't count toward an automatic merge (see the + // verified gate below). + // + // ROLLOUT: runs dispatched before this deploy carry a random `rtok_` + // from `task_board_review_claims`, so fall back to the table lookup — also + // cycle-scoped, or a token kept across a bounce re-approves with no review. + // Delete that branch (and the table, its migration, and + // `resolveReviewClaimByToken`) once every in-flight review cycle has + // drained — a day is ample. + const currentCycleAt = reviewToken ? reviewCycleStart( await ctx.storage.taskBoard.listActivity( taskBoardItemId, @@ -128,7 +135,22 @@ export const TASK_BOARD_REVIEW_DECISION = defineTool({ ), ) : 0; - const verified = reviewTokenVerified(claim, reviewer, currentCycleAt); + const verified = + !!reviewToken && + (verifyReviewToken( + reviewToken, + taskBoardItemId, + reviewer, + new Date(currentCycleAt), + ) || + reviewTokenVerified( + await ctx.storage.taskBoard.resolveReviewClaimByToken( + taskBoardItemId, + reviewToken, + ), + reviewer, + currentCycleAt, + )); if (decision === "request_changes") { // Break a runaway review loop BEFORE bouncing. A reviewer that keeps diff --git a/apps/api/src/tools/task-board/review-sweeper.ts b/apps/api/src/tools/task-board/review-sweeper.ts index 2afa386083..78abf5065a 100644 --- a/apps/api/src/tools/task-board/review-sweeper.ts +++ b/apps/api/src/tools/task-board/review-sweeper.ts @@ -22,7 +22,7 @@ * `DBOS.startWorkflow`, which DBOS rejects from a step: * `DBOSInvalidWorkflowTransitionError` ("Invalid call to a `workflow` * function from within a `step` or `transaction`", code 21). The per-reviewer - * `.catch` logged it and released the claim, so it retried and failed + * `.catch` logged it, so it retried and failed * forever. Even a task WITH a linked PR never got a reviewer. * * A sweeper fixes both at once and is the only shape that can: it runs on a @@ -30,9 +30,9 @@ * also makes the pipeline self-healing — every card already stranded is picked * up on the next tick, with no backfill. * - * Everything it calls is idempotent: `enqueueEnabledReviewers` claims per - * (task, reviewer, cycle), so re-running every tick cannot spawn duplicate - * reviewer runs. + * Everything it calls is idempotent: `enqueueEnabledReviewers` fences on a + * thread id derived from (task, reviewer, cycle, attempt) and a matching run + * workflow id, so re-running every tick cannot spawn duplicate reviewer runs. * * Idempotent is not the same as terminating, though, and conflating the two is * what took out the GitHub App's rate limit: a card whose checks never go green @@ -49,7 +49,7 @@ * It later grew a third job for the same reason — **retrying a failed merge**. * A card whose reviewers all approved but whose merge was refused (GitHub down, * the repo's connection deleted, a transient 500) is left In Review by - * `TASK_BOARD_REVIEW_DECISION`, and the cycle's reviewer claims are already + * `TASK_BOARD_REVIEW_DECISION`, and the cycle's reviewer attempts are already * spent, so nothing re-dispatches and nothing re-merges: the card is stranded * forever on a failure that has usually fixed itself minutes later. The sweeper * is again the only place that can retry, for the same reason as above. @@ -488,7 +488,7 @@ export class TaskBoardReviewSweeper { } // A card whose review already COMPLETED but whose merge failed is stranded: - // the cycle's reviewer claims are spent, so the dispatch below is a no-op + // the cycle's reviewer attempts are spent, so the dispatch below is a no-op // and nothing else ever retries. Retry the merge first — it's the only path // back out of In Review for these, and it's why the sweep must keep visiting // a card that has no reviewer left to enqueue. Gated on verified approval + diff --git a/apps/api/src/tools/task-board/review-token.test.ts b/apps/api/src/tools/task-board/review-token.test.ts new file mode 100644 index 0000000000..9d60b7415a --- /dev/null +++ b/apps/api/src/tools/task-board/review-token.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "bun:test"; +import { mintReviewToken, verifyReviewToken } from "./review-token"; + +const ITEM = "tbi_1"; +const CYCLE = new Date("2026-01-01T00:00:00.000Z"); + +describe("review token", () => { + it("round-trips the tuple it was minted for", () => { + const token = mintReviewToken(ITEM, "qa", CYCLE); + expect(token.startsWith("rtok_")).toBe(true); + expect(verifyReviewToken(token, ITEM, "qa", CYCLE)).toBe(true); + }); + + it("rejects another reviewer, task, or cycle", () => { + const token = mintReviewToken(ITEM, "qa", CYCLE); + expect(verifyReviewToken(token, ITEM, "code_review", CYCLE)).toBe(false); + expect(verifyReviewToken(token, "tbi_2", "qa", CYCLE)).toBe(false); + expect( + verifyReviewToken(token, ITEM, "qa", new Date(CYCLE.getTime() + 1)), + ).toBe(false); + }); + + it("rejects a tampered or empty token", () => { + const token = mintReviewToken(ITEM, "qa", CYCLE); + expect(verifyReviewToken(`${token}x`, ITEM, "qa", CYCLE)).toBe(false); + expect(verifyReviewToken(`rtok_${"A".repeat(43)}`, ITEM, "qa", CYCLE)).toBe( + false, + ); + expect(verifyReviewToken("", ITEM, "qa", CYCLE)).toBe(false); + }); +}); diff --git a/apps/api/src/tools/task-board/review-token.ts b/apps/api/src/tools/task-board/review-token.ts new file mode 100644 index 0000000000..989a81c42d --- /dev/null +++ b/apps/api/src/tools/task-board/review-token.ts @@ -0,0 +1,54 @@ +/** + * Reviewer-identity tokens: an HMAC over (task, reviewer, review cycle). + * + * The token is handed to a reviewer run in its prompt and echoed back to + * `TASK_BOARD_REVIEW_DECISION`, which is how that tool knows the caller really + * is the reviewer it says it is — without it `reviewer` is self-asserted and + * one agent could forge the "both reviewers approved" auto-merge gate. + * + * A signature, not a stored row: the tuple it signs is already derivable at + * verify time, so there is nothing to persist. Same signing key and compare as + * `file-storage/share-password.ts`. + */ + +import { createHmac, randomBytes, timingSafeEqual } from "node:crypto"; +import type { ReviewerKind } from "@decocms/shared/task-board"; +import { getSettings } from "@/settings"; + +let signingKey: Buffer | null = null; +function getSigningKey(): Buffer { + if (signingKey) return signingKey; + let secret: string | undefined; + try { + const settings = getSettings(); + secret = settings.studioJwtSecret ?? settings.betterAuthSecret; + } catch { + // Settings not initialized (e.g. unit tests) — fall back like auth/jwt.ts. + secret = undefined; + } + signingKey = secret ? Buffer.from(secret) : randomBytes(32); + return signingKey; +} + +/** The `rtok_` prefix is quoted verbatim in reviewer prompts — keep it. */ +export function mintReviewToken( + itemId: string, + reviewer: ReviewerKind, + cycleAt: Date, +): string { + const mac = createHmac("sha256", getSigningKey()) + .update(`${itemId}:${reviewer}:${cycleAt.toISOString()}`) + .digest("base64url"); + return `rtok_${mac}`; +} + +export function verifyReviewToken( + token: string, + itemId: string, + reviewer: ReviewerKind, + cycleAt: Date, +): boolean { + const a = Buffer.from(token); + const b = Buffer.from(mintReviewToken(itemId, reviewer, cycleAt)); + return a.length === b.length && timingSafeEqual(a, b); +}