diff --git a/server/src/services/issue-execution-policy.ts b/server/src/services/issue-execution-policy.ts index 0bde880c9d1..16d3e8b45f0 100644 --- a/server/src/services/issue-execution-policy.ts +++ b/server/src/services/issue-execution-policy.ts @@ -1237,6 +1237,39 @@ export function applyIssueExecutionPolicyTransition(input: TransitionInput): Tra return stageResult; } +/** Status decision reason when verified native work enters a configured review stage. */ +export const COMPLETION_REVIEW_STAGE_REASON = "execution_stage_review_required"; +/** Status decision reason when the configured review policy refuses the completion, as the route would. */ +export const COMPLETION_REVIEW_STAGE_REJECTED_REASON = "execution_stage_transition_rejected"; + +export type CompletionReviewStageEntry = + | { kind: "none" } + | { kind: "starts"; patch: Record; state: IssueExecutionState } + | { kind: "rejected"; message: string }; + +/** + * What a worker's completion does to the task's configured execution policy, + * exactly as the issue route would apply `status: "done"`: start the next + * pending stage (with the monitor transition for the new assignee), leave the + * task free to close, or refuse the transition. Refusals fail closed. + */ +export function completionReviewStageEntry(issue: IssueLike, actorAgentId: string): CompletionReviewStageEntry { + let patch: Record; + try { + const policy = normalizeIssueExecutionPolicy(issue.executionPolicy ?? null); + if (!policy) return { kind: "none" }; + ({ patch } = applyIssueExecutionPolicyTransition({ + issue, policy, previousPolicy: policy, requestedStatus: "done", + requestedAssigneePatch: {}, actor: { agentId: actorAgentId, userId: null }, + })); + } catch (error) { + return { kind: "rejected", message: error instanceof Error ? error.message : String(error) }; + } + const state = patch.executionState as IssueExecutionState | null | undefined; + if (patch.status !== "in_review" || state?.status !== "pending" || !state.currentParticipant) return { kind: "none" }; + return { kind: "starts", patch, state }; +} + export function applyIssueMonitorPolicyTransition(input: TransitionInput): TransitionResult { return { patch: applyMonitorTransition(input, {}) }; } diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index dadd829fbd9..20398e40e37 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1,3 +1,4 @@ +import { completionReviewStageEntry } from "./issue-execution-policy.js"; import { agentConfigurationChangeTargetId } from "./change-consent-gate.js"; import { assertAgentInteraction } from "./agent-interaction-policy.js"; import { resolveNewTaskParent } from "./issue-hierarchy.js"; @@ -524,6 +525,8 @@ type IssueResolutionContext = { reviewPolicy: IssueReviewPolicy | null; createdByAgentId: string | null; createdByUserId: string | null; + executionPolicy?: unknown; + executionState?: unknown; }; async function assertRequestConfirmationResolutionAllowedUnderLock( @@ -2462,6 +2465,8 @@ export function issueThreadInteractionService( reviewPolicy: issues.reviewPolicy, createdByAgentId: issues.createdByAgentId, createdByUserId: issues.createdByUserId, + executionPolicy: issues.executionPolicy, + executionState: issues.executionState, }) .from(issues) .where(eq(issues.id, args.issue.id)) @@ -2586,10 +2591,18 @@ export function issueThreadInteractionService( )).limit(1); // Each explicit reviewer must be able to answer independently. Completing // on the first answer would cancel the other pending decisions. + // A task whose configured review stage has not run cannot be closed by + // this card (for example an early review of unfinished work): the + // answer resumes the worker, whose finished work then enters that stage. + const configuredReviewPending = completionReviewStageEntry({ + status: issueContext.status, assigneeAgentId: issueContext.assigneeAgentId, assigneeUserId: issueContext.assigneeUserId, + executionPolicy: issueContext.executionPolicy as Record | null, + executionState: issueContext.executionState as Record | null, + }, issueContext.assigneeAgentId ?? "").kind !== "none"; const completedIssue = otherPending.length > 0 || issueContext.status !== "in_review" ? null : await issueService(db).update( args.issue.id, { - status: "done", + status: configuredReviewPending ? "todo" : "done", actorAgentId: args.actor.agentId ?? null, actorUserId: args.actor.userId ?? null, }, @@ -2755,6 +2768,8 @@ export function issueThreadInteractionService( reviewPolicy: issues.reviewPolicy, createdByAgentId: issues.createdByAgentId, createdByUserId: issues.createdByUserId, + executionPolicy: issues.executionPolicy, + executionState: issues.executionState, }) .from(issues) .where(eq(issues.id, args.issue.id)) diff --git a/server/src/services/native-runtime/native-completion-feedback.ts b/server/src/services/native-runtime/native-completion-feedback.ts index 18969541880..e091aa4db8a 100644 --- a/server/src/services/native-runtime/native-completion-feedback.ts +++ b/server/src/services/native-runtime/native-completion-feedback.ts @@ -10,6 +10,7 @@ import { findSupersededExternalReviewRequests } from "./superseded-external-revi import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js"; import { issueService } from "../issues.js"; import { isConversation } from "../agent-conversations.js"; +import { completionReviewStageEntry } from "../issue-execution-policy.js"; import { issueTreeControlService } from "../issue-tree-control.js"; import { isPublishedGitHubReviewCompletionContinuation, readPublishedGitHubReviewCompletionContinuation, resolveExternalChatResponseWaitAuthorization } from "./chat-attachment-reuse.js"; import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; @@ -381,6 +382,21 @@ export async function nativeCompletionFeedback( ); } } + // An approval request can be a governed action card, so only a review + // request duplicates the configured stage. + const reviewStage = issue.executionState?.status !== "pending" && !isConversation(issue) && + completionReviewStageEntry(issue, run.agentId).kind === "starts"; + // Only a review of finished work belongs to the configured stage; a review + // needed before the work can continue keeps its own request path. + const workFinished = result.completionClaim.objectiveSatisfied && + !result.completionClaim.remainingWork.some(work => work.blocksCompletion); + if (reviewStage && workFinished && result.reportedWorkDisposition === "needs_review" && + (signals.actionableAttentionRequests.length === 0 || + signals.actionableAttentionRequests.some(request => request.kind === "review"))) { + throw new Error( + "This task has a configured review stage with its own reviewer. Do not request a separate review. Report done once the work is complete and verified; Paperclip then starts the configured review, and the task stays in review until that reviewer decides.", + ); + } if ( result.reportedWorkDisposition === "needs_review" && signals.actionableAttentionRequests.length === 0 @@ -407,6 +423,9 @@ export async function nativeCompletionFeedback( if (result.reportedWorkDisposition === "blocked") { return "Blocker report accepted. Explain why work cannot continue, name the blocker owner and give the unblock action in your final response; do not describe the task as completed."; } + if (reviewStage && result.reportedWorkDisposition === "done") { + return `Completion report accepted. This task's configured review stage starts after this turn, and the task stays in review until its reviewer decides. Describe the delivered work and say it awaits that review on [this task](/issues/${issue.identifier ?? issue.id}); do not say the task is done.`; + } const links = await savedDocumentLinks(db, run, issue); const documentGuidance = links.length ? ` When compatible with the requested response format, include these clickable links to this run's saved documents in your final response: ${links.join(" ")}` : ""; diff --git a/server/src/services/native-runtime/native-execution-policy-review-stage-race.test.ts b/server/src/services/native-runtime/native-execution-policy-review-stage-race.test.ts new file mode 100644 index 00000000000..18d71af4ee8 --- /dev/null +++ b/server/src/services/native-runtime/native-execution-policy-review-stage-race.test.ts @@ -0,0 +1,96 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agentWakeupRequests, agents, companies, companyMemberships, completionContracts, createDb, heartbeatRuns, issueWorkProducts, issues, statusDecisions } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { nativeCompletionFeedback } from "./native-completion-feedback.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../../vendor/paperclip-runner/testing.js"; +import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js"; + + +const staleRead = vi.hoisted(() => ({ remaining: 0 })); +vi.mock("../issue-execution-policy.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, completionReviewStageEntry: (...args: Parameters) => { + if (staleRead.remaining > 0) { staleRead.remaining -= 1; return { kind: "none" as const }; } + return actual.completionReviewStageEntry(...args); + } }; +}); + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("native completion when the review policy changes before commit", () => { + let temporary: Awaited>; + let db: ReturnType; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-review-stage-race-"); + db = createDb(temporary.connectionString); + }, 90_000); + afterAll(async () => { await temporary?.cleanup(); }); + + type Options = { stage?: "user" | "agent"; monitor?: boolean; liveCycleWithRemovedStage?: boolean; unprovenEvidence?: boolean; + beforeFinish?: (runId: string, result: PrpStructuredRunResult) => Promise }; + async function finishVerifiedDone(options: Options = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const reviewerId = `reviewer-${randomUUID()}`, reviewerAgentId = randomUUID(), stageId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Review stage", issuePrefix: `R${companyId.slice(0, 7)}` }); + await db.insert(agents).values([{ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner", status: "active" }, + { id: reviewerAgentId, companyId, name: "Reviewer", adapterType: "paperclip_runner", status: "active" }]); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: reviewerId, status: "active", membershipRole: "operator" }); + const participant = options.stage === "agent" ? { id: randomUUID(), type: "agent", agentId: reviewerAgentId } : { id: randomUUID(), type: "user", userId: reviewerId }; + const nextCheckAt = new Date(Date.now() + 3_600_000); + await db.insert(issues).values({ id: issueId, companyId, title: "Build the site", status: "in_progress", assigneeAgentId: agentId, + ...(options.stage || options.liveCycleWithRemovedStage ? { executionPolicy: { mode: "normal", commentRequired: true, + stages: [{ id: stageId, type: "review", approvalsNeeded: 1, participants: [participant] }], + ...(options.monitor ? { monitor: { nextCheckAt: nextCheckAt.toISOString(), notes: "Check the deploy.", scheduledBy: "assignee" } } : {}) } } : {}), + ...(options.monitor ? { monitorNextCheckAt: nextCheckAt } : {}), + ...(options.liveCycleWithRemovedStage ? { executionState: { status: "changes_requested", currentStageId: randomUUID(), currentStageIndex: 0, + currentStageType: "review", currentParticipant: { type: "user", userId: reviewerId }, returnAssignee: { type: "agent", agentId }, + completedStageIds: [], lastDecisionId: randomUUID(), lastDecisionOutcome: "changes_requested" } } : {}) }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, status: "running", + runtimeMode: "native", contextSnapshot: { issueId } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const contractId = randomUUID(), sessionId = randomUUID(), runnerId = randomUUID(), contractSha256 = `contract-${runId}`; + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "standalone-v1", objective: "Build the site", criteria: [{ id: "objective", requirement: "Build the site" }] }, + canonicalSha256: contractSha256, createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ completionContractId: contractId, completionContractSha256: contractSha256, + nativeSessionId: sessionId, runnerInstanceId: runnerId }).where(eq(heartbeatRuns.id, runId)); + const [verified] = await db.insert(issueWorkProducts).values({ companyId, issueId, + type: "artifact", provider: "test", title: "Site check", status: "completed", reviewState: "approved" }).returning(); + const evidenceRef = `work_product:${verified!.id}`; + // The recorded failing report: a file name and an unbound attachment ref, + // and a passed check with no durable artifact. The worker cannot make its + // own unreviewed work product authoritative; the configured reviewer does. + const refs = options.unprovenEvidence ? ["website.html", `attachment:${randomUUID()}`] : [evidenceRef]; + const result: PrpStructuredRunResult = { ...CONTROL_PLANE_CONFORMANCE_RESULT, + completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, + criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: refs }] }, + evidence: options.unprovenEvidence ? refs.map(ref => ({ ref })) : [{ kind: "work_product", ref: evidenceRef }], + verification: [options.unprovenEvidence ? { commandOrCheck: "python3 html.parser verification of website.html", status: "passed" } + : { commandOrCheck: "Site check", status: "passed", artifactRef: evidenceRef }] } as PrpStructuredRunResult; + if (options.beforeFinish) await options.beforeFinish(runId, result); + const identity = { companyId, agentId, issueId, runId }; + const port = new PaperclipControlPlanePort(db, { ...identity, sessionId, completionContractId: contractId, + completionContractSha256: contractSha256, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: randomUUID() }); + await port.openRun({ identity: { ...identity, sessionId }, backendKind: "mock", sourceInstanceId: runnerId }); + await port.completeRun({ result, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, callerResultId: randomUUID() }); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" }); + const [task] = await db.select().from(issues).where(eq(issues.id, issueId)); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, issueId)); + const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + return { task: task!, decisions, wakes, reviewerId, reviewerAgentId, stageId, agentId }; + } + + it("does not close a task whose policy gained a review stage after the finalizer read it", async () => { + // The finalizer reads the task as having no stage (a stale read); the + // committer then sees the configured stage under the lock. + staleRead.remaining = 1; + const r = await finishVerifiedDone({ stage: "user" }); + expect(r.task.status).not.toBe("done"); + expect(r.decisions.some(row => row.toStatus === "done")).toBe(false); + }); +}); diff --git a/server/src/services/native-runtime/native-execution-policy-review-stage.test.ts b/server/src/services/native-runtime/native-execution-policy-review-stage.test.ts new file mode 100644 index 00000000000..a518ec9119a --- /dev/null +++ b/server/src/services/native-runtime/native-execution-policy-review-stage.test.ts @@ -0,0 +1,201 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agentWakeupRequests, agents, companies, companyMemberships, completionContracts, createDb, heartbeatRuns, issueThreadInteractions, issueWorkProducts, issues, statusDecisions } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { completionReviewStageEntry } from "../issue-execution-policy.js"; +import { nativeCompletionFeedback } from "./native-completion-feedback.js"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; +import { gateDoneOnConfiguredReview, supersedeStaleOwnerCompletion, type NativeStatusDecision } from "./status-arbiter.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../../vendor/paperclip-runner/testing.js"; +import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("native completion with a configured review stage", () => { + let temporary: Awaited>; + let db: ReturnType; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-review-stage-"); + db = createDb(temporary.connectionString); + }, 90_000); + afterAll(async () => { await temporary?.cleanup(); }); + + type Options = { stage?: "user" | "agent"; monitor?: boolean; liveCycleWithRemovedStage?: boolean; unprovenEvidence?: boolean; earlyReview?: boolean; + beforeFinish?: (runId: string, result: PrpStructuredRunResult) => Promise }; + async function finishVerifiedDone(options: Options = {}) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const reviewerId = `reviewer-${randomUUID()}`, reviewerAgentId = randomUUID(), stageId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Review stage", issuePrefix: `R${companyId.slice(0, 7)}` }); + await db.insert(agents).values([{ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner", status: "active" }, + { id: reviewerAgentId, companyId, name: "Reviewer", adapterType: "paperclip_runner", status: "active" }]); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: reviewerId, status: "active", membershipRole: "operator" }); + const participant = options.stage === "agent" ? { id: randomUUID(), type: "agent", agentId: reviewerAgentId } : { id: randomUUID(), type: "user", userId: reviewerId }; + const nextCheckAt = new Date(Date.now() + 3_600_000); + await db.insert(issues).values({ id: issueId, companyId, title: "Build the site", status: "in_progress", assigneeAgentId: agentId, + ...(options.stage || options.liveCycleWithRemovedStage ? { executionPolicy: { mode: "normal", commentRequired: true, + stages: [{ id: stageId, type: "review", approvalsNeeded: 1, participants: [participant] }], + ...(options.monitor ? { monitor: { nextCheckAt: nextCheckAt.toISOString(), notes: "Check the deploy.", scheduledBy: "assignee" } } : {}) } } : {}), + ...(options.monitor ? { monitorNextCheckAt: nextCheckAt } : {}), + ...(options.liveCycleWithRemovedStage ? { executionState: { status: "changes_requested", currentStageId: randomUUID(), currentStageIndex: 0, + currentStageType: "review", currentParticipant: { type: "user", userId: reviewerId }, returnAssignee: { type: "agent", agentId }, + completedStageIds: [], lastDecisionId: randomUUID(), lastDecisionOutcome: "changes_requested" } } : {}) }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, status: "running", + runtimeMode: "native", contextSnapshot: { issueId } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const contractId = randomUUID(), sessionId = randomUUID(), runnerId = randomUUID(), contractSha256 = `contract-${runId}`; + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "standalone-v1", objective: "Build the site", criteria: [{ id: "objective", requirement: "Build the site" }] }, + canonicalSha256: contractSha256, createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ completionContractId: contractId, completionContractSha256: contractSha256, + nativeSessionId: sessionId, runnerInstanceId: runnerId }).where(eq(heartbeatRuns.id, runId)); + const [verified] = await db.insert(issueWorkProducts).values({ companyId, issueId, + type: "artifact", provider: "test", title: "Site check", status: "completed", reviewState: "approved" }).returning(); + const evidenceRef = `work_product:${verified!.id}`; + // The recorded failing report: a file name and an unbound attachment ref, + // and a passed check with no durable artifact. The worker cannot make its + // own unreviewed work product authoritative; the configured reviewer does. + const refs = options.unprovenEvidence ? ["website.html", `attachment:${randomUUID()}`] : [evidenceRef]; + const result: PrpStructuredRunResult = { ...CONTROL_PLANE_CONFORMANCE_RESULT, + completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, + criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: refs }] }, + evidence: options.unprovenEvidence ? refs.map(ref => ({ ref })) : [{ kind: "work_product", ref: evidenceRef }], + verification: [options.unprovenEvidence ? { commandOrCheck: "python3 html.parser verification of website.html", status: "passed" } + : { commandOrCheck: "Site check", status: "passed", artifactRef: evidenceRef }] } as PrpStructuredRunResult; + const reported: PrpStructuredRunResult = options.earlyReview ? { ...result, reportedWorkDisposition: "needs_review", + completionClaim: { ...result.completionClaim, objectiveSatisfied: false, + criteria: [{ criterionId: "objective", status: "not_satisfied", evidenceRefs: [] }] as never, + remainingWork: [{ description: "Build the pages after the layout review", blocksCompletion: true }] }, + attentionRequests: [{ kind: "review", summary: "Riley: review the layout before I build the pages.", ownerClass: "human" }] } : result; + if (options.beforeFinish) await options.beforeFinish(runId, reported); + const identity = { companyId, agentId, issueId, runId }; + const port = new PaperclipControlPlanePort(db, { ...identity, sessionId, completionContractId: contractId, + completionContractSha256: contractSha256, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: randomUUID() }); + await port.openRun({ identity: { ...identity, sessionId }, backendKind: "mock", sourceInstanceId: runnerId }); + await port.completeRun({ result: reported, terminal: options.earlyReview ? { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, reportedWorkDisposition: "needs_review" } : CONTROL_PLANE_CONFORMANCE_TERMINAL, callerResultId: randomUUID() }); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" }); + const [task] = await db.select().from(issues).where(eq(issues.id, issueId)); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, issueId)); + const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + return { task: task!, decisions, wakes, reviewerId, reviewerAgentId, stageId, agentId, companyId, issueId }; + } + + it("closes verified work on a task without a configured review stage", async () => { + const r = await finishVerifiedDone(); + expect(r.task.status).toBe("done"); + expect(r.decisions.map(row => row.reasonCode)).toEqual(["completion_contract_satisfied"]); + }); + + it("starts the configured human review stage instead of closing verified work", async () => { + const r = await finishVerifiedDone({ stage: "user", monitor: true }); + expect(r.task).toMatchObject({ status: "in_review", assigneeUserId: r.reviewerId, assigneeAgentId: null, monitorNextCheckAt: null }); + expect(r.task.executionState).toMatchObject({ status: "pending", currentStageId: r.stageId, currentStageType: "review", + currentParticipant: { type: "user", userId: r.reviewerId }, returnAssignee: { type: "agent", agentId: r.agentId } }); + expect((r.task.executionPolicy as Record).monitor ?? null).toBeNull(); + expect(r.decisions.map(row => [row.toStatus, row.reasonCode])).toEqual([["in_review", "execution_stage_review_required"]]); + expect(r.wakes.filter(wake => wake.reason === "execution_review_requested")).toEqual([]); + }); + + it("starts the configured review for a clean done claim whose evidence only the reviewer can accept", async () => { + const r = await finishVerifiedDone({ stage: "user", unprovenEvidence: true }); + expect(r.task).toMatchObject({ status: "in_review", assigneeUserId: r.reviewerId }); + expect(r.task.executionState).toMatchObject({ status: "pending", currentParticipant: { type: "user", userId: r.reviewerId } }); + expect(r.decisions.map(row => [row.toStatus, row.reasonCode])).toEqual([["in_review", "execution_stage_review_required"]]); + expect(r.wakes.filter(wake => wake.agentId === r.agentId)).toEqual([]); + }); + + it("still requires proven evidence to close a task without a configured review", async () => { + const r = await finishVerifiedDone({ unprovenEvidence: true }); + expect(r.task.status).toBe("in_progress"); + expect(r.decisions.map(row => row.reasonCode)).toEqual(["completion_evidence_incomplete"]); + }); + + it("starts a configured agent review stage and wakes that reviewer", async () => { + const r = await finishVerifiedDone({ stage: "agent" }); + expect(r.task).toMatchObject({ status: "in_review", assigneeAgentId: r.reviewerAgentId, assigneeUserId: null }); + expect(r.task.executionState).toMatchObject({ status: "pending", currentParticipant: { type: "agent", agentId: r.reviewerAgentId } }); + expect(r.wakes.filter(wake => wake.reason === "execution_review_requested")) + .toEqual([expect.objectContaining({ agentId: r.reviewerAgentId, status: "queued" })]); + }); + + it("keeps the task open when the policy refuses Done for a live review cycle", async () => { + const r = await finishVerifiedDone({ liveCycleWithRemovedStage: true }); + expect(r.task.status).toBe("in_progress"); + expect(r.decisions.map(row => [row.toStatus, row.reasonCode])).toEqual([["in_progress", "execution_stage_transition_rejected"]]); + }); + + it("tells the worker the configured stage requests the review", async () => { + await finishVerifiedDone({ stage: "user", beforeFinish: async (runId, result) => { + const accepted = await nativeCompletionFeedback(db, runId, result); + expect(accepted).toContain("configured review stage starts after this turn"); + expect(accepted).toContain("do not say the task is done"); + const review = { kind: "review" as const, summary: "Riley: please review the site.", ownerClass: "human" as const }; + await expect(nativeCompletionFeedback(db, runId, { ...result, reportedWorkDisposition: "needs_review", attentionRequests: [review] })) + .rejects.toThrow("Report done once the work is complete and verified"); + await expect(nativeCompletionFeedback(db, runId, { ...result, reportedWorkDisposition: "needs_review", + attentionRequests: [review, { kind: "human_input", summary: "Which signup URL?", ownerClass: "human" }] })) + .rejects.toThrow("configured review stage with its own reviewer"); + await expect(nativeCompletionFeedback(db, runId, { ...result, reportedWorkDisposition: "needs_review" })) + .rejects.toThrow("configured review stage with its own reviewer"); + await expect(nativeCompletionFeedback(db, runId, { ...result, reportedWorkDisposition: "needs_review", + attentionRequests: [{ kind: "approval", summary: "Approve the page-service call.", ownerClass: "human" }] })) + .resolves.not.toContain("configured review stage"); + } }); + }); + + it("keeps the review request path for a review needed before the work can continue", async () => { + await finishVerifiedDone({ stage: "user", beforeFinish: async (runId, result) => { + const early = { ...result, reportedWorkDisposition: "needs_review" as const, + completionClaim: { ...result.completionClaim, objectiveSatisfied: false, + remainingWork: [{ description: "Build the pages after the layout review", blocksCompletion: true }] }, + attentionRequests: [{ kind: "review" as const, summary: "Riley: review the layout before I build the pages.", ownerClass: "human" as const }] }; + await expect(nativeCompletionFeedback(db, runId, early)).resolves.not.toContain("configured review stage"); + } }); + }); + + it("resumes the worker instead of closing when an early review card is accepted on a configured-review task", async () => { + const r = await finishVerifiedDone({ stage: "user", earlyReview: true }); + expect(r.task.status).toBe("in_review"); + const [card] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, r.issueId)); + expect(card).toMatchObject({ kind: "request_confirmation", status: "pending" }); + await issueThreadInteractionService(db).acceptInteraction( + { id: r.issueId, companyId: r.companyId, projectId: null, goalId: null, status: "in_review" }, card!.id, {}, { userId: r.reviewerId }); + const [after] = await db.select().from(issues).where(eq(issues.id, r.issueId)); + expect(after).toMatchObject({ status: "todo", assigneeAgentId: r.agentId }); + }); + + it("gates every Done-producing decision on the configured review", () => { + const done = { policyVersion: "test", statusAction: "done", toStatus: "done", reasonCode: "decision_superseded_by_new_evidence", + unblockDescriptor: null, effects: [{ kind: "release_checkout" }] } as unknown as NativeStatusDecision; + expect(gateDoneOnConfiguredReview(done, { kind: "none" }, "worker")).toBe(done); + expect(gateDoneOnConfiguredReview(done, { kind: "starts", patch: {}, state: {} as never }, "worker")) + .toMatchObject({ statusAction: "in_review", reasonCode: "execution_stage_review_required" }); + expect(gateDoneOnConfiguredReview(done, { kind: "rejected", message: "No eligible review participant." }, "worker")) + .toMatchObject({ statusAction: "in_progress", reasonCode: "execution_stage_transition_rejected" }); + const open = { ...done, statusAction: "in_progress", toStatus: "in_progress" } as NativeStatusDecision; + expect(gateDoneOnConfiguredReview(open, { kind: "starts", patch: {}, state: {} as never }, "worker")).toBe(open); + }); + + it("lets a newer run's ownership win before the review policy maps a stale Done", () => { + const done = { policyVersion: "test", statusAction: "done", toStatus: "done", reasonCode: "decision_superseded_by_new_evidence", + unblockDescriptor: null, effects: [{ kind: "release_checkout" }] } as unknown as NativeStatusDecision; + const heldByNewer = { status: "in_progress", assigneeAgentId: "worker", assigneeUserId: null, executionRunId: "newer-run" }; + const superseded = supersedeStaleOwnerCompletion(done, heldByNewer, "worker", "older-run"); + expect(superseded.statusAction).toBe("preserve"); + expect(gateDoneOnConfiguredReview(superseded, { kind: "rejected", message: "No eligible review participant." }, "worker")).toBe(superseded); + }); + + it("fails closed when the stored policy cannot be read", () => { + expect(completionReviewStageEntry({ status: "in_progress", assigneeAgentId: "worker", + executionPolicy: { stages: "unreadable" } as unknown as Record }, "worker").kind).toBe("rejected"); + }); + + it("keeps ordinary completion feedback when no review stage is configured", async () => { + await finishVerifiedDone({ beforeFinish: async (runId, result) => { + expect(await nativeCompletionFeedback(db, runId, result)).not.toContain("configured review stage"); + } }); + }); +}); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index a2a8c17e0c8..6ec13047d12 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -38,6 +38,7 @@ import { findSupersededExternalReviewRequests } from "./superseded-external-revi import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; import { conversationNativeDecision, isConversation, settleConversationTurnForRun } from "../agent-conversations.js"; import { issueTreeControlService } from "../issue-tree-control.js"; +import { COMPLETION_REVIEW_STAGE_REASON, completionReviewStageEntry } from "../issue-execution-policy.js"; import { randomUUID } from "node:crypto"; import { preserveNativeWorkspaceExportLease } from "./native-workspace-export-resume.js"; import { and, desc, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; @@ -66,6 +67,7 @@ import { supersedeHandedOffDecision, supersedeStaleOwnerCompletion, type NativeAuthoritativeIssueStatus, + gateDoneOnConfiguredReview, } from "./status-arbiter.js"; import { recordNativeWorkAssessment } from "./work-assessments.js"; import { @@ -1644,9 +1646,14 @@ export async function finalizeNativeRun(input: { }, authoritativeIssue) : null; if (supersededMonitorWait && supersededMonitorWait.resultId !== resultRow.id) throw new NativeStatusRaceError(); const qualityReview = await readRequiredQualityReview(input.db, authoritativeIssue); + // The configured review stage a completion would start, decided as the + // issue route decides `status: "done"`. Review runs keep their own path. + const reviewStage = isConversation(authoritativeIssue) || reviewContext + ? { kind: "none" as const } : completionReviewStageEntry(authoritativeIssue, run.agentId); const statusInput: Parameters[0] = { requiredQualityReviewSatisfied: qualityReview.satisfied, executionStageApprovedByRun: await executionStageApprovedByRun(input.db, run, authoritativeIssue), + executionReviewStageStarts: reviewStage.kind === "starts", caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)), monitorWaitDispatched: monitorDelivery !== null, monitorWaitPreserved: preservedMonitorWait !== null, @@ -1717,7 +1724,7 @@ export async function finalizeNativeRun(input: { proposedDecision = { ...proposedDecision, statusAction: "in_review", toStatus: "in_review", reasonCode: OWN_QUESTION_UNANSWERED_REASON, unblockDescriptor: null, effects: [{ kind: "release_checkout" }] }; } - if (proposedDecision.statusAction === "done") { + if (proposedDecision.statusAction === "done" || proposedDecision.reasonCode === COMPLETION_REVIEW_STAGE_REASON) { const continuation = record(run.contextSnapshot?.executionContinuation); const objective = typeof continuation.objective === "string" ? continuation.objective : [authoritativeIssue.title, authoritativeIssue.description].filter(Boolean).join("\n"); @@ -1733,6 +1740,7 @@ export async function finalizeNativeRun(input: { : effect.kind === "record_finalization_error" ? { ...effect, nextAction: error.message } : effect) }; } } + proposedDecision = gateDoneOnConfiguredReview(proposedDecision, reviewStage, run.agentId); const decision = conversationNativeDecision({ conversation: isConversation(authoritativeIssue), terminalState, workspaceFinalizeStatus: input.workspaceFinalizeStatus, hasGovernanceGate: !!governanceGate, diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index 225d979dc35..b2dca3eaa56 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -1,4 +1,5 @@ import { readRequiredQualityReview } from "../required-quality-review.js"; +import { completionReviewStageEntry } from "../issue-execution-policy.js"; import { hasNativeWorkspaceFinalizationOwner } from "./native-workspace-finalization-recovery-state.js"; import { and, desc, eq, gt, or } from "drizzle-orm"; import { completionContracts, heartbeatRuns, issueWorkProducts, issues, nativeRunResults, @@ -9,7 +10,7 @@ import { classifyNativeEvidence } from "./evidence-classifier.js"; import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js"; import { resolveNativeReconciliationStatus, type NativeReconciliationFacts } from "./native-reconciliation-status.js"; import { pendingNativeGovernance, resolveNativeFinalizerStatus } from "./native-run-finalizer.js"; -import { NATIVE_STATUS_ARBITER_POLICY_VERSION, supersedeStaleOwnerCompletion, type NativeAuthoritativeIssueStatus, type NativeStatusDecision } from "./status-arbiter.js"; +import { gateDoneOnConfiguredReview, NATIVE_STATUS_ARBITER_POLICY_VERSION, supersedeStaleOwnerCompletion, type NativeAuthoritativeIssueStatus, type NativeStatusDecision } from "./status-arbiter.js"; import { commitNativeStatusDecision, NativeStatusRaceError } from "./status-decision-committer.js"; import { recordNativeWorkAssessment } from "./work-assessments.js"; import { readPreservedNativeResultFinalization } from "./preserved-native-result-finalization.js"; @@ -215,9 +216,13 @@ export async function reassessNativeStatusFromSource( ? await pendingNativeGovernance({ db, companyId: row.companyId, issueId: row.issueId, runId: row.runId, executionState: record(currentIssue.executionState) }) : null; const qualityReview = currentIssue ? await readRequiredQualityReview(db, currentIssue) : null; + // The configured review policy gates Done here exactly as in the finalizer. + const reviewStage = currentIssue && !currentIssue.conversationAgentId + ? completionReviewStageEntry(currentIssue, row.agentId) : { kind: "none" as const }; let decision = retiredAutomaticReview && currentIssue ? resolveNativeFinalizerStatus({ hasPendingChildCompletion, + executionReviewStageStarts: reviewStage.kind === "starts", requiredQualityReviewSatisfied: qualityReview?.satisfied, assessment: reassessment, terminalState: "succeeded", workspaceFinalizeStatus: "succeeded", governanceGate, @@ -235,9 +240,12 @@ export async function reassessNativeStatusFromSource( requiredQualityReviewSatisfied: false, priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, agentId: row.agentId }); } // New evidence from a reassigned-away run cannot close the new owner's task. + // Ownership is checked first, as in the finalizer, so a stale run cannot + // turn its result into a review or a recovery for itself. if (currentIssue && !currentIssue.conversationAgentId) { decision = supersedeStaleOwnerCompletion(decision, { ...currentIssue, status: row.issueStatus }, row.agentId, row.runId); } + decision = gateDoneOnConfiguredReview(decision, reviewStage, row.agentId); let committed: Awaited>; try { committed = await commitNativeStatusDecision({ diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index 594f7338b21..f3b6fd85f3d 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -1,3 +1,4 @@ +import { COMPLETION_REVIEW_STAGE_REASON, COMPLETION_REVIEW_STAGE_REJECTED_REASON, type CompletionReviewStageEntry } from "../issue-execution-policy.js"; import type { NativeEvidenceAssessment } from "./evidence-classifier.js"; import { NATIVE_MODEL_REJECTION_MESSAGE, NATIVE_PROVIDER_CAPACITY_MAX_RETRIES, NATIVE_PROVIDER_OVERLOADED_CODE, NATIVE_PROVIDER_OVERLOADED_MESSAGE } from "./native-provider-failure.js"; @@ -113,6 +114,27 @@ export function heldByAnotherRun(issue: { executionRunId?: string | null }, runI return Boolean(issue.executionRunId && issue.executionRunId !== runId); } +/** + * The configured review policy decides a Done exactly as the issue route does: + * a stage that starts turns Done into its review, and a refused transition + * keeps the task open with a recovery note. Every Done-producing path applies it. + */ +export function gateDoneOnConfiguredReview( + decision: NativeStatusDecision, + stage: CompletionReviewStageEntry, + agentId: string, +): NativeStatusDecision { + if (decision.statusAction !== "done" || stage.kind === "none") return decision; + if (stage.kind === "starts") { + return { ...decision, statusAction: "in_review", toStatus: "in_review", reasonCode: COMPLETION_REVIEW_STAGE_REASON, + unblockDescriptor: null, effects: [{ kind: "release_checkout" }] }; + } + return { ...decision, statusAction: "in_progress", toStatus: "in_progress", reasonCode: COMPLETION_REVIEW_STAGE_REJECTED_REASON, + unblockDescriptor: null, effects: [{ kind: "release_checkout" }, { kind: "record_recovery", cause: COMPLETION_REVIEW_STAGE_REJECTED_REASON, + nextAction: `The task's configured review policy cannot start its review: ${stage.message} Correct the review stages or participants; do not mark the task done.`, + agentId }] }; +} + export function supersedeStaleOwnerCompletion( decision: NativeStatusDecision, issue: { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; @@ -120,7 +142,7 @@ export function supersedeStaleOwnerCompletion( agentId: string, runId?: string, ): NativeStatusDecision { - if (decision.statusAction !== "done") return decision; + if (decision.statusAction !== "done" && decision.reasonCode !== COMPLETION_REVIEW_STAGE_REASON) return decision; const ownerSuperseded = !ownsNativeTaskCompletion(issue, agentId); if (!ownerSuperseded && !(runId && heldByAnotherRun(issue, runId))) return decision; return { @@ -223,6 +245,11 @@ export function arbitrateNativeStatus(input: { * the completion authority for its done report, as the policy intends. */ executionStageApprovedByRun?: boolean; + /** + * Done would start the task's configured review stage. That reviewer, not + * the worker's evidence references, accepts the work, as on the issue route. + */ + executionReviewStageStarts?: boolean; /** Re-derived from the committed runner terminal and pinned execution identity. */ providerModelRejected?: boolean; providerOverloaded?: boolean; @@ -517,7 +544,14 @@ export function arbitrateNativeStatus(input: { !input.assessment.hasFailedVerification && input.assessment.attentionRequests.length === 0 && !input.assessment.hasBlockingRemainingWork; - const complete = evidenceComplete || policyClaimComplete || stageApprovalComplete; + const reviewStageClaim = + input.executionReviewStageStarts === true && + input.assessment.reportedDisposition === "done" && + input.assessment.objectiveClaimSatisfied && + !input.assessment.hasFailedVerification && + input.assessment.attentionRequests.length === 0 && + !input.assessment.hasBlockingRemainingWork; + const complete = evidenceComplete || policyClaimComplete || stageApprovalComplete || reviewStageClaim; if (complete && input.hasPendingChildCompletion) { return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, @@ -529,6 +563,16 @@ export function arbitrateNativeStatus(input: { effects: [{ kind: "release_checkout" }], }; } + if (complete && input.executionReviewStageStarts === true) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_review", + toStatus: "in_review", + reasonCode: COMPLETION_REVIEW_STAGE_REASON, + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }], + }; + } if (complete) { return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 52f69c7b5ac..f7ab8b98188 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -74,6 +74,7 @@ import { resolveExternalChatResponseWaitAuthorizationInTransaction, } from "./chat-attachment-reuse.js"; import { issueService } from "../issues.js"; +import { COMPLETION_REVIEW_STAGE_REASON, COMPLETION_REVIEW_STAGE_REJECTED_REASON, completionReviewStageEntry } from "../issue-execution-policy.js"; import { companySkillService } from "../company-skills.js"; import { syncRoutineRunStatusForIssue } from "../routine-run-status.js"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; @@ -1847,7 +1848,16 @@ export async function commitNativeStatusDecision(input: { if (input.decision.statusAction === "done" && !await onboardingFirstTaskCompletionAllowed(tx as unknown as Db, { companyId: input.companyId, issueId: input.issueId, runId: input.runId, })) throw new NativeStatusRaceError(); - if (input.decision.statusAction === "done" && !isConversation(issue) && heldByAnotherRun(issue, input.runId)) { + if ((input.decision.statusAction === "done" || input.decision.reasonCode === COMPLETION_REVIEW_STAGE_REASON || + input.decision.reasonCode === COMPLETION_REVIEW_STAGE_REJECTED_REASON) && + !isConversation(issue) && heldByAnotherRun(issue, input.runId)) { + throw new NativeStatusRaceError(); + } + // Every Done path (finalizer, reassessment, retired review) passes this + // lock. A policy that now requires or refuses review cannot be closed by a + // decision made without it; the finalizer recomputes against the policy. + if (input.decision.statusAction === "done" && !isConversation(issue) && + completionReviewStageEntry(issue, originRun?.agentId ?? "").kind !== "none") { throw new NativeStatusRaceError(); } if (input.decision.statusAction === "done") { @@ -2343,9 +2353,15 @@ export async function commitNativeStatusDecision(input: { if (!updated) throw new NativeStatusRaceError(); } else { failAt("status_projection", input.failpoint); + // Start the configured review stage from the locked task, as the issue + // route does; a changed policy makes the saved decision stale. + const reviewStage = input.decision.reasonCode === COMPLETION_REVIEW_STAGE_REASON + ? completionReviewStageEntry(issue, originRun?.agentId ?? "") : null; + if (reviewStage && reviewStage.kind !== "starts") throw new NativeStatusRaceError(); const projected = await issueService(tx as unknown as Db).update( input.issueId, { + ...(reviewStage?.kind === "starts" ? reviewStage.patch : {}), status: input.decision.toStatus, statusVersion: input.priorStatusVersion + 1, lastStatusDecisionId: decisionRow.id, @@ -2358,6 +2374,21 @@ export async function commitNativeStatusDecision(input: { ); if (!projected) throw new NativeStatusRaceError(); updated = projected; + // The route wakes an agent participant when its stage starts; a person + // acts from the task. This is a workflow wake, not the worker's message. + const participant = reviewStage?.kind === "starts" ? reviewStage.state.currentParticipant : null; + if (reviewStage?.kind === "starts" && participant?.type === "agent" && participant.agentId) { + const state = reviewStage.state; + const reason = state.currentStageType === "approval" ? "execution_approval_requested" : "execution_review_requested"; + const executionStage = { wakeRole: state.currentStageType === "approval" ? "approver" : "reviewer", + stageId: state.currentStageId, stageType: state.currentStageType, currentParticipant: participant, + returnAssignee: state.returnAssignee, reviewRequest: state.reviewRequest ?? null, + lastDecisionOutcome: state.lastDecisionOutcome, allowedActions: ["approve", "request_changes"] }; + await enqueueWake({ tx: tx as unknown as Db, companyId: input.companyId, issueId: input.issueId, + agentId: participant.agentId, reason, idempotencyKey: `native-status:${decisionRow.id}:execution-stage`, + payload: { mutation: "update", executionStage }, + contextSnapshot: { executionStage, executionStageSource: "issue.execution_stage" } }); + } if (updated.originKind === "routine_execution") { await syncRoutineRunStatusForIssue(tx as unknown as Db, input.issueId); }