diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md index 470f27e..191bc92 100644 --- a/.github/PULL_REQUEST_TEMPLATE.md +++ b/.github/PULL_REQUEST_TEMPLATE.md @@ -5,6 +5,7 @@ ## Validation - [ ] `npm test` +- [ ] `npm run test:e2e` - [ ] `npm run lint` - [ ] `npm run typecheck` - [ ] `npm run build` diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9fe958e..f432082 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -23,10 +23,10 @@ jobs: timeout-minutes: 15 steps: - name: Checkout - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Set up Node.js - uses: actions/setup-node@v4 + uses: actions/setup-node@v5 with: node-version: 22 cache: npm @@ -37,6 +37,9 @@ jobs: - name: Test run: npm test + - name: End-to-end test + run: npm run test:e2e + - name: Lint run: npm run lint diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bc13f5..58e0648 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,18 @@ All notable changes to HiveForge are documented here. ## [Unreleased] -- Continue the V0.1 local-first MVP toward a production-readiness review. +- Reconcile the authoritative Master Source V0.1.4 when it is supplied. + +## [0.1.0-alpha.2] + +- Close the real Codex turn-to-validation-to-review loop with durable SQLite turn state. +- Align the adapter with codex-cli 0.152.0 App Server stdio, account/read and turn/completed. +- Add canonical RCP v0.1, immutable base/head Git evidence, validation persistence, recovery fencing, dynamic mailbox watchers and cross-process SSE replay. +- Add dashboard evidence, integration coverage and a real Codex App Server smoke path. + +## [0.1.0-alpha.3] + +- Keep the canonical Review ZIP finalizer compatible with Windows PowerShell 5.1. ## [0.1.0-alpha.1] @@ -12,5 +23,7 @@ All notable changes to HiveForge are documented here. - Add CI, contribution guidance, security reporting, issue forms, and release governance. - Add a Windows-first execution finalizer for repeatable validation and publication. -[Unreleased]: https://github.com/KayzenRoot/hiveforge/compare/v0.1.0-alpha.1...HEAD +[Unreleased]: https://github.com/KayzenRoot/hiveforge/compare/v0.1.0-alpha.2...HEAD [0.1.0-alpha.1]: https://github.com/KayzenRoot/hiveforge/releases/tag/v0.1.0-alpha.1 +[0.1.0-alpha.2]: https://github.com/KayzenRoot/hiveforge/releases/tag/v0.1.0-alpha.2 +[0.1.0-alpha.3]: https://github.com/KayzenRoot/hiveforge/releases/tag/v0.1.0-alpha.3 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 8bc2ae5..5e64ea6 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -14,6 +14,7 @@ Install dependencies with `npm ci`, then run the full validation set before open ```powershell npm test +npm run test:e2e npm run lint npm run typecheck npm run build diff --git a/README.md b/README.md index 0f5b4d2..c7bf8d4 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,7 @@ npm run worker ```powershell npm test +npm run test:e2e npm run lint npm run typecheck npm run build diff --git a/app/api/events/route.ts b/app/api/events/route.ts index ba8cedb..1698fc8 100644 --- a/app/api/events/route.ts +++ b/app/api/events/route.ts @@ -5,7 +5,7 @@ import type { RunEvent } from "../../../src/core/types"; export const runtime = "nodejs"; function encode(event: RunEvent | Record): Uint8Array { - return new TextEncoder().encode(`data: ${JSON.stringify(event)}\n\n`); + return new TextEncoder().encode("data: " + JSON.stringify(event) + "\n\n"); } export function GET(request: NextRequest) { @@ -17,17 +17,34 @@ export function GET(request: NextRequest) { let unsubscribe: () => void = () => undefined; let heartbeat: ReturnType | undefined; + let poller: ReturnType | undefined; let closed = false; + let lastCreatedAt = ""; + let lastEventId = ""; + const seen = new Set(); + const stream = new ReadableStream({ start(controller) { - for (const event of db.listEvents(runId)) controller.enqueue(encode(event)); - unsubscribe = events.subscribe(runId, (event) => { if (!closed) controller.enqueue(encode(event)); }); + const enqueue = (event: RunEvent) => { + if (closed || seen.has(event.id)) return; + seen.add(event.id); + lastCreatedAt = event.createdAt; + lastEventId = event.id; + controller.enqueue(encode(event)); + }; + for (const event of db.listEvents(runId, 250)) enqueue(event); + unsubscribe = events.subscribe(runId, enqueue); + poller = setInterval(() => { + if (closed) return; + for (const event of db.listEventsAfter(runId, lastCreatedAt, lastEventId, 250)) enqueue(event); + }, 750); heartbeat = setInterval(() => { if (!closed) controller.enqueue(new TextEncoder().encode(": heartbeat\n\n")); }, 15_000); }, cancel() { closed = true; unsubscribe(); if (heartbeat) clearInterval(heartbeat); + if (poller) clearInterval(poller); } }); return new Response(stream, { headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-transform", Connection: "keep-alive" } }); diff --git a/app/api/projects/route.ts b/app/api/projects/route.ts index fde575c..82921d4 100644 --- a/app/api/projects/route.ts +++ b/app/api/projects/route.ts @@ -8,7 +8,8 @@ export function GET() { const { db } = getRuntime(); const projects = db.listProjects().map((project) => { const run = db.getLatestRun(project.id); - return { ...project, latestRun: run ? { id: run.id, state: run.state, iterationCount: run.iterationCount, updatedAt: run.updatedAt } : undefined }; + const overview = run ? db.getRunOverview(run.id) : undefined; + return { ...project, latestRun: overview ?? undefined }; }); return NextResponse.json({ projects }); } diff --git a/app/api/runs/route.ts b/app/api/runs/route.ts index 8b20881..52d0a54 100644 --- a/app/api/runs/route.ts +++ b/app/api/runs/route.ts @@ -4,6 +4,18 @@ import type { AutonomyMode } from "../../../src/core/types"; export const runtime = "nodejs"; +export function GET(request: NextRequest) { + const { db } = getRuntime(); + const runId = request.nextUrl.searchParams.get("runId"); + const projectId = request.nextUrl.searchParams.get("projectId"); + if (runId) { + const overview = db.getRunOverview(runId); + return overview ? NextResponse.json({ overview }) : NextResponse.json({ error: "Run not found" }, { status: 404 }); + } + if (projectId) return NextResponse.json({ run: db.getLatestRun(projectId) ?? null }); + return NextResponse.json({ error: "runId or projectId is required" }, { status: 400 }); +} + export async function POST(request: NextRequest) { const { db, codex, engine } = getRuntime(); const body = await request.json() as { projectId?: string; prompt?: string }; diff --git a/app/dashboard.tsx b/app/dashboard.tsx index f156d02..967679c 100644 --- a/app/dashboard.tsx +++ b/app/dashboard.tsx @@ -3,7 +3,9 @@ import { FormEvent, useCallback, useEffect, useMemo, useState } from "react"; type Health = { status: string; codex: { available: boolean; version: string | null; appServer: boolean; authentication: string; reason?: string } }; -type Project = { id: string; name: string; localPath: string; autonomyMode: string; workingBranch: string; reviewMailboxPath: string; latestRun?: { id: string; state: string; iterationCount: number; updatedAt: string } }; +type Validation = { kind: string; command: string | null; status: string; exitCode: number | null; stdout: string; stderr: string }; +type RunOverview = { run: { id: string; workOrderId: string; state: string; autonomyMode: string; iterationCount: number; progressPercent: number; currentBlocker: string | null; runBaseSha: string | null; runBaseBranch: string | null; expectedBaseSha: string | null; expectedHeadSha: string | null; lastReviewStatus: string | null; lastReviewVerdict: string | null; lastReviewSummary: string | null; lastCheckpointNote: string | null; updatedAt: string }; gitEvidence: { baseSha: string; headSha: string; branch: string; status: string; isClean: boolean; changedFiles: string[]; diffSummary: string } | null; validations: Validation[]; review: { status: string; review: { reviewId: string; verdict: string; summary: string; progressPercent: number; checkpointNote: string } } | null }; +type Project = { id: string; name: string; localPath: string; autonomyMode: string; workingBranch: string; reviewMailboxPath: string; latestRun?: RunOverview }; type RunEvent = { id: string; type: string; state: string | null; message: string; createdAt: string }; const modeCopy: Record = { @@ -22,7 +24,8 @@ export default function Dashboard() { const [form, setForm] = useState({ name: "", localPath: "", defaultBranch: "main", workingBranch: "main", testCommand: "npm test", autonomyMode: "AUTONOMOUS", reviewMailboxPath: "" }); const selected = useMemo(() => projects.find((project) => project.id === selectedId) ?? projects[0], [projects, selectedId]); - const latestRun = selected?.latestRun; + const overview = selected?.latestRun; + const latestRun = overview?.run; const load = useCallback(async () => { const [projectsResponse, healthResponse] = await Promise.all([fetch("/api/projects"), fetch("/api/health")]); @@ -101,10 +104,10 @@ export default function Dashboard() {
-

ACTIVE WORK ORDER

{selected ? selected.name : "No project selected"}

{status}
Autonomy{selected ? modeCopy[selected.autonomyMode] : "—"}
Branch{selected?.workingBranch ?? "—"}
Work order{latestRun ? latestRun.id.slice(0, 12) : "Awaiting run"}
{message &&

{message}

}
+

ACTIVE WORK ORDER

{selected ? selected.name : "No project selected"}

{status}
Autonomy{selected ? modeCopy[selected.autonomyMode] : "—"}
Branch{latestRun?.runBaseBranch ?? selected?.workingBranch ?? "—"}
Work order{latestRun?.workOrderId ?? "Awaiting run"}
{overview &&
Git evidence{overview.gitEvidence?.isClean ? "CLEAN" : overview.gitEvidence ? "DIRTY" : "—"}{overview.gitEvidence?.headSha?.slice(0, 12) ?? "not captured"}
Validation{overview.validations.length ? overview.validations.map((item) => item.kind + ":" + item.status).join(" · ") : "pending"}{overview.review ? overview.review.status + " / " + overview.review.review.verdict : "review pending"}
Progress{latestRun?.progressPercent ?? 0}%{latestRun?.currentBlocker ?? latestRun?.lastCheckpointNote ?? "No active blocker"}
}{message &&

{message}

}

PERSISTED EVENT STREAM

Timeline

live
{events.length === 0 ?
Events will appear here as the worker moves the run.
: events.slice().reverse().map((event) =>
{event.type.replaceAll("_", " ")}

{event.message}

{event.state && {event.state}}
)}
HIVEFORGE / runtime truth lives in SQLite{health?.codex.version ?? "codex —"} · evidence lock enabled
diff --git a/app/globals.css b/app/globals.css index beade56..20b9160 100644 --- a/app/globals.css +++ b/app/globals.css @@ -17,3 +17,4 @@ button:disabled { cursor: not-allowed; opacity: .35; } @media (max-width: 820px) { .app-shell { padding: 0 20px 20px; }.hero-grid { grid-template-columns: 1fr; padding: 60px 0 44px; }.hero-signal { border-left: 0; border-top: 1px solid var(--line); padding: 20px 0 0; }.workspace-grid { grid-template-columns: 1fr; }.footer { gap: 9px; flex-direction: column; }.topbar-actions { font-size: 0; }.topbar-actions .icon-button { font-size: 16px; } } @media (max-width: 500px) { .run-meta { gap: 15px; grid-template-columns: 1fr; }.run-meta div + div { border-left: 0; padding-left: 0; }.run-actions { flex-wrap: wrap; }.hero-copy h1 { font-size: 56px; }.panel-heading { padding: 20px 18px 15px; }.run-meta, .run-actions, .timeline { padding-left: 18px; padding-right: 18px; } } @media (prefers-reduced-motion: reduce) { *, *:before, *:after { animation-duration: .01ms !important; animation-iteration-count: 1 !important; scroll-behavior: auto !important; } } +.evidence-strip { display: grid; gap: 1px; grid-template-columns: repeat(3, 1fr); margin: 14px 25px 0; }.evidence-strip > div { background: rgba(0,0,0,.12); min-width: 0; padding: 12px; }.evidence-strip span, .evidence-strip strong, .evidence-strip small { display: block; }.evidence-strip span { color: var(--muted); font-size: 9px; letter-spacing: .08em; text-transform: uppercase; }.evidence-strip strong { font-size: 12px; margin-top: 6px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }.evidence-strip small { color: var(--muted); font-size: 9px; line-height: 1.4; margin-top: 5px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } diff --git a/docs/CHECKPOINT.md b/docs/CHECKPOINT.md index e15bcaf..c03cd20 100644 --- a/docs/CHECKPOINT.md +++ b/docs/CHECKPOINT.md @@ -1,7 +1,9 @@ -# Checkpoint — Prompt 001 +# Checkpoint — Prompt 001B -HiveForge V0.1 has a local Next.js dashboard, SQLite runtime truth, explicit workflow transitions, Git evidence capture, RCP validation, mailbox consumption, a real Codex App Server JSONL adapter, FakeCodexAdapter tests, SSE event streaming and restart recovery checks. +Status: CORRECTION_IN_PROGRESS -The authoritative Master Source was not supplied. The repository keeps a clearly marked bootstrap note at `docs/HIVEFORGE-MASTER-SOURCE-V0.1.md`; reconcile it before starting the next increment. +The closed loop now has a production completion path: Codex terminal turn events advance into persisted validation, immutable Git evidence, canonical RCP validation, Evidence Lock, autonomy gates and durable next dispatches. SQLite-backed event replay and dynamic mailbox reconciliation cover process boundaries and newly registered projects. -Validation completed on 2026-09-01: tests, E2E, lint, typecheck and production build passed. Proposed checkpoint: `READY_FOR_SOL_AUDIT`. +The authoritative HIVEFORGE-MASTER-SOURCE-V0.1.4.md required by the 001B governance finding was not present in the supplied Downloads folder or repository. It was not invented or replaced with a local approximation. The checkpoint therefore remains blocked until the authoritative source is supplied and reconciled. + +Functional correction checkpoint: READY_FOR_SOURCE_RECONCILIATION. diff --git a/docs/CODEX-APP-SERVER.md b/docs/CODEX-APP-SERVER.md new file mode 100644 index 0000000..79983c5 --- /dev/null +++ b/docs/CODEX-APP-SERVER.md @@ -0,0 +1,28 @@ +# Codex App Server integration + +The correction increment was verified against the installed `codex-cli 0.152.0`. + +Regenerate the official protocol artifacts locally with: + +```powershell +.\scripts\verify-codex-protocol.ps1 +``` + +The adapter uses the official stdio lifecycle: + +- launch: `codex app-server` (stdio is the default); +- initialization: `initialize`, `initialized`; +- authentication/account: `account/read` with `refreshToken: false`; +- thread: `thread/start` with `cwd` and `ephemeral`; +- turn: `turn/start` with `threadId`, `input` and `cwd`; +- terminal event: `turn/completed`, mapped only when the returned turn status is `completed`, `failed` or `interrupted`. + +No legacy `getAuthStatus` call, undocumented project field or raw authentication token is used. Every returned turn ID is persisted before the run can advance, and a missing ID blocks the dispatch. + +The real smoke command is: + +```powershell +npx tsx scripts/codex-smoke.ts +``` + +It uses a temporary Git repository and asks Codex to reply without modifying files. diff --git a/docs/HIVEFORGE-GITHUB-GOVERNANCE-V0.1.md b/docs/HIVEFORGE-GITHUB-GOVERNANCE-V0.1.md index 924bfa6..ca68ac2 100644 --- a/docs/HIVEFORGE-GITHUB-GOVERNANCE-V0.1.md +++ b/docs/HIVEFORGE-GITHUB-GOVERNANCE-V0.1.md @@ -15,7 +15,7 @@ The approved bootstrap may land directly on an empty `main`. All later changes u ## Required validation -The CI workflow runs `npm ci`, `npm test`, `npm run lint`, `npm run typecheck`, and `npm run build` on pushes to `main` and work branches and on pull requests targeting `main`. A configured command failing blocks acceptance. +The CI workflow runs `npm ci`, `npm test`, `npm run test:e2e`, `npm run lint`, `npm run typecheck`, and `npm run build` on pushes to `main` and work branches and on pull requests targeting `main`. A configured command failing blocks acceptance. ## Protection and permissions diff --git a/docs/RCP-EXAMPLE.json b/docs/RCP-EXAMPLE.json index 119da63..12a554e 100644 --- a/docs/RCP-EXAMPLE.json +++ b/docs/RCP-EXAMPLE.json @@ -1,15 +1,15 @@ { - "protocol": "RCP", - "version": "0.1", + "schema_version": "0.1", "review_id": "review-001", "project_id": "project-uuid", - "run_id": "run-uuid", "work_order_id": "WO-ABC12345", - "decision": "APPROVED", - "action": "CONTINUE", - "head_sha": "0123456789abcdef0123456789abcdef01234567", "base_sha": "fedcba9876543210fedcba9876543210fedcba98", + "head_sha": "0123456789abcdef0123456789abcdef01234567", + "verdict": "APPROVED", + "progress_percent": 50, + "summary": "Increment reviewed against the locked Git evidence.", "findings": [], + "next_action": "CONTINUE", "executor_prompt": "Continue with the next safe increment and return validation evidence.", - "created_at": "2026-09-01T00:00:00.000Z" + "checkpoint_note": "Keep the work order base and head SHAs unchanged for this review." } diff --git a/docs/SETUP.md b/docs/SETUP.md index 944a395..21dc137 100644 --- a/docs/SETUP.md +++ b/docs/SETUP.md @@ -21,7 +21,7 @@ Start the separate mailbox worker in another terminal: npm run worker ``` -The worker watches each registered project's `reviewMailboxPath`. Only top-level `.json` files are consumed. Processed files move to `.processed`; invalid files move to `.rejected`. +The worker watches each registered project's `reviewMailboxPath` and reconciles the registry every five seconds, so projects registered after worker startup are picked up without a restart. Only top-level `.json` files are consumed. Processed files move to `.processed`; invalid files move to `.rejected`. ## Validation commands @@ -35,8 +35,10 @@ npm run build ## Review contract -See `docs/RCP-EXAMPLE.json`. A valid contract must include project, run, work order, base SHA and head SHA. Evidence Lock rejects a review whose Git identity does not match the captured run evidence. +See `docs/RCP-EXAMPLE.json`. The canonical contract uses `schema_version`, `review_id`, `project_id`, `work_order_id`, `base_sha`, `head_sha`, `verdict`, `progress_percent`, `summary`, `findings`, `next_action`, `executor_prompt` and `checkpoint_note`. A full Git SHA is required; `work_order_id` uniquely resolves the run. Evidence Lock also requires clean working-tree evidence. + +Validation runs the configured project `testCommand` and any configured optional lint, typecheck and build commands after every terminal Codex turn. Each result is persisted in SQLite with command, timestamps, exit code, stdout, stderr and `PASS`, `FAIL`, `NOT_CONFIGURED` or `ERROR`. ## Environment notes -No secret, cookie or raw authentication token is stored by HiveForge. The Codex adapter asks App Server for auth status with `includeToken: false` when a real process is started. +No secret, cookie or raw authentication token is stored by HiveForge. The Codex adapter uses the installed official App Server lifecycle and asks `account/read` with `refreshToken: false`; it does not call the legacy `getAuthStatus` method. diff --git a/package-lock.json b/package-lock.json index 719ddce..36b8f0f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "hiveforge", - "version": "0.1.0-alpha.1", + "version": "0.1.0-alpha.3", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "hiveforge", - "version": "0.1.0-alpha.1", + "version": "0.1.0-alpha.3", "dependencies": { "next": "15.5.9", "react": "19.1.0", diff --git a/package.json b/package.json index 4c07dcc..dc4de4c 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "hiveforge", - "version": "0.1.0-alpha.1", + "version": "0.1.0-alpha.3", "private": true, "engines": { "node": ">=22" diff --git a/scripts/codex-smoke.ts b/scripts/codex-smoke.ts new file mode 100644 index 0000000..c4b012c --- /dev/null +++ b/scripts/codex-smoke.ts @@ -0,0 +1,41 @@ +import { execFileSync } from "node:child_process"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { CodexAppServerAdapter } from "../src/adapters/codex"; + +function git(cwd: string, args: string[]) { + return execFileSync("git", args, { cwd, encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] }).trim(); +} + +async function main() { + const cwd = mkdtempSync(join(tmpdir(), "hiveforge-codex-smoke-")); + git(cwd, ["init", "-b", "main"]); + git(cwd, ["config", "user.email", "smoke@hiveforge.local"]); + git(cwd, ["config", "user.name", "HiveForge Smoke"]); + writeFileSync(join(cwd, "README.md"), "Codex App Server smoke fixture.\n"); + git(cwd, ["add", "."]); + git(cwd, ["commit", "-m", "smoke fixture"]); + + const adapter = new CodexAppServerAdapter(); + const completed = new Promise<{ turnId: string; status: string; error?: string }>((resolve, reject) => { + const timeout = setTimeout(() => reject(new Error("turn/completed smoke timeout")), 90_000); + adapter.setEventHandler((event) => { + if (event.kind !== "turn") return; + clearTimeout(timeout); + resolve({ turnId: event.turnId, status: event.status, error: event.error }); + }); + }); + try { + const thread = await adapter.startThread({ cwd }); + const turn = await adapter.sendPrompt({ threadId: thread.threadId, cwd, prompt: "Reply with exactly HIVEFORGE_SMOKE_OK. Do not modify files, run commands, or create commits." }); + if (!turn.turnId) throw new Error("Codex App Server returned no turn id"); + const result = await completed; + if (result.status !== "completed") throw new Error("Codex smoke ended with " + result.status + (result.error ? ": " + result.error : "")); + console.log("CODEX APP SERVER SMOKE PASS: " + result.turnId); + } finally { + await adapter.close(); + } +} + +void main().catch((error) => { console.error("CODEX APP SERVER SMOKE BLOCKED: " + (error instanceof Error ? error.message : String(error))); process.exitCode = 1; }); diff --git a/scripts/create-review-zip.ps1 b/scripts/create-review-zip.ps1 index dba8615..c13a59f 100644 --- a/scripts/create-review-zip.ps1 +++ b/scripts/create-review-zip.ps1 @@ -56,13 +56,21 @@ function Invoke-CapturedCommand { $display = $Executable if ($Arguments.Count -gt 0) { $display += " " + ($Arguments -join " ") } $exitCode = 0 + $previousErrorActionPreference = $ErrorActionPreference + $nativePreferenceExists = $null -ne (Get-Variable -Name PSNativeCommandUseErrorActionPreference -ErrorAction SilentlyContinue) + $previousNativePreference = if ($nativePreferenceExists) { $PSNativeCommandUseErrorActionPreference } else { $false } try { + $ErrorActionPreference = "Continue" + if ($nativePreferenceExists) { $PSNativeCommandUseErrorActionPreference = $false } & $Executable @Arguments 1> $stdoutPath 2> $stderrPath $exitCode = if ($null -eq $LASTEXITCODE) { 0 } else { [int]$LASTEXITCODE } } catch { $exitCode = 127 Write-Utf8File -Path $stdoutPath -Content "" Write-Utf8File -Path $stderrPath -Content $_.Exception.Message + } finally { + $ErrorActionPreference = $previousErrorActionPreference + if ($nativePreferenceExists) { $PSNativeCommandUseErrorActionPreference = $previousNativePreference } } $stdout = "" $stderr = "" @@ -129,7 +137,7 @@ function Assert-ZipIsValid { "evidence/remote.txt", "evidence/compare-base.txt", "evidence/pr.txt", "evidence/release.txt", "evidence/ci-status.txt", "evidence/commands-run.txt", "validation/tests.txt", "validation/lint.txt", - "validation/typecheck.txt", "validation/build.txt" + "validation/typecheck.txt", "validation/build.txt", "validation/e2e.txt" ) foreach ($entry in $required) { if ($names -notcontains $entry) { throw "ZIP is missing required entry: $entry" } @@ -158,7 +166,7 @@ try { $remoteUrlRecord = Invoke-CapturedCommand -Label "git remote origin" -Executable "git" -Arguments @("remote", "get-url", "origin") -AllowFailure $fetchMainRecord = Invoke-CapturedCommand -Label "git fetch origin main" -Executable "git" -Arguments @("fetch", "origin", "main", "--quiet") -AllowFailure $remoteMainRecord = Invoke-CapturedCommand -Label "git remote main" -Executable "git" -Arguments @("rev-parse", "--verify", "origin/main") -AllowFailure - $baseRecord = if ($remoteMainRecord.ExitCode -eq 0) { $remoteMainRecord } elseif ($headRecord.ExitCode -eq 0) { Invoke-CapturedCommand -Label "git base parent" -Executable "git" -Arguments @("rev-parse", "HEAD^") -AllowFailure } else { $null } + $baseRecord = if ($remoteMainRecord.ExitCode -eq 0) { $remoteMainRecord } else { $null } $logRecord = Invoke-CapturedCommand -Label "git log" -Executable "git" -Arguments @("log", "-10", "--oneline", "--decorate") -AllowFailure $branchesRecord = Invoke-CapturedCommand -Label "git branches" -Executable "git" -Arguments @("branch", "--all", "--verbose") -AllowFailure $headSha = if ($headRecord.ExitCode -eq 0) { $headRecord.Stdout.Trim() } else { $null } @@ -237,7 +245,7 @@ try { $packagePath = Join-Path $RepoRoot "package.json" $package = if (Test-Path -LiteralPath $packagePath) { Get-Content -LiteralPath $packagePath -Raw | ConvertFrom-Json } else { $null } - $scriptNames = [ordered]@{ tests = "test"; lint = "lint"; typecheck = "typecheck"; build = "build" } + $scriptNames = [ordered]@{ tests = "test"; e2e = "test:e2e"; lint = "lint"; typecheck = "typecheck"; build = "build" } $validation = [ordered]@{} foreach ($label in $scriptNames.Keys) { $scriptName = $scriptNames[$label] @@ -281,7 +289,7 @@ try { "REVIEW-MANIFEST.md", "REVIEW-MANIFEST.json", "evidence/git-status.txt", "evidence/git-log.txt", "evidence/git-branches.txt", "evidence/git-head.txt", "evidence/git-diff-stat.txt", "evidence/git-diff.patch", "evidence/changed-files.txt", "evidence/repository-tree.txt", "evidence/environment.txt", "evidence/commands-run.txt", "evidence/remote.txt", "evidence/compare-base.txt", "evidence/pr.txt", "evidence/release.txt", "evidence/ci-status.txt", - "validation/tests.txt", "validation/lint.txt", "validation/typecheck.txt", "validation/build.txt", "repository/" + "validation/tests.txt", "validation/e2e.txt", "validation/lint.txt", "validation/typecheck.txt", "validation/build.txt", "repository/" ) Write-Utf8File -Path (Join-Path $RunDirectory "evidence/changed-files.txt") -Content $(if ($changedFiles.Count -gt 0) { $changedFiles -join "`r`n" } else { "" }) @@ -339,7 +347,9 @@ try { ) foreach ($label in $validation.Keys) { $item = $validation[$label] - $markdown += "| $label | $($item.command ?? 'not configured') | $($item.exit_code ?? '-') | $($item.status) |" + $commandText = if ($null -eq $item.command) { "not configured" } else { [string]$item.command } + $exitText = if ($null -eq $item.exit_code) { "-" } else { [string]$item.exit_code } + $markdown += "| $label | $commandText | $exitText | $($item.status) |" } $markdown += @( "", @@ -367,7 +377,8 @@ try { if ($SimulateFailureBeforeReplace) { throw "Simulated failure before canonical replacement." } Replace-CanonicalZip -Source $TemporaryZip -Destination $CanonicalZip Assert-ZipIsValid -ZipPath $CanonicalZip - Write-Host "Review ZIP ready: $([System.IO.Path]::GetRelativePath($RepoRoot, $CanonicalZip).Replace('\', '/'))" + $relativeZip = $CanonicalZip.Substring($RepoRoot.Length + 1).Replace('\', '/') + Write-Host "Review ZIP ready: $relativeZip" } catch { Write-Error $_ exit 1 diff --git a/scripts/e2e.ts b/scripts/e2e.ts index 080ce5d..6637322 100644 --- a/scripts/e2e.ts +++ b/scripts/e2e.ts @@ -1,11 +1,12 @@ import assert from "node:assert/strict"; import { execFileSync } from "node:child_process"; -import { mkdtempSync, mkdirSync, writeFileSync } from "node:fs"; +import { mkdtempSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { FakeCodexAdapter } from "../src/adapters/fake-codex"; import { GitAdapter } from "../src/adapters/git"; import { EventBus } from "../src/core/event-bus"; +import { serializeReviewContract } from "../src/core/rcp"; import type { ReviewContract } from "../src/core/types"; import { Database } from "../src/db/database"; import { ReviewMailboxWatcher } from "../src/mailbox/watcher"; @@ -23,16 +24,21 @@ function createFixture() { writeFileSync(join(directory, "README.md"), "base\n"); git(directory, ["add", "."]); git(directory, ["commit", "-m", "base"]); - writeFileSync(join(directory, "README.md"), "head\n"); - git(directory, ["add", "."]); - git(directory, ["commit", "-m", "head"]); - return { directory, headSha: git(directory, ["rev-parse", "HEAD"]) }; + return { directory }; +} + +async function settle(db: Database, runId: string) { + for (let index = 0; index < 100; index += 1) { + const state = db.getRun(runId)?.state; + if (state === "WAITING_REVIEW" || state === "BLOCKED") return; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + throw new Error("workflow did not settle for run " + runId); } async function main() { const fixture = createFixture(); - const mailbox = join(fixture.directory, ".mailbox"); - mkdirSync(mailbox); + const mailbox = mkdtempSync(join(tmpdir(), "hiveforge-e2e-mailbox-")); const db = new Database(":memory:"); const events = new EventBus(); const fake = new FakeCodexAdapter(); @@ -41,30 +47,35 @@ async function main() { const run = db.createRun({ projectId: project.id, autonomyMode: "AUTONOMOUS" }); await engine.start(run.id, "first turn"); - await engine.markValidation(run.id); + fake.completeTurn(); + await settle(db, run.id); const waiting = db.getRun(run.id)!; assert.equal(waiting.state, "WAITING_REVIEW"); const evidence = db.getLatestGitEvidence(run.id)!; - const review: ReviewContract = { protocol: "RCP", version: "0.1", review_id: "review-valid", project_id: project.id, run_id: run.id, work_order_id: waiting.workOrderId, decision: "APPROVED", action: "CONTINUE", head_sha: evidence.headSha, base_sha: evidence.baseSha, findings: [], executor_prompt: "second turn without copy/paste", created_at: new Date().toISOString() }; - const watcher = new ReviewMailboxWatcher(mailbox, { onReview: async (value) => { const result = await engine.processReview(value); if (value.review_id === "review-valid") assert.equal(result.accepted, true); else assert.equal(result.accepted, false); }, onRejected: ({ errors }) => { throw new Error(errors.join("; ")); } }); - writeFileSync(join(mailbox, "review-valid.json"), JSON.stringify(review)); + const review: ReviewContract = { schemaVersion: "0.1", reviewId: "review-valid", projectId: project.id, workOrderId: waiting.workOrderId, verdict: "APPROVED", progressPercent: 70, summary: "First increment is valid", findings: [], nextAction: "CONTINUE", executorPrompt: "second turn without copy/paste", checkpointNote: "Continue from the locked evidence", headSha: evidence.headSha, baseSha: evidence.baseSha }; + const watcher = new ReviewMailboxWatcher(mailbox, { + onReview: async (value) => { const result = await engine.processReview(value); if (value.reviewId === "review-valid") assert.equal(result.accepted, true); else assert.equal(result.accepted, false); }, + onRejected: ({ errors }) => { throw new Error(errors.join("; ")); } + }); + writeFileSync(join(mailbox, "review-valid.json"), JSON.stringify(serializeReviewContract(review))); await watcher.scan(); assert.equal(fake.prompts.length, 2, "valid review must produce the second dispatch automatically"); assert.equal(db.getRun(run.id)?.state, "CODEX_RUNNING"); const negativeRun = db.createRun({ projectId: project.id, autonomyMode: "AUTONOMOUS" }); await engine.start(negativeRun.id, "negative first turn"); - await engine.markValidation(negativeRun.id); + fake.completeTurn(); + await settle(db, negativeRun.id); const negativeWaiting = db.getRun(negativeRun.id)!; const negativeEvidence = db.getLatestGitEvidence(negativeRun.id)!; - const invalidReview = { ...review, review_id: "review-wrong-sha", run_id: negativeRun.id, work_order_id: negativeWaiting.workOrderId, head_sha: "0000000000000000000000000000000000000000", base_sha: negativeEvidence.baseSha }; - writeFileSync(join(mailbox, "review-wrong-sha.json"), JSON.stringify(invalidReview)); + const invalidReview: ReviewContract = { ...review, reviewId: "review-wrong-sha", workOrderId: negativeWaiting.workOrderId, headSha: "0000000000000000000000000000000000000000", baseSha: negativeEvidence.baseSha }; + writeFileSync(join(mailbox, "review-wrong-sha.json"), JSON.stringify(serializeReviewContract(invalidReview))); await watcher.scan(); assert.equal(fake.prompts.length, 3, "wrong SHA must not dispatch a second prompt"); assert.equal(db.getRun(negativeRun.id)?.state, "WAITING_REVIEW"); watcher.stop(); db.close(); - console.log("E2E PASS: valid review dispatched automatically; wrong SHA stayed waiting with no dispatch."); + console.log("E2E PASS: real terminal callback advanced validation; canonical review dispatched automatically; wrong SHA stayed waiting with no dispatch."); } void main().catch((error) => { console.error(error); process.exitCode = 1; }); diff --git a/scripts/finalize-execution.ps1 b/scripts/finalize-execution.ps1 index 316db41..8e1f178 100644 --- a/scripts/finalize-execution.ps1 +++ b/scripts/finalize-execution.ps1 @@ -22,7 +22,7 @@ Set-Location $RepoRoot $CanonicalRemote = "https://github.com/KayzenRoot/hiveforge.git" $Repository = "KayzenRoot/hiveforge" -$RequiredChecks = @("test", "lint", "typecheck", "build") +$RequiredChecks = @("test", "test:e2e", "lint", "typecheck", "build") function Get-CommandText { param([Parameter(Mandatory = $true)][string[]]$Command) @@ -104,7 +104,6 @@ function Try-Protect-Main { } enforce_admins = $true required_pull_request_reviews = [ordered]@{ - dismissal_restrictions = $null dismiss_stale_reviews = $true require_code_owner_reviews = $true required_approving_review_count = 1 @@ -256,7 +255,7 @@ if (-not $Bootstrap) { $prUrl = Get-CommandText @("gh", "pr", "list", "--repo", $Repository, "--head", $currentBranch, "--base", "main", "--state", "open", "--json", "url", "--jq", ".[0].url") if (-not $prUrl) { $lineBreak = [Environment]::NewLine - $prBody = "Work order: $WorkOrderId$lineBreak$lineBreak" + "Validation: npm test, npm run lint, npm run typecheck, npm run build.$lineBreak$lineBreak" + "This pull request is not auto-merged." + $prBody = "Work order: $WorkOrderId$lineBreak$lineBreak" + "Validation: npm test, npm run test:e2e, npm run lint, npm run typecheck, npm run build.$lineBreak$lineBreak" + "This pull request is not auto-merged." $prUrl = Invoke-Checked @("gh", "pr", "create", "--repo", $Repository, "--base", "main", "--head", $currentBranch, "--title", $CommitMessage, "--body", $prBody) "create pull request" } Write-Host "Pull request: $prUrl" diff --git a/scripts/verify-codex-protocol.ps1 b/scripts/verify-codex-protocol.ps1 new file mode 100644 index 0000000..26c9fc3 --- /dev/null +++ b/scripts/verify-codex-protocol.ps1 @@ -0,0 +1,10 @@ +param( + [string]$OutputDirectory = (Join-Path ([System.IO.Path]::GetTempPath()) ("hiveforge-protocol-" + [guid]::NewGuid().ToString("N"))) +) + +New-Item -ItemType Directory -Path $OutputDirectory -Force | Out-Null +codex app-server generate-ts --out $OutputDirectory +if ($LASTEXITCODE -ne 0) { throw "Could not generate Codex TypeScript protocol" } +codex app-server generate-json-schema --out $OutputDirectory +if ($LASTEXITCODE -ne 0) { throw "Could not generate Codex JSON schema" } +Write-Output ("Codex protocol artifacts generated at " + $OutputDirectory) diff --git a/src/adapters/codex.ts b/src/adapters/codex.ts index 0b95f27..603018f 100644 --- a/src/adapters/codex.ts +++ b/src/adapters/codex.ts @@ -1,13 +1,17 @@ import { execFileSync, spawn, spawnSync, type ChildProcessWithoutNullStreams } from "node:child_process"; import { createInterface, type Interface } from "node:readline"; -import type { CodexAdapter } from "../core/types"; +import type { CodexAdapter, CodexLifecycleEvent } from "../core/types"; +// Compatibility types mirror the generated protocol from codex-cli 0.152.0. +// Regenerate with: +// codex app-server generate-ts --out +// codex app-server generate-json-schema --out interface WireMessage { id?: number | string; method?: string; params?: Record; - result?: Record; - error?: { message?: string }; + result?: unknown; + error?: { message?: string; code?: number; data?: unknown }; } interface PendingRequest { @@ -24,6 +28,15 @@ export interface CodexProbe { reason?: string; } +function asRecord(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +function errorText(value: unknown): string { + if (value instanceof Error) return value.message; + return typeof value === "string" ? value : JSON.stringify(value); +} + export class CodexAppServerAdapter implements CodexAdapter { readonly provider = "codex-app-server" as const; private child: ChildProcessWithoutNullStreams | null = null; @@ -32,10 +45,15 @@ export class CodexAppServerAdapter implements CodexAdapter { private initialized = false; private authentication = "UNKNOWN"; private readonly pending = new Map(); - private readonly onNotification?: (message: WireMessage) => void; + private readonly terminalTurns = new Set(); + private eventHandler: (event: CodexLifecycleEvent) => void = () => undefined; - constructor(onNotification?: (message: WireMessage) => void) { - this.onNotification = onNotification; + constructor(onEvent?: (event: CodexLifecycleEvent) => void) { + if (onEvent) this.eventHandler = onEvent; + } + + setEventHandler(handler: (event: CodexLifecycleEvent) => void): void { + this.eventHandler = handler; } probe(): CodexProbe { @@ -45,7 +63,7 @@ export class CodexAppServerAdapter implements CodexAdapter { let authentication = this.authentication; try { const loginResult = spawnSync("codex", ["login", "status"], { encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] }); - const loginStatus = `${loginResult.stdout ?? ""}\n${loginResult.stderr ?? ""}`.trim().toLowerCase(); + const loginStatus = (String(loginResult.stdout ?? "") + "\n" + String(loginResult.stderr ?? "")).trim().toLowerCase(); if (loginStatus.includes("not logged") || loginStatus.includes("not authenticated")) authentication = "NOT_AUTHENTICATED"; else if (loginStatus.includes("logged in")) authentication = loginStatus.includes("chatgpt") ? "CHATGPT" : "AUTHENTICATED"; } catch { @@ -53,7 +71,7 @@ export class CodexAppServerAdapter implements CodexAdapter { } return { available: true, version, appServer: true, authentication }; } catch (error) { - return { available: false, version: null, appServer: false, authentication: "UNKNOWN", reason: error instanceof Error ? error.message : "codex unavailable" }; + return { available: false, version: null, appServer: false, authentication: "UNKNOWN", reason: errorText(error) }; } } @@ -61,7 +79,10 @@ export class CodexAppServerAdapter implements CodexAdapter { if (this.child && this.initialized) return; const probe = this.probe(); if (!probe.available) throw new Error("Codex App Server is not configured"); - this.child = spawn("codex", ["app-server", "--stdio"], { stdio: ["pipe", "pipe", "pipe"] }); + + // The installed server defaults to stdio. Keep the invocation aligned with + // the official CLI lifecycle instead of relying on an undocumented flag. + this.child = spawn("codex", ["app-server"], { stdio: ["pipe", "pipe", "pipe"], windowsHide: true }); this.lines = createInterface({ input: this.child.stdout }); this.lines.on("line", (line) => this.handleLine(line)); this.child.stderr.on("data", () => undefined); @@ -71,18 +92,26 @@ export class CodexAppServerAdapter implements CodexAdapter { this.child = null; this.lines?.close(); this.lines = null; - this.failPending(new Error(`Codex App Server exited (${code ?? "unknown"}/${signal ?? "no signal"})`)); + this.failPending(new Error("Codex App Server exited (" + (code ?? "unknown") + "/" + (signal ?? "no signal") + ")")); }); await this.request("initialize", { clientInfo: { name: "hiveforge", title: "HiveForge", version: "0.1.0" }, - capabilities: null + capabilities: { experimentalApi: false, requestAttestation: false } }); this.send({ method: "initialized" }); this.initialized = true; + try { - const auth = await this.request("getAuthStatus", { includeToken: false, refreshToken: false }); - this.authentication = typeof auth.authMethod === "string" ? auth.authMethod : "NOT_AUTHENTICATED"; + // account/read is the documented account/auth lifecycle method in the + // generated protocol. Do not probe the legacy getAuthStatus method. + const auth = await this.request("account/read", { refreshToken: false }); + const account = asRecord(auth.account); + if (Object.keys(account).length > 0) { + this.authentication = String(account.type ?? account.email ?? "AUTHENTICATED").toUpperCase(); + } else { + this.authentication = auth.requiresOpenaiAuth === true ? "NOT_AUTHENTICATED" : "UNKNOWN"; + } } catch { this.authentication = "UNKNOWN"; } @@ -96,21 +125,46 @@ export class CodexAppServerAdapter implements CodexAdapter { } catch { return; } - if (message.id !== undefined && (message.result || message.error)) { + + if (message.id !== undefined && (message.result !== undefined || message.error !== undefined)) { const pending = this.pending.get(message.id); if (!pending) return; clearTimeout(pending.timeout); this.pending.delete(message.id); if (message.error) pending.reject(new Error(message.error.message ?? "Codex App Server request failed")); - else pending.resolve(message.result ?? {}); + else pending.resolve(asRecord(message.result)); return; } - this.onNotification?.(message); + + const params = message.params ?? {}; + if (message.method === "turn/completed") { + const turn = asRecord(params.turn); + const threadId = typeof params.threadId === "string" ? params.threadId : ""; + const turnId = typeof turn.id === "string" ? turn.id : ""; + const status = turn.status; + if (threadId && turnId && (status === "completed" || status === "failed" || status === "interrupted") && !this.terminalTurns.has(turnId)) { + this.terminalTurns.add(turnId); + this.eventHandler({ + kind: "turn", + threadId, + turnId, + status, + error: turn.error ? errorText(turn.error) : undefined, + payload: params + }); + } + } + this.eventHandler({ + kind: "notification", + method: message.method ?? "unknown", + threadId: typeof params.threadId === "string" ? params.threadId : undefined, + payload: params + }); } private send(message: Record): void { if (!this.child?.stdin.writable) throw new Error("Codex App Server stdin is not writable"); - this.child.stdin.write(`${JSON.stringify(message)}\n`); + this.child.stdin.write(JSON.stringify(message) + "\n"); } private request(method: string, params: Record): Promise> { @@ -118,7 +172,7 @@ export class CodexAppServerAdapter implements CodexAdapter { return new Promise((resolve, reject) => { const timeout = setTimeout(() => { this.pending.delete(id); - reject(new Error(`Codex App Server timeout: ${method}`)); + reject(new Error("Codex App Server timeout: " + method)); }, 30_000); this.pending.set(id, { resolve, reject, timeout }); try { @@ -141,9 +195,9 @@ export class CodexAppServerAdapter implements CodexAdapter { async startThread(input: { cwd: string; projectId?: string }): Promise<{ threadId: string }> { await this.ensureStarted(); - const result = await this.request("thread/start", { cwd: input.cwd, projectId: input.projectId ?? null, ephemeral: false }); - const thread = result.thread as { id?: string } | undefined; - if (!thread?.id) throw new Error("Codex App Server did not return a thread id"); + const result = await this.request("thread/start", { cwd: input.cwd, ephemeral: false }); + const thread = asRecord(result.thread); + if (typeof thread.id !== "string" || thread.id.length === 0) throw new Error("Codex App Server did not return a thread id"); return { threadId: thread.id }; } @@ -154,8 +208,8 @@ export class CodexAppServerAdapter implements CodexAdapter { input: [{ type: "text", text: input.prompt, text_elements: [] }], cwd: input.cwd }); - const turn = result.turn as { id?: string } | undefined; - return { turnId: turn?.id ?? null }; + const turn = asRecord(result.turn); + return { turnId: typeof turn.id === "string" ? turn.id : null }; } async close(): Promise { @@ -166,5 +220,6 @@ export class CodexAppServerAdapter implements CodexAdapter { this.lines = null; this.initialized = false; this.authentication = "UNKNOWN"; + this.terminalTurns.clear(); } } diff --git a/src/adapters/fake-codex.ts b/src/adapters/fake-codex.ts index f15919e..5354fbe 100644 --- a/src/adapters/fake-codex.ts +++ b/src/adapters/fake-codex.ts @@ -1,19 +1,54 @@ import { randomUUID } from "node:crypto"; -import type { CodexAdapter } from "../core/types"; +import type { CodexAdapter, CodexLifecycleEvent } from "../core/types"; export class FakeCodexAdapter implements CodexAdapter { readonly provider = "fake" as const; readonly prompts: Array<{ threadId: string; prompt: string; cwd: string }> = []; readonly threads: string[] = []; + readonly turns: Array<{ turnId: string; threadId: string; prompt: string; cwd: string }> = []; + private eventHandler: (event: CodexLifecycleEvent) => void = () => undefined; + + setEventHandler(handler: (event: CodexLifecycleEvent) => void): void { + this.eventHandler = handler; + } async startThread(): Promise<{ threadId: string }> { - const threadId = `fake-thread-${randomUUID()}`; + const threadId = "fake-thread-" + randomUUID(); this.threads.push(threadId); return { threadId }; } async sendPrompt(input: { threadId: string; prompt: string; cwd: string }): Promise<{ turnId: string }> { this.prompts.push(input); - return { turnId: `fake-turn-${randomUUID()}` }; + const turnId = "fake-turn-" + randomUUID(); + this.turns.push({ turnId, ...input }); + return { turnId }; + } + + completeTurn(turnId = this.turns.at(-1)?.turnId): void { + if (!turnId) throw new Error("No fake Codex turn is available"); + const turn = this.turns.find((candidate) => candidate.turnId === turnId); + if (!turn) throw new Error("Fake Codex turn not found: " + turnId); + this.eventHandler({ + kind: "turn", + threadId: turn.threadId, + turnId, + status: "completed", + payload: { threadId: turn.threadId, turn: { id: turnId, status: "completed" } } + }); + } + + failTurn(turnId = this.turns.at(-1)?.turnId, status: "failed" | "interrupted" = "failed"): void { + if (!turnId) throw new Error("No fake Codex turn is available"); + const turn = this.turns.find((candidate) => candidate.turnId === turnId); + if (!turn) throw new Error("Fake Codex turn not found: " + turnId); + this.eventHandler({ + kind: "turn", + threadId: turn.threadId, + turnId, + status, + error: status === "failed" ? "fake Codex failure" : "fake Codex interruption", + payload: { threadId: turn.threadId, turnId, status } + }); } } diff --git a/src/adapters/git.ts b/src/adapters/git.ts index 41a3466..24d0ec5 100644 --- a/src/adapters/git.ts +++ b/src/adapters/git.ts @@ -13,6 +13,7 @@ export interface GitSnapshot { status: string; changedFiles: string[]; diffSummary: string; + isClean: boolean; } export class GitAdapter { @@ -26,26 +27,23 @@ export class GitAdapter { } assertRepository(localPath: string): void { - if (!this.isRepository(localPath)) throw new Error(`Not a Git repository: ${localPath}`); + if (!this.isRepository(localPath)) throw new Error("Not a Git repository: " + localPath); } - snapshot(localPath: string): GitSnapshot { + snapshot(localPath: string, capturedBaseSha?: string): GitSnapshot { this.assertRepository(localPath); const headSha = git(localPath, ["rev-parse", "HEAD"]); - let baseSha = headSha; - try { - baseSha = git(localPath, ["rev-parse", "HEAD^"]); - } catch { - // A first commit has no parent; using HEAD as base is deterministic. - } + const baseSha = capturedBaseSha ?? headSha; const branch = git(localPath, ["branch", "--show-current"]) || "DETACHED"; - const status = git(localPath, ["status", "--short"]) || "clean"; - const changedFiles = git(localPath, ["diff", "--name-only", `${baseSha}..${headSha}`]).split(/\r?\n/).filter(Boolean); - const diffSummary = git(localPath, ["diff", "--stat", `${baseSha}..${headSha}`]) || "No committed changes"; - return { baseSha, headSha, branch, status, changedFiles, diffSummary }; + const statusOutput = git(localPath, ["status", "--short"]); + const isClean = statusOutput.length === 0; + const status = isClean ? "clean" : statusOutput; + const changedFiles = git(localPath, ["diff", "--name-only", baseSha + ".." + headSha]).split(/\r?\n/).filter(Boolean); + const diffSummary = git(localPath, ["diff", "--stat", baseSha + ".." + headSha]) || "No committed changes"; + return { baseSha, headSha, branch, status, changedFiles, diffSummary, isClean }; } - capture(runId: string, localPath: string): Omit { - return { runId, ...this.snapshot(localPath) }; + capture(runId: string, localPath: string, capturedBaseSha?: string): Omit { + return { runId, ...this.snapshot(localPath, capturedBaseSha) }; } } diff --git a/src/core/evidence-lock.ts b/src/core/evidence-lock.ts index 3f7f25c..f450ffe 100644 --- a/src/core/evidence-lock.ts +++ b/src/core/evidence-lock.ts @@ -7,10 +7,12 @@ export interface EvidenceLockResult { export function verifyEvidenceLock(review: ReviewContract, project: Project, run: Run, evidence: GitEvidence): EvidenceLockResult { const errors: string[] = []; - if (review.project_id !== project.id) errors.push("project_id mismatch"); - if (review.run_id !== run.id) errors.push("run_id mismatch"); - if (review.work_order_id !== run.workOrderId) errors.push("work_order_id mismatch"); - if (review.head_sha !== evidence.headSha) errors.push("head_sha does not match captured Git evidence"); - if (review.base_sha !== evidence.baseSha) errors.push("base_sha does not match captured Git evidence"); + if (review.projectId !== project.id) errors.push("project_id mismatch"); + if (review.workOrderId !== run.workOrderId) errors.push("work_order_id mismatch"); + if (review.headSha !== evidence.headSha) errors.push("head_sha does not match captured Git evidence"); + if (review.baseSha !== evidence.baseSha) errors.push("base_sha does not match captured Git evidence"); + if (run.expectedHeadSha !== evidence.headSha) errors.push("stored run head SHA does not match captured Git evidence"); + if (run.expectedBaseSha !== evidence.baseSha) errors.push("stored run base SHA does not match captured Git evidence"); + if (!evidence.isClean) errors.push("working tree is not clean"); return { locked: errors.length === 0, errors }; } diff --git a/src/core/loop-guard.ts b/src/core/loop-guard.ts index 573bd00..0d61fa1 100644 --- a/src/core/loop-guard.ts +++ b/src/core/loop-guard.ts @@ -10,7 +10,7 @@ export interface LoopGuardResult { export function evaluateLoopGuard(run: Run, review: ReviewContract): LoopGuardResult { const nextIterationCount = run.iterationCount + 1; - const nextCorrectionCycles = run.correctionCycles + (review.action === "EXECUTE_CORRECTION" ? 1 : 0); + const nextCorrectionCycles = run.correctionCycles + (review.nextAction === "EXECUTE_CORRECTION" ? 1 : 0); const nextRepeatedFindings = { ...run.repeatedFindings }; for (const finding of review.findings) nextRepeatedFindings[finding.id] = (nextRepeatedFindings[finding.id] ?? 0) + 1; diff --git a/src/core/rcp.ts b/src/core/rcp.ts index 8bd3c35..d0c351e 100644 --- a/src/core/rcp.ts +++ b/src/core/rcp.ts @@ -1,8 +1,15 @@ -import type { ReviewAction, ReviewContract, ReviewDecision, ReviewFinding } from "./types"; +import type { ReviewContract, ReviewFinding, ReviewNextAction, ReviewVerdict } from "./types"; -const decisions = new Set(["APPROVED", "CORRECTION_REQUIRED", "BLOCKED", "COMPLETE"]); -const actions = new Set(["CONTINUE", "EXECUTE_CORRECTION", "BLOCKED", "COMPLETE"]); -const severities = new Set(["INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL"]); +const verdicts = new Set(["APPROVED", "CORRECTION_REQUIRED", "BLOCKED", "COMPLETE"]); +const actions = new Set(["CONTINUE", "EXECUTE_CORRECTION", "WAIT", "STOP"]); +const severities = new Set(["CRITICAL", "HIGH", "MEDIUM", "LOW"]); +const shaPattern = /^(?:[0-9a-f]{40}|[0-9a-f]{64})$/i; +const actionByVerdict: Record = { + APPROVED: "CONTINUE", + CORRECTION_REQUIRED: "EXECUTE_CORRECTION", + BLOCKED: "WAIT", + COMPLETE: "STOP" +}; export interface RcpValidation { valid: boolean; @@ -17,38 +24,48 @@ function isRecord(value: unknown): value is Record { function requiredString(data: Record, field: string, errors: string[]): string { const value = data[field]; if (typeof value !== "string" || value.trim().length === 0) { - errors.push(`${field} must be a non-empty string`); + errors.push(field + " must be a non-empty string"); return ""; } return value; } +function requiredSha(data: Record, field: string, errors: string[]): string { + const value = requiredString(data, field, errors); + if (value && !shaPattern.test(value)) errors.push(field + " must be a full 40 or 64 character Git SHA"); + return value; +} + export function validateReviewContract(input: unknown): RcpValidation { const errors: string[] = []; if (!isRecord(input)) return { valid: false, errors: ["RCP must be a JSON object"] }; - const protocol = input.protocol; - const version = input.version; - if (protocol !== "RCP") errors.push("protocol must be RCP"); - if (version !== "0.1") errors.push("version must be 0.1"); + for (const legacyField of ["protocol", "version", "run_id", "decision", "action"]) { + if (legacyField in input) errors.push(legacyField + " is not part of canonical RCP v0.1"); + } + + if (input.schema_version !== "0.1") errors.push("schema_version must be 0.1"); + const reviewId = requiredString(input, "review_id", errors); + const projectId = requiredString(input, "project_id", errors); + const workOrderId = requiredString(input, "work_order_id", errors); + const baseSha = requiredSha(input, "base_sha", errors); + const headSha = requiredSha(input, "head_sha", errors); + const summary = requiredString(input, "summary", errors); + const executorPrompt = requiredString(input, "executor_prompt", errors); + const checkpointNote = requiredString(input, "checkpoint_note", errors); - const review_id = requiredString(input, "review_id", errors); - const project_id = requiredString(input, "project_id", errors); - const run_id = requiredString(input, "run_id", errors); - const work_order_id = requiredString(input, "work_order_id", errors); - const head_sha = requiredString(input, "head_sha", errors); - const base_sha = requiredString(input, "base_sha", errors); - const executor_prompt = requiredString(input, "executor_prompt", errors); - const created_at = requiredString(input, "created_at", errors); + const progressPercent = input.progress_percent; + if (typeof progressPercent !== "number" || !Number.isFinite(progressPercent) || !Number.isInteger(progressPercent) || progressPercent < 0 || progressPercent > 100) { + errors.push("progress_percent must be an integer between 0 and 100"); + } - const decision = input.decision; - const action = input.action; - if (!decisions.has(decision as ReviewDecision)) errors.push("decision is not supported"); - if (!actions.has(action as ReviewAction)) errors.push("action is not supported"); - if (decision === "APPROVED" && action !== "CONTINUE") errors.push("APPROVED requires CONTINUE"); - if (decision === "CORRECTION_REQUIRED" && action !== "EXECUTE_CORRECTION") errors.push("CORRECTION_REQUIRED requires EXECUTE_CORRECTION"); - if (decision === "BLOCKED" && action !== "BLOCKED") errors.push("BLOCKED requires BLOCKED"); - if (decision === "COMPLETE" && action !== "COMPLETE") errors.push("COMPLETE requires COMPLETE"); + const verdict = input.verdict; + const nextAction = input.next_action; + if (!verdicts.has(verdict as ReviewVerdict)) errors.push("verdict is not supported"); + if (!actions.has(nextAction as ReviewNextAction)) errors.push("next_action is not supported"); + if (verdicts.has(verdict as ReviewVerdict) && actions.has(nextAction as ReviewNextAction) && actionByVerdict[verdict as ReviewVerdict] !== nextAction) { + errors.push(String(verdict) + " requires next_action " + actionByVerdict[verdict as ReviewVerdict]); + } const rawFindings = input.findings; const findings: ReviewFinding[] = []; @@ -57,14 +74,18 @@ export function validateReviewContract(input: unknown): RcpValidation { } else { rawFindings.forEach((raw, index) => { if (!isRecord(raw)) { - errors.push(`findings[${index}] must be an object`); + errors.push("findings[" + index + "] must be an object"); return; } - const id = requiredString(raw, "id", errors); - const summary = requiredString(raw, "summary", errors); + const id = typeof raw.id === "string" && raw.id.trim().length > 0 ? raw.id : ""; + const title = typeof raw.title === "string" && raw.title.trim().length > 0 ? raw.title : ""; + const details = typeof raw.details === "string" && raw.details.trim().length > 0 ? raw.details : ""; + if (!id) errors.push("findings[" + index + "].id must be a non-empty string"); + if (!title) errors.push("findings[" + index + "].title must be a non-empty string"); + if (!details) errors.push("findings[" + index + "].details must be a non-empty string"); const severity = raw.severity; - if (!severities.has(severity as ReviewFinding["severity"])) errors.push(`findings[${index}].severity is not supported`); - findings.push({ id, summary, severity: severity as ReviewFinding["severity"] }); + if (!severities.has(severity as ReviewFinding["severity"])) errors.push("findings[" + index + "].severity is not supported"); + findings.push({ id, title, details, severity: severity as ReviewFinding["severity"] }); }); } @@ -73,19 +94,46 @@ export function validateReviewContract(input: unknown): RcpValidation { valid: true, errors: [], value: { - protocol: "RCP", - version: "0.1", - review_id, - project_id, - run_id, - work_order_id, - decision: decision as ReviewDecision, - action: action as ReviewAction, - head_sha, - base_sha, + schemaVersion: "0.1", + reviewId, + projectId, + workOrderId, + baseSha, + headSha, + verdict: verdict as ReviewVerdict, + progressPercent: progressPercent as number, + summary, findings, - executor_prompt, - created_at + nextAction: nextAction as ReviewNextAction, + executorPrompt, + checkpointNote } }; } + +export function serializeReviewContract(review: ReviewContract): Record { + return { + schema_version: review.schemaVersion, + review_id: review.reviewId, + project_id: review.projectId, + work_order_id: review.workOrderId, + base_sha: review.baseSha, + head_sha: review.headSha, + verdict: review.verdict, + progress_percent: review.progressPercent, + summary: review.summary, + findings: review.findings.map((finding) => ({ id: finding.id, severity: finding.severity, title: finding.title, details: finding.details })), + next_action: review.nextAction, + executor_prompt: review.executorPrompt, + checkpoint_note: review.checkpointNote + }; +} + +export function reviewSemantics(): Record { + return { + APPROVED: "The reviewed increment is acceptable; CONTINUE may dispatch the executor subject to autonomy policy.", + CORRECTION_REQUIRED: "The reviewed increment needs correction; EXECUTE_CORRECTION may dispatch the supplied executor prompt subject to autonomy policy.", + BLOCKED: "The workflow is blocked and remains safe; WAIT records the blocker without autonomous dispatch.", + COMPLETE: "The work order is complete; STOP closes the run without dispatch." + }; +} diff --git a/src/core/state-machine.ts b/src/core/state-machine.ts index 26d4769..9deab63 100644 --- a/src/core/state-machine.ts +++ b/src/core/state-machine.ts @@ -3,13 +3,13 @@ import type { RunState } from "./types"; const transitions: Record = { IDLE: ["QUEUED", "FAILED"], QUEUED: ["CODEX_RUNNING", "PAUSED", "FAILED"], - CODEX_RUNNING: ["VALIDATING", "PAUSED", "FAILED"], - VALIDATING: ["GIT_SYNC", "FAILED"], - GIT_SYNC: ["WAITING_REVIEW", "FAILED"], + CODEX_RUNNING: ["VALIDATING", "PAUSED", "BLOCKED", "FAILED"], + VALIDATING: ["GIT_SYNC", "BLOCKED", "FAILED"], + GIT_SYNC: ["WAITING_REVIEW", "BLOCKED", "FAILED"], WAITING_REVIEW: ["REVIEW_RECEIVED", "PAUSED", "BLOCKED", "FAILED"], REVIEW_RECEIVED: ["WAITING_APPROVAL", "DISPATCHING_NEXT", "BLOCKED", "COMPLETE", "FAILED"], WAITING_APPROVAL: ["DISPATCHING_NEXT", "PAUSED", "BLOCKED", "FAILED"], - DISPATCHING_NEXT: ["CODEX_RUNNING", "FAILED"], + DISPATCHING_NEXT: ["CODEX_RUNNING", "BLOCKED", "FAILED"], PAUSED: ["QUEUED", "CODEX_RUNNING", "FAILED"], BLOCKED: ["QUEUED", "FAILED"], COMPLETE: [], diff --git a/src/core/types.ts b/src/core/types.ts index d64469a..4e5c883 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -18,9 +18,12 @@ export const RUN_STATES = [ ] as const; export type RunState = (typeof RUN_STATES)[number]; -export type ReviewDecision = "APPROVED" | "CORRECTION_REQUIRED" | "BLOCKED" | "COMPLETE"; -export type ReviewAction = "CONTINUE" | "EXECUTE_CORRECTION" | "BLOCKED" | "COMPLETE"; +export type ReviewVerdict = "APPROVED" | "CORRECTION_REQUIRED" | "BLOCKED" | "COMPLETE"; +export type ReviewNextAction = "CONTINUE" | "EXECUTE_CORRECTION" | "WAIT" | "STOP"; export type ReviewStatus = "RECEIVED" | "ACCEPTED" | "REJECTED" | "CONSUMED"; +export type ValidationKind = "test" | "lint" | "typecheck" | "build"; +export type ValidationStatus = "PASS" | "FAIL" | "NOT_CONFIGURED" | "ERROR"; +export type CodexTurnStatus = "PREPARED" | "IN_PROGRESS" | "COMPLETED" | "FAILED" | "INTERRUPTED" | "CANCELLED" | "ORPHANED"; export interface ProjectInput { name: string; @@ -55,9 +58,17 @@ export interface Run { correctionCycles: number; repeatedFindings: Record; currentPrompt: string | null; + runBaseSha: string | null; + runBaseBranch: string | null; expectedBaseSha: string | null; expectedHeadSha: string | null; lastReviewId: string | null; + lastReviewStatus: ReviewStatus | null; + lastReviewVerdict: ReviewVerdict | null; + lastReviewSummary: string | null; + lastCheckpointNote: string | null; + progressPercent: number; + currentBlocker: string | null; createdAt: string; updatedAt: string; } @@ -81,29 +92,31 @@ export interface GitEvidence { status: string; changedFiles: string[]; diffSummary: string; + isClean: boolean; capturedAt: string; } export interface ReviewFinding { id: string; - severity: "INFO" | "LOW" | "MEDIUM" | "HIGH" | "CRITICAL"; - summary: string; + severity: "CRITICAL" | "HIGH" | "MEDIUM" | "LOW"; + title: string; + details: string; } export interface ReviewContract { - protocol: "RCP"; - version: "0.1"; - review_id: string; - project_id: string; - run_id: string; - work_order_id: string; - decision: ReviewDecision; - action: ReviewAction; - head_sha: string; - base_sha: string; + schemaVersion: "0.1"; + reviewId: string; + projectId: string; + workOrderId: string; + baseSha: string; + headSha: string; + verdict: ReviewVerdict; + progressPercent: number; + summary: string; findings: ReviewFinding[]; - executor_prompt: string; - created_at: string; + nextAction: ReviewNextAction; + executorPrompt: string; + checkpointNote: string; } export interface CodexThread { @@ -116,8 +129,70 @@ export interface CodexThread { updatedAt: string; } +export interface CodexTurn { + id: string; + runId: string; + threadId: string; + turnId: string; + dispatchKey: string; + prompt: string; + status: CodexTurnStatus; + startedAt: string; + finishedAt: string | null; + error: string | null; + validationStartedAt: string | null; + createdAt: string; + updatedAt: string; +} + +export interface ValidationResult { + id: string; + runId: string; + turnId: string; + kind: ValidationKind; + command: string | null; + startedAt: string; + finishedAt: string; + exitCode: number | null; + stdout: string; + stderr: string; + status: ValidationStatus; +} + +export interface ReviewSnapshot { + status: ReviewStatus; + review: ReviewContract; +} + +export interface CodexTurnLifecycleEvent { + kind: "turn"; + threadId: string; + turnId: string; + status: "completed" | "failed" | "interrupted"; + error?: string; + payload: Record; +} + +export interface CodexNotificationEvent { + kind: "notification"; + method: string; + threadId?: string; + payload: Record; +} + +export type CodexLifecycleEvent = CodexTurnLifecycleEvent | CodexNotificationEvent; + +export interface RunOverview { + run: Run; + gitEvidence: GitEvidence | null; + validations: ValidationResult[]; + review: ReviewSnapshot | null; + events: RunEvent[]; +} + export interface CodexAdapter { readonly provider: "codex-app-server" | "fake"; + setEventHandler(handler: (event: CodexLifecycleEvent) => void): void; startThread(input: { cwd: string; projectId?: string }): Promise<{ threadId: string }>; sendPrompt(input: { threadId: string; prompt: string; cwd: string }): Promise<{ turnId: string | null }>; close?(): Promise; diff --git a/src/db/database.ts b/src/db/database.ts index 8e7aa2e..b0b2df6 100644 --- a/src/db/database.ts +++ b/src/db/database.ts @@ -3,7 +3,8 @@ import { mkdirSync } from "node:fs"; import { dirname, join } from "node:path"; import { DatabaseSync, type SQLInputValue } from "node:sqlite"; import { migrations } from "./migrations"; -import type { AutonomyMode, CodexThread, GitEvidence, Project, ProjectInput, ReviewContract, ReviewStatus, Run, RunEvent, RunState } from "../core/types"; +import type { AutonomyMode, CodexThread, CodexTurn, CodexTurnStatus, GitEvidence, Project, ProjectInput, ReviewContract, ReviewSnapshot, ReviewStatus, Run, RunEvent, RunOverview, RunState, ValidationKind, ValidationResult, ValidationStatus } from "../core/types"; +import { serializeReviewContract, validateReviewContract } from "../core/rcp"; type Row = Record; @@ -30,6 +31,15 @@ function mapProject(row: Row): Project { }; } +function parseJson(value: unknown, fallback: T): T { + if (typeof value !== "string") return fallback; + try { + return JSON.parse(value) as T; + } catch { + return fallback; + } +} + function mapRun(row: Row): Run { return { id: String(row.id), @@ -42,11 +52,19 @@ function mapRun(row: Row): Run { repeatedFindingThreshold: Number(row.repeated_finding_threshold), iterationCount: Number(row.iteration_count), correctionCycles: Number(row.correction_cycles), - repeatedFindings: JSON.parse(String(row.repeated_findings_json)) as Record, + repeatedFindings: parseJson>(row.repeated_findings_json, {}), currentPrompt: row.current_prompt ? String(row.current_prompt) : null, + runBaseSha: row.run_base_sha ? String(row.run_base_sha) : null, + runBaseBranch: row.run_base_branch ? String(row.run_base_branch) : null, expectedBaseSha: row.expected_base_sha ? String(row.expected_base_sha) : null, expectedHeadSha: row.expected_head_sha ? String(row.expected_head_sha) : null, lastReviewId: row.last_review_id ? String(row.last_review_id) : null, + lastReviewStatus: row.last_review_status ? row.last_review_status as ReviewStatus : null, + lastReviewVerdict: row.last_review_verdict ? String(row.last_review_verdict) as Run["lastReviewVerdict"] : null, + lastReviewSummary: row.last_review_summary ? String(row.last_review_summary) : null, + lastCheckpointNote: row.last_checkpoint_note ? String(row.last_checkpoint_note) : null, + progressPercent: Number(row.progress_percent ?? 0), + currentBlocker: row.current_blocker ? String(row.current_blocker) : null, createdAt: String(row.created_at), updatedAt: String(row.updated_at) }; @@ -132,6 +150,11 @@ export class Database { return row ? mapRun(row) : undefined; } + getRunByWorkOrderId(workOrderId: string): Run | undefined { + const row = this.get("SELECT * FROM runs WHERE work_order_id = ?", workOrderId); + return row ? mapRun(row) : undefined; + } + listNonTerminalRuns(): Run[] { return this.all("SELECT * FROM runs WHERE state NOT IN ('COMPLETE', 'FAILED') ORDER BY updated_at ASC").map(mapRun); } @@ -144,18 +167,27 @@ export class Database { updateRun(run: Run): Run { const timestamp = now(); this.run( - `UPDATE runs SET state = ?, iteration_count = ?, correction_cycles = ?, repeated_findings_json = ?, current_prompt = ?, expected_base_sha = ?, expected_head_sha = ?, last_review_id = ?, updated_at = ? WHERE id = ?`, - run.state, run.iterationCount, run.correctionCycles, JSON.stringify(run.repeatedFindings), run.currentPrompt, run.expectedBaseSha, run.expectedHeadSha, run.lastReviewId, timestamp, run.id + `UPDATE runs SET state = ?, iteration_count = ?, correction_cycles = ?, repeated_findings_json = ?, current_prompt = ?, run_base_sha = ?, run_base_branch = ?, expected_base_sha = ?, expected_head_sha = ?, last_review_id = ?, last_review_status = ?, last_review_verdict = ?, last_review_summary = ?, last_checkpoint_note = ?, progress_percent = ?, current_blocker = ?, updated_at = ? WHERE id = ?`, + run.state, run.iterationCount, run.correctionCycles, JSON.stringify(run.repeatedFindings), run.currentPrompt, run.runBaseSha, run.runBaseBranch, run.expectedBaseSha, run.expectedHeadSha, run.lastReviewId, run.lastReviewStatus, run.lastReviewVerdict, run.lastReviewSummary, run.lastCheckpointNote, run.progressPercent, run.currentBlocker, timestamp, run.id ); return this.getRun(run.id)!; } + setRunBase(runId: string, baseSha: string, branch: string): Run { + this.run("UPDATE runs SET run_base_sha = ?, run_base_branch = ?, updated_at = ? WHERE id = ?", baseSha, branch, now(), runId); + return this.getRun(runId)!; + } + appendEvent(runId: string, type: string, message: string, state: RunState | null, payload: Record = {}): RunEvent { const event: RunEvent = { id: randomUUID(), runId, type, state, message, payload, createdAt: now() }; this.run("INSERT INTO run_events(id, run_id, type, state, message, payload_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", event.id, runId, type, state, message, JSON.stringify(payload), event.createdAt); return event; } + hasEvent(runId: string, type: string): boolean { + return Boolean(this.get("SELECT id FROM run_events WHERE run_id = ? AND type = ? LIMIT 1", runId, type)); + } + appendMailboxEvent(filePath: string, kind: string, message: string): void { this.run("INSERT INTO mailbox_events(id, file_path, kind, message, created_at) VALUES (?, ?, ?, ?, ?)", randomUUID(), filePath, kind, message, now()); } @@ -166,6 +198,12 @@ export class Database { })); } + listEventsAfter(runId: string, createdAt: string, eventId: string, limit = 100): RunEvent[] { + return this.all("SELECT * FROM run_events WHERE run_id = ? AND (created_at > ? OR (created_at = ? AND id > ?)) ORDER BY created_at ASC, id ASC LIMIT ?", runId, createdAt, createdAt, eventId, limit).map((row) => ({ + id: String(row.id), runId: String(row.run_id), type: String(row.type), state: row.state ? row.state as RunState : null, message: String(row.message), payload: JSON.parse(String(row.payload_json)) as Record, createdAt: String(row.created_at) + })); + } + saveThread(input: { runId: string; provider: CodexThread["provider"]; threadId: string; status: string }): CodexThread { const id = randomUUID(); const timestamp = now(); @@ -175,19 +213,19 @@ export class Database { saveGitEvidence(input: Omit): GitEvidence { const evidence: GitEvidence = { ...input, id: randomUUID(), capturedAt: now() }; - this.run("INSERT INTO git_evidence(id, run_id, base_sha, head_sha, branch, status, changed_files_json, diff_summary, captured_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", evidence.id, evidence.runId, evidence.baseSha, evidence.headSha, evidence.branch, evidence.status, JSON.stringify(evidence.changedFiles), evidence.diffSummary, evidence.capturedAt); + this.run("INSERT INTO git_evidence(id, run_id, base_sha, head_sha, branch, status, changed_files_json, diff_summary, is_clean, captured_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", evidence.id, evidence.runId, evidence.baseSha, evidence.headSha, evidence.branch, evidence.status, JSON.stringify(evidence.changedFiles), evidence.diffSummary, evidence.isClean ? 1 : 0, evidence.capturedAt); return evidence; } getLatestGitEvidence(runId: string): GitEvidence | undefined { const row = this.get("SELECT * FROM git_evidence WHERE run_id = ? ORDER BY captured_at DESC LIMIT 1", runId); if (!row) return undefined; - return { id: String(row.id), runId: String(row.run_id), baseSha: String(row.base_sha), headSha: String(row.head_sha), branch: String(row.branch), status: String(row.status), changedFiles: JSON.parse(String(row.changed_files_json)) as string[], diffSummary: String(row.diff_summary), capturedAt: String(row.captured_at) }; + return { id: String(row.id), runId: String(row.run_id), baseSha: String(row.base_sha), headSha: String(row.head_sha), branch: String(row.branch), status: String(row.status), changedFiles: parseJson(row.changed_files_json, []), diffSummary: String(row.diff_summary), isClean: Number(row.is_clean ?? 1) === 1, capturedAt: String(row.captured_at) }; } saveReview(review: ReviewContract, status: ReviewStatus): boolean { try { - this.run("INSERT INTO reviews(id, run_id, project_id, work_order_id, decision, action, head_sha, base_sha, status, payload_json, processed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", review.review_id, review.run_id, review.project_id, review.work_order_id, review.decision, review.action, review.head_sha, review.base_sha, status, JSON.stringify(review), now()); + this.run("INSERT INTO reviews(id, run_id, project_id, work_order_id, decision, action, head_sha, base_sha, status, payload_json, processed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", review.reviewId, this.getRunByWorkOrderId(review.workOrderId)?.id ?? null, review.projectId, review.workOrderId, review.verdict, review.nextAction, review.headSha, review.baseSha, status, JSON.stringify(serializeReviewContract(review)), now()); return true; } catch (error) { if (error instanceof Error && error.message.includes("UNIQUE")) return false; @@ -198,7 +236,21 @@ export class Database { getReview(reviewId: string): { review: ReviewContract; status: ReviewStatus } | undefined { const row = this.get("SELECT payload_json, status FROM reviews WHERE id = ?", reviewId); if (!row) return undefined; - return { review: JSON.parse(String(row.payload_json)) as ReviewContract, status: row.status as ReviewStatus }; + const parsed = validateReviewContract(parseJson(row.payload_json, null)); + if (!parsed.valid || !parsed.value) throw new Error(`Stored review ${reviewId} is not canonical: ${parsed.errors.join(", ")}`); + return { review: parsed.value, status: row.status as ReviewStatus }; + } + + getLatestReview(runId: string): ReviewSnapshot | null { + const row = this.get("SELECT id, payload_json, status FROM reviews WHERE run_id = ? ORDER BY processed_at DESC LIMIT 1", runId); + if (!row) return null; + const parsed = validateReviewContract(parseJson(row.payload_json, null)); + if (!parsed.valid || !parsed.value) return null; + return { review: parsed.value, status: row.status as ReviewStatus }; + } + + updateReviewStatus(reviewId: string, status: ReviewStatus): void { + this.run("UPDATE reviews SET status = ? WHERE id = ?", status, reviewId); } createApproval(reviewId: string): void { @@ -214,6 +266,24 @@ export class Database { return Boolean(this.get("SELECT id FROM dispatches WHERE run_id = ? AND dispatch_key = ?", runId, dispatchKey)); } + getDispatch(runId: string, dispatchKey: string): { id: string; status: string; threadId: string | null; turnId: string | null; reviewId: string | null; prompt: string } | undefined { + const row = this.get("SELECT * FROM dispatches WHERE run_id = ? AND dispatch_key = ?", runId, dispatchKey); + if (!row) return undefined; + return { + id: String(row.id), status: String(row.status), threadId: row.thread_id ? String(row.thread_id) : null, + turnId: row.turn_id ? String(row.turn_id) : null, reviewId: row.review_id ? String(row.review_id) : null, prompt: String(row.prompt) + }; + } + + getThreadForRun(runId: string): CodexThread | undefined { + const row = this.get("SELECT * FROM codex_threads WHERE run_id = ? ORDER BY created_at ASC LIMIT 1", runId); + if (!row) return undefined; + return { + id: String(row.id), runId: String(row.run_id), provider: row.provider as CodexThread["provider"], threadId: String(row.thread_id), + status: String(row.status), createdAt: String(row.created_at), updatedAt: String(row.updated_at) + }; + } + recordDispatch(runId: string, reviewId: string | null, dispatchKey: string, prompt: string): boolean { try { this.run("INSERT INTO dispatches(id, run_id, review_id, dispatch_key, prompt, status, sent_at) VALUES (?, ?, ?, ?, ?, 'SENT', ?)", randomUUID(), runId, reviewId, dispatchKey, prompt, now()); @@ -223,4 +293,89 @@ export class Database { throw error; } } + + prepareDispatch(runId: string, reviewId: string | null, dispatchKey: string, prompt: string, threadId: string): boolean { + try { + this.run("INSERT INTO dispatches(id, run_id, review_id, dispatch_key, prompt, status, sent_at, thread_id) VALUES (?, ?, ?, ?, ?, 'PREPARED', ?, ?)", randomUUID(), runId, reviewId, dispatchKey, prompt, now(), threadId); + return true; + } catch (error) { + if (error instanceof Error && error.message.includes("UNIQUE")) return false; + throw error; + } + } + + markDispatchSent(runId: string, dispatchKey: string, turnId: string): boolean { + const result = this.run("UPDATE dispatches SET status = 'SENT', turn_id = ?, sent_at = ? WHERE run_id = ? AND dispatch_key = ? AND status = 'PREPARED'", turnId, now(), runId, dispatchKey); + return result.changes > 0; + } + + markDispatchFailed(runId: string, dispatchKey: string, message: string): boolean { + const result = this.run("UPDATE dispatches SET status = 'FAILED', prompt = prompt || ? WHERE run_id = ? AND dispatch_key = ? AND status = 'PREPARED'", `\n[dispatch-error] ${message}`, runId, dispatchKey); + return result.changes > 0; + } + + createCodexTurn(input: { runId: string; threadId: string; turnId: string; dispatchKey: string; prompt: string; status?: CodexTurnStatus }): CodexTurn { + const id = randomUUID(); + const timestamp = now(); + const status = input.status ?? "IN_PROGRESS"; + this.run("INSERT INTO codex_turns(id, run_id, thread_id, turn_id, dispatch_key, prompt, status, started_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", id, input.runId, input.threadId, input.turnId, input.dispatchKey, input.prompt, status, timestamp, timestamp, timestamp); + return this.getCodexTurn(input.turnId)!; + } + + getCodexTurn(turnId: string): CodexTurn | undefined { + const row = this.get("SELECT * FROM codex_turns WHERE turn_id = ?", turnId); + return row ? this.mapCodexTurn(row) : undefined; + } + + getCodexTurnByIdentity(threadId: string, turnId: string): CodexTurn | undefined { + const row = this.get("SELECT * FROM codex_turns WHERE thread_id = ? AND turn_id = ?", threadId, turnId); + return row ? this.mapCodexTurn(row) : undefined; + } + + getActiveCodexTurn(runId: string): CodexTurn | undefined { + const row = this.get("SELECT * FROM codex_turns WHERE run_id = ? AND status IN ('PREPARED', 'IN_PROGRESS') ORDER BY started_at DESC LIMIT 1", runId); + return row ? this.mapCodexTurn(row) : undefined; + } + + updateCodexTurnStatus(turnId: string, status: CodexTurnStatus, error: string | null = null): CodexTurn | undefined { + const finishedAt = ["COMPLETED", "FAILED", "INTERRUPTED", "CANCELLED", "ORPHANED"].includes(status) ? now() : null; + this.run("UPDATE codex_turns SET status = ?, error = ?, finished_at = COALESCE(?, finished_at), updated_at = ? WHERE turn_id = ?", status, error, finishedAt, now(), turnId); + return this.getCodexTurn(turnId); + } + + beginTurnValidation(turnId: string): boolean { + const result = this.run("UPDATE codex_turns SET validation_started_at = ?, updated_at = ? WHERE turn_id = ? AND validation_started_at IS NULL", now(), now(), turnId); + return result.changes > 0; + } + + saveValidationResult(input: Omit): ValidationResult { + const result: ValidationResult = { ...input, id: randomUUID() }; + this.run("INSERT INTO validation_results(id, run_id, turn_id, kind, command, started_at, finished_at, exit_code, stdout, stderr, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", result.id, result.runId, result.turnId, result.kind, result.command, result.startedAt, result.finishedAt, result.exitCode, result.stdout, result.stderr, result.status); + return result; + } + + getValidationResults(runId: string, turnId?: string): ValidationResult[] { + const rows = turnId + ? this.all("SELECT * FROM validation_results WHERE run_id = ? AND turn_id = ? ORDER BY finished_at ASC", runId, turnId) + : this.all("SELECT * FROM validation_results WHERE run_id = ? ORDER BY finished_at ASC", runId); + return rows.map((row) => ({ + id: String(row.id), runId: String(row.run_id), turnId: String(row.turn_id), kind: row.kind as ValidationKind, + command: row.command ? String(row.command) : null, startedAt: String(row.started_at), finishedAt: String(row.finished_at), + exitCode: row.exit_code === null || row.exit_code === undefined ? null : Number(row.exit_code), stdout: String(row.stdout ?? ""), stderr: String(row.stderr ?? ""), status: row.status as ValidationStatus + })); + } + + getRunOverview(runId: string): RunOverview | undefined { + const run = this.getRun(runId); + if (!run) return undefined; + return { run, gitEvidence: this.getLatestGitEvidence(runId) ?? null, validations: this.getValidationResults(runId), review: this.getLatestReview(runId), events: this.listEvents(runId, 250) }; + } + + private mapCodexTurn(row: Row): CodexTurn { + return { + id: String(row.id), runId: String(row.run_id), threadId: String(row.thread_id), turnId: String(row.turn_id), dispatchKey: String(row.dispatch_key), prompt: String(row.prompt), + status: row.status as CodexTurnStatus, startedAt: String(row.started_at), finishedAt: row.finished_at ? String(row.finished_at) : null, error: row.error ? String(row.error) : null, + validationStartedAt: row.validation_started_at ? String(row.validation_started_at) : null, createdAt: String(row.created_at), updatedAt: String(row.updated_at) + }; + } } diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 600b2cf..3de5fe6 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -115,5 +115,53 @@ export const migrations: Array<{ version: number; sql: string }> = [ created_at TEXT NOT NULL ); ` + }, + { + version: 3, + sql: ` + ALTER TABLE runs ADD COLUMN run_base_sha TEXT; + ALTER TABLE runs ADD COLUMN run_base_branch TEXT; + ALTER TABLE runs ADD COLUMN last_review_status TEXT; + ALTER TABLE runs ADD COLUMN last_review_verdict TEXT; + ALTER TABLE runs ADD COLUMN last_review_summary TEXT; + ALTER TABLE runs ADD COLUMN last_checkpoint_note TEXT; + ALTER TABLE runs ADD COLUMN progress_percent INTEGER NOT NULL DEFAULT 0; + ALTER TABLE runs ADD COLUMN current_blocker TEXT; + ALTER TABLE git_evidence ADD COLUMN is_clean INTEGER NOT NULL DEFAULT 1; + ALTER TABLE dispatches ADD COLUMN thread_id TEXT; + ALTER TABLE dispatches ADD COLUMN turn_id TEXT; + CREATE TABLE IF NOT EXISTS codex_turns ( + id TEXT PRIMARY KEY, + run_id TEXT NOT NULL REFERENCES runs(id), + thread_id TEXT NOT NULL, + turn_id TEXT NOT NULL UNIQUE, + dispatch_key TEXT NOT NULL, + prompt TEXT NOT NULL, + status TEXT NOT NULL, + started_at TEXT NOT NULL, + finished_at TEXT, + error TEXT, + validation_started_at TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(run_id, dispatch_key) + ); + CREATE INDEX IF NOT EXISTS idx_codex_turns_run_status ON codex_turns(run_id, status); + CREATE TABLE IF NOT EXISTS validation_results ( + id TEXT PRIMARY KEY, + run_id TEXT NOT NULL REFERENCES runs(id), + turn_id TEXT NOT NULL REFERENCES codex_turns(turn_id), + kind TEXT NOT NULL, + command TEXT, + started_at TEXT NOT NULL, + finished_at TEXT NOT NULL, + exit_code INTEGER, + stdout TEXT NOT NULL DEFAULT '', + stderr TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + UNIQUE(run_id, turn_id, kind) + ); + CREATE INDEX IF NOT EXISTS idx_validation_results_run ON validation_results(run_id, finished_at); + ` } ]; diff --git a/src/mailbox/registry.ts b/src/mailbox/registry.ts new file mode 100644 index 0000000..24a6f9f --- /dev/null +++ b/src/mailbox/registry.ts @@ -0,0 +1,43 @@ +import { ReviewMailboxWatcher, type MailboxCallbacks } from "./watcher"; +import type { Project } from "../core/types"; + +interface RegisteredWatcher { + mailboxPath: string; + watcher: ReviewMailboxWatcher; +} + +export class MailboxWatcherRegistry { + private readonly watchers = new Map(); + + constructor(private readonly callbacksFor: (project: Project) => MailboxCallbacks) {} + + async reconcile(projects: Project[]): Promise { + const wanted = new Map(projects.map((project) => [project.id, project])); + for (const [projectId, registered] of this.watchers) { + const project = wanted.get(projectId); + if (!project || project.reviewMailboxPath !== registered.mailboxPath) { + registered.watcher.stop(); + this.watchers.delete(projectId); + } + } + for (const project of projects) { + if (this.watchers.has(project.id)) continue; + const watcher = new ReviewMailboxWatcher(project.reviewMailboxPath, this.callbacksFor(project)); + await watcher.start(); + this.watchers.set(project.id, { mailboxPath: project.reviewMailboxPath, watcher }); + } + } + + async scanAll(): Promise { + await Promise.all([...this.watchers.values()].map(({ watcher }) => watcher.scan())); + } + + stopAll(): void { + for (const { watcher } of this.watchers.values()) watcher.stop(); + this.watchers.clear(); + } + + get size(): number { + return this.watchers.size; + } +} diff --git a/src/server/runtime.ts b/src/server/runtime.ts index 93f3cad..c599c60 100644 --- a/src/server/runtime.ts +++ b/src/server/runtime.ts @@ -20,13 +20,7 @@ export function getRuntime(): HiveforgeRuntime { const events = new EventBus(); const db = new Database(); const git = new GitAdapter(); - const codex = new CodexAppServerAdapter((message) => { - const params = message.params ?? {}; - const runId = typeof params.threadId === "string" ? db.get<{ run_id: string }>("SELECT run_id FROM codex_threads WHERE thread_id = ?", params.threadId)?.run_id : undefined; - if (!runId) return; - const event = db.appendEvent(runId, `CODEX_${(message.method ?? "EVENT").replace(/[^A-Z0-9]+/gi, "_").toUpperCase()}`, `Codex event: ${message.method ?? "unknown"}`, db.getRun(runId)?.state ?? null, params); - events.publish(event); - }); + const codex = new CodexAppServerAdapter(); const engine = new WorkflowEngine(db, git, events, codex); globalRuntime.__hiveforgeRuntime = { db, events, git, codex, engine }; engine.recover(); diff --git a/src/worker.ts b/src/worker.ts index efa2796..719b5c3 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -1,30 +1,39 @@ -import { ReviewMailboxWatcher } from "./mailbox/watcher"; +import { MailboxWatcherRegistry } from "./mailbox/registry"; import { getRuntime } from "./server/runtime"; async function main() { const runtime = getRuntime(); - const watchers: ReviewMailboxWatcher[] = []; - for (const project of runtime.db.listProjects()) { - const watcher = new ReviewMailboxWatcher(project.reviewMailboxPath, { - onReview: async (review) => { - const result = await runtime.engine.processReview(review); - if (!result.accepted && result.run) { - const event = runtime.db.appendEvent(result.run.id, "REVIEW_REJECTED", result.errors.join("; "), result.run.state); - runtime.events.publish(event); - } - }, - onRejected: async (input) => { - runtime.db.appendMailboxEvent(input.file, "REJECTED", input.errors.join("; ")); - console.warn(`[hiveforge] mailbox rejected ${input.file}: ${input.errors.join("; ")}`); + const registry = new MailboxWatcherRegistry(() => ({ + onReview: async (review) => { + const result = await runtime.engine.processReview(review); + if (!result.accepted && result.run) { + const event = runtime.db.appendEvent(result.run.id, "REVIEW_REJECTED", result.errors.join("; "), result.run.state); + runtime.events.publish(event); } - }); - await watcher.start(); - watchers.push(watcher); - } - process.on("SIGINT", () => { watchers.forEach((watcher) => watcher.stop()); process.exit(0); }); - process.on("SIGTERM", () => { watchers.forEach((watcher) => watcher.stop()); process.exit(0); }); - console.log(`HiveForge worker watching ${watchers.length} mailbox(es)`); - setInterval(() => { for (const watcher of watchers) void watcher.scan(); }, 5_000); + }, + onRejected: async (input) => { + runtime.db.appendMailboxEvent(input.file, "REJECTED", input.errors.join("; ")); + console.warn("[hiveforge] mailbox rejected " + input.file + ": " + input.errors.join("; ")); + } + })); + + const reconcile = async () => { + try { + await registry.reconcile(runtime.db.listProjects()); + } catch (error) { + console.error("[hiveforge] mailbox reconciliation failed", error); + } + }; + await reconcile(); + const interval = setInterval(() => { void reconcile(); void registry.scanAll(); }, 5_000); + const shutdown = () => { + clearInterval(interval); + registry.stopAll(); + process.exit(0); + }; + process.on("SIGINT", shutdown); + process.on("SIGTERM", shutdown); + console.log("HiveForge worker watching " + registry.size + " mailbox(es); dynamic reconciliation enabled"); } void main().catch((error) => { console.error(error); process.exitCode = 1; }); diff --git a/src/workflow/engine.ts b/src/workflow/engine.ts index 1bec872..f7f5fb5 100644 --- a/src/workflow/engine.ts +++ b/src/workflow/engine.ts @@ -1,19 +1,28 @@ -import type { CodexAdapter } from "../core/types"; +import type { CodexAdapter, CodexLifecycleEvent, CodexTurn, ReviewContract, Run, RunEvent } from "../core/types"; import { evaluateLoopGuard } from "../core/loop-guard"; -import { validateReviewContract } from "../core/rcp"; +import { serializeReviewContract, validateReviewContract } from "../core/rcp"; +import { verifyEvidenceLock } from "../core/evidence-lock"; import { transition } from "../core/state-machine"; -import type { ReviewContract, Run, RunEvent } from "../core/types"; import { Database } from "../db/database"; import { GitAdapter } from "../adapters/git"; import { EventBus } from "../core/event-bus"; +import { ValidationRunner } from "./validation"; + +const terminalTurnStatuses = new Set(["COMPLETED", "FAILED", "INTERRUPTED", "CANCELLED", "ORPHANED"]); export class WorkflowEngine { + private readonly validation: ValidationRunner; + constructor( private readonly db: Database, private readonly git: GitAdapter, private readonly events: EventBus, - private readonly adapter: CodexAdapter - ) {} + private readonly adapter: CodexAdapter, + validation = new ValidationRunner(db) + ) { + this.validation = validation; + this.adapter.setEventHandler((event) => this.handleAdapterEvent(event)); + } private event(runId: string, type: string, message: string, state: Run["state"] | null, payload: Record = {}): RunEvent { const event = this.db.appendEvent(runId, type, message, state, payload); @@ -22,98 +31,214 @@ export class WorkflowEngine { } private move(run: Run, next: Run["state"], message: string, type = "STATE_CHANGED"): Run { + if (run.state === next) return run; run.state = transition(run.state, next); const updated = this.db.updateRun(run); this.event(run.id, type, message, next); return updated; } + private block(run: Run, message: string, type = "RUN_BLOCKED"): Run { + run.currentBlocker = message; + run = this.db.updateRun(run); + if (run.state !== "BLOCKED" && run.state !== "COMPLETE" && run.state !== "FAILED") { + run = this.move(run, "BLOCKED", message, type); + } else { + this.event(run.id, type, message, run.state); + } + return run; + } + async start(runId: string, prompt: string): Promise { let run = this.requireRun(runId); const project = this.requireProject(run.projectId); - if (run.state !== "IDLE" && run.state !== "PAUSED" && run.state !== "BLOCKED") throw new Error(`Run cannot start from ${run.state}`); + if (run.state !== "IDLE" && run.state !== "PAUSED" && run.state !== "BLOCKED") throw new Error("Run cannot start from " + run.state); + + const snapshot = this.git.snapshot(project.localPath, run.runBaseSha ?? undefined); + if (!snapshot.isClean) throw new Error("Run must start from a clean committed Git state"); + if (!run.runBaseSha) { + run = this.db.setRunBase(run.id, snapshot.headSha, snapshot.branch); + this.event(run.id, "RUN_BASE_CAPTURED", "Captured immutable Git base at work-order start", run.state, { baseSha: snapshot.headSha, branch: snapshot.branch }); + } + run = this.move(run, "QUEUED", "Run queued"); run = this.move(run, "CODEX_RUNNING", "Codex turn started"); - const existingThread = this.db.get<{ thread_id: string }>("SELECT thread_id FROM codex_threads WHERE run_id = ? ORDER BY created_at ASC LIMIT 1", run.id); - const thread = existingThread ? { threadId: existingThread.thread_id } : await this.adapter.startThread({ cwd: project.localPath, projectId: project.id }); - if (!existingThread) this.db.saveThread({ runId: run.id, provider: this.adapter.provider, threadId: thread.threadId, status: "ACTIVE" }); - const dispatchKey = run.iterationCount === 0 ? "initial" : `iteration-${run.iterationCount + 1}`; - if (!this.db.hasDispatch(run.id, dispatchKey)) { - await this.adapter.sendPrompt({ threadId: thread.threadId, prompt, cwd: project.localPath }); - this.db.recordDispatch(run.id, null, dispatchKey, prompt); - this.event(run.id, "DISPATCH_SENT", "Prompt dispatched to Codex", run.state, { dispatchKey, provider: this.adapter.provider }); - } - run.currentPrompt = prompt; - run.iterationCount += 1; - run = this.db.updateRun(run); - return run; + const storedThread = this.db.getThreadForRun(run.id); + const thread = storedThread ?? await this.adapter.startThread({ cwd: project.localPath }); + if (!storedThread) this.db.saveThread({ runId: run.id, provider: this.adapter.provider, threadId: thread.threadId, status: "ACTIVE" }); + + const dispatchKey = run.iterationCount === 0 ? "initial" : "iteration-" + (run.iterationCount + 1); + const existing = this.db.getDispatch(run.id, dispatchKey); + if (existing) { + if (existing.status === "SENT" && existing.turnId) return run; + return this.block(run, "An ambiguous prepared dispatch exists; recovery will not resend it", "DISPATCH_AMBIGUOUS"); + } + + if (!this.db.prepareDispatch(run.id, null, dispatchKey, prompt, thread.threadId)) { + return this.block(run, "Dispatch intent already exists; recovery will not resend it", "DISPATCH_AMBIGUOUS"); + } + + try { + const response = await this.adapter.sendPrompt({ threadId: thread.threadId, prompt, cwd: project.localPath }); + if (!response.turnId) throw new Error("Codex App Server did not return a turn id"); + this.db.createCodexTurn({ runId: run.id, threadId: thread.threadId, turnId: response.turnId, dispatchKey, prompt }); + this.db.markDispatchSent(run.id, dispatchKey, response.turnId); + run.currentPrompt = prompt; + run.iterationCount += 1; + run.currentBlocker = null; + run = this.db.updateRun(run); + this.event(run.id, "DISPATCH_SENT", "Prompt dispatched to Codex", run.state, { dispatchKey, provider: this.adapter.provider, threadId: thread.threadId, turnId: response.turnId }); + return run; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.db.markDispatchFailed(run.id, dispatchKey, message); + return this.block(run, "Initial Codex dispatch failed: " + message, "DISPATCH_FAILED"); + } } + /** + * Compatibility entry point retained for callers migrating from 001A. + * Production completion is driven by the adapter's turn/completed event. + */ async markValidation(runId: string): Promise { - let run = this.requireRun(runId); + const run = this.requireRun(runId); + const turn = this.db.getActiveCodexTurn(run.id); + if (!turn) throw new Error("No active durable Codex turn; manual validation is not a valid completion path"); + await this.handleCodexTurn({ kind: "turn", threadId: turn.threadId, turnId: turn.turnId, status: "completed", payload: {} }); + return this.requireRun(runId); + } + + private handleAdapterEvent(event: CodexLifecycleEvent): void { + if (event.kind === "turn") { + void this.handleCodexTurn(event).catch((error: unknown) => { + const turn = this.db.getCodexTurnByIdentity(event.threadId, event.turnId); + if (turn) this.failRun(turn.runId, "Codex turn handler failed: " + (error instanceof Error ? error.message : String(error))); + }); + return; + } + const runId = event.threadId ? this.db.get<{ run_id: string }>("SELECT run_id FROM codex_threads WHERE thread_id = ?", event.threadId)?.run_id : undefined; + if (runId) this.event(runId, "CODEX_" + event.method.replace(/[^A-Z0-9]+/gi, "_").toUpperCase(), "Codex event: " + event.method, this.db.getRun(runId)?.state ?? null, event.payload); + } + + async handleCodexTurn(event: { kind: "turn"; threadId: string; turnId: string; status: "completed" | "failed" | "interrupted"; error?: string; payload: Record }): Promise { + const turn = this.db.getCodexTurnByIdentity(event.threadId, event.turnId); + if (!turn) return; + if (terminalTurnStatuses.has(turn.status)) return; + + const nextStatus = event.status === "completed" ? "COMPLETED" : event.status === "failed" ? "FAILED" : "INTERRUPTED"; + this.db.updateCodexTurnStatus(event.turnId, nextStatus, event.error ?? null); + const run = this.requireRun(turn.runId); + if (event.status !== "completed") { + this.failRun(run.id, "Codex turn " + event.status + (event.error ? ": " + event.error : "")); + this.event(run.id, "CODEX_TURN_TERMINAL", "Codex turn ended without a completed result", this.db.getRun(run.id)?.state ?? null, { turnId: event.turnId, status: event.status, error: event.error ?? null }); + return; + } + + this.event(run.id, "CODEX_TURN_COMPLETED", "Codex turn completed; deterministic validation started", run.state, { threadId: event.threadId, turnId: event.turnId }); + await this.completeCodexTurn(turn); + } + + private async completeCodexTurn(turn: CodexTurn): Promise { + if (!this.db.beginTurnValidation(turn.turnId)) return; + let run = this.requireRun(turn.runId); const project = this.requireProject(run.projectId); - run = this.move(run, "VALIDATING", "Codex turn completed; validation started"); - run = this.move(run, "GIT_SYNC", "Capturing deterministic Git evidence"); - const evidence = this.db.saveGitEvidence(this.git.capture(run.id, project.localPath)); - run.expectedBaseSha = evidence.baseSha; - run.expectedHeadSha = evidence.headSha; - run = this.db.updateRun(run); - this.event(run.id, "GIT_EVIDENCE_CAPTURED", "Git evidence locked for review", run.state, { headSha: evidence.headSha, baseSha: evidence.baseSha, branch: evidence.branch }); - return this.move(run, "WAITING_REVIEW", "Waiting for a review contract"); + if (run.state !== "CODEX_RUNNING") { + this.block(run, "Completed Codex turn arrived while run was in " + run.state, "TURN_STATE_MISMATCH"); + return; + } + + try { + run = this.move(run, "VALIDATING", "Codex turn completed; validation started"); + const validations = await this.validation.run(project, run.id, turn.turnId); + this.event(run.id, "VALIDATION_COMPLETED", "Configured validation commands completed", run.state, { + turnId: turn.turnId, + results: validations.map((result) => ({ kind: result.kind, status: result.status, exitCode: result.exitCode })) + }); + run = this.move(run, "GIT_SYNC", "Capturing deterministic Git evidence"); + const evidence = this.db.saveGitEvidence(this.git.capture(run.id, project.localPath, run.runBaseSha ?? undefined)); + run.expectedBaseSha = evidence.baseSha; + run.expectedHeadSha = evidence.headSha; + run.currentBlocker = null; + run = this.db.updateRun(run); + this.event(run.id, "GIT_EVIDENCE_CAPTURED", "Git evidence captured from immutable work-order base", run.state, { + headSha: evidence.headSha, baseSha: evidence.baseSha, branch: evidence.branch, isClean: evidence.isClean + }); + + const test = validations.find((result) => result.kind === "test"); + const failed = validations.filter((result) => result.status === "FAIL" || result.status === "ERROR"); + const testMissing = !test || test.status === "NOT_CONFIGURED"; + if (!evidence.isClean || testMissing || failed.length > 0) { + const blocker = !evidence.isClean + ? "Working tree is dirty; review evidence is fail-closed" + : testMissing + ? "Required test command is not configured" + : "Validation failed: " + failed.map((result) => result.kind + "=" + result.status).join(", "); + this.block(run, blocker, "VALIDATION_BLOCKED"); + return; + } + run = this.db.updateRun(run); + this.move(run, "WAITING_REVIEW", "Waiting for a canonical review contract"); + } catch (error) { + this.failRun(run.id, "Validation/evidence pipeline failed: " + (error instanceof Error ? error.message : String(error))); + } } pause(runId: string): Run { const run = this.requireRun(runId); - if (!["QUEUED", "CODEX_RUNNING", "WAITING_REVIEW", "WAITING_APPROVAL"].includes(run.state)) throw new Error(`Run cannot pause from ${run.state}`); + if (!["QUEUED", "CODEX_RUNNING", "WAITING_REVIEW", "WAITING_APPROVAL"].includes(run.state)) throw new Error("Run cannot pause from " + run.state); return this.move(run, "PAUSED", "Run paused by operator", "RUN_PAUSED"); } async resume(runId: string): Promise { const run = this.requireRun(runId); - if (run.state !== "PAUSED") throw new Error(`Run cannot resume from ${run.state}`); + if (run.state !== "PAUSED") throw new Error("Run cannot resume from " + run.state); return this.start(run.id, run.currentPrompt ?? "Resume the current work order and report evidence."); } stop(runId: string): Run { const run = this.requireRun(runId); - if (run.state === "COMPLETE" || run.state === "FAILED") throw new Error(`Run cannot stop from ${run.state}`); + if (run.state === "COMPLETE" || run.state === "FAILED") throw new Error("Run cannot stop from " + run.state); return this.move(run, "FAILED", "Run stopped by operator", "RUN_STOPPED"); } async processReview(input: unknown): Promise<{ accepted: boolean; run: Run | null; errors: string[] }> { - const parsed = validateReviewContract(input); + const internalInput = typeof input === "object" && input !== null && !Array.isArray(input) && "schemaVersion" in input; + const boundaryInput = internalInput ? serializeReviewContract(input as ReviewContract) : input; + const parsed = validateReviewContract(boundaryInput); + const candidateWorkOrderId = typeof boundaryInput === "object" && boundaryInput !== null && !Array.isArray(boundaryInput) ? (boundaryInput as Record).work_order_id : undefined; + const candidateRun = typeof candidateWorkOrderId === "string" ? this.db.getRunByWorkOrderId(candidateWorkOrderId) ?? null : null; if (!parsed.valid || !parsed.value) { - const candidateRunId = typeof input === "object" && input !== null && !Array.isArray(input) ? (input as Record).run_id : undefined; - const run = typeof candidateRunId === "string" ? this.db.getRun(candidateRunId) ?? null : null; - if (run) this.event(run.id, "REVIEW_REJECTED", "Invalid review contract", run.state, { errors: parsed.errors }); - return { accepted: false, run, errors: parsed.errors }; + if (candidateRun) this.event(candidateRun.id, "REVIEW_REJECTED", "Invalid canonical review contract", candidateRun.state, { errors: parsed.errors }); + return { accepted: false, run: candidateRun, errors: parsed.errors }; } const review = parsed.value; - const run = this.requireRun(review.run_id); + const run = this.db.getRunByWorkOrderId(review.workOrderId) ?? null; + if (!run) return { accepted: false, run: null, errors: ["work_order_id does not resolve to an active work order"] }; const project = this.requireProject(run.projectId); - if (run.state !== "WAITING_REVIEW") return { accepted: false, run, errors: [`run is not waiting for review (${run.state})`] }; + if (run.state !== "WAITING_REVIEW") return { accepted: false, run, errors: ["run is not waiting for review (" + run.state + ")"] }; if (!this.db.saveReview(review, "RECEIVED")) { - this.event(run.id, "REVIEW_DUPLICATE", "Review id was already consumed", run.state, { reviewId: review.review_id }); + this.event(run.id, "REVIEW_DUPLICATE", "Review id was already consumed", run.state, { reviewId: review.reviewId }); return { accepted: false, run, errors: ["review id already consumed"] }; } const evidence = this.db.getLatestGitEvidence(run.id); - if (!evidence) return this.rejectReview(run, review.review_id, "Git evidence is missing"); - const lockErrors: string[] = []; - if (review.project_id !== project.id) lockErrors.push("project_id mismatch"); - if (review.run_id !== run.id) lockErrors.push("run_id mismatch"); - if (review.work_order_id !== run.workOrderId) lockErrors.push("work_order_id mismatch"); - if (review.head_sha !== evidence.headSha) lockErrors.push("head_sha does not match captured Git evidence"); - if (review.base_sha !== evidence.baseSha) lockErrors.push("base_sha does not match captured Git evidence"); - if (lockErrors.length > 0) return this.rejectReview(run, review.review_id, `Evidence Lock rejected: ${lockErrors.join(", ")}`, { errors: lockErrors }); - - this.db.run("UPDATE reviews SET status = 'ACCEPTED' WHERE id = ?", review.review_id); - let current = this.move(run, "REVIEW_RECEIVED", "Valid review received", "REVIEW_ACCEPTED"); - current.lastReviewId = review.review_id; + if (!evidence) return this.rejectReview(run, review.reviewId, "Git evidence is missing"); + const lock = verifyEvidenceLock(review, project, run, evidence); + if (!lock.locked) return this.rejectReview(run, review.reviewId, "Evidence Lock rejected: " + lock.errors.join(", "), { errors: lock.errors }); + + this.db.updateReviewStatus(review.reviewId, "ACCEPTED"); + let current = this.move(run, "REVIEW_RECEIVED", "Valid canonical review received", "REVIEW_ACCEPTED"); + current.lastReviewId = review.reviewId; + current.lastReviewStatus = "ACCEPTED"; + current.lastReviewVerdict = review.verdict; + current.lastReviewSummary = review.summary; + current.lastCheckpointNote = review.checkpointNote; + current.progressPercent = review.progressPercent; + current.currentBlocker = review.verdict === "BLOCKED" ? review.summary : null; current = this.db.updateRun(current); - this.event(current.id, "EVIDENCE_LOCKED", "Review matches project, work order and Git SHA", current.state, { reviewId: review.review_id }); + this.event(current.id, "EVIDENCE_LOCKED", "Review matches project, work order and Git evidence", current.state, { reviewId: review.reviewId, headSha: review.headSha, baseSha: review.baseSha }); - if (review.action === "BLOCKED") return { accepted: true, run: this.move(current, "BLOCKED", "Review blocked the workflow"), errors: [] }; - if (review.action === "COMPLETE") return { accepted: true, run: this.move(current, "COMPLETE", "Review completed the workflow"), errors: [] }; + if (review.nextAction === "WAIT") return { accepted: true, run: this.block(current, review.summary, "REVIEW_BLOCKED"), errors: [] }; + if (review.nextAction === "STOP") return { accepted: true, run: this.move(current, "COMPLETE", "Review completed the workflow", "RUN_COMPLETE"), errors: [] }; const guard = evaluateLoopGuard(current, review); if (!guard.allowed) { @@ -121,17 +246,16 @@ export class WorkflowEngine { current.correctionCycles = guard.nextCorrectionCycles; current.repeatedFindings = guard.nextRepeatedFindings; current = this.db.updateRun(current); - this.event(current.id, "LOOP_GUARD_BLOCKED", guard.reason ?? "Loop guard blocked dispatch", current.state); - return { accepted: true, run: this.move(current, "BLOCKED", guard.reason ?? "Loop guard blocked dispatch"), errors: [] }; + return { accepted: true, run: this.block(current, guard.reason ?? "Loop guard blocked dispatch", "LOOP_GUARD_BLOCKED"), errors: [] }; } current.iterationCount = guard.nextIterationCount; current.correctionCycles = guard.nextCorrectionCycles; current.repeatedFindings = guard.nextRepeatedFindings; current = this.db.updateRun(current); if (current.autonomyMode !== "AUTONOMOUS") { - this.db.createApproval(review.review_id); - this.event(current.id, "APPROVAL_REQUIRED", `${current.autonomyMode} requires approval before dispatch`, current.state, { reviewId: review.review_id }); - return { accepted: true, run: this.move(current, "WAITING_APPROVAL", "Waiting for human approval") , errors: [] }; + this.db.createApproval(review.reviewId); + this.event(current.id, "APPROVAL_REQUIRED", current.autonomyMode + " requires approval before dispatch", current.state, { reviewId: review.reviewId }); + return { accepted: true, run: this.move(current, "WAITING_APPROVAL", "Waiting for human approval"), errors: [] }; } return { accepted: true, run: await this.dispatchAcceptedReview(current, review), errors: [] }; } @@ -140,53 +264,94 @@ export class WorkflowEngine { const stored = this.db.getReview(reviewId); if (!stored) throw new Error("Review not found"); if (!this.db.approveReview(reviewId)) throw new Error("Review is not awaiting approval"); - const run = this.requireRun(stored.review.run_id); - if (run.state !== "WAITING_APPROVAL") throw new Error(`Run is not waiting for approval (${run.state})`); + const run = this.db.getRunByWorkOrderId(stored.review.workOrderId); + if (!run) throw new Error("Review work order not found"); + if (run.state !== "WAITING_APPROVAL") throw new Error("Run is not waiting for approval (" + run.state + ")"); this.event(run.id, "APPROVAL_GRANTED", "Human approval granted", run.state, { reviewId }); return this.dispatchAcceptedReview(run, stored.review); } recover(): Run[] { const runs = this.db.listNonTerminalRuns(); - for (const run of runs) this.event(run.id, "RECOVERY_CHECKED", `Recovered run in ${run.state}; no duplicate dispatch issued`, run.state); - return runs; + for (const run of runs) { + if (run.state === "CODEX_RUNNING" || run.state === "DISPATCHING_NEXT") { + const turn = this.db.getActiveCodexTurn(run.id); + if (turn) { + if (!this.db.hasEvent(run.id, "RECOVERY_ORPHANED")) { + this.db.updateCodexTurnStatus(turn.turnId, "ORPHANED", "Turn ownership could not be proven after process recovery"); + this.block(run, "Active Codex turn became orphaned during recovery", "RECOVERY_ORPHANED"); + } + } else if (!this.db.hasEvent(run.id, "RECOVERY_NO_DURABLE_TURN")) { + this.block(run, "Run was active without a durable Codex turn; recovery will not resend", "RECOVERY_NO_DURABLE_TURN"); + } + } else if (!this.db.hasEvent(run.id, "RECOVERY_RECONCILED")) { + this.event(run.id, "RECOVERY_RECONCILED", "Run recovery checked without dispatch", run.state); + } + } + return runs.map((run) => this.db.getRun(run.id)!).filter(Boolean); } private async dispatchAcceptedReview(run: Run, review: ReviewContract): Promise { const project = this.requireProject(run.projectId); - let current = this.move(run, "DISPATCHING_NEXT", "Preparing next dispatch"); - const dispatchKey = `review-${review.review_id}`; - if (this.db.hasDispatch(current.id, dispatchKey)) return this.move(current, "CODEX_RUNNING", "Existing dispatch reused after recovery"); - const threadRow = this.db.get<{ thread_id: string }>("SELECT thread_id FROM codex_threads WHERE run_id = ? ORDER BY created_at ASC LIMIT 1", current.id); - if (!threadRow) throw new Error("Codex thread missing for next dispatch"); + let current = this.move(run, "DISPATCHING_NEXT", "Preparing next durable dispatch"); + const dispatchKey = "review-" + review.reviewId; + const existing = this.db.getDispatch(current.id, dispatchKey); + if (existing) { + if (existing.status === "SENT" && existing.turnId) { + current.currentPrompt = review.executorPrompt; + current = this.db.updateRun(current); + return this.move(current, "CODEX_RUNNING", "Existing dispatch reused after recovery"); + } + return this.block(current, "An ambiguous review dispatch exists; recovery will not resend it", "DISPATCH_AMBIGUOUS"); + } + const thread = this.db.getThreadForRun(current.id); + if (!thread) return this.block(current, "Codex thread missing for next dispatch", "DISPATCH_FAILED"); + if (!this.db.prepareDispatch(current.id, review.reviewId, dispatchKey, review.executorPrompt, thread.threadId)) { + return this.block(current, "Dispatch intent already exists; recovery will not resend it", "DISPATCH_AMBIGUOUS"); + } try { - await this.adapter.sendPrompt({ threadId: threadRow.thread_id, prompt: review.executor_prompt, cwd: project.localPath }); - this.db.recordDispatch(current.id, review.review_id, dispatchKey, review.executor_prompt); - this.event(current.id, "DISPATCH_SENT", "Next prompt dispatched automatically", "CODEX_RUNNING", { dispatchKey, reviewId: review.review_id }); - } catch (error: unknown) { - this.event(current.id, "DISPATCH_FAILED", error instanceof Error ? error.message : String(error), "FAILED"); - current = this.move(current, "FAILED", "Next dispatch failed", "DISPATCH_FAILED"); - return current; + const response = await this.adapter.sendPrompt({ threadId: thread.threadId, prompt: review.executorPrompt, cwd: project.localPath }); + if (!response.turnId) throw new Error("Codex App Server did not return a turn id"); + this.db.createCodexTurn({ runId: current.id, threadId: thread.threadId, turnId: response.turnId, dispatchKey, prompt: review.executorPrompt }); + this.db.markDispatchSent(current.id, dispatchKey, response.turnId); + current.currentPrompt = review.executorPrompt; + current.currentBlocker = null; + current = this.db.updateRun(current); + this.event(current.id, "DISPATCH_SENT", "Next prompt dispatched to Codex", "DISPATCHING_NEXT", { dispatchKey, reviewId: review.reviewId, turnId: response.turnId }); + return this.move(current, "CODEX_RUNNING", "Codex is running the next prompt"); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + this.db.markDispatchFailed(current.id, dispatchKey, message); + return this.block(current, "Next Codex dispatch failed: " + message, "DISPATCH_FAILED"); } - current.currentPrompt = review.executor_prompt; - return this.move(current, "CODEX_RUNNING", "Codex is running the next prompt"); + } + + private failRun(runId: string, message: string): Run { + let run = this.requireRun(runId); + run.currentBlocker = message; + run = this.db.updateRun(run); + if (run.state === "CODEX_RUNNING" || run.state === "VALIDATING" || run.state === "GIT_SYNC" || run.state === "DISPATCHING_NEXT") { + return this.block(run, message, "RUN_FAILED_SAFE"); + } + this.event(run.id, "RUN_FAILED_SAFE", message, run.state); + return run; } private rejectReview(run: Run, reviewId: string, message: string, payload: Record = {}): { accepted: false; run: Run; errors: string[] } { - this.db.run("UPDATE reviews SET status = 'REJECTED' WHERE id = ?", reviewId); + this.db.updateReviewStatus(reviewId, "REJECTED"); this.event(run.id, "REVIEW_REJECTED", message, run.state, payload); return { accepted: false, run, errors: [message] }; } private requireRun(id: string): Run { const run = this.db.getRun(id); - if (!run) throw new Error(`Run not found: ${id}`); + if (!run) throw new Error("Run not found: " + id); return run; } private requireProject(id: string) { const project = this.db.getProject(id); - if (!project) throw new Error(`Project not found: ${id}`); + if (!project) throw new Error("Project not found: " + id); return project; } } diff --git a/src/workflow/validation.ts b/src/workflow/validation.ts new file mode 100644 index 0000000..20f57df --- /dev/null +++ b/src/workflow/validation.ts @@ -0,0 +1,64 @@ +import { exec } from "node:child_process"; +import { promisify } from "node:util"; +import type { Project, ValidationKind, ValidationResult, ValidationStatus } from "../core/types"; +import { Database } from "../db/database"; + +const execAsync = promisify(exec); +const validationKinds: ValidationKind[] = ["test", "lint", "typecheck", "build"]; + +function statusFor(exitCode: number | null, error: unknown): ValidationStatus { + if (error && exitCode === null) return "ERROR"; + return exitCode === 0 ? "PASS" : "FAIL"; +} + +export class ValidationRunner { + constructor(private readonly db: Database) {} + + async run(project: Project, runId: string, turnId: string): Promise { + const results: ValidationResult[] = []; + for (const kind of validationKinds) { + const command = project[kind + "Command" as "testCommand" | "lintCommand" | "typecheckCommand" | "buildCommand"] as string | undefined; + if (!command?.trim()) { + const timestamp = new Date().toISOString(); + results.push(this.db.saveValidationResult({ + runId, turnId, kind, command: null, startedAt: timestamp, finishedAt: timestamp, + exitCode: null, stdout: "", stderr: "", status: "NOT_CONFIGURED" + })); + continue; + } + + const startedAt = new Date().toISOString(); + let stdout = ""; + let stderr = ""; + let exitCode: number | null = null; + let error: unknown = null; + try { + const result = await execAsync(command, { + cwd: project.localPath, + windowsHide: true, + maxBuffer: 10 * 1024 * 1024, + shell: process.platform === "win32" ? (process.env.ComSpec ?? "cmd.exe") : "/bin/sh" + }); + stdout = result.stdout; + stderr = result.stderr; + exitCode = 0; + } catch (caught: unknown) { + error = caught; + const failure = caught as { code?: number | string; stdout?: string; stderr?: string; message?: string }; + stdout = String(failure.stdout ?? ""); + stderr = String(failure.stderr ?? failure.message ?? caught); + exitCode = typeof failure.code === "number" + ? failure.code + : typeof failure.code === "string" && /^-?\d+$/.test(failure.code) + ? Number(failure.code) + : null; + } + const finishedAt = new Date().toISOString(); + results.push(this.db.saveValidationResult({ + runId, turnId, kind, command, startedAt, finishedAt, exitCode, stdout, stderr, + status: statusFor(exitCode, error) + })); + } + return results; + } +} diff --git a/tests/core.test.ts b/tests/core.test.ts index 21e0d4f..e4250c4 100644 --- a/tests/core.test.ts +++ b/tests/core.test.ts @@ -1,16 +1,22 @@ import { describe, expect, it } from "vitest"; import { verifyEvidenceLock } from "../src/core/evidence-lock"; import { evaluateLoopGuard } from "../src/core/loop-guard"; -import { validateReviewContract } from "../src/core/rcp"; +import { validateReviewContract, serializeReviewContract } from "../src/core/rcp"; import { allowedTransitions, canTransition, transition } from "../src/core/state-machine"; import type { GitEvidence, Project, ReviewContract, Run } from "../src/core/types"; +const shaBase = "0123456789abcdef0123456789abcdef01234567"; +const shaHead = "fedcba9876543210fedcba9876543210fedcba98"; const project: Project = { id: "project-1", name: "Test", localPath: "C:/repo", defaultBranch: "main", workingBranch: "main", testCommand: "npm test", autonomyMode: "AUTONOMOUS", reviewMailboxPath: "C:/mailbox", createdAt: "now", updatedAt: "now" }; -const run: Run = { id: "run-1", projectId: project.id, workOrderId: "WO-123", state: "WAITING_REVIEW", autonomyMode: "AUTONOMOUS", maxIterations: 3, maxCorrectionCycles: 2, repeatedFindingThreshold: 2, iterationCount: 1, correctionCycles: 0, repeatedFindings: {}, currentPrompt: "first", expectedBaseSha: "base", expectedHeadSha: "head", lastReviewId: null, createdAt: "now", updatedAt: "now" }; -const evidence: GitEvidence = { id: "evidence-1", runId: run.id, baseSha: "base", headSha: "head", branch: "main", status: "clean", changedFiles: [], diffSummary: "", capturedAt: "now" }; +const run: Run = { id: "run-1", projectId: project.id, workOrderId: "WO-123", state: "WAITING_REVIEW", autonomyMode: "AUTONOMOUS", maxIterations: 3, maxCorrectionCycles: 2, repeatedFindingThreshold: 2, iterationCount: 1, correctionCycles: 0, repeatedFindings: {}, currentPrompt: "first", runBaseSha: shaBase, runBaseBranch: "main", expectedBaseSha: shaBase, expectedHeadSha: shaHead, lastReviewId: null, lastReviewStatus: null, lastReviewVerdict: null, lastReviewSummary: null, lastCheckpointNote: null, progressPercent: 0, currentBlocker: null, createdAt: "now", updatedAt: "now" }; +const evidence: GitEvidence = { id: "evidence-1", runId: run.id, baseSha: shaBase, headSha: shaHead, branch: "main", status: "clean", changedFiles: [], diffSummary: "", isClean: true, capturedAt: "now" }; function review(overrides: Partial = {}): ReviewContract { - return { protocol: "RCP", version: "0.1", review_id: "review-1", project_id: project.id, run_id: run.id, work_order_id: run.workOrderId, decision: "APPROVED", action: "CONTINUE", head_sha: "head", base_sha: "base", findings: [], executor_prompt: "next", created_at: "now", ...overrides }; + return { schemaVersion: "0.1", reviewId: "review-1", projectId: project.id, workOrderId: run.workOrderId, baseSha: shaBase, headSha: shaHead, verdict: "APPROVED", progressPercent: 50, summary: "Increment reviewed", findings: [], nextAction: "CONTINUE", executorPrompt: "next", checkpointNote: "Evidence locked", ...overrides }; +} + +function canonical(overrides: Record = {}) { + return { ...serializeReviewContract(review()), ...overrides }; } describe("workflow state machine", () => { @@ -23,29 +29,37 @@ describe("workflow state machine", () => { }); }); -describe("RCP v0.1", () => { - it("accepts a complete contract", () => { - const result = validateReviewContract(review()); +describe("canonical RCP v0.1", () => { + it("accepts and serializes the exact external contract", () => { + const result = validateReviewContract(canonical()); expect(result.valid).toBe(true); - expect(result.value?.action).toBe("CONTINUE"); + expect(result.value?.nextAction).toBe("CONTINUE"); + expect(serializeReviewContract(result.value!)).toEqual(canonical()); + }); + + it("rejects legacy aliases and incomplete evidence fields", () => { + const result = validateReviewContract({ ...canonical(), protocol: "RCP", run_id: run.id, head_sha: "", findings: [{ id: "F1", severity: "HIGH" }] }); + expect(result.valid).toBe(false); + expect(result.errors).toEqual(expect.arrayContaining(["protocol is not part of canonical RCP v0.1", "run_id is not part of canonical RCP v0.1", "head_sha must be a non-empty string", "findings[0].title must be a non-empty string", "findings[0].details must be a non-empty string"])); }); - it("rejects missing critical fields and mismatched actions", () => { - const result = validateReviewContract({ ...review(), head_sha: "", action: "EXECUTE_CORRECTION" }); + it("enforces verdict and next_action coherence", () => { + const result = validateReviewContract(canonical({ verdict: "APPROVED", next_action: "EXECUTE_CORRECTION" })); expect(result.valid).toBe(false); - expect(result.errors).toEqual(expect.arrayContaining(["head_sha must be a non-empty string", "APPROVED requires CONTINUE"])); + expect(result.errors).toContain("APPROVED requires next_action CONTINUE"); }); }); describe("evidence lock and loop guards", () => { - it("locks matching project, work order and SHAs", () => { + it("locks matching project, work order, base/head SHAs and clean state", () => { expect(verifyEvidenceLock(review(), project, run, evidence).locked).toBe(true); - expect(verifyEvidenceLock(review({ head_sha: "wrong" }), project, run, evidence).locked).toBe(false); + expect(verifyEvidenceLock(review({ headSha: shaBase }), project, run, evidence).locked).toBe(false); + expect(verifyEvidenceLock(review(), project, run, { ...evidence, isClean: false }).locked).toBe(false); }); it("blocks max iterations, correction cycles and repeated findings", () => { expect(evaluateLoopGuard({ ...run, iterationCount: 3 }, review()).allowed).toBe(false); - expect(evaluateLoopGuard({ ...run, correctionCycles: 2 }, review({ decision: "CORRECTION_REQUIRED", action: "EXECUTE_CORRECTION" })).allowed).toBe(false); - expect(evaluateLoopGuard({ ...run, repeatedFindings: { F1: 2 } }, review({ findings: [{ id: "F1", severity: "HIGH", summary: "still present" }] })).allowed).toBe(false); + expect(evaluateLoopGuard({ ...run, correctionCycles: 2 }, review({ verdict: "CORRECTION_REQUIRED", nextAction: "EXECUTE_CORRECTION" })).allowed).toBe(false); + expect(evaluateLoopGuard({ ...run, repeatedFindings: { F1: 2 } }, review({ findings: [{ id: "F1", severity: "HIGH", title: "Still present", details: "Still present" }] })).allowed).toBe(false); }); }); diff --git a/tests/integration.test.ts b/tests/integration.test.ts index 84630d7..fbfcf14 100644 --- a/tests/integration.test.ts +++ b/tests/integration.test.ts @@ -1,4 +1,3 @@ -import assert from "node:assert/strict"; import { execFileSync } from "node:child_process"; import { mkdtempSync, mkdirSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -7,6 +6,8 @@ import { describe, expect, it } from "vitest"; import { FakeCodexAdapter } from "../src/adapters/fake-codex"; import { GitAdapter } from "../src/adapters/git"; import { EventBus } from "../src/core/event-bus"; +import { serializeReviewContract } from "../src/core/rcp"; +import type { ReviewContract } from "../src/core/types"; import { Database } from "../src/db/database"; import { ReviewMailboxWatcher } from "../src/mailbox/watcher"; import { WorkflowEngine } from "../src/workflow/engine"; @@ -15,58 +16,137 @@ function fixture() { const directory = mkdtempSync(join(tmpdir(), "hiveforge-test-")); const git = (args: string[]) => execFileSync("git", args, { cwd: directory, encoding: "utf8" }).trim(); git(["init", "-b", "main"]); git(["config", "user.email", "test@hiveforge.local"]); git(["config", "user.name", "HiveForge Test"]); - writeFileSync(join(directory, "README.md"), "one\n"); git(["add", "."]); git(["commit", "-m", "one"]); - writeFileSync(join(directory, "README.md"), "two\n"); git(["add", "."]); git(["commit", "-m", "two"]); - return directory; + writeFileSync(join(directory, "README.md"), "base\n"); git(["add", "."]); git(["commit", "-m", "base"]); + return { directory, git, baseSha: git(["rev-parse", "HEAD"]) }; +} + +async function settle(db: Database, runId: string) { + for (let index = 0; index < 100; index += 1) { + const state = db.getRun(runId)?.state; + if (state === "WAITING_REVIEW" || state === "BLOCKED") return; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + throw new Error("workflow did not settle for run " + runId); } async function readyRun(mode: "AUTONOMOUS" | "GUARDED" | "CONTROLLED") { - const directory = fixture(); + const context = fixture(); + const mailbox = join(context.directory, ".mailbox"); + mkdirSync(mailbox); const db = new Database(":memory:"); const fake = new FakeCodexAdapter(); const engine = new WorkflowEngine(db, new GitAdapter(), new EventBus(), fake); - const project = db.createProject({ name: mode, localPath: directory, defaultBranch: "main", workingBranch: "main", testCommand: "node --version", autonomyMode: mode, reviewMailboxPath: join(directory, ".mailbox") }); + const project = db.createProject({ name: mode, localPath: context.directory, defaultBranch: "main", workingBranch: "main", testCommand: "node --version", autonomyMode: mode, reviewMailboxPath: mailbox }); const run = db.createRun({ projectId: project.id, autonomyMode: mode }); - await engine.start(run.id, "first"); await engine.markValidation(run.id); - return { db, engine, fake, project, run: db.getRun(run.id)! }; + await engine.start(run.id, "first"); + fake.completeTurn(); + await settle(db, run.id); + return { ...context, db, engine, fake, project, run: db.getRun(run.id)! }; +} + +function makeReview(run: Awaited>, overrides: Partial = {}): ReviewContract { + const evidence = run.db.getLatestGitEvidence(run.run.id)!; + return { schemaVersion: "0.1", reviewId: "review-" + run.run.id, projectId: run.project.id, workOrderId: run.run.workOrderId, baseSha: evidence.baseSha, headSha: evidence.headSha, verdict: "APPROVED", progressPercent: 60, summary: "Increment reviewed", findings: [], nextAction: "CONTINUE", executorPrompt: "approved next", checkpointNote: "Evidence locked", ...overrides }; } describe("SQLite, Git, mailbox and workflow integration", () => { - it("runs migrations and persists projects, runs and events", () => { + it("persists projects, runs, durable turns, validation results and events", async () => { + const context = await readyRun("AUTONOMOUS"); + expect(context.db.getRun(context.run.id)?.state).toBe("WAITING_REVIEW"); + expect(context.db.getActiveCodexTurn(context.run.id)).toBeUndefined(); + expect(context.db.getValidationResults(context.run.id).find((item) => item.kind === "test")?.status).toBe("PASS"); + expect(context.db.listEvents(context.run.id).map((item) => item.type)).toEqual(expect.arrayContaining(["RUN_BASE_CAPTURED", "CODEX_TURN_COMPLETED", "VALIDATION_COMPLETED", "GIT_EVIDENCE_CAPTURED"])); + context.db.close(); + }); + + it("captures the complete multi-commit work-order diff from the base SHA", async () => { + const context = fixture(); const db = new Database(":memory:"); - const project = db.createProject({ name: "Persisted", localPath: "C:/repo", defaultBranch: "main", workingBranch: "main", testCommand: "npm test", autonomyMode: "AUTONOMOUS", reviewMailboxPath: "C:/mailbox" }); + const fake = new FakeCodexAdapter(); + const engine = new WorkflowEngine(db, new GitAdapter(), new EventBus(), fake); + const project = db.createProject({ name: "Multi commit", localPath: context.directory, defaultBranch: "main", workingBranch: "main", testCommand: "node --version", autonomyMode: "AUTONOMOUS", reviewMailboxPath: join(context.directory, ".mailbox") }); const run = db.createRun({ projectId: project.id, autonomyMode: "AUTONOMOUS" }); - db.appendEvent(run.id, "TEST", "persisted", "IDLE"); - expect(db.getProject(project.id)?.name).toBe("Persisted"); - expect(db.listEvents(run.id)).toHaveLength(1); - expect(db.listNonTerminalRuns().map((item) => item.id)).toContain(run.id); + await engine.start(run.id, "multi-commit"); + writeFileSync(join(context.directory, "README.md"), "commit one\n"); context.git(["add", "."]); context.git(["commit", "-m", "commit one"]); + writeFileSync(join(context.directory, "README.md"), "commit two\n"); context.git(["add", "."]); context.git(["commit", "-m", "commit two"]); + fake.completeTurn(); await settle(db, run.id); + const evidence = db.getLatestGitEvidence(run.id)!; + expect(evidence.baseSha).toBe(context.baseSha); + expect(evidence.headSha).toBe(context.git(["rev-parse", "HEAD"])); + expect(evidence.changedFiles).toContain("README.md"); + expect(evidence.isClean).toBe(true); db.close(); }); - it("captures deterministic Git evidence and consumes mailbox JSON", async () => { - const directory = fixture(); const mailbox = join(directory, ".mailbox"); mkdirSync(mailbox); - const db = new Database(":memory:"); const fake = new FakeCodexAdapter(); const engine = new WorkflowEngine(db, new GitAdapter(), new EventBus(), fake); - const project = db.createProject({ name: "Mailbox", localPath: directory, defaultBranch: "main", workingBranch: "main", testCommand: "node --version", autonomyMode: "AUTONOMOUS", reviewMailboxPath: mailbox }); - const run = db.createRun({ projectId: project.id, autonomyMode: "AUTONOMOUS" }); await engine.start(run.id, "first"); await engine.markValidation(run.id); - const waiting = db.getRun(run.id)!; const evidence = db.getLatestGitEvidence(run.id)!; - writeFileSync(join(mailbox, "valid.json"), JSON.stringify({ protocol: "RCP", version: "0.1", review_id: "mailbox-review", project_id: project.id, run_id: run.id, work_order_id: waiting.workOrderId, decision: "APPROVED", action: "CONTINUE", head_sha: evidence.headSha, base_sha: evidence.baseSha, findings: [], executor_prompt: "second", created_at: new Date().toISOString() })); - const watcher = new ReviewMailboxWatcher(mailbox, { onReview: async (review) => { const result = await engine.processReview(review); assert.equal(result.accepted, true); }, onRejected: () => { throw new Error("unexpected mailbox rejection"); } }); - await watcher.scan(); - expect(fake.prompts).toHaveLength(2); expect(db.getRun(run.id)?.state).toBe("CODEX_RUNNING"); + it("rejects a dirty starting tree", async () => { + const context = fixture(); + const db = new Database(":memory:"); + const fake = new FakeCodexAdapter(); + const engine = new WorkflowEngine(db, new GitAdapter(), new EventBus(), fake); + const project = db.createProject({ name: "Dirty", localPath: context.directory, defaultBranch: "main", workingBranch: "main", testCommand: "node --version", autonomyMode: "AUTONOMOUS", reviewMailboxPath: join(context.directory, ".mailbox") }); + const run = db.createRun({ projectId: project.id, autonomyMode: "AUTONOMOUS" }); + writeFileSync(join(context.directory, "README.md"), "uncommitted\n"); + await expect(engine.start(run.id, "must fail closed")).rejects.toThrow("clean committed Git state"); + expect(db.getRun(run.id)?.state).toBe("IDLE"); db.close(); }); + it("consumes a canonical mailbox review and dispatches the next turn", async () => { + const context = await readyRun("AUTONOMOUS"); + const mailbox = join(context.directory, ".mailbox"); + const review = makeReview(context, { reviewId: "mailbox-review" }); + const watcher = new ReviewMailboxWatcher(mailbox, { + onReview: async (value) => { const result = await context.engine.processReview(value); if (!result.accepted) throw new Error(result.errors.join("; ")); }, + onRejected: ({ errors }) => { throw new Error("unexpected mailbox rejection: " + errors.join("; ")); } + }); + writeFileSync(join(mailbox, "valid.json"), JSON.stringify(serializeReviewContract(review))); + await watcher.scan(); + expect(context.fake.prompts).toHaveLength(2); + expect(context.db.getRun(context.run.id)?.state).toBe("CODEX_RUNNING"); + watcher.stop(); context.db.close(); + }); + it("uses approval gates for GUARDED and CONTROLLED", async () => { for (const mode of ["GUARDED", "CONTROLLED"] as const) { - const context = await readyRun(mode); const { db, engine, project, run, fake } = context; const evidence = db.getLatestGitEvidence(run.id)!; - const result = await engine.processReview({ protocol: "RCP", version: "0.1", review_id: `review-${mode}`, project_id: project.id, run_id: run.id, work_order_id: run.workOrderId, decision: "APPROVED", action: "CONTINUE", head_sha: evidence.headSha, base_sha: evidence.baseSha, findings: [], executor_prompt: "approved next", created_at: new Date().toISOString() }); - expect(result.run?.state).toBe("WAITING_APPROVAL"); expect(fake.prompts).toHaveLength(1); - const approved = await engine.approveReview(`review-${mode}`); expect(approved.state).toBe("CODEX_RUNNING"); expect(fake.prompts).toHaveLength(2); db.close(); + const context = await readyRun(mode); + const review = makeReview(context, { reviewId: "review-" + mode }); + const result = await context.engine.processReview(serializeReviewContract(review)); + expect(result.run?.state).toBe("WAITING_APPROVAL"); + expect(context.fake.prompts).toHaveLength(1); + const approved = await context.engine.approveReview(review.reviewId); + expect(approved.state).toBe("CODEX_RUNNING"); + expect(context.fake.prompts).toHaveLength(2); + context.db.close(); } }); - it("recovery does not issue duplicate dispatches", async () => { - const context = await readyRun("AUTONOMOUS"); const { db, engine, run, fake } = context; const before = fake.prompts.length; const recovered = engine.recover(); - expect(recovered.some((item) => item.id === run.id)).toBe(true); expect(fake.prompts.length).toBe(before); db.close(); + it("blocks failed validation and records the failure", async () => { + const context = fixture(); + const db = new Database(":memory:"); + const fake = new FakeCodexAdapter(); + const engine = new WorkflowEngine(db, new GitAdapter(), new EventBus(), fake); + const project = db.createProject({ name: "Failing validation", localPath: context.directory, defaultBranch: "main", workingBranch: "main", testCommand: "node -e \"process.exit(1)\"", autonomyMode: "AUTONOMOUS", reviewMailboxPath: join(context.directory, ".mailbox") }); + const run = db.createRun({ projectId: project.id, autonomyMode: "AUTONOMOUS" }); + await engine.start(run.id, "validation failure"); fake.completeTurn(); await settle(db, run.id); + expect(db.getRun(run.id)?.state).toBe("BLOCKED"); + expect(db.getValidationResults(run.id).find((item) => item.kind === "test")?.status).toBe("FAIL"); + expect(fake.prompts).toHaveLength(1); + db.close(); + }); + + it("recovery orphans an active turn without resending it", async () => { + const context = fixture(); + const db = new Database(":memory:"); + const fake = new FakeCodexAdapter(); + const engine = new WorkflowEngine(db, new GitAdapter(), new EventBus(), fake); + const project = db.createProject({ name: "Recovery", localPath: context.directory, defaultBranch: "main", workingBranch: "main", testCommand: "node --version", autonomyMode: "AUTONOMOUS", reviewMailboxPath: join(context.directory, ".mailbox") }); + const run = db.createRun({ projectId: project.id, autonomyMode: "AUTONOMOUS" }); + await engine.start(run.id, "recovery"); + const before = fake.prompts.length; + const recovered = engine.recover(); + expect(recovered.find((item) => item.id === run.id)?.state).toBe("BLOCKED"); + expect(fake.prompts).toHaveLength(before); + expect(db.getActiveCodexTurn(run.id)).toBeUndefined(); + db.close(); }); }); diff --git a/tests/mailbox-registry.test.ts b/tests/mailbox-registry.test.ts new file mode 100644 index 0000000..9fdb952 --- /dev/null +++ b/tests/mailbox-registry.test.ts @@ -0,0 +1,39 @@ +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import { MailboxWatcherRegistry } from "../src/mailbox/registry"; +import type { Project } from "../src/core/types"; + +function project(id: string, mailboxPath: string): Project { + return { id, name: id, localPath: mailboxPath, defaultBranch: "main", workingBranch: "main", testCommand: "node --version", autonomyMode: "AUTONOMOUS", reviewMailboxPath: mailboxPath, createdAt: "now", updatedAt: "now" }; +} + +describe("dynamic mailbox watcher registry", () => { + it("reconciles projects created after worker startup without duplicate watchers", async () => { + const firstPath = mkdtempSync(join(tmpdir(), "hiveforge-mailbox-")); + const secondPath = mkdtempSync(join(tmpdir(), "hiveforge-mailbox-")); + const seen: string[] = []; + const registry = new MailboxWatcherRegistry((item) => ({ + onReview: (review) => { seen.push(item.id + ":" + review.reviewId); }, + onRejected: () => { throw new Error("unexpected rejection"); } + })); + const first = project("first", firstPath); + const second = project("second", secondPath); + await registry.reconcile([first]); + expect(registry.size).toBe(1); + await registry.reconcile([first, second]); + expect(registry.size).toBe(2); + await registry.reconcile([first, second]); + expect(registry.size).toBe(2); + writeFileSync(join(secondPath, "review.json"), JSON.stringify({ + schema_version: "0.1", review_id: "dynamic-review", project_id: "second", work_order_id: "WO-DYNAMIC", + base_sha: "0123456789abcdef0123456789abcdef01234567", head_sha: "fedcba9876543210fedcba9876543210fedcba98", + verdict: "BLOCKED", progress_percent: 0, summary: "waiting", findings: [], next_action: "WAIT", + executor_prompt: "wait", checkpoint_note: "wait" + })); + await registry.scanAll(); + expect(seen).toEqual(["second:dynamic-review"]); + registry.stopAll(); + }); +}); diff --git a/vitest.config.ts b/vitest.config.ts index 651c979..6723713 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -3,6 +3,8 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ test: { environment: "node", - include: ["tests/**/*.test.ts"] + include: ["tests/**/*.test.ts"], + testTimeout: 30_000, + fileParallelism: false } });