diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 1e997c3398..be983f67d1 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -259,6 +259,7 @@ Invariants: - task must trace to company goal chain via `goal_id`, `parent_id`, or project-goal linkage - `in_progress` requires assignee - an `in_review -> done | cancelled` verdict is authorized against the current review policy while the issue row is locked; a policy change in the same request or a concurrent request cannot relax that verdict gate +- accepting or rejecting the review-confirmation interaction locks the issue row before resolving the interaction and reauthorizes against the current review policy in that transaction - while a restrictive review policy is stored, changing it requires an actor who is allowed by that row-locked policy - the transition into `in_review` and its requester activity record commit atomically, including transitions without an explicit review-interaction binding - terminal states: `done | cancelled` diff --git a/server/src/__tests__/issue-thread-interactions-service.test.ts b/server/src/__tests__/issue-thread-interactions-service.test.ts index f01204f5da..f284c1e328 100644 --- a/server/src/__tests__/issue-thread-interactions-service.test.ts +++ b/server/src/__tests__/issue-thread-interactions-service.test.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { + activityLog, agents, companies, createDb, @@ -48,6 +49,7 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { afterEach(async () => { await db.delete(issueThreadInteractions); + await db.delete(activityLog); await db.delete(issueComments); await db.delete(issueDocuments); await db.delete(documentRevisions); @@ -100,6 +102,27 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { return { companyId, goalId, issueId }; } + async function recordReviewTransition(args: { + companyId: string; + issueId: string; + interactionId: string; + actorId?: string; + }) { + await db.insert(activityLog).values({ + companyId: args.companyId, + actorType: "user", + actorId: args.actorId ?? "local-board", + action: "issue.updated", + entityType: "issue", + entityId: args.issueId, + details: { + status: "in_review", + reviewInteractionId: args.interactionId, + _previous: { status: "in_progress" }, + }, + }); + } + it("persists addressees without allowing them to bypass board-only governance", async () => { const { companyId, issueId } = await seedConfirmationIssue("Agent-addressed interaction"); const creatorAgentId = randomUUID(); @@ -1687,6 +1710,7 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { }, { userId: "local-board", }); + await recordReviewTransition({ companyId, issueId, interactionId: created.id }); const accepted = await interactionsSvc.acceptInteraction({ id: issueId, @@ -1707,10 +1731,113 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { }); }); + it.each(["accept", "reject"] as const)( + "revalidates review policy under the issue lock before interaction %s", + async (action) => { + const { companyId, goalId, issueId } = await seedConfirmationIssue(`Locked ${action} policy`); + const resolverAgentId = randomUUID(); + const resolverRunId = randomUUID(); + await db.insert(agents).values({ + id: resolverAgentId, + companyId, + name: "Review agent", + role: "reviewer", + status: "active", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }); + await db.insert(heartbeatRuns).values({ + id: resolverRunId, + companyId, + agentId: resolverAgentId, + invocationSource: "manual", + status: "running", + startedAt: new Date(), + }); + const created = await interactionsSvc.create({ id: issueId, companyId }, { + kind: "request_confirmation", + payload: { version: 1, prompt: "Approve this review?" }, + }, { + userId: "local-board", + }); + await db.update(issues) + .set({ status: "in_review", reviewPolicy: "anyone" }) + .where(eq(issues.id, issueId)); + await recordReviewTransition({ companyId, issueId, interactionId: created.id }); + + let releasePolicyLock!: () => void; + let policyLockReady!: () => void; + const holdPolicyLock = new Promise((resolve) => { + releasePolicyLock = resolve; + }); + const policyLocked = new Promise((resolve) => { + policyLockReady = resolve; + }); + const tightenPolicy = db.transaction(async (tx) => { + await tx.select({ id: issues.id }) + .from(issues) + .where(eq(issues.id, issueId)) + .for("update"); + await tx.update(issues) + .set({ reviewPolicy: "human_only" }) + .where(eq(issues.id, issueId)); + policyLockReady(); + await holdPolicyLock; + }); + await policyLocked; + + const actor = { + agentId: resolverAgentId, + runId: resolverRunId, + reviewVerdictAuthorized: true, + }; + const verdict = action === "accept" + ? interactionsSvc.acceptInteraction({ + id: issueId, + companyId, + goalId, + projectId: null, + status: "in_review", + }, created.id, {}, actor) + : interactionsSvc.rejectInteraction({ + id: issueId, + companyId, + status: "in_review", + }, created.id, { reason: "Needs changes" }, actor); + let verdictSettled = false; + void verdict.then( + () => { verdictSettled = true; }, + () => { verdictSettled = true; }, + ); + const denied = expect(verdict).rejects.toMatchObject({ + status: 403, + details: expect.objectContaining({ + code: "review_policy_denied", + policy: "human_only", + }), + }); + + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(verdictSettled).toBe(false); + releasePolicyLock(); + await tightenPolicy; + await denied; + + const persisted = await db.select({ status: issueThreadInteractions.status }) + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, created.id)) + .then((rows) => rows[0]); + expect(persisted?.status).toBe("pending"); + }, + ); + it("preserves creator and same-run guards for authorized agent review verdicts", async () => { const { companyId, goalId, issueId } = await seedConfirmationIssue("Guard agent review verdicts"); const resolverAgentId = randomUUID(); const resolverRunId = randomUUID(); + await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, issueId)); await db.insert(agents).values({ id: resolverAgentId, companyId, @@ -1740,6 +1867,7 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { await db.update(issueThreadInteractions) .set({ createdByAgentId: resolverAgentId }) .where(eq(issueThreadInteractions.id, createdByResolver.id)); + await recordReviewTransition({ companyId, issueId, interactionId: createdByResolver.id }); const createdBySameRun = await interactionsSvc.create({ id: issueId, companyId }, { kind: "request_checkbox_confirmation", @@ -1763,6 +1891,13 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => { }; await expect(interactionsSvc.acceptInteraction(issue, createdByResolver.id, {}, actor)) .rejects.toThrow("Agents cannot resolve interactions they created"); + await db.update(activityLog).set({ + details: { + status: "in_review", + reviewInteractionId: createdBySameRun.id, + _previous: { status: "in_progress" }, + }, + }).where(eq(activityLog.entityId, issueId)); await expect(interactionsSvc.acceptInteraction(issue, createdBySameRun.id, { selectedOptionIds: ["approve"], }, actor)).rejects.toThrow("Agents cannot resolve interactions created by the same run"); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 84f0cd667a..50ae5ccd19 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -4155,6 +4155,7 @@ export function issueRoutes( assertBoard(req); if (isReviewConfirmationVerdict) { await assertPendingReviewInteractionVerdictAllowed(req, issue, interaction); + return "review_verdict" as const; } return "standard" as const; } @@ -10682,7 +10683,7 @@ export function issueRoutes( agentId: actor.agentId, runId: actor.runId, userId: actor.actorType === "user" ? actor.actorId : null, - ...(actor.actorType === "agent" && resolutionAuthorization === "review_verdict" + ...(resolutionAuthorization === "review_verdict" ? { reviewVerdictAuthorized: true } : {}), }); @@ -10838,7 +10839,7 @@ export function issueRoutes( agentId: actor.agentId, runId: actor.runId, userId: actor.actorType === "user" ? actor.actorId : null, - ...(actor.actorType === "agent" && resolutionAuthorization === "review_verdict" + ...(resolutionAuthorization === "review_verdict" ? { reviewVerdictAuthorized: true } : {}), }); diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 11ffee195e..7bbe2888ab 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -20,6 +20,7 @@ import type { CancelIssueThreadInteraction, CreateIssueThreadInteraction, InteractionResolverGovernance, + IssueReviewPolicy, IssueThreadInteraction, IssueThreadInteractionKind, IssueThreadInteractionResolverPolicy, @@ -59,6 +60,10 @@ import { conflict, forbidden, notFound, unprocessable } from "../errors.js"; import { getTelemetryClient } from "../telemetry.js"; import { logActivity } from "./activity-log.js"; import { evaluateAgentInvokabilityFromDb } from "./agent-invokability.js"; +import { + assertIssueReviewVerdictActorAllowed, + isIssueReviewVerdictInteraction, +} from "./issue-review-policy.js"; import { issueService, runWorkspaceIsFinalized } from "./issues.js"; import { createPullRequestMergeStateResolver, @@ -286,8 +291,48 @@ type IssueResolutionContext = { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; + reviewPolicy: IssueReviewPolicy | null; + createdByAgentId: string | null; + createdByUserId: string | null; }; +async function assertRequestConfirmationResolutionAllowedUnderLock( + tx: Db, + issue: IssueResolutionContext, + interaction: IssueThreadInteractionRow, + actor: InteractionActor, +) { + if (isTerminalIssueStatus(issue.status)) { + throw conflict("Interaction is no longer actionable because the issue is closed"); + } + + const isReviewVerdict = issue.status === "in_review" + && isRequestConfirmationLikeKind(interaction.kind) + && await isIssueReviewVerdictInteraction(tx, { issue, interaction }); + + if (!isReviewVerdict) { + assertAgentResolutionAllowed(interaction, { + ...actor, + reviewVerdictAuthorized: false, + }); + return; + } + + if (actor.agentId) assertAgentInteractionActorAllowed(interaction, actor); + const verdictActor = actor.agentId + ? { type: "agent" as const, id: actor.agentId } + : actor.userId + ? { type: "user" as const, id: actor.userId } + : null; + if (!verdictActor) { + throw forbidden("A review verdict requires an authenticated agent or user"); + } + await assertIssueReviewVerdictActorAllowed(tx, { + issue, + actor: verdictActor, + }); +} + const REQUEST_CONFIRMATION_INTERACTION_KINDS = [ "request_confirmation", "request_checkbox_confirmation", @@ -1330,17 +1375,62 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti return { interaction: expired, continuationIssue: null }; } - const interaction = hydrateInteraction(args.current); - const selectedOptionIds = - interaction.kind === "request_checkbox_confirmation" + const now = new Date(); + const result = await db.transaction(async (tx) => { + // Lock the issue before claiming the interaction. Policy mutations and + // review transitions use the same issue-row lock, so the authoritative + // review policy and requester are stable through the verdict write. + const issueContext = await tx + .select({ + id: issues.id, + companyId: issues.companyId, + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + reviewPolicy: issues.reviewPolicy, + createdByAgentId: issues.createdByAgentId, + createdByUserId: issues.createdByUserId, + }) + .from(issues) + .where(eq(issues.id, args.issue.id)) + .for("update") + .then((rows: IssueResolutionContext[]) => rows[0] ?? null); + + if (!issueContext || issueContext.companyId !== args.issue.companyId) { + throw notFound("Issue not found"); + } + + const lockedCurrent = await tx + .select() + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, args.current.id)) + .for("update") + .then((rows) => rows[0] ?? null); + if ( + !lockedCurrent + || lockedCurrent.companyId !== args.issue.companyId + || lockedCurrent.issueId !== args.issue.id + ) { + throw notFound("Interaction not found"); + } + if (lockedCurrent.status !== "pending") { + throw conflict("Interaction has already been resolved"); + } + await assertRequestConfirmationResolutionAllowedUnderLock( + tx as unknown as Db, + issueContext, + lockedCurrent, + args.actor, + ); + + const interaction = hydrateInteraction(lockedCurrent); + const selectedOptionIds = interaction.kind === "request_checkbox_confirmation" ? resolveSelectedCheckboxConfirmationOptions({ interaction, selectedOptionIds: args.input.selectedOptionIds, }) : undefined; - const now = new Date(); - const result = await db.transaction(async (tx) => { const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -1357,7 +1447,7 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti updatedAt: now, }) .where(and( - eq(issueThreadInteractions.id, args.current.id), + eq(issueThreadInteractions.id, lockedCurrent.id), eq(issueThreadInteractions.status, "pending"), )) .returning(); @@ -1366,32 +1456,16 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti throw conflict("Interaction has already been resolved"); } - const issueContext = await tx - .select({ - id: issues.id, - companyId: issues.companyId, - status: issues.status, - assigneeAgentId: issues.assigneeAgentId, - assigneeUserId: issues.assigneeUserId, - }) - .from(issues) - .where(eq(issues.id, args.issue.id)) - .then((rows: IssueResolutionContext[]) => rows[0] ?? null); - - if (!issueContext || issueContext.companyId !== args.issue.companyId) { - throw notFound("Issue not found"); - } - let continuationIssue: IssueWakeTarget | null = null; if (shouldReturnAcceptedConfirmationToCreatorAgent({ issue: issueContext, - current: args.current, + current: lockedCurrent, actor: args.actor, })) { const returnStatus = issueContext.status === "blocked" ? "blocked" : "todo"; const returnedIssue = await issueService(db).update(args.issue.id, { status: returnStatus, - assigneeAgentId: args.current.createdByAgentId, + assigneeAgentId: lockedCurrent.createdByAgentId, assigneeUserId: null, actorAgentId: args.actor.agentId ?? null, actorUserId: args.actor.userId ?? null, @@ -1420,12 +1494,12 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti entityType: "issue", entityId: args.issue.id, details: { - interactionId: args.current.id, - interactionKind: args.current.kind, + interactionId: lockedCurrent.id, + interactionKind: lockedCurrent.kind, interactionStatus: "accepted", resolutionActorKind: "system", - requestedResolverPolicy: args.current.requestedResolverPolicy, - effectiveResolverPolicy: args.current.effectiveResolverPolicy, + requestedResolverPolicy: lockedCurrent.requestedResolverPolicy, + effectiveResolverPolicy: lockedCurrent.effectiveResolverPolicy, ...(args.actor.resolutionDetails ?? {}), }, }); @@ -1461,31 +1535,77 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti } const now = new Date(); - const [updated] = await db - .update(issueThreadInteractions) - .set({ - status: "rejected", - result: { - version: 1, - outcome: "rejected", - reason: reason || null, - }, - resolvedByAgentId: args.actor.agentId ?? null, - resolvedByRunId: args.actor.runId ?? null, - resolvedByUserId: args.actor.userId ?? null, - resolvedAt: now, - updatedAt: now, - }) - .where(and( - eq(issueThreadInteractions.id, args.current.id), - eq(issueThreadInteractions.status, "pending"), - )) - .returning(); + const updated = await db.transaction(async (tx) => { + const issueContext = await tx + .select({ + id: issues.id, + companyId: issues.companyId, + status: issues.status, + assigneeAgentId: issues.assigneeAgentId, + assigneeUserId: issues.assigneeUserId, + reviewPolicy: issues.reviewPolicy, + createdByAgentId: issues.createdByAgentId, + createdByUserId: issues.createdByUserId, + }) + .from(issues) + .where(eq(issues.id, args.issue.id)) + .for("update") + .then((rows: IssueResolutionContext[]) => rows[0] ?? null); + if (!issueContext || issueContext.companyId !== args.issue.companyId) { + throw notFound("Issue not found"); + } + + const lockedCurrent = await tx + .select() + .from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, args.current.id)) + .for("update") + .then((rows) => rows[0] ?? null); + if ( + !lockedCurrent + || lockedCurrent.companyId !== args.issue.companyId + || lockedCurrent.issueId !== args.issue.id + ) { + throw notFound("Interaction not found"); + } + if (lockedCurrent.status !== "pending") { + throw conflict("Interaction has already been resolved"); + } + await assertRequestConfirmationResolutionAllowedUnderLock( + tx as unknown as Db, + issueContext, + lockedCurrent, + args.actor, + ); + + const [resolved] = await tx + .update(issueThreadInteractions) + .set({ + status: "rejected", + result: { + version: 1, + outcome: "rejected", + reason: reason || null, + }, + resolvedByAgentId: args.actor.agentId ?? null, + resolvedByRunId: args.actor.runId ?? null, + resolvedByUserId: args.actor.userId ?? null, + resolvedAt: now, + updatedAt: now, + }) + .where(and( + eq(issueThreadInteractions.id, lockedCurrent.id), + eq(issueThreadInteractions.status, "pending"), + )) + .returning(); + + if (!resolved) { + throw conflict("Interaction has already been resolved"); + } + await touchIssue(tx, args.issue.id); + return resolved; + }); - if (!updated) { - throw conflict("Interaction has already been resolved"); - } - await touchIssue(db, args.issue.id); const rejected = hydrateInteraction(updated); await emitInteractionResolvedTelemetry(db, rejected); return rejected;