import { readFile } from 'node:fs/promises'; import { join } from 'node:path'; import { Trace } from '../harness/trace.ts'; import { disposeWorkspacePath } from '../workspace/repo-store.ts'; import { patchpawPaths } from '../config/paths.ts'; import { readApprovalPlan, updateApprovalPlan } from './approval-plans.ts'; import { applyRunPhase } from './phases.ts'; import { readPaused, savePaused } from './resume.ts'; import { readState, statePath, writeState } from './state.ts'; import type { CommunicationStore } from './communication-store.ts'; import type { OutboundConnection, OutboundItem, StoredItem } from './outbound.ts'; export type FinalizationKind = | 'stale_review' | 'review' | 'run_notice' | 'approval_plan' | 'legacy_conflict_proposal' | 'legacy_conflict_repair' | 'close_refusal' | 'close_completion' | 'close_start' | 'none'; function sourceRun(item: Pick) { return typeof item.source.run_id === 'string' ? item.source.run_id : undefined; } /** Classify only from durable outbox identity; handlers own all I/O and state transitions. */ export function classifyFinalization(item: Pick): FinalizationKind { const runId = sourceRun(item); if (item.kind === 'review' && item.status === 'cancelled_stale') return 'stale_review'; if (item.status !== 'delivered') return 'none'; if (item.kind === 'review' && runId) return 'review'; if (runId && item.purpose === 'run_notice') return 'run_notice'; if (item.purpose === 'approval_plan') return 'approval_plan'; if (item.purpose === 'conflict_proposal') return 'legacy_conflict_proposal'; if (item.purpose === 'conflict_repair') return 'legacy_conflict_repair'; if (item.purpose === 'close_refusal') return 'close_refusal'; if (item.purpose === 'close_completion') return 'close_completion'; if (item.purpose === 'close_start') return 'close_start'; return 'none'; } export type FinalizationConnection = OutboundConnection | (() => Promise); export interface FinalizationConfig { root: string; legacyHome?: string; snapshotRoot: string; } export interface FinalizationContext { config: FinalizationConfig; stored: StoredItem; store: CommunicationStore; resolveConnection: () => Promise; done: () => Promise; } type JsonObject = Record; function isJsonObject(value: unknown): value is JsonObject { return typeof value === 'object' && value !== null && !Array.isArray(value); } async function json(path: string): Promise { try { const value: unknown = JSON.parse(await readFile(path, 'utf8')); if (!isJsonObject(value)) throw new Error(`Expected JSON object at ${path}`); return value; } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null; throw error; } } function stringValue(value: unknown) { return typeof value === 'string' ? value : undefined; } function runDir(root: string, runId: string) { return join(patchpawPaths(root).runs, runId); } export async function finalizeStaleReview(context: FinalizationContext) { const { config, stored, resolveConnection, done } = context; const item = stored.item; const source = sourceRun(item); if (!source) return done(); const dir = runDir(config.root, source); const review = await json(join(dir, 'review.json')); if (!review || await json(join(dir, 'review-stale.json'))) return done(); let actualHead = ''; let prState = 'unknown'; try { const resolved = await resolveConnection(); if (!resolved) throw new Error('SCM connection required for stale Review finalization'); if (resolved.adapter) { const projectId = typeof item.source.project_id === 'string' || typeof item.source.project_id === 'number' ? String(item.source.project_id) : item.repo.match(/^gitlab:(.+):project:(.+)$/)?.[2]; if (!projectId) throw new Error('GitLab stale Review is missing its project id'); const changeRequest = await resolved.adapter.readChangeRequest(projectId, item.pr_number, { allowClosed: true }); actualHead = changeRequest.source.sha; prState = changeRequest.state; } else { if (!resolved.client) throw new Error('GitHub client required for stale Review finalization'); const [owner, repo] = item.repo.split('/'); const { data } = await resolved.client.rest.pulls.get({ owner, repo, pull_number: item.pr_number }); actualHead = data.head.sha; prState = data.state; } } catch { // The cancellation record remains durable if the remote read is unavailable. } const trace = new Trace(dir); trace.save('review-stale.json', { expected_head: String(item.source.head_sha ?? ''), actual_head: actualHead, pr_state: prState }); const path = statePath(patchpawPaths(config.root).state, item.repo, item.pr_number); const state = await readState(path); if (state?.run_id === source) { trace.save('result.json', { status: 'review_stale', run_id: source, repo: item.repo, pr_number: item.pr_number, final_head_sha: String(item.source.head_sha ?? ''), recovered: true, expected_head: String(item.source.head_sha ?? ''), actual_head: actualHead, pr_state: prState }); await writeState(path, applyRunPhase({ ...state, active: false }, 'review_stale')); } return done(); } export async function finalizeReviewDelivery(context: FinalizationContext) { const { config, stored, done } = context; const item = stored.item; const source = sourceRun(item); if (!source) return done(); const dir = runDir(config.root, source); const review = await json(join(dir, 'review.json')); if (!review || await json(join(dir, 'review-publication.json'))) return done(); const headSha = stringValue(review.head_sha); if (!headSha) throw new Error('Review artifact is missing head_sha'); const publication = { ...(item.receipt ?? {}), run_id: source, kind: 'review', recovered: item.source.recovered === 1, reused: item.receipt?.reused ?? false }; const trace = new Trace(dir); trace.save('review-publication.json', publication); const manifest = await json(join(dir, 'manifest.json')); const path = statePath(patchpawPaths(config.root).state, item.repo, item.pr_number); const state = await readState(path); if (state?.run_id === source) { const existing = await json(join(dir, 'result.json')); if (!existing || existing.status !== 'review_completed') trace.save('result.json', { status: 'review_completed', run_id: source, repo: item.repo, pr_number: item.pr_number, final_head_sha: headSha, recovered: item.source.recovered === 1, review, publication }); await writeState(path, applyRunPhase({ ...state, active: false, current_head_sha: headSha }, 'review_completed')); const workspacePath = stringValue(manifest?.workspace_path); if (workspacePath) { try { await disposeWorkspacePath(config.root, item.repo, workspacePath, trace); } catch { /* next lifecycle can retry cleanup */ } } trace.emit('review_delayed_delivery_finalized', { delivery_id: item.delivery_id, run_id: source }); } else if (manifest) trace.emit('review_delayed_delivery_preserved', { delivery_id: item.delivery_id, run_id: source, current_run_id: state?.run_id }); return done(); } export async function finalizeRunNotice(context: FinalizationContext) { const { config, stored, done } = context; const item = stored.item; const source = sourceRun(item); if (!source) return done(); const dir = runDir(config.root, source); if (!await json(join(dir, 'manifest.json')) && !await json(join(dir, 'run-notice.json'))) return done(); const trace = new Trace(dir); trace.save('notification.json', { status: 'published', ...(item.receipt ?? {}), delivery_id: item.delivery_id }); trace.emit('run_notice_delayed_delivery_finalized', { delivery_id: item.delivery_id }); return done(); } export async function finalizeApprovalPlanPublication(context: FinalizationContext) { const { config, stored, done } = context; const item = stored.item; // A Plan body is published through the durable outbox. If the original run could not // confirm the remote receipt, this mechanical finalizer records it here so `/approval` // can later verify a real publication. The Agent body is never re-read or re-shaped. const path = statePath(patchpawPaths(config.root).state, item.repo, item.pr_number); const revision = Number(item.source.plan_revision); const planId = typeof item.source.plan_id === 'string' ? item.source.plan_id : undefined; const bodySha256 = typeof item.source.body_sha256 === 'string' ? item.source.body_sha256 : undefined; const plan = Number.isSafeInteger(revision) ? await readApprovalPlan(path, revision) : null; if (plan && plan.plan_id === planId && plan.body_sha256 === bodySha256 && plan.status === 'publication_pending') { const publication = { delivery_id: item.delivery_id, remote_id: item.receipt?.id, remote_url: item.receipt?.html_url, published_at: item.receipt?.published_at }; await updateApprovalPlan(path, revision, { status: 'published', publication }); const state = await readState(path); if (state?.approval_plan?.plan_id === planId && state.approval_plan.plan_revision === revision) { await writeState(path, applyRunPhase({ ...state, active: false, approval_plan: { ...state.approval_plan, status: 'published', ...publication } }, 'awaiting_approval')); } const paused = await readPaused(path); if (paused?.status === 'publication_pending') { await savePaused(path, { ...paused, status: 'awaiting_approval', pause_phase: 'awaiting_approval' }); } new Trace(runDir(config.root, typeof item.source.run_id === 'string' ? item.source.run_id : '')).emit('approval_plan_publication_finalized', { delivery_id: item.delivery_id, plan_id: planId, plan_revision: revision, remote_id: item.receipt?.id }); } return done(); }