import { createHash, timingSafeEqual } from "node:crypto"; import { authorityContext, unsafeOpaqueId } from "@awp/contracts"; import type { AgentRunId, AttemptId, AuditRecordId, ChangeSetId, ConnectionId, CorrelationId, CredentialReferenceId, EventId, MutationContext, OutboxMessageId, OperationId, PrincipalId, ProviderId, ProviderOperationContext, ReviewId, ReviewFindingId, VerificationEvidenceId, WorkspaceId, } from "@awp/contracts"; import type { AgentRun, Attempt, AttemptToolCall, ChangeSet, FactoryRun, Review, ReviewFinding, Task, VerificationEvidence, Workspace, } from "@awp/domain"; import { createAttemptSelectionProvenance, createCandidateManifest } from "@awp/domain"; import type { AgentProvider, ForgeProvider, RequiredChecksProvider, TrustedMerger, TrustedPublisher, WorkspaceProvider, } from "./ports/providers.js"; import { executeTrustedMerge } from "./trusted-merge.js"; import type { ApplicationTransaction, UnitOfWork } from "./ports/repositories.js"; import type { Clock, IdGenerator } from "./ports/runtime.js"; import { WorkflowStepTransactionRunner, runMarkedWorkflowTransaction, type DurableWorkflowStepRunner, } from "./durable-step.js"; export interface ExecutionSelection { readonly accountId: string; readonly model?: string; } export interface ExecutionDispatchRequest { readonly factoryRun: FactoryRun; readonly task: Task; readonly context: MutationContext; readonly selection?: ExecutionSelection; } export interface DurableWorkflowStepContext { readonly operationId: OperationId; readonly workflowSteps: DurableWorkflowStepRunner; readonly iteration?: number; } export interface ExecutionDispatcher { dispatch(request: ExecutionDispatchRequest): Promise; dispatchDurably?( request: ExecutionDispatchRequest, durable: DurableWorkflowStepContext, ): Promise; } export interface AutoMergeContinuation { schedule(changeSetId: ChangeSetId, context: MutationContext): Promise; } export interface AutoMergeReconciliationResult { readonly state: "waiting" | "merged" | "terminal"; readonly changeSetId: ChangeSetId; readonly reason?: string; } export interface AutoMergeReconciler { reconcileAutoMerge( changeSetId: ChangeSetId, context: MutationContext, ): Promise; reconcileAutoMergeDurably?( changeSetId: ChangeSetId, context: MutationContext, durable: DurableWorkflowStepContext, ): Promise; } export interface ExecutionVerificationEvidenceInput { readonly name: string; readonly state: "passed" | "failed" | "stale" | "missing"; readonly source: string; readonly observedAt: string; readonly required?: boolean; readonly details?: Readonly>; } export interface ExecutionCompletionInput { readonly attemptId: AttemptId; readonly token: string; readonly diff: string; readonly baseRevision: string; readonly candidateTreeDigest: string; readonly workspaceCheckpointDigest: string; readonly workspaceCheckpointedAt: string; readonly changedPaths: readonly string[]; readonly changes: readonly { readonly path: string; readonly kind: "add" | "modify" | "delete"; }[]; readonly toolCalls: readonly AttemptToolCall[]; readonly evidence: readonly ExecutionVerificationEvidenceInput[]; } export interface RecordVerificationEvidenceInput extends ExecutionVerificationEvidenceInput { readonly changeSetId: ChangeSetId; } export interface RecordReviewFindingInput { readonly changeSetId: ChangeSetId; readonly severity: "blocking" | "warning" | "recommendation" | "info"; readonly summary: string; readonly source?: string; } export interface ExecutionFailureInput { readonly attemptId: AttemptId; readonly token: string; readonly reason: string; readonly workspaceCheckpointDigest: string; readonly workspaceCheckpointedAt: string; } export interface ExecutionReviewCompletionInput { readonly attemptId: AttemptId; readonly reviewId: ReviewId; readonly token: string; readonly candidateDigest: string; readonly workspaceCheckpointDigest: string; readonly workspaceCheckpointedAt: string; readonly disposition: "approved" | "changes-requested" | "blocked"; readonly toolCalls: readonly AttemptToolCall[]; readonly findings: readonly { readonly severity: "blocking" | "warning" | "recommendation" | "info"; readonly summary: string; }[]; } export interface TrustedRepositoryBoundary { readonly forge: ForgeProvider; readonly publisher: TrustedPublisher; readonly merger: TrustedMerger; readonly repositoryKey: (repositoryUrl: string) => string; readonly connectionId: ConnectionId; readonly credentialReferenceId: CredentialReferenceId; } export interface RequiredChecksBoundary { readonly provider: RequiredChecksProvider; readonly connectionId: ConnectionId; readonly credentialReferenceId: CredentialReferenceId; } export class MergeRefusedError extends Error { constructor(message: string) { super(message); this.name = "MergeRefusedError"; } } const REPOSITORY_REQUIRED_CHECK_SOURCE_PREFIX = "repository-required-checks:"; function rejectReservedEvidenceSource(source: string): void { if (source.trim().startsWith(REPOSITORY_REQUIRED_CHECK_SOURCE_PREFIX)) { throw new Error( "Repository-required check evidence can only be recorded by the CI provider boundary", ); } } function verificationSatisfied( changeSet: ChangeSet, evidence: readonly VerificationEvidence[], findings: readonly ReviewFinding[], ): boolean { const latestByCheck = new Map(); for (const item of evidence) { if (item.changeSetId !== changeSet.id || item.candidateDigest !== changeSet.candidateDigest) continue; const key = `${item.source}\u0000${item.name}`; const previous = latestByCheck.get(key); if (!previous || Date.parse(item.observedAt) >= Date.parse(previous.observedAt)) { latestByCheck.set(key, item); } } const required = [...latestByCheck.values()].filter((item) => item.required); if (required.length === 0 || required.some((item) => item.state !== "passed")) return false; if (changeSet.repositoryKey && changeSet.publicationReference) { const policy = latestByCheck.get( `${REPOSITORY_REQUIRED_CHECK_SOURCE_PREFIX}${changeSet.repositoryKey}\u0000policy`, ); if (!policy || policy.state !== "passed" || !policy.required) return false; } return !findings.some( (finding) => finding.changeSetId === changeSet.id && finding.candidateDigest === changeSet.candidateDigest && finding.severity === "blocking" && !finding.resolved, ); } function validatedWorkspaceCheckpoint( digest: string, observedAt: string, ): { readonly digest: string; readonly observedAt: string; } { const normalizedDigest = digest.trim(); if (!/^[0-9a-f]{40,64}$/iu.test(normalizedDigest)) { throw new Error("Workspace checkpoint requires an immutable Git tree digest"); } if (!Number.isFinite(Date.parse(observedAt))) { throw new Error("Workspace checkpoint requires a valid observation timestamp"); } return { digest: normalizedDigest, observedAt: new Date(observedAt).toISOString() }; } function withWorkspaceCheckpoint( workspace: Workspace, digest: string, observedAt: string, collectedAt?: string, ): Workspace { if (workspace.cleanedAt) throw new Error("Cleaned Workspace cannot accept a new checkpoint"); if ( workspace.checkpointDigest !== undefined && workspace.checkpointDigest !== digest && workspace.checkpointCollectedAt !== undefined ) { throw new Error("Collected Workspace checkpoint identity is immutable"); } const nextCollectedAt = workspace.checkpointDigest === digest ? (workspace.checkpointCollectedAt ?? collectedAt) : collectedAt; const same = workspace.checkpointDigest === digest && workspace.checkpointedAt === observedAt && workspace.checkpointSource === "git-tree" && workspace.checkpointCollectedAt === nextCollectedAt; if (same) return workspace; const next = { ...workspace, checkpointDigest: digest, checkpointedAt: observedAt, checkpointSource: "git-tree", revision: workspace.revision + 1, }; if (nextCollectedAt !== undefined) return { ...next, checkpointCollectedAt: nextCollectedAt }; const withoutCollectedCheckpoint = { ...next }; delete withoutCollectedCheckpoint.checkpointCollectedAt; return withoutCollectedCheckpoint; } export function executionCallbackToken(secret: string, attemptId: string): string { if (!secret.trim()) throw new Error("Execution callback secret must not be empty"); if (!attemptId.trim()) throw new Error("Execution callback token requires Attempt identity"); return createHash("sha256").update(`${secret}:${attemptId}`).digest("hex"); } export interface WorkspaceDispatchConfiguration { readonly profileKey: string; readonly callbackBaseUrl: string; readonly callbackSecret: string; readonly agentProviderId: ProviderId; readonly accountId?: string; readonly model?: string; readonly connectionId: ConnectionId; readonly credentialReferenceId: CredentialReferenceId; readonly forceFirstAttemptFailure?: boolean; readonly agentProvider?: AgentProvider; readonly agentConnectionId?: ConnectionId; readonly agentCredentialReferenceId?: CredentialReferenceId; } export class WorkspaceExecutionDispatcher implements ExecutionDispatcher { constructor( private readonly uow: UnitOfWork, private readonly workspaceProvider: WorkspaceProvider, private readonly ids: IdGenerator, private readonly clock: Clock, private readonly configuration: WorkspaceDispatchConfiguration, private readonly trustedRepository?: TrustedRepositoryBoundary, private readonly requiredChecks?: RequiredChecksBoundary, private readonly autoMergeContinuation?: AutoMergeContinuation, private readonly nextTaskDispatcher?: ExecutionDispatcher, ) {} async dispatch(request: ExecutionDispatchRequest): Promise { return this.dispatchWithSteps(request); } async dispatchDurably( request: ExecutionDispatchRequest, durable: DurableWorkflowStepContext, ): Promise { return this.dispatchWithSteps(request, durable); } private async dispatchWithSteps( request: ExecutionDispatchRequest, durable?: DurableWorkflowStepContext, ): Promise { const selectedAccountId = request.selection?.accountId ?? this.configuration.accountId; if (!selectedAccountId) throw new Error("Execution requires an explicit provider account selection"); const selectedModel = request.selection?.model ?? this.configuration.model ?? "provider-default"; const effectiveContext = durable ? { ...request.context, operationId: durable.operationId, idempotencyKey: `dispatch:${request.factoryRun.id}:${request.task.id}`, } : request.context; const effectiveRequest: ExecutionDispatchRequest = { ...request, context: effectiveContext }; const stepTransactions = durable ? new WorkflowStepTransactionRunner(this.uow, this.clock) : undefined; const domainStep = async ( stepName: string, stepKey: string, work: (tx: ApplicationTransaction) => Promise, ): Promise => { if (!durable || !stepTransactions) return this.uow.transaction(work); return runMarkedWorkflowTransaction( durable.workflowSteps, stepTransactions, { operationId: durable.operationId, stepName, stepKey }, work, ); }; const externalStep = async ( stepName: string, stepKey: string, work: () => Promise, ): Promise => (durable ? durable.workflowSteps.run(stepName, stepKey, work) : work()); const execution = await domainStep( "awp-i1-task-dispatch-state-prepare", String(request.task.id), async (tx) => { const existingAgentRun = ( await tx.agentRuns.listByProject(request.factoryRun.projectId) ).find( (candidate) => candidate.factoryRunId === request.factoryRun.id && candidate.taskId === request.task.id && candidate.role === "coder", ); if (existingAgentRun) { const attempts = await tx.attempts.listByAgentRunIds([existingAgentRun.id]); const initialAttempt = attempts.find( (candidate) => candidate.selection.kind === "initial", ); if (!initialAttempt) { throw new Error("Existing AgentRun is missing its initial Attempt"); } const existingWorkspace = await tx.workspaces.getById(initialAttempt.workspaceId); if (!existingWorkspace) { throw new Error("Existing AgentRun is missing its Workspace"); } return { workspace: existingWorkspace, agentRun: existingAgentRun, attempt: initialAttempt, shouldProvision: initialAttempt.status === "created" && (existingAgentRun.status === "queued" || existingAgentRun.status === "waiting"), }; } const workspace: Workspace = { id: this.ids.next(), projectId: request.factoryRun.projectId, revision: 1, }; const agentRunId = this.ids.next(); const agentRun: AgentRun = { id: agentRunId, factoryRunId: request.factoryRun.id, taskId: request.task.id, agentPrincipalId: `principal:agent:${agentRunId}` as PrincipalId, role: "coder", status: "queued", revision: 1, }; const attempt: Attempt = { id: this.ids.next(), agentRunId: agentRun.id, workspaceId: workspace.id, status: "created", providerId: this.configuration.agentProviderId, accountId: selectedAccountId, model: selectedModel, selection: createAttemptSelectionProvenance({ kind: "initial", reason: "automatic first legal Task dispatch", }), revision: 1, }; await tx.workspaces.insert(workspace); await tx.agentRuns.insert(agentRun); await tx.attempts.insert(attempt); const currentFactoryRun = (await tx.factoryRuns.getById(request.factoryRun.id)) ?? request.factoryRun; await tx.factoryRuns.update({ ...currentFactoryRun, status: "starting", reason: "Provisioning K3s Workspace", revision: currentFactoryRun.revision + 1, }); await this.record( tx, effectiveRequest, "FactoryRunStarted", "FactoryRun", currentFactoryRun.id, currentFactoryRun.revision + 1, { taskId: request.task.id, agentRunId: agentRun.id, workspaceId: workspace.id }, ); await this.record(tx, effectiveRequest, "AgentRunQueued", "AgentRun", agentRun.id, 1, { taskId: request.task.id, workspaceId: workspace.id, attemptId: attempt.id, }); return { workspace, agentRun, attempt, shouldProvision: true }; }, ); if (!execution.shouldProvision) return; const { workspace, agentRun, attempt } = execution; try { const fixtureEnvironment = { AWP_AGENT_RUN_ID: agentRun.id, AWP_ATTEMPT_ID: attempt.id, AWP_TASK_ID: request.task.id, AWP_AGENT_ROLE: "coder", AWP_BASE_IDENTITY: request.factoryRun.projectId, AWP_SELECTION_KIND: "initial", AWP_FORCE_FIRST_ATTEMPT_FAILURE: this.configuration.forceFirstAttemptFailure ? "1" : "0", AWP_CALLBACK_URL: `${this.configuration.callbackBaseUrl}/internal/execution/complete`, AWP_FAILURE_URL: `${this.configuration.callbackBaseUrl}/internal/execution/fail`, AWP_CALLBACK_TOKEN: this.completionToken(attempt.id), }; const workspaceResult = await externalStep( "awp-i1-task-dispatch-workspace-provision", String(workspace.id), () => this.workspaceProvider.create( { operationId: effectiveContext.operationId, idempotencyKey: `workspace:${String(workspace.id)}:provision`, correlationId: effectiveContext.correlationId, authority: effectiveContext.authority, connectionId: this.configuration.connectionId, credentialReferenceId: this.configuration.credentialReferenceId, }, workspace.id, this.configuration.profileKey, agentRun.id, { environment: this.configuration.agentProvider === undefined ? fixtureEnvironment : { AWP_AGENT_RUN_ID: agentRun.id, AWP_ATTEMPT_ID: attempt.id, AWP_TASK_ID: request.task.id, AWP_AGENT_ROLE: "coder", }, }, ), ); const agentResult = await externalStep( "awp-i1-task-dispatch-agent-start", String(attempt.id), () => this.startAgentAttempt( effectiveContext, attempt, agentRun.id, request.task.id, workspace.id, "initial", ), ); await domainStep("awp-i1-task-dispatch-state-started", String(attempt.id), async (tx) => { const currentFactoryRun = await tx.factoryRuns.getById(request.factoryRun.id); const currentAgentRun = await tx.agentRuns.getById(agentRun.id); const currentAttempt = await tx.attempts.getById(attempt.id); if (!currentFactoryRun || !currentAgentRun || !currentAttempt) { throw new Error("Execution state disappeared during Workspace provisioning"); } if (currentAgentRun.status === "active" && currentAttempt.status === "running") { return { state: "already-running" as const }; } await tx.factoryRuns.update({ ...currentFactoryRun, status: "running", reason: "K3s Workspace accepted", revision: currentFactoryRun.revision + 1, }); await tx.agentRuns.update({ ...currentAgentRun, status: "active", revision: currentAgentRun.revision + 1, }); await tx.attempts.update({ ...currentAttempt, status: "running", revision: currentAttempt.revision + 1, }); await this.record( tx, effectiveRequest, "AgentRunStarted", "AgentRun", agentRun.id, currentAgentRun.revision + 1, { taskId: request.task.id, workspaceId: workspace.id, attemptId: attempt.id, workspaceProviderState: workspaceResult.value.state, workspaceProviderReferences: workspaceResult.references, ...(agentResult === undefined ? {} : { agentProviderState: agentResult.value.state, agentProviderReferences: agentResult.references, }), }, ); return { state: "running" as const }; }); } catch (error) { const reason = error instanceof Error ? error.message : "Workspace provisioning failed"; await domainStep("awp-i1-task-dispatch-state-failed", String(attempt.id), async (tx) => { const currentFactoryRun = await tx.factoryRuns.getById(request.factoryRun.id); const currentAgentRun = await tx.agentRuns.getById(agentRun.id); const currentAttempt = await tx.attempts.getById(attempt.id); if (currentFactoryRun) { await tx.factoryRuns.update({ ...currentFactoryRun, status: "retryable", reason, revision: currentFactoryRun.revision + 1, }); } if (currentAgentRun) { await tx.agentRuns.update({ ...currentAgentRun, status: "waiting", reason, revision: currentAgentRun.revision + 1, }); } if (currentAttempt && currentAttempt.status !== "terminal") { await tx.attempts.update({ ...currentAttempt, status: "created", reason, revision: currentAttempt.revision + 1, }); } await this.record( tx, effectiveRequest, "AgentRunProvisioningFailed", "AgentRun", agentRun.id, (currentAgentRun?.revision ?? 1) + 1, { taskId: request.task.id, workspaceId: workspace.id, attemptId: attempt.id, reason, }, ); return { state: "retryable" as const, reason }; }); throw error; } } async reconcilePendingReviews( projectId: FactoryRun["projectId"], context: MutationContext, ): Promise { const changeSetIds = await this.uow.transaction(async (tx) => { const changeSets = await tx.changeSets.listByProject(projectId); return changeSets .filter( (changeSet) => changeSet.publicationReference !== undefined && !["merged", "superseded", "cancelled"].includes(changeSet.status), ) .map((changeSet) => changeSet.id); }); let reconciled = 0; for (const changeSetId of changeSetIds) { await this.ensureIndependentReview(changeSetId, context); reconciled += 1; } return reconciled; } async failAttempt(input: ExecutionFailureInput, context: MutationContext): Promise { const expected = this.completionToken(input.attemptId); const suppliedBuffer = Buffer.from(input.token); const expectedBuffer = Buffer.from(expected); if ( suppliedBuffer.length !== expectedBuffer.length || !timingSafeEqual(suppliedBuffer, expectedBuffer) ) { throw new Error("Invalid Attempt failure token"); } const reason = input.reason.trim(); if (!reason) throw new Error("Attempt failure reason must not be empty"); const checkpoint = validatedWorkspaceCheckpoint( input.workspaceCheckpointDigest, input.workspaceCheckpointedAt, ); if (!this.workspaceProvider.replaceCompute || !this.workspaceProvider.attestCheckpoint) { throw new Error("Workspace provider does not support content-attested preserved-WIP retry"); } const retryDecision = await this.uow.transaction(async (tx) => { const failed = await tx.attempts.getById(input.attemptId); if (!failed) throw new Error("Attempt does not exist"); const workspace = await tx.workspaces.getById(failed.workspaceId); if (!workspace) throw new Error("Attempt Workspace does not exist"); const checkpointedWorkspace = withWorkspaceCheckpoint( workspace, checkpoint.digest, checkpoint.observedAt, ); if (checkpointedWorkspace !== workspace) await tx.workspaces.update(checkpointedWorkspace); const history = await tx.attempts.listByAgentRunIds([failed.agentRunId]); const existing = history.find( (candidate) => candidate.selection.previousAttemptId === failed.id, ); if (existing) { return { retry: existing, shouldLaunch: existing.status === "created" }; } const agentRun = await tx.agentRuns.getById(failed.agentRunId); if (!agentRun) throw new Error("AgentRun does not exist"); const factoryRun = await tx.factoryRuns.getById(agentRun.factoryRunId); const task = await tx.tasks.getById(agentRun.taskId); if (!factoryRun || !task) throw new Error("Execution hierarchy does not exist"); const next: Attempt = { id: this.ids.next(), agentRunId: failed.agentRunId, workspaceId: failed.workspaceId, status: "created", ...(failed.providerId === undefined ? {} : { providerId: failed.providerId }), ...(failed.accountId === undefined ? {} : { accountId: failed.accountId }), ...(failed.model === undefined ? {} : { model: failed.model }), selection: createAttemptSelectionProvenance({ kind: "retry", reason: "Automatic retry with preserved Workspace WIP", previousAttemptId: failed.id, }), revision: 1, }; await tx.attempts.update({ ...failed, status: "terminal", reason, revision: failed.revision + 1, }); await tx.attempts.insert(next); await tx.agentRuns.update({ ...agentRun, status: "waiting", reason: `Retrying after Attempt failure: ${reason}`, revision: agentRun.revision + 1, }); await tx.factoryRuns.update({ ...factoryRun, status: "retryable", reason, revision: factoryRun.revision + 1, }); await this.record( tx, { factoryRun, task, context }, "AttemptFailed", "Attempt", failed.id, failed.revision + 1, { reason, retryAttemptId: next.id, workspaceId: next.workspaceId }, ); return { retry: next, shouldLaunch: true }; }); const retry = retryDecision.retry; await this.attestWorkspaceCheckpoint( retry.workspaceId, checkpoint.digest, checkpoint.observedAt, context, `retry-checkpoint:${String(retry.id)}`, ); if (!retryDecision.shouldLaunch) return retry; const agentRun = await this.uow.transaction((tx) => tx.agentRuns.getById(retry.agentRunId)); if (!agentRun) throw new Error("AgentRun disappeared before retry"); const operationContext = { operationId: context.operationId, idempotencyKey: `workspace:${String(retry.workspaceId)}:retry:${String(retry.id)}:replace-compute`, correlationId: context.correlationId, authority: context.authority, connectionId: this.configuration.connectionId, credentialReferenceId: this.configuration.credentialReferenceId, }; await this.uow.transaction(async (tx) => { const current = await tx.attempts.getById(retry.id); if (!current) throw new Error("Retry Attempt disappeared before compute replacement"); if (current.status !== "created") return; await tx.attempts.update({ ...current, status: "running", revision: current.revision + 1, }); }); try { const replacementResult = await this.workspaceProvider.replaceCompute( operationContext, retry.workspaceId, this.configuration.profileKey, retry.agentRunId, { environment: this.configuration.agentProvider === undefined ? { AWP_AGENT_RUN_ID: retry.agentRunId, AWP_ATTEMPT_ID: retry.id, AWP_TASK_ID: agentRun.taskId, AWP_SELECTION_KIND: "retry", AWP_CALLBACK_URL: `${this.configuration.callbackBaseUrl}/internal/execution/complete`, AWP_FAILURE_URL: `${this.configuration.callbackBaseUrl}/internal/execution/fail`, AWP_CALLBACK_TOKEN: this.completionToken(retry.id), } : { AWP_AGENT_RUN_ID: retry.agentRunId, AWP_ATTEMPT_ID: retry.id, AWP_TASK_ID: agentRun.taskId, }, }, ); await this.uow.transaction(async (tx) => { const currentAgentRun = await tx.agentRuns.getById(retry.agentRunId); const factoryRun = currentAgentRun ? await tx.factoryRuns.getById(currentAgentRun.factoryRunId) : undefined; const task = await tx.tasks.getById(agentRun.taskId); if (!currentAgentRun || !factoryRun || !task) { throw new Error("Retry execution hierarchy disappeared after compute replacement"); } await this.record( tx, { factoryRun, task, context }, "RetryComputeReplaced", "Attempt", retry.id, retry.revision + 1, { previousAttemptId: retry.selection.previousAttemptId, agentRunId: retry.agentRunId, taskId: agentRun.taskId, workspaceId: retry.workspaceId, workspaceProviderState: replacementResult.value.state, workspaceProviderDetails: replacementResult.value.details, workspaceProviderReferences: replacementResult.references, workspaceProviderObservedAt: replacementResult.observedAt, ...(replacementResult.reconciliationToken === undefined ? {} : { replacementReconciliationToken: replacementResult.reconciliationToken }), }, ); }); await this.startAgentAttempt( context, retry, retry.agentRunId, agentRun.taskId, retry.workspaceId, "retry", ); } catch (error) { await this.uow.transaction(async (tx) => { const current = await tx.attempts.getById(retry.id); if (!current || current.status !== "running") return; await tx.attempts.update({ ...current, status: "created", revision: current.revision + 1, }); }); throw error; } return this.uow.transaction(async (tx) => { const currentAttempt = await tx.attempts.getById(retry.id); const currentAgentRun = await tx.agentRuns.getById(retry.agentRunId); const factoryRun = currentAgentRun ? await tx.factoryRuns.getById(currentAgentRun.factoryRunId) : undefined; if (!currentAttempt || !currentAgentRun || !factoryRun) { throw new Error("Retry state disappeared"); } await tx.agentRuns.update({ ...currentAgentRun, status: "active", reason: "Retry Attempt running with preserved Workspace WIP", revision: currentAgentRun.revision + 1, }); await tx.factoryRuns.update({ ...factoryRun, status: "running", reason: "Retry Attempt running", revision: factoryRun.revision + 1, }); return currentAttempt; }); } async complete(input: ExecutionCompletionInput, context: MutationContext): Promise { const expected = this.completionToken(input.attemptId); const suppliedBuffer = Buffer.from(input.token); const expectedBuffer = Buffer.from(expected); if ( suppliedBuffer.length !== expectedBuffer.length || !timingSafeEqual(suppliedBuffer, expectedBuffer) ) { throw new Error("Invalid Attempt completion token"); } if (!input.diff.trim()) throw new Error("Attempt completion diff must not be empty"); if ( input.toolCalls.length === 0 || input.toolCalls.some((call) => !call.name.trim() || !call.summary.trim()) ) { throw new Error("Attempt completion requires at least one named tool call"); } if ( input.evidence.length === 0 || !input.evidence.some((item) => item.required !== false) || input.evidence.some( (item) => !item.name.trim() || !item.source.trim() || !Number.isFinite(Date.parse(item.observedAt)), ) ) { throw new Error("Attempt completion requires timestamped required VerificationEvidence"); } for (const item of input.evidence) rejectReservedEvidenceSource(item.source); const baseRevision = input.baseRevision.trim(); const candidateTreeDigest = input.candidateTreeDigest.trim(); if (!/^[0-9a-f]{40,64}$/i.test(baseRevision)) { throw new Error("Attempt completion requires a valid Git base revision"); } if (!/^[0-9a-f]{40,64}$/i.test(candidateTreeDigest)) { throw new Error("Attempt completion requires a valid candidate tree digest"); } const checkpoint = validatedWorkspaceCheckpoint( input.workspaceCheckpointDigest, input.workspaceCheckpointedAt, ); if (checkpoint.digest !== candidateTreeDigest) { throw new Error( "Attempt completion Workspace checkpoint must equal the collected candidate tree", ); } if (!this.workspaceProvider.attestCheckpoint) { throw new Error("Workspace provider does not support content checkpoint attestation"); } const patchDigest = createHash("sha256").update(input.diff).digest("hex"); const manifest = createCandidateManifest({ treeDigest: candidateTreeDigest, patchDigest, changedPaths: input.changedPaths.map((path) => path.trim()), changes: input.changes.map((change) => ({ path: change.path.trim(), kind: change.kind, })), }); if (manifest.changedPaths.length === 0) { throw new Error("Attempt completion must identify at least one changed path"); } const completed = await this.uow.transaction(async (tx) => { const attempt = await tx.attempts.getById(input.attemptId); if (!attempt) throw new Error("Attempt does not exist"); const agentRun = await tx.agentRuns.getById(attempt.agentRunId); if (!agentRun) throw new Error("AgentRun does not exist"); const task = await tx.tasks.getById(agentRun.taskId); if (!task) throw new Error("Task does not exist"); const factoryRun = await tx.factoryRuns.getById(agentRun.factoryRunId); if (!factoryRun) throw new Error("FactoryRun does not exist"); const workspace = await tx.workspaces.getById(attempt.workspaceId); if (!workspace) throw new Error("Attempt Workspace does not exist"); const checkpointedWorkspace = withWorkspaceCheckpoint( workspace, checkpoint.digest, checkpoint.observedAt, this.clock.now().toISOString(), ); if (checkpointedWorkspace !== workspace) await tx.workspaces.update(checkpointedWorkspace); const existing = (await tx.changeSets.listByProject(factoryRun.projectId)).find( (candidate) => candidate.producerAttemptId === attempt.id, ); if (existing) { if ( existing.baseIdentity !== baseRevision || existing.candidateDigest !== candidateTreeDigest || existing.candidateManifest.patchDigest !== manifest.patchDigest || existing.diff !== input.diff ) { throw new Error( "Attempt completion replay does not match the durable ChangeSet identity", ); } return { changeSet: existing, workspaceId: attempt.workspaceId }; } const changeSet: ChangeSet = { id: this.ids.next(), projectId: factoryRun.projectId, taskId: task.id, producerAttemptId: attempt.id, baseIdentity: baseRevision, candidateDigest: candidateTreeDigest, candidateManifest: manifest, diff: input.diff, revision: 1, status: "publishing", }; await tx.changeSets.insert(changeSet); for (const item of input.evidence) { const evidence: VerificationEvidence = { id: this.ids.next(), changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, name: item.name.trim(), state: item.state, source: item.source.trim(), observedAt: new Date(item.observedAt).toISOString(), required: item.required !== false, ...(item.details === undefined ? {} : { details: item.details }), }; await tx.verificationEvidence.insert(evidence); } await tx.attempts.update({ ...attempt, status: "terminal", toolCalls: input.toolCalls.map((call) => ({ name: call.name.trim(), summary: call.summary.trim(), })), revision: attempt.revision + 1, }); await tx.agentRuns.update({ ...agentRun, status: "completed", reason: "ChangeSet durably collected; trusted publication pending", revision: agentRun.revision + 1, }); await tx.factoryRuns.update({ ...factoryRun, status: "running", reason: "ChangeSet collected; repository-scoped trusted publication pending", revision: factoryRun.revision + 1, }); await tx.tasks.update({ ...task, status: "review", revision: task.revision + 1 }); await this.record( tx, { factoryRun, task, context }, "ChangeSetCreated", "ChangeSet", changeSet.id, 1, { attemptId: attempt.id, agentRunId: agentRun.id, candidateDigest: changeSet.candidateDigest, verificationEvidenceCount: input.evidence.length, publicationState: "publishing", }, ); return { changeSet, workspaceId: attempt.workspaceId }; }); await this.attestWorkspaceCheckpoint( completed.workspaceId, checkpoint.digest, checkpoint.observedAt, context, `collected-checkpoint:${String(completed.changeSet.id)}`, ); let published: ChangeSet; try { published = await this.publishForReview(completed.changeSet.id, context); } catch (error) { await this.uow.transaction(async (tx) => { const current = await tx.changeSets.getById(completed.changeSet.id); if (!current || current.publicationReference || current.status === "merged") return; if (current.status !== "publication-failed") { const failed: ChangeSet = { ...current, status: "publication-failed", revision: current.revision + 1, }; await tx.changeSets.update(failed); await this.recordChangeSetMutation( tx, current, context, "ChangeSetPublicationFailed", "ChangeSet", current.id, { candidateDigest: current.candidateDigest, reason: error instanceof Error ? error.message : "trusted publication failed", }, ); } }); throw error; } await this.workspaceProvider.destroy( { operationId: context.operationId, idempotencyKey: `workspace:${String(completed.workspaceId)}:producer-destroy`, correlationId: context.correlationId, authority: context.authority, connectionId: this.configuration.connectionId, credentialReferenceId: this.configuration.credentialReferenceId, }, completed.workspaceId, ); return published; } async refreshRequiredChecks( changeSetId: ChangeSetId, context: MutationContext, durable?: DurableWorkflowStepContext, ): Promise { const boundary = this.requiredChecks; if (!boundary) return []; const basis = await this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); if ( !changeSet.repositoryKey || !changeSet.publicationReference?.nativeRevision || !changeSet.targetReference ) { throw new Error("Repository-required checks require a trusted published ChangeSet"); } if (!changeSet.targetReference.startsWith("refs/heads/")) { throw new Error("Repository-required checks require a canonical target branch reference"); } const project = await tx.projects.getById(changeSet.projectId); if (!project) throw new Error("ChangeSet Project does not exist"); return { changeSet, repositoryKey: changeSet.repositoryKey, branch: changeSet.targetReference.slice("refs/heads/".length), publicationRevision: changeSet.publicationReference.nativeRevision, projectRequiredChecks: project.requiredChecks, }; }); const providerContext: ProviderOperationContext = { operationId: context.operationId, correlationId: context.correlationId, idempotencyKey: `changeset:${String(changeSetId)}:required-checks`, authority: context.authority, connectionId: boundary.connectionId, credentialReferenceId: boundary.credentialReferenceId, }; const pollKey = `${String(changeSetId)}:poll:${durable?.iteration ?? 0}:${basis.publicationRevision}`; const snapshot: | { readonly ok: true; readonly observedAt: string; readonly requiredChecks: readonly string[]; readonly checks: readonly { readonly name: string; readonly state: "passed" | "failed" | "missing"; readonly observedAt: string; readonly details?: Readonly>; }[]; readonly providerReference?: unknown; } | { readonly ok: false; readonly observedAt: string } = await this.workflowExternalStep( durable, "awp-i1-automerge-required-checks-observe", pollKey, async () => { try { const result = await boundary.provider.observeRequiredChecks( providerContext, basis.repositoryKey, basis.branch, basis.publicationRevision, ); if ( result.value.repositoryKey !== basis.repositoryKey || result.value.branch !== basis.branch || result.value.revision !== basis.publicationRevision ) { throw new Error("Repository-required check observation identity mismatch"); } return { ok: true as const, observedAt: result.observedAt, requiredChecks: result.value.requiredChecks, checks: result.value.checks, ...(result.references[0] === undefined ? {} : { providerReference: result.references[0] }), }; } catch { return { ok: false as const, observedAt: this.clock.now().toISOString() }; } }, ); return this.workflowDomainStep( durable, "awp-i1-automerge-required-checks-record", pollKey, async (tx) => { const current = await tx.changeSets.getById(changeSetId); if (!current) throw new Error("ChangeSet disappeared while refreshing required checks"); if ( current.candidateDigest !== basis.changeSet.candidateDigest || current.repositoryKey !== basis.repositoryKey || current.publicationReference?.nativeRevision !== basis.publicationRevision || current.targetReference !== basis.changeSet.targetReference ) { throw new Error("ChangeSet identity changed while refreshing repository-required checks"); } const source = `${REPOSITORY_REQUIRED_CHECK_SOURCE_PREFIX}${current.repositoryKey}`; const existing = await tx.verificationEvidence.listByChangeSetIds([current.id]); const inserted: VerificationEvidence[] = []; const add = async (input: { name: string; state: VerificationEvidence["state"]; observedAt: string; required: boolean; details: Readonly>; }) => { const evidence: VerificationEvidence = { id: this.ids.next(), changeSetId: current.id, candidateDigest: current.candidateDigest, name: input.name, state: input.state, source, observedAt: new Date(input.observedAt).toISOString(), required: input.required, details: input.details, }; await tx.verificationEvidence.insert(evidence); inserted.push(evidence); }; const common = { repositoryKey: current.repositoryKey, branch: basis.branch, candidateDigest: current.candidateDigest, publicationRevision: basis.publicationRevision, }; if (!snapshot.ok) { await add({ name: "policy", state: "missing", observedAt: snapshot.observedAt, required: true, details: { ...common, providerState: "unavailable", projectRequiredChecks: [...basis.projectRequiredChecks], }, }); for (const name of basis.projectRequiredChecks) { await add({ name: `required:${name}`, state: "missing", observedAt: snapshot.observedAt, required: true, details: { ...common, checkName: name, projectRequired: true, providerRequired: false, providerState: "unavailable", }, }); } } else { const providerRequiredChecks = new Set(snapshot.requiredChecks); const projectRequiredChecks = new Set(basis.projectRequiredChecks); const requiredNames = new Set([...providerRequiredChecks, ...projectRequiredChecks]); await add({ name: "policy", state: "passed", observedAt: snapshot.observedAt, required: true, details: { ...common, providerState: "observed", providerRequiredChecks: [...providerRequiredChecks].sort(), projectRequiredChecks: [...projectRequiredChecks].sort(), requiredChecks: [...requiredNames].sort(), ...(snapshot.providerReference === undefined ? {} : { providerReference: snapshot.providerReference }), }, }); const observedByName = new Map( snapshot.checks.map((check) => [check.name, check] as const), ); for (const name of [...requiredNames].sort()) { const observation = observedByName.get(name); await add({ name: `required:${name}`, state: observation?.state ?? "missing", observedAt: snapshot.observedAt, required: true, details: { ...common, checkName: name, projectRequired: projectRequiredChecks.has(name), providerRequired: providerRequiredChecks.has(name), ...(observation === undefined ? {} : { providerObservedAt: observation.observedAt }), ...(observation?.details === undefined ? {} : { provider: observation.details }), }, }); } const previousNames = new Set( existing .filter( (item) => item.candidateDigest === current.candidateDigest && item.source === source && item.name.startsWith("required:"), ) .map((item) => item.name.slice("required:".length)), ); for (const previousName of previousNames) { if (requiredNames.has(previousName)) continue; await add({ name: `required:${previousName}`, state: "stale", observedAt: snapshot.observedAt, required: false, details: { ...common, checkName: previousName, staleReason: "repository-policy-no-longer-requires-check", }, }); } } const allEvidence = await tx.verificationEvidence.listByChangeSetIds([current.id]); const findings = await tx.reviewFindings.listByChangeSetIds([current.id]); const approved = await this.hasApprovedIndependentReview(tx, current); const ready = approved && verificationSatisfied(current, allEvidence, findings); if (current.status !== "merged") { const nextStatus: ChangeSet["status"] = ready ? "ready-to-merge" : current.status === "reviewing" || current.status === "changes-requested" ? current.status : "verifying"; if (nextStatus !== current.status) { await tx.changeSets.update({ ...current, status: nextStatus, revision: current.revision + 1, }); } } const attempt = await tx.attempts.getById(current.producerAttemptId); const agentRun = attempt ? await tx.agentRuns.getById(attempt.agentRunId) : undefined; const factoryRun = agentRun ? await tx.factoryRuns.getById(agentRun.factoryRunId) : undefined; const task = await tx.tasks.getById(current.taskId); if (factoryRun && task) { await this.record( tx, { factoryRun, task, context }, "RepositoryRequiredChecksRefreshed", "ChangeSet", current.id, current.revision, { candidateDigest: current.candidateDigest, repositoryKey: current.repositoryKey, branch: basis.branch, publicationRevision: basis.publicationRevision, policyObserved: snapshot.ok, evidenceCount: inserted.length, mergeGateSatisfied: ready, }, ); } return inserted; }, ); } async recordVerificationEvidence( input: RecordVerificationEvidenceInput, context: MutationContext, ): Promise { if ( !input.name.trim() || !input.source.trim() || !Number.isFinite(Date.parse(input.observedAt)) ) { throw new Error("VerificationEvidence requires name, source, and observedAt"); } rejectReservedEvidenceSource(input.source); return this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(input.changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); const evidence: VerificationEvidence = { id: this.ids.next(), changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, name: input.name.trim(), state: input.state, source: input.source.trim(), observedAt: new Date(input.observedAt).toISOString(), required: input.required !== false, ...(input.details === undefined ? {} : { details: input.details }), }; await tx.verificationEvidence.insert(evidence); const allEvidence = await tx.verificationEvidence.listByChangeSetIds([changeSet.id]); const findings = await tx.reviewFindings.listByChangeSetIds([changeSet.id]); const approved = await this.hasApprovedIndependentReview(tx, changeSet); const ready = approved && verificationSatisfied(changeSet, allEvidence, findings); if (changeSet.status !== "merged") { const nextStatus: ChangeSet["status"] = ready ? "ready-to-merge" : "verifying"; if (changeSet.status !== nextStatus) { await tx.changeSets.update({ ...changeSet, status: nextStatus, revision: changeSet.revision + 1, }); } } await this.recordChangeSetMutation( tx, changeSet, context, "VerificationEvidenceRecorded", "VerificationEvidence", evidence.id, { name: evidence.name, state: evidence.state, source: evidence.source, required: evidence.required, candidateDigest: evidence.candidateDigest, mergeGateSatisfied: ready, }, ); return evidence; }); } async recordReviewFinding( input: RecordReviewFindingInput, context: MutationContext, ): Promise { const summary = input.summary.trim(); if (!summary) throw new Error("Review finding summary must not be empty"); return this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(input.changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); const finding: ReviewFinding = { id: this.ids.next(), changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, severity: input.severity, summary, ...(input.source?.trim() ? { source: input.source.trim() } : {}), resolved: false, }; await tx.reviewFindings.insert(finding); if (input.severity === "blocking" && changeSet.status === "ready-to-merge") { await tx.changeSets.update({ ...changeSet, status: "verifying", revision: changeSet.revision + 1, }); } await this.recordChangeSetMutation( tx, changeSet, context, "ReviewFindingRecorded", "ReviewFinding", finding.id, { severity: finding.severity, summary: finding.summary, candidateDigest: finding.candidateDigest, }, ); return finding; }); } async resolveReviewFinding( findingId: ReviewFindingId, context: MutationContext, ): Promise { return this.uow.transaction(async (tx) => { const finding = await tx.reviewFindings.getById(findingId); if (!finding) throw new Error("ReviewFinding does not exist"); if (finding.resolved) return finding; const changeSet = await tx.changeSets.getById(finding.changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); if (finding.candidateDigest !== changeSet.candidateDigest) { throw new Error("ReviewFinding candidate identity is stale"); } const resolved: ReviewFinding = { ...finding, resolved: true }; await tx.reviewFindings.update(resolved); const evidence = await tx.verificationEvidence.listByChangeSetIds([changeSet.id]); const findings = (await tx.reviewFindings.listByChangeSetIds([changeSet.id])).map((item) => item.id === resolved.id ? resolved : item, ); const approved = await this.hasApprovedIndependentReview(tx, changeSet); const ready = approved && verificationSatisfied(changeSet, evidence, findings); if (changeSet.status !== "merged" && ready && changeSet.status !== "ready-to-merge") { await tx.changeSets.update({ ...changeSet, status: "ready-to-merge", revision: changeSet.revision + 1, }); } await this.recordChangeSetMutation( tx, changeSet, context, "ReviewFindingResolved", "ReviewFinding", finding.id, { candidateDigest: finding.candidateDigest, mergeGateSatisfied: ready }, ); return resolved; }); } async completeReview( input: ExecutionReviewCompletionInput, context: MutationContext, ): Promise { const expected = this.completionToken(input.attemptId); const suppliedBuffer = Buffer.from(input.token); const expectedBuffer = Buffer.from(expected); if ( suppliedBuffer.length !== expectedBuffer.length || !timingSafeEqual(suppliedBuffer, expectedBuffer) ) { throw new Error("Invalid reviewer Attempt completion token"); } if ( input.toolCalls.length === 0 || input.toolCalls.some((call) => !call.name.trim() || !call.summary.trim()) ) { throw new Error("Reviewer completion requires non-empty tool activity provenance"); } if ( input.disposition !== "approved" && input.disposition !== "changes-requested" && input.disposition !== "blocked" ) { throw new Error("Reviewer disposition is invalid"); } const checkpoint = validatedWorkspaceCheckpoint( input.workspaceCheckpointDigest, input.workspaceCheckpointedAt, ); if (checkpoint.digest !== input.candidateDigest) { throw new Error("Reviewer Workspace checkpoint must equal the immutable ChangeSet candidate"); } if (!this.workspaceProvider.attestCheckpoint) { throw new Error("Workspace provider does not support content checkpoint attestation"); } if ( input.findings.some( (finding) => !finding.summary.trim() || !["blocking", "warning", "recommendation", "info"].includes(finding.severity), ) ) { throw new Error("Reviewer findings are invalid"); } if (input.disposition !== "approved" && input.findings.length === 0) { throw new Error("Non-approved reviewer disposition requires at least one finding"); } const completed = await this.uow.transaction(async (tx) => { const review = await tx.reviews.getById(input.reviewId); if (!review) throw new Error("Review does not exist"); const changeSet = await tx.changeSets.getById(review.changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); if ( review.candidateDigest !== changeSet.candidateDigest || input.candidateDigest !== changeSet.candidateDigest ) { throw new Error("Reviewer completion candidate identity is stale"); } const reviewerAttempt = await tx.attempts.getById(input.attemptId); if (!reviewerAttempt) throw new Error("Reviewer Attempt does not exist"); const reviewerWorkspace = await tx.workspaces.getById(reviewerAttempt.workspaceId); if (!reviewerWorkspace) throw new Error("Reviewer Workspace does not exist"); const checkpointedWorkspace = withWorkspaceCheckpoint( reviewerWorkspace, checkpoint.digest, checkpoint.observedAt, this.clock.now().toISOString(), ); if (checkpointedWorkspace !== reviewerWorkspace) await tx.workspaces.update(checkpointedWorkspace); const reviewerAgentRun = await tx.agentRuns.getById(reviewerAttempt.agentRunId); if ( !reviewerAgentRun || reviewerAgentRun.role !== "reviewer" || review.reviewerAgentRunId !== reviewerAgentRun.id || review.reviewerPrincipalId !== reviewerAgentRun.agentPrincipalId ) { throw new Error("Reviewer Attempt is not the assigned independent reviewer"); } const producerAttempt = await tx.attempts.getById(changeSet.producerAttemptId); const producerAgentRun = producerAttempt ? await tx.agentRuns.getById(producerAttempt.agentRunId) : undefined; if ( !producerAttempt || !producerAgentRun || producerAgentRun.role !== "coder" || producerAgentRun.agentPrincipalId === reviewerAgentRun.agentPrincipalId || producerAgentRun.id === reviewerAgentRun.id ) { throw new Error("Review does not satisfy coder/reviewer independence"); } const factoryRun = await tx.factoryRuns.getById(reviewerAgentRun.factoryRunId); const task = await tx.tasks.getById(changeSet.taskId); if (!factoryRun || !task) throw new Error("Review execution hierarchy does not exist"); if (review.status === "submitted") { if (review.disposition !== input.disposition) { throw new Error("Reviewer completion replay attempted to change disposition"); } return { review, workspaceId: reviewerAttempt.workspaceId, reviewerContext: context }; } if (review.status !== "requested" && review.status !== "reviewing") { throw new Error("Review is not pending"); } const reviewerContext: MutationContext = { operationId: context.operationId, correlationId: context.correlationId, idempotencyKey: context.idempotencyKey, authority: authorityContext( { id: reviewerAgentRun.agentPrincipalId, kind: "agent", capabilities: [], }, [], changeSet.projectId, ), }; const submitted: Review = { ...review, status: "submitted", disposition: input.disposition, }; await tx.reviews.update(submitted); await tx.attempts.update({ ...reviewerAttempt, status: "terminal", reason: `Independent review submitted: ${input.disposition}`, toolCalls: input.toolCalls.map((call) => ({ name: call.name.trim(), summary: call.summary.trim(), })), revision: reviewerAttempt.revision + 1, }); await tx.agentRuns.update({ ...reviewerAgentRun, status: "completed", reason: `Independent review submitted: ${input.disposition}`, revision: reviewerAgentRun.revision + 1, }); for (const findingInput of input.findings) { const finding: ReviewFinding = { id: this.ids.next(), changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, severity: findingInput.severity, summary: findingInput.summary.trim(), source: `reviewer-agent:${String(reviewerAgentRun.id)}`, resolved: false, }; await tx.reviewFindings.insert(finding); } const evidence = await tx.verificationEvidence.listByChangeSetIds([changeSet.id]); const findings = await tx.reviewFindings.listByChangeSetIds([changeSet.id]); const verified = verificationSatisfied(changeSet, evidence, findings); const nextChangeSetStatus: ChangeSet["status"] = input.disposition === "approved" && verified ? "ready-to-merge" : input.disposition === "approved" ? "verifying" : "changes-requested"; await tx.changeSets.update({ ...changeSet, status: nextChangeSetStatus, revision: changeSet.revision + 1, }); if (input.disposition !== "approved" && task.status !== "correction") { await tx.tasks.update({ ...task, status: "correction", revision: task.revision + 1 }); } await tx.factoryRuns.update({ ...factoryRun, status: "running", reason: input.disposition === "approved" ? "Independent review submitted; MergeGate evaluation active" : "Independent review requested correction", revision: factoryRun.revision + 1, }); await this.record( tx, { factoryRun, task, context: reviewerContext }, "ReviewSubmitted", "Review", submitted.id, 1, { changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, reviewerAgentRunId: reviewerAgentRun.id, reviewerPrincipalId: reviewerAgentRun.agentPrincipalId, disposition: input.disposition, findingCount: input.findings.length, verificationSatisfied: verified, }, ); return { review: submitted, workspaceId: reviewerAttempt.workspaceId, reviewerContext }; }); await this.attestWorkspaceCheckpoint( completed.workspaceId, checkpoint.digest, checkpoint.observedAt, completed.reviewerContext, `review-checkpoint:${String(completed.review.id)}`, ); await this.refreshRequiredChecks(completed.review.changeSetId, completed.reviewerContext); if (input.disposition === "approved" && this.autoMergeContinuation) { await this.autoMergeContinuation.schedule(completed.review.changeSetId, context); } // Keep the reviewer Workspace until ChangeSet terminal cleanup. Destroying the Pod inside // its own callback can kill the reviewer before it receives the durable submission response. return completed.review; } async approveReview(reviewId: ReviewId, context: MutationContext): Promise { return this.uow.transaction(async (tx) => { const review = await tx.reviews.getById(reviewId); if (!review) throw new Error("Review does not exist"); if (review.reviewerAgentRunId) { throw new Error("Agent-assigned Review must be submitted by its reviewer Attempt callback"); } if (review.reviewerPrincipalId !== context.authority.principal.id) { throw new Error("Review approval requires the assigned independent reviewer"); } if (review.status === "submitted" && review.disposition === "approved") return review; if (review.status !== "requested" && review.status !== "reviewing") { throw new Error("Review is not pending"); } const changeSet = await tx.changeSets.getById(review.changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); if (changeSet.candidateDigest !== review.candidateDigest) { throw new Error("Review candidate identity is stale"); } const attempt = await tx.attempts.getById(changeSet.producerAttemptId); if (!attempt) throw new Error("Producing Attempt does not exist"); const agentRun = await tx.agentRuns.getById(attempt.agentRunId); if (!agentRun || agentRun.role !== "coder") throw new Error("Producing AgentRun does not exist"); if (agentRun.agentPrincipalId === review.reviewerPrincipalId) { throw new Error("Reviewer Principal must differ from coder Principal"); } const factoryRun = await tx.factoryRuns.getById(agentRun.factoryRunId); const task = await tx.tasks.getById(changeSet.taskId); if (!factoryRun || !task) throw new Error("Execution hierarchy does not exist"); const approved: Review = { ...review, status: "submitted", disposition: "approved", }; await tx.reviews.update(approved); const evidence = await tx.verificationEvidence.listByChangeSetIds([changeSet.id]); const findings = await tx.reviewFindings.listByChangeSetIds([changeSet.id]); const verified = verificationSatisfied(changeSet, evidence, findings); const independentDogfoodReview = await this.hasApprovedIndependentReview(tx, changeSet); await tx.changeSets.update({ ...changeSet, status: verified && independentDogfoodReview ? "ready-to-merge" : "verifying", revision: changeSet.revision + 1, }); await this.record( tx, { factoryRun, task, context }, "ReviewApproved", "Review", review.id, 1, { changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, reviewerPrincipalId: approved.reviewerPrincipalId, verificationSatisfied: verified, independentDogfoodReview, }, ); return approved; }); } async reconcileAutoMerge( changeSetId: ChangeSetId, context: MutationContext, ): Promise { return this.reconcileAutoMergeWithSteps(changeSetId, context); } async reconcileAutoMergeDurably( changeSetId: ChangeSetId, context: MutationContext, durable: DurableWorkflowStepContext, ): Promise { return this.reconcileAutoMergeWithSteps(changeSetId, context, durable); } private async reconcileAutoMergeWithSteps( changeSetId: ChangeSetId, context: MutationContext, durable?: DurableWorkflowStepContext, ): Promise { const snapshot = await this.uow.transaction((tx) => tx.changeSets.getById(changeSetId)); if (!snapshot) throw new Error("ChangeSet does not exist"); const trustedContext = this.controlPlaneContinuationContext( context, snapshot.projectId, changeSetId, ); if (snapshot.status === "merged") { await this.reconcileMergedFactoryRunDispatches(changeSetId, trustedContext, durable); await this.cleanupTerminalChangeSetWorkspaces(changeSetId, trustedContext, durable); return { state: "merged", changeSetId }; } if ( ["changes-requested", "superseded", "cancelled", "publication-failed"].includes( snapshot.status, ) ) { return { state: "terminal", changeSetId, reason: `ChangeSet cannot auto-merge from ${snapshot.status}`, }; } await this.refreshRequiredChecks(changeSetId, trustedContext, durable); const refreshed = await this.uow.transaction((tx) => tx.changeSets.getById(changeSetId)); if (!refreshed) throw new Error("ChangeSet disappeared during auto-merge reconciliation"); if (refreshed.status === "merged") { await this.reconcileMergedFactoryRunDispatches(changeSetId, trustedContext, durable); await this.cleanupTerminalChangeSetWorkspaces(changeSetId, trustedContext, durable); return { state: "merged", changeSetId }; } if ( ["changes-requested", "superseded", "cancelled", "publication-failed"].includes( refreshed.status, ) ) { return { state: "terminal", changeSetId, reason: `ChangeSet cannot auto-merge from ${refreshed.status}`, }; } if (refreshed.status !== "ready-to-merge") { return { state: "waiting", changeSetId, reason: "Independent Review or repository-required checks are still pending", }; } await this.executeMerge(changeSetId, trustedContext, false, durable); return { state: "merged", changeSetId }; } async requestMerge(changeSetId: ChangeSetId, context: MutationContext): Promise { return this.executeMerge(changeSetId, context, true); } private async executeMerge( changeSetId: ChangeSetId, context: MutationContext, refreshChecks: boolean, durable?: DurableWorkflowStepContext, ): Promise { const trusted = this.trustedRepository; if (!trusted) throw new MergeRefusedError("Repository-scoped trusted merge is not configured"); const beforeRefresh = await this.uow.transaction((tx) => tx.changeSets.getById(changeSetId)); if (!beforeRefresh) throw new Error("ChangeSet does not exist"); if (beforeRefresh.status === "merged") { await this.reconcileMergedFactoryRunDispatches(changeSetId, context, durable); await this.cleanupTerminalChangeSetWorkspaces(changeSetId, context, durable); return ( (await this.uow.transaction((tx) => tx.changeSets.getById(changeSetId))) ?? beforeRefresh ); } if (refreshChecks) await this.refreshRequiredChecks(changeSetId, context, durable); const input = await this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); if (changeSet.status === "merged") return { alreadyMerged: changeSet } as const; const approved = await this.hasApprovedIndependentReview(tx, changeSet); if (!approved || changeSet.status !== "ready-to-merge") { throw new MergeRefusedError( "Merge requires an approved independent reviewer AgentRun for the immutable candidate", ); } const evidence = await tx.verificationEvidence.listByChangeSetIds([changeSet.id]); const findings = await tx.reviewFindings.listByChangeSetIds([changeSet.id]); if (!verificationSatisfied(changeSet, evidence, findings)) { throw new MergeRefusedError( "Merge requires current passing required VerificationEvidence and no unresolved blocking findings", ); } if ( !changeSet.repositoryKey || !changeSet.publicationReference || !changeSet.targetReference || !changeSet.targetRevision ) { throw new MergeRefusedError( "Merge requires a durable repository-scoped trusted publication identity", ); } const project = await tx.projects.getById(changeSet.projectId); if (!project) throw new Error("ChangeSet Project does not exist"); const repositoryKey = trusted.repositoryKey(project.repositoryUrl); if (repositoryKey !== changeSet.repositoryKey) { throw new MergeRefusedError("Project repository no longer matches the published ChangeSet"); } const attempt = await tx.attempts.getById(changeSet.producerAttemptId); const agentRun = attempt ? await tx.agentRuns.getById(attempt.agentRunId) : undefined; const factoryRun = agentRun ? await tx.factoryRuns.getById(agentRun.factoryRunId) : undefined; const task = await tx.tasks.getById(changeSet.taskId); if (!attempt || !agentRun || !factoryRun || !task) { throw new Error("Execution hierarchy does not exist"); } return { changeSet, factoryRun, task, attempt, repositoryKey } as const; }); if ("alreadyMerged" in input) { await this.reconcileMergedFactoryRunDispatches(changeSetId, context, durable); await this.cleanupTerminalChangeSetWorkspaces(changeSetId, context, durable); return input.alreadyMerged; } const providerContext = this.trustedProviderContext( context, `changeset:${String(changeSetId)}:merge`, ); const mergeRequest = { context: providerContext, changeSetId: input.changeSet.id, repositoryKey: input.repositoryKey, publicationReference: input.changeSet.publicationReference!, candidateDigest: input.changeSet.candidateDigest, candidateManifest: input.changeSet.candidateManifest, expectedBase: { reference: input.changeSet.targetReference!, revision: input.changeSet.baseIdentity, }, expectedTarget: { reference: input.changeSet.targetReference!, revision: input.changeSet.targetRevision!, }, }; const providerMerge = await this.workflowExternalStep( durable, "awp-i1-automerge-trusted-merge", String(changeSetId), () => executeTrustedMerge(trusted.merger, input.changeSet, mergeRequest, async () => { const observed = await trusted.forge.inspectRepository( providerContext, input.repositoryKey, ); return { reference: `refs/heads/${observed.value.defaultBranch}`, revision: observed.value.headRevision, }; }), ); const resultingRevision = providerMerge.value.resultingRevision; const mergeReference = providerMerge.value.mergeReference; const merged = await this.workflowDomainStep( durable, "awp-i1-automerge-merge-record", String(changeSetId), async (tx) => { const current = await tx.changeSets.getById(changeSetId); const task = await tx.tasks.getById(input.task.id); const currentFactoryRun = await tx.factoryRuns.getById(input.factoryRun.id); if (!current || !task || !currentFactoryRun) throw new Error("Merge state disappeared"); if (current.status === "merged") { if ( current.resultingRevision !== resultingRevision || current.mergeReference?.nativeId !== mergeReference.nativeId || current.mergeReference?.nativeRevision !== mergeReference.nativeRevision ) { throw new MergeRefusedError( "Merged ChangeSet provider identity was concurrently rebound", ); } return current; } if (current.status !== "ready-to-merge") { throw new MergeRefusedError("ChangeSet is no longer ready to merge"); } if ( current.repositoryKey !== input.repositoryKey || current.publicationReference?.nativeId !== input.changeSet.publicationReference?.nativeId || current.targetReference !== input.changeSet.targetReference || current.targetRevision !== input.changeSet.targetRevision ) { throw new MergeRefusedError( "Published ChangeSet identity changed before merge persistence", ); } const merged: ChangeSet = { ...current, mergeReference, resultingRevision, status: "merged", revision: current.revision + 1, }; await tx.changeSets.update(merged); const completedTask: Task = { ...task, status: "completed", revision: task.revision + 1 }; await tx.tasks.update(completedTask); const projectTasks = await tx.tasks.listByProject(input.factoryRun.projectId); const planTasks = projectTasks .filter((candidate) => candidate.planRevisionId === task.planRevisionId) .map((candidate) => (candidate.id === task.id ? completedTask : candidate)); const completedIds = new Set( planTasks .filter((candidate) => candidate.status === "completed") .map((candidate) => candidate.id), ); const activeStatuses = new Set([ "completed", "cancelled", "dispatched", "queued", "executing", "review", ]); const newlyReady = planTasks .filter( (candidate) => !activeStatuses.has(candidate.status) && candidate.dependencyIds.every((dependencyId) => completedIds.has(dependencyId)), ) .sort((left, right) => (left.position ?? 0) - (right.position ?? 0)); const newlyReadyIds = new Set(newlyReady.map((candidate) => candidate.id)); for (const candidate of planTasks) { if (candidate.id === task.id || activeStatuses.has(candidate.status)) continue; const nextStatus: Task["status"] = newlyReadyIds.has(candidate.id) ? "dispatched" : "blocked"; const updated: Task = { ...candidate, status: nextStatus, revision: candidate.revision + 1, }; await tx.tasks.update(updated); } const planComplete = planTasks.every((candidate) => candidate.status === "completed"); if (planComplete) { const revision = await tx.planRevisions.getById(task.planRevisionId); const plan = revision ? await tx.plans.getById(revision.planId) : undefined; if (plan) { await tx.plans.update({ ...plan, status: "completed", revision: plan.revision + 1 }); } await tx.factoryRuns.update({ ...currentFactoryRun, status: "completed", reason: "All Tasks in the PlanRevision completed through trusted merge", revision: currentFactoryRun.revision + 1, }); } else if (currentFactoryRun.status !== "running") { await tx.factoryRuns.update({ ...currentFactoryRun, status: "running", reason: "Plan graph still has active or dependency-blocked Tasks", revision: currentFactoryRun.revision + 1, }); } await this.record( tx, { factoryRun: currentFactoryRun, task, context }, "MergeCompleted", "ChangeSet", merged.id, merged.revision, { resultingRevision, mergeReference, repositoryKey: input.repositoryKey, mergedBy: context.authority.principal.id, }, ); return merged; }, ); await this.reconcileFactoryRunDispatches(input.factoryRun.id, context, durable); await this.cleanupTerminalChangeSetWorkspaces(changeSetId, context, durable); return merged; } private async reconcileMergedFactoryRunDispatches( changeSetId: ChangeSetId, context: MutationContext, durable?: DurableWorkflowStepContext, ): Promise { const factoryRunId = await this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(changeSetId); if (!changeSet || changeSet.status !== "merged") return undefined; const attempt = await tx.attempts.getById(changeSet.producerAttemptId); const agentRun = attempt ? await tx.agentRuns.getById(attempt.agentRunId) : undefined; return agentRun?.factoryRunId; }); if (factoryRunId) await this.reconcileFactoryRunDispatches(factoryRunId, context, durable); } private async reconcileFactoryRunDispatches( factoryRunId: FactoryRun["id"], context: MutationContext, durable?: DurableWorkflowStepContext, ): Promise { const requests = await this.uow.transaction(async (tx) => { const factoryRun = await tx.factoryRuns.getById(factoryRunId); if (!factoryRun) throw new Error("FactoryRun disappeared during dispatch reconciliation"); const tasks = (await tx.tasks.listByProject(factoryRun.projectId)).filter( (task) => task.planRevisionId === factoryRun.planRevisionId && task.status === "dispatched", ); if (tasks.length > 0 && ["completed", "cancelled", "failed"].includes(factoryRun.status)) { throw new Error("Terminal FactoryRun cannot retain dispatched Tasks"); } return tasks.map((task) => ({ factoryRun, task, context, ...(factoryRun.accountId === undefined ? {} : { selection: { accountId: factoryRun.accountId, ...(factoryRun.model === undefined ? {} : { model: factoryRun.model }), }, }), })); }); const dispatcher = this.nextTaskDispatcher ?? this; await Promise.all( requests.map((request) => this.workflowExternalStep( durable, "awp-i1-automerge-next-task-dispatch", String(request.task.id), () => dispatcher.dispatch(request), ), ), ); } private async workflowDomainStep( durable: DurableWorkflowStepContext | undefined, stepName: string, stepKey: string, work: (tx: ApplicationTransaction) => Promise, ): Promise { if (!durable) return this.uow.transaction(work); return runMarkedWorkflowTransaction( durable.workflowSteps, new WorkflowStepTransactionRunner(this.uow, this.clock), { operationId: durable.operationId, stepName, stepKey }, work, ); } private async workflowExternalStep( durable: DurableWorkflowStepContext | undefined, stepName: string, stepKey: string, work: () => Promise, ): Promise { return durable ? durable.workflowSteps.run(stepName, stepKey, work) : work(); } private async hasApprovedIndependentReview( tx: ApplicationTransaction, changeSet: ChangeSet, ): Promise { const producerAttempt = await tx.attempts.getById(changeSet.producerAttemptId); const producerAgentRun = producerAttempt ? await tx.agentRuns.getById(producerAttempt.agentRunId) : undefined; if (!producerAgentRun || producerAgentRun.role !== "coder") return false; const reviews = await tx.reviews.listByChangeSetIds([changeSet.id]); for (const review of reviews) { if ( review.candidateDigest !== changeSet.candidateDigest || review.status !== "submitted" || review.disposition !== "approved" || !review.reviewerAgentRunId ) { continue; } const reviewerAgentRun = await tx.agentRuns.getById(review.reviewerAgentRunId); if ( reviewerAgentRun?.role === "reviewer" && reviewerAgentRun.agentPrincipalId === review.reviewerPrincipalId && reviewerAgentRun.id !== producerAgentRun.id && reviewerAgentRun.agentPrincipalId !== producerAgentRun.agentPrincipalId ) { return true; } } return false; } private workspaceProviderContext( context: MutationContext, idempotencyKey: string, ): ProviderOperationContext { return { operationId: context.operationId, idempotencyKey, correlationId: context.correlationId, authority: context.authority, connectionId: this.configuration.connectionId, credentialReferenceId: this.configuration.credentialReferenceId, }; } private async attestWorkspaceCheckpoint( workspaceId: WorkspaceId, digest: string, observedAt: string, context: MutationContext, idempotencyKey: string, ): Promise { const attest = this.workspaceProvider.attestCheckpoint; if (!attest) throw new Error("Workspace provider does not support content checkpoint attestation"); await attest.call( this.workspaceProvider, this.workspaceProviderContext(context, idempotencyKey), workspaceId, this.configuration.profileKey, digest, observedAt, ); } private async cleanupTerminalChangeSetWorkspaces( changeSetId: ChangeSetId, context: MutationContext, durable?: DurableWorkflowStepContext, ): Promise { const cleanup = this.workspaceProvider.cleanupCheckpoint; if (!cleanup) return; const cleanupBasis = await this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(changeSetId); if (!changeSet || changeSet.status !== "merged") return undefined; const workspaceIds = new Set(); const producerAttempt = await tx.attempts.getById(changeSet.producerAttemptId); if (producerAttempt) workspaceIds.add(producerAttempt.workspaceId); const reviews = await tx.reviews.listByChangeSetIds([changeSet.id]); for (const review of reviews) { if (!review.reviewerAgentRunId) continue; const attempts = await tx.attempts.listByAgentRunIds([review.reviewerAgentRunId]); for (const attempt of attempts) workspaceIds.add(attempt.workspaceId); } const workspaces: Workspace[] = []; for (const workspaceId of workspaceIds) { const workspace = await tx.workspaces.getById(workspaceId); if (workspace) workspaces.push(workspace); } return { changeSet, workspaces }; }); if (!cleanupBasis) return; for (const workspace of cleanupBasis.workspaces) { if (workspace.cleanedAt) continue; if (!workspace.checkpointDigest || !workspace.checkpointCollectedAt) { throw new Error( `Merged ChangeSet Workspace ${String(workspace.id)} has no collected content checkpoint`, ); } const providerContext = this.workspaceProviderContext( context, `terminal-workspace-cleanup:${String(changeSetId)}:${String(workspace.id)}`, ); await this.workflowExternalStep( durable, "awp-i1-automerge-workspace-destroy", String(workspace.id), () => this.workspaceProvider.destroy(providerContext, workspace.id), ); await this.workflowExternalStep( durable, "awp-i1-automerge-workspace-checkpoint-cleanup", String(workspace.id), () => cleanup.call( this.workspaceProvider, providerContext, workspace.id, this.configuration.profileKey, workspace.checkpointDigest!, true, ), ); await this.workflowDomainStep( durable, "awp-i1-automerge-workspace-cleanup-record", String(workspace.id), async (tx) => { const current = await tx.workspaces.getById(workspace.id); if (!current || current.cleanedAt) { return { workspaceId: workspace.id, cleanedAt: current?.cleanedAt ?? workspace.cleanedAt ?? null, }; } if ( current.checkpointDigest !== workspace.checkpointDigest || !current.checkpointCollectedAt ) { throw new Error("Workspace checkpoint identity changed during terminal cleanup"); } const cleanedAt = this.clock.now().toISOString(); await tx.workspaces.update({ ...current, cleanedAt, revision: current.revision + 1, }); if (durable) { await this.recordChangeSetMutation( tx, cleanupBasis.changeSet, context, "WorkspaceTerminalCleanupCompleted", "Workspace", workspace.id, { checkpointDigest: workspace.checkpointDigest, cleanedAt }, ); } return { workspaceId: workspace.id, cleanedAt }; }, ); } } private controlPlaneContinuationContext( source: MutationContext, projectId: FactoryRun["projectId"], changeSetId: ChangeSetId, ): MutationContext { const operationId = unsafeOpaqueId(`automerge:${String(changeSetId)}`); return { operationId, correlationId: source.correlationId, idempotencyKey: `automerge:${String(changeSetId)}`, authority: authorityContext( { id: unsafeOpaqueId("principal:system:control-plane"), kind: "system", capabilities: [], }, [], projectId, ), }; } private trustedProviderContext( context: MutationContext, idempotencyKey: string, ): ProviderOperationContext { const trusted = this.trustedRepository; if (!trusted) throw new Error("Repository-scoped trusted publication is not configured"); return { operationId: context.operationId, idempotencyKey, correlationId: context.correlationId, authority: context.authority, connectionId: trusted.connectionId, credentialReferenceId: trusted.credentialReferenceId, }; } private async publishForReview( changeSetId: ChangeSetId, context: MutationContext, ): Promise { const trusted = this.trustedRepository; if (!trusted) throw new Error("Repository-scoped trusted publication is not configured"); const snapshot = await this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(changeSetId); if (!changeSet) throw new Error("ChangeSet does not exist"); const project = await tx.projects.getById(changeSet.projectId); if (!project) throw new Error("ChangeSet Project does not exist"); const producerAttempt = await tx.attempts.getById(changeSet.producerAttemptId); const producerAgentRun = producerAttempt ? await tx.agentRuns.getById(producerAttempt.agentRunId) : undefined; if (!producerAttempt || !producerAgentRun || producerAgentRun.role !== "coder") { throw new Error("Trusted publication requires the producing coder AgentRun"); } const repositoryKey = trusted.repositoryKey(project.repositoryUrl); if (changeSet.repositoryKey !== undefined && changeSet.repositoryKey !== repositoryKey) { throw new Error("ChangeSet repository identity no longer matches its Project"); } const reviews = await tx.reviews.listByChangeSetIds([changeSet.id]); return { changeSet, project, repositoryKey, reviews, factoryRunId: producerAgentRun.factoryRunId, }; }); if (snapshot.changeSet.publicationReference) { if (!snapshot.changeSet.targetReference || !snapshot.changeSet.targetRevision) { throw new Error("Published ChangeSet is missing immutable target identity"); } if (snapshot.reviews.length > 0) { await this.ensureIndependentReview(snapshot.changeSet.id, context); return ( (await this.uow.transaction((tx) => tx.changeSets.getById(snapshot.changeSet.id))) ?? snapshot.changeSet ); } return this.finalizePublication( snapshot.changeSet, snapshot.repositoryKey, snapshot.changeSet.publicationReference, snapshot.changeSet.targetReference, snapshot.changeSet.targetRevision, context, ); } const providerContext = this.trustedProviderContext( context, `changeset:${String(changeSetId)}:publish`, ); const inspected = await trusted.forge.inspectRepository( providerContext, snapshot.repositoryKey, ); const targetReference = `refs/heads/${inspected.value.defaultBranch}`; const publication = await trusted.publisher.publish({ context: providerContext, changeSetId: snapshot.changeSet.id, factoryRunId: snapshot.factoryRunId, repositoryKey: snapshot.repositoryKey, baseRevision: snapshot.changeSet.baseIdentity, candidateDigest: snapshot.changeSet.candidateDigest, candidateManifest: snapshot.changeSet.candidateManifest, diff: snapshot.changeSet.diff, }); const publicationReference = publication.references.find( (reference) => reference.resourceType === "publication", ); if (!publicationReference || !publicationReference.nativeRevision) { throw new Error("Trusted GitHub publication returned no immutable publication reference"); } return this.finalizePublication( snapshot.changeSet, snapshot.repositoryKey, publicationReference, targetReference, snapshot.changeSet.baseIdentity, context, ); } private async finalizePublication( publishedSnapshot: ChangeSet, repositoryKey: string, publicationReference: NonNullable, targetReference: string, targetRevision: string, context: MutationContext, ): Promise { const published = await this.uow.transaction(async (tx) => { let current = await tx.changeSets.getById(publishedSnapshot.id); if (!current) throw new Error("ChangeSet disappeared during trusted publication"); if (current.repositoryKey !== undefined && current.repositoryKey !== repositoryKey) { throw new Error("ChangeSet repository identity was concurrently rebound"); } if (current.publicationReference !== undefined) { if ( current.publicationReference.providerId !== publicationReference.providerId || current.publicationReference.nativeId !== publicationReference.nativeId || current.publicationReference.nativeRevision !== publicationReference.nativeRevision || current.targetReference !== targetReference || current.targetRevision !== targetRevision ) { throw new Error("ChangeSet publication identity was concurrently rebound"); } return current; } const next: ChangeSet = { ...current, repositoryKey, publicationReference, targetReference, targetRevision, status: "reviewing", revision: current.revision + 1, }; await tx.changeSets.update(next); await this.recordChangeSetMutation( tx, current, context, "ChangeSetPublished", "ChangeSet", current.id, { repositoryKey, publicationReference, targetReference, targetRevision, candidateDigest: current.candidateDigest, }, ); current = next; return current; }); await this.refreshRequiredChecks(published.id, context); await this.ensureIndependentReview(published.id, context); return (await this.uow.transaction((tx) => tx.changeSets.getById(published.id))) ?? published; } private async ensureIndependentReview( changeSetId: ChangeSetId, context: MutationContext, ): Promise { const assignment = await this.uow.transaction(async (tx) => { const changeSet = await tx.changeSets.getById(changeSetId); if (!changeSet) throw new Error("ChangeSet disappeared before independent Review assignment"); const producerAttempt = await tx.attempts.getById(changeSet.producerAttemptId); if (!producerAttempt) throw new Error("Producing Attempt does not exist"); const producerAgentRun = await tx.agentRuns.getById(producerAttempt.agentRunId); if (!producerAgentRun || producerAgentRun.role !== "coder") { throw new Error("Independent Review requires a coder AgentRun producer"); } const factoryRun = await tx.factoryRuns.getById(producerAgentRun.factoryRunId); const task = await tx.tasks.getById(changeSet.taskId); if (!factoryRun || !task) throw new Error("Execution hierarchy does not exist"); const reviews = await tx.reviews.listByChangeSetIds([changeSet.id]); const existingReview = reviews.find( (candidate) => candidate.candidateDigest === changeSet.candidateDigest, ); if (existingReview) { if (!existingReview.reviewerAgentRunId) { throw new Error("Dogfood Review is missing its independent reviewer AgentRun binding"); } const reviewerAgentRun = await tx.agentRuns.getById(existingReview.reviewerAgentRunId); if ( !reviewerAgentRun || reviewerAgentRun.role !== "reviewer" || reviewerAgentRun.agentPrincipalId !== existingReview.reviewerPrincipalId || reviewerAgentRun.agentPrincipalId === producerAgentRun.agentPrincipalId ) { throw new Error("Dogfood Review independent reviewer identity is invalid"); } const attempts = await tx.attempts.listByAgentRunIds([reviewerAgentRun.id]); const attempt = attempts.find((candidate) => candidate.selection.kind === "initial"); if (!attempt) throw new Error("Reviewer AgentRun is missing its initial Attempt"); const workspace = await tx.workspaces.getById(attempt.workspaceId); if (!workspace) throw new Error("Reviewer AgentRun is missing its Workspace"); return { changeSet, factoryRun, task, review: existingReview, reviewerAgentRun, attempt, workspace, shouldProvision: existingReview.status !== "submitted" && attempt.status === "created" && (reviewerAgentRun.status === "queued" || reviewerAgentRun.status === "waiting"), }; } if (!producerAttempt.accountId) { throw new Error("Independent reviewer requires explicit provider account provenance"); } const workspace: Workspace = { id: this.ids.next(), projectId: changeSet.projectId, revision: 1, }; const reviewerAgentRunId = this.ids.next(); const reviewerPrincipalId = unsafeOpaqueId( `principal:agent:${String(reviewerAgentRunId)}`, ); if (reviewerPrincipalId === producerAgentRun.agentPrincipalId) { throw new Error("Independent reviewer Principal must differ from coder Principal"); } const reviewerAgentRun: AgentRun = { id: reviewerAgentRunId, factoryRunId: factoryRun.id, taskId: task.id, agentPrincipalId: reviewerPrincipalId, role: "reviewer", status: "queued", revision: 1, }; const attempt: Attempt = { id: this.ids.next(), agentRunId: reviewerAgentRun.id, workspaceId: workspace.id, status: "created", providerId: producerAttempt.providerId ?? this.configuration.agentProviderId, accountId: producerAttempt.accountId, model: producerAttempt.model ?? this.configuration.model ?? "provider-default", selection: createAttemptSelectionProvenance({ kind: "initial", reason: "independent reviewer assignment for the exact published candidate", }), revision: 1, }; const review: Review = { id: unsafeOpaqueId(`review:${String(changeSet.id)}`), changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, reviewerPrincipalId, reviewerAgentRunId: reviewerAgentRun.id, status: "requested", }; await tx.workspaces.insert(workspace); await tx.agentRuns.insert(reviewerAgentRun); await tx.attempts.insert(attempt); await tx.reviews.insert(review); const reread = await tx.reviews.getById(review.id); if ( !reread || reread.candidateDigest !== changeSet.candidateDigest || reread.reviewerAgentRunId !== reviewerAgentRun.id ) { throw new Error("Review could not be durably bound to the independent reviewer AgentRun"); } await this.record( tx, { factoryRun, task, context }, "ReviewAgentAssigned", "Review", review.id, 1, { changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, reviewerAgentRunId: reviewerAgentRun.id, reviewerPrincipalId, producerAgentRunId: producerAgentRun.id, producerPrincipalId: producerAgentRun.agentPrincipalId, }, ); return { changeSet, factoryRun, task, review, reviewerAgentRun, attempt, workspace, shouldProvision: true, }; }); if (!assignment.shouldProvision) return; const { review, reviewerAgentRun, attempt, workspace, factoryRun, task, changeSet } = assignment; try { const fixtureEnvironment = { AWP_AGENT_RUN_ID: reviewerAgentRun.id, AWP_ATTEMPT_ID: attempt.id, AWP_TASK_ID: task.id, AWP_AGENT_ROLE: "reviewer", AWP_REVIEW_ID: review.id, AWP_CANDIDATE_DIGEST: changeSet.candidateDigest, AWP_REVIEW_CALLBACK_URL: `${this.configuration.callbackBaseUrl}/internal/execution/review`, AWP_FAILURE_URL: `${this.configuration.callbackBaseUrl}/internal/execution/fail`, AWP_CALLBACK_TOKEN: this.completionToken(attempt.id), }; const workspaceResult = await this.workspaceProvider.create( { operationId: context.operationId, idempotencyKey: `review:${String(review.id)}:workspace:${String(workspace.id)}:provision`, correlationId: context.correlationId, authority: context.authority, connectionId: this.configuration.connectionId, credentialReferenceId: this.configuration.credentialReferenceId, }, workspace.id, this.configuration.profileKey, reviewerAgentRun.id, { environment: this.configuration.agentProvider === undefined ? fixtureEnvironment : { AWP_AGENT_RUN_ID: reviewerAgentRun.id, AWP_ATTEMPT_ID: attempt.id, AWP_TASK_ID: task.id, AWP_AGENT_ROLE: "reviewer", AWP_REVIEW_ID: review.id, }, }, ); const agentResult = await this.startAgentAttempt( context, attempt, reviewerAgentRun.id, task.id, workspace.id, "initial", ); await this.uow.transaction(async (tx) => { const currentRun = await tx.agentRuns.getById(reviewerAgentRun.id); const currentAttempt = await tx.attempts.getById(attempt.id); const currentReview = await tx.reviews.getById(review.id); const currentFactoryRun = await tx.factoryRuns.getById(factoryRun.id); if (!currentRun || !currentAttempt || !currentReview || !currentFactoryRun) { throw new Error("Independent reviewer state disappeared during provisioning"); } if (currentReview.status === "submitted") return; await tx.agentRuns.update({ ...currentRun, status: "active", reason: "Reviewing exact published ChangeSet candidate", revision: currentRun.revision + 1, }); await tx.attempts.update({ ...currentAttempt, status: "running", revision: currentAttempt.revision + 1, }); await tx.reviews.update({ ...currentReview, status: "reviewing" }); await tx.factoryRuns.update({ ...currentFactoryRun, status: "running", reason: "Independent reviewer AgentRun active", revision: currentFactoryRun.revision + 1, }); await this.record( tx, { factoryRun: currentFactoryRun, task, context }, "ReviewAgentStarted", "AgentRun", currentRun.id, currentRun.revision + 1, { reviewId: currentReview.id, changeSetId: changeSet.id, candidateDigest: changeSet.candidateDigest, workspaceId: workspace.id, workspaceProviderState: workspaceResult.value.state, ...(agentResult === undefined ? {} : { agentProviderState: agentResult.value.state, agentProviderReferences: agentResult.references, }), }, ); }); } catch (error) { const reason = error instanceof Error ? error.message : "Independent reviewer provisioning failed"; await this.uow.transaction(async (tx) => { const currentRun = await tx.agentRuns.getById(reviewerAgentRun.id); const currentAttempt = await tx.attempts.getById(attempt.id); const currentFactoryRun = await tx.factoryRuns.getById(factoryRun.id); if (currentRun) { await tx.agentRuns.update({ ...currentRun, status: "waiting", reason, revision: currentRun.revision + 1, }); } if (currentAttempt && currentAttempt.status !== "terminal") { await tx.attempts.update({ ...currentAttempt, status: "created", reason, revision: currentAttempt.revision + 1, }); } if (currentFactoryRun) { await tx.factoryRuns.update({ ...currentFactoryRun, status: "retryable", reason, revision: currentFactoryRun.revision + 1, }); } }); throw error; } } private async startAgentAttempt( context: MutationContext, attempt: Attempt, agentRunId: AgentRunId, taskId: Task["id"], workspaceId: WorkspaceId, phase: "initial" | "retry", ) { const provider = this.configuration.agentProvider; if (!provider) return undefined; const connectionId = this.configuration.agentConnectionId; const credentialReferenceId = this.configuration.agentCredentialReferenceId; if (!connectionId || !credentialReferenceId) { throw new Error( "Agent provider execution requires connection and credential-reference authority", ); } return provider.startAttempt( { operationId: context.operationId, idempotencyKey: `attempt:${String(attempt.id)}:agent:${phase}`, correlationId: context.correlationId, authority: context.authority, connectionId, credentialReferenceId, }, attempt.id, agentRunId, taskId, workspaceId, ); } private async recordChangeSetMutation( tx: ApplicationTransaction, changeSet: ChangeSet, context: MutationContext, type: string, aggregateType: string, aggregateId: string, payload: unknown, ): Promise { const attempt = await tx.attempts.getById(changeSet.producerAttemptId); const agentRun = attempt ? await tx.agentRuns.getById(attempt.agentRunId) : undefined; const factoryRun = agentRun ? await tx.factoryRuns.getById(agentRun.factoryRunId) : undefined; const task = await tx.tasks.getById(changeSet.taskId); if (!attempt || !agentRun || !factoryRun || !task) { throw new Error("ChangeSet execution hierarchy does not exist"); } await this.record( tx, { factoryRun, task, context }, type, aggregateType, aggregateId, 1, payload, ); } private completionToken(attemptId: AttemptId): string { return executionCallbackToken(this.configuration.callbackSecret, String(attemptId)); } private async record( tx: ApplicationTransaction, request: ExecutionDispatchRequest, type: string, aggregateType: string, aggregateId: string, aggregateRevision: number, payload: unknown, ): Promise { const occurredAt = this.clock.now().toISOString(); const projectId = request.factoryRun.projectId; const principalId = request.context.authority.principal.id; await tx.events.append({ id: this.ids.next(), type, schemaVersion: 1, occurredAt, aggregateType, aggregateId, aggregateRevision, projectId, principalId, correlationId: request.context.correlationId as CorrelationId, payload, }); await tx.audit.append({ id: this.ids.next(), occurredAt, principalId, action: type, targetType: aggregateType, targetId: aggregateId, projectId, disposition: "allowed", correlationId: request.context.correlationId, safeMetadata: { operationId: request.context.operationId }, }); await tx.outbox.append({ id: this.ids.next(), topic: type, payload: { aggregateType, aggregateId, aggregateRevision, projectId, payload }, occurredAt, }); } }