From 5af12655a2555e6202606a5e12f5cb1589e4da96 Mon Sep 17 00:00:00 2001 From: andreihasna Date: Thu, 16 Jul 2026 18:46:17 +0300 Subject: [PATCH] Add audited branch adjudication --- src/cli/index.tsx | 105 +++++ src/db/branch-adjudication.test.ts | 190 +++++++++ src/db/branch-adjudication.ts | 658 +++++++++++++++++++++++++++++ src/db/database.test.ts | 14 +- src/db/database.ts | 21 + src/index.ts | 12 + 6 files changed, 996 insertions(+), 4 deletions(-) create mode 100644 src/db/branch-adjudication.test.ts create mode 100644 src/db/branch-adjudication.ts diff --git a/src/cli/index.tsx b/src/cli/index.tsx index c4be631..71b9bdf 100644 --- a/src/cli/index.tsx +++ b/src/cli/index.tsx @@ -1,6 +1,7 @@ #!/usr/bin/env bun import { registerEventsCommands } from "@hasna/events/commander"; import { execSync } from "node:child_process"; +import { readFileSync } from "node:fs"; import { program } from "commander"; import { getCliVersion } from "./version.js"; import { parseIntOption } from "./args.js"; @@ -17,6 +18,12 @@ import { getRepoStats, AmbiguousRepoNameError, } from "../db/repos.js"; +import { + BranchAdjudicationError, + adjudicateBranches, + type BranchAdjudicationRequest, + type BranchAdjudicationRowSpec, +} from "../db/branch-adjudication.js"; import { PrimaryRelocationError, relocatePrimaryRepo, @@ -481,6 +488,104 @@ registry } }); +function branchAdjudicationSpecFromFile(path: string): Omit & { + actor?: string; + idempotency_key?: string; + idempotencyKey?: string; +} { + const parsed = JSON.parse(readFileSync(path, "utf8")) as unknown; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + throw new BranchAdjudicationError("INVALID_REQUEST", "spec file must contain a JSON object"); + } + const object = parsed as Record; + if (!Array.isArray(object["rows"])) { + throw new BranchAdjudicationError("INVALID_REQUEST", "spec file must contain rows[]"); + } + const rows = object["rows"].map((value, index): BranchAdjudicationRowSpec => { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new BranchAdjudicationError("INVALID_REQUEST", `rows[${index}] must be an object`); + } + const row = value as Record; + const expectedIsRemote = row["expected_is_remote"] ?? row["expectedIsRemote"]; + return { + id: Number(row["id"]), + repoId: Number(row["repo_id"] ?? row["repoId"]), + name: String(row["name"] ?? ""), + action: "reclassify-local", + expectedIsRemote: expectedIsRemote === undefined + ? undefined + : typeof expectedIsRemote === "boolean" || typeof expectedIsRemote === "number" + ? expectedIsRemote + : Number.NaN, + expectedLastCommitSha: String(row["expected_last_commit_sha"] ?? row["expectedLastCommitSha"] ?? ""), + expectedRepoRevision: typeof (row["expected_repo_revision"] ?? row["expectedRepoRevision"]) === "string" + ? String(row["expected_repo_revision"] ?? row["expectedRepoRevision"]) + : undefined, + evidenceRepoPath: String(row["evidence_repo_path"] ?? row["evidenceRepoPath"] ?? ""), + evidenceRef: typeof (row["evidence_ref"] ?? row["evidenceRef"]) === "string" + ? String(row["evidence_ref"] ?? row["evidenceRef"]) + : undefined, + }; + }); + return { + rows, + actor: typeof object["actor"] === "string" ? object["actor"] : undefined, + idempotency_key: typeof object["idempotency_key"] === "string" ? object["idempotency_key"] : undefined, + idempotencyKey: typeof object["idempotencyKey"] === "string" ? object["idempotencyKey"] : undefined, + }; +} + +registry + .command("adjudicate-branches") + .description("Dry-run or apply exact audited branch-row adjudications") + .requiredOption("--spec ", "JSON spec containing exact guarded branch row adjudications") + .option("--actor ", "Auditable operator or workflow identity; overrides spec actor") + .option("--idempotency-key ", "Stable unique key for this logical adjudication; overrides spec key") + .option("--expected-plan-hash ", "Exact plan hash emitted by reviewed dry run; required with --apply") + .option("--dry-run", "Plan exact row adjudication without writing (default)") + .option("--apply", "Atomically apply exact row adjudication using the reviewed plan hash") + .option("--json", "Output the versioned JSON result") + .action((opts) => { + const json = Boolean(opts.json); + try { + if (opts.apply && opts.dryRun) { + throw new BranchAdjudicationError("INVALID_REQUEST", "--apply and --dry-run are mutually exclusive"); + } + const spec = branchAdjudicationSpecFromFile(opts.spec); + const actor = opts.actor ?? spec.actor; + const idempotencyKey = opts.idempotencyKey ?? spec.idempotencyKey ?? spec.idempotency_key; + const result = adjudicateBranches({ + rows: spec.rows, + actor, + idempotencyKey, + apply: Boolean(opts.apply), + expectedPlanHash: opts.expectedPlanHash, + databasePath: getDbPath(), + }); + if (json) { + console.log(JSON.stringify(result, null, 2)); + } else if (result.applied) { + const replayed = result.replayed ? "replayed " : ""; + console.log(chalk.green(`✓ ${replayed}branch adjudication applied for ${result.plan.rows.length} row(s)`)); + console.log(chalk.dim(` Receipt: ${result.receipt!.id}`)); + } else { + console.log(chalk.yellow(`Dry run: ${result.plan.rows.length} branch row(s) can be adjudicated`)); + console.log(chalk.dim(` Plan: ${result.plan.plan_hash}`)); + console.log(chalk.dim(" Re-run with --apply --expected-plan-hash after review.")); + } + } catch (error) { + const code = error instanceof BranchAdjudicationError ? error.code : "UNEXPECTED_ERROR"; + const message = error instanceof Error ? error.message : "unknown branch adjudication error"; + if (json) { + const details = error instanceof BranchAdjudicationError ? error.details : undefined; + console.log(JSON.stringify({ schema: "open-repos.branch-adjudication.v1", ok: false, error: { code, message, details } }, null, 2)); + } else { + console.error(chalk.red(`${code}: ${message}`)); + } + process.exitCode = 1; + } + }); + // ── Commits ── program .command("commits") diff --git a/src/db/branch-adjudication.test.ts b/src/db/branch-adjudication.test.ts new file mode 100644 index 0000000..32bc3da --- /dev/null +++ b/src/db/branch-adjudication.test.ts @@ -0,0 +1,190 @@ +import { describe, it, expect, beforeEach, afterAll } from "bun:test"; +import { execFileSync } from "node:child_process"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { closeDb, getDb } from "./database"; +import { + BranchAdjudicationError, + adjudicateBranches, + type BranchAdjudicationRequest, +} from "./branch-adjudication"; + +let root = ""; +let repoPath = ""; +let head = ""; +let legacyHead = ""; + +function git(...args: string[]): string { + return execFileSync("git", ["-C", repoPath, ...args], { + encoding: "utf8", + stdio: ["ignore", "pipe", "pipe"], + }).trim(); +} + +function seedGitRepo() { + repoPath = join(root, "repo"); + execFileSync("git", ["init", "-b", "main", repoPath], { stdio: ["ignore", "pipe", "pipe"] }); + git("config", "user.email", "repos-test@invalid.example"); + git("config", "user.name", "Repos Test"); + writeFileSync(join(repoPath, "README.md"), "main\n"); + git("add", "README.md"); + git("commit", "-m", "main"); + head = git("rev-parse", "HEAD"); + git("checkout", "-b", "legacy"); + writeFileSync(join(repoPath, "legacy.txt"), "legacy\n"); + git("add", "legacy.txt"); + git("commit", "-m", "legacy"); + legacyHead = git("rev-parse", "HEAD"); + git("update-ref", "refs/heads/legacy-preserved/build/test", legacyHead); + git("checkout", "main"); + git("update-ref", "refs/heads/build/test", head); +} + +function seedDb() { + const db = getDb(":memory:"); + db.query(`INSERT INTO repos ( + id, path, name, org, remote_url, default_branch, description, + last_scanned, commit_count, branch_count, tag_count, updated_at + ) VALUES + (1, ?, 'legacy', 'hasna', 'github.com/hasna/legacy', 'main', 'fixture', '2026-07-16', 0, 0, 0, 'legacy-rev'), + (2, ?, 'target', 'hasna', 'github.com/hasna/target', 'main', 'fixture', '2026-07-16', 0, 0, 0, 'target-rev')`) + .run(join(root, "missing-legacy"), repoPath); + db.query("INSERT INTO branches (id, repo_id, name, is_remote, last_commit_sha, ahead, behind) VALUES (101, 1, 'build/test', 1, ?, 3, 4)") + .run(legacyHead.slice(0, 9)); + db.query("INSERT INTO branches (id, repo_id, name, is_remote, last_commit_sha, ahead, behind) VALUES (102, 2, 'build/test', 1, ?, 5, 6)") + .run(head.slice(0, 9)); +} + +function request(changes: Partial = {}): BranchAdjudicationRequest { + return { + actor: "test:branch-adjudication", + idempotencyKey: "test-branch-adjudication", + rows: [ + { + id: 101, + repoId: 1, + name: "build/test", + action: "reclassify-local", + expectedIsRemote: 1, + expectedLastCommitSha: legacyHead.slice(0, 9), + expectedRepoRevision: "legacy-rev", + evidenceRepoPath: repoPath, + evidenceRef: "refs/heads/legacy-preserved/build/test", + }, + { + id: 102, + repoId: 2, + name: "build/test", + action: "reclassify-local", + expectedIsRemote: 1, + expectedLastCommitSha: head.slice(0, 9), + expectedRepoRevision: "target-rev", + evidenceRepoPath: repoPath, + evidenceRef: "refs/heads/build/test", + }, + ], + ...changes, + }; +} + +function expectCode(action: () => unknown, code: string) { + try { + action(); + throw new Error(`expected ${code}`); + } catch (error) { + expect(error).toBeInstanceOf(BranchAdjudicationError); + expect((error as BranchAdjudicationError).code).toBe(code); + } +} + +beforeEach(() => { + closeDb(); + process.env["HASNA_REPOS_DB_PATH"] = ":memory:"; + root = mkdtempSync(join(tmpdir(), "repos-branch-adjudication-")); + seedGitRepo(); + seedDb(); +}); + +afterAll(() => { + closeDb(); + delete process.env["HASNA_REPOS_DB_PATH"]; +}); + +describe("branch adjudication", () => { + it("plans exact branch row reclassification without writing", () => { + const dry = adjudicateBranches(request()); + expect(dry.applied).toBe(false); + expect(dry.plan.can_apply).toBe(true); + expect(dry.plan.rows).toHaveLength(2); + expect(dry.plan.rows[0]!.before.is_remote).toBe(1); + expect(dry.plan.rows[0]!.after.is_remote).toBe(0); + const rows = getDb(":memory:").query("SELECT id, is_remote, ahead, behind FROM branches ORDER BY id").all(); + expect(rows).toEqual([ + { id: 101, is_remote: 1, ahead: 3, behind: 4 }, + { id: 102, is_remote: 1, ahead: 5, behind: 6 }, + ]); + }); + + it("fails dry-run when a duplicate local branch row already exists", () => { + getDb(":memory:").query("INSERT INTO branches (repo_id, name, is_remote, last_commit_sha) VALUES (1, 'build/test', 0, ?)") + .run(legacyHead.slice(0, 9)); + expectCode(() => adjudicateBranches(request()), "DUPLICATE_LOCAL_ROW"); + }); + + it("fails dry-run when evidence ref does not match stored sha", () => { + const bad = request({ rows: [{ ...request().rows[0]!, evidenceRef: "refs/heads/build/test" }] }); + expectCode(() => adjudicateBranches(bad), "EVIDENCE_REF_MISMATCH"); + }); + + it("rejects revision expressions instead of treating them as refs", () => { + const bad = request({ rows: [{ ...request().rows[0]!, evidenceRef: "refs/heads/legacy-preserved/build/test~0" }] }); + expectCode(() => adjudicateBranches(bad), "EVIDENCE_REF_MISMATCH"); + }); + + it("rejects stored commit prefixes that do not disambiguate to the evidence ref", () => { + const bad = request({ rows: [{ ...request().rows[0]!, expectedLastCommitSha: "dead" }] }); + expectCode(() => adjudicateBranches(bad), "ROW_MISMATCH"); + }); + + it("requires the reviewed plan hash for apply", () => { + expectCode(() => adjudicateBranches(request({ apply: true })), "PLAN_HASH_REQUIRED"); + }); + + it("applies only exact guarded rows and writes an audit receipt", () => { + const dry = adjudicateBranches(request()); + const applied = adjudicateBranches(request({ apply: true, expectedPlanHash: dry.plan.plan_hash })); + expect(applied.applied).toBe(true); + expect(applied.replayed).toBe(false); + expect(applied.receipt?.row_count).toBe(2); + const rows = getDb(":memory:").query("SELECT id, is_remote, ahead, behind FROM branches ORDER BY id").all(); + expect(rows).toEqual([ + { id: 101, is_remote: 0, ahead: 0, behind: 0 }, + { id: 102, is_remote: 0, ahead: 0, behind: 0 }, + ]); + const audits = getDb(":memory:").query("SELECT idempotency_key, plan_hash, row_count FROM branch_adjudication_audit").all(); + expect(audits).toEqual([{ idempotency_key: "test-branch-adjudication", plan_hash: dry.plan.plan_hash, row_count: 2 }]); + }); + + it("replays an idempotent apply and rejects conflicting reuse", () => { + const dry = adjudicateBranches(request()); + adjudicateBranches(request({ apply: true, expectedPlanHash: dry.plan.plan_hash })); + const replay = adjudicateBranches(request({ apply: true, expectedPlanHash: dry.plan.plan_hash })); + expect(replay.replayed).toBe(true); + expectCode( + () => adjudicateBranches(request({ + idempotencyKey: "test-branch-adjudication", + rows: [{ ...request().rows[0]! }], + apply: true, + expectedPlanHash: dry.plan.plan_hash, + })), + "IDEMPOTENCY_CONFLICT", + ); + }); + + it("fails apply when the reviewed plan hash is stale", () => { + const dry = adjudicateBranches(request()); + getDb(":memory:").query("UPDATE branches SET ahead = 99 WHERE id = 101").run(); + expectCode(() => adjudicateBranches(request({ apply: true, expectedPlanHash: dry.plan.plan_hash })), "PLAN_HASH_MISMATCH"); + }); +}); diff --git a/src/db/branch-adjudication.ts b/src/db/branch-adjudication.ts new file mode 100644 index 0000000..38e46c1 --- /dev/null +++ b/src/db/branch-adjudication.ts @@ -0,0 +1,658 @@ +import { execFileSync } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import type { Database } from "bun:sqlite"; +import { isAbsolute } from "node:path"; +import { getDb, openNonMigratingDb } from "./database.js"; + +const SCHEMA = "open-repos.branch-adjudication.v1" as const; +const AUDIT_SCHEMA = "open-repos.branch-adjudication-receipt.v1" as const; +const OPERATION = "branch_adjudication" as const; +const HASH_PATTERN = /^[0-9a-f]{64}$/; +const SAFE_KEY_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,199}$/; +const ABBREVIATED_SHA_PATTERN = /^[0-9a-f]{4,64}$/; +const MAX_ROWS = 100; + +export type BranchAdjudicationErrorCode = + | "INVALID_REQUEST" + | "PLAN_HASH_REQUIRED" + | "PLAN_HASH_MISMATCH" + | "IDEMPOTENCY_CONFLICT" + | "ROW_NOT_FOUND" + | "ROW_MISMATCH" + | "REPO_NOT_FOUND" + | "STALE_REPO_ROW" + | "DUPLICATE_LOCAL_ROW" + | "EVIDENCE_REF_MISMATCH" + | "TARGET_NOT_GIT_CHECKOUT" + | "TRANSACTION_CONFLICT"; + +export class BranchAdjudicationError extends Error { + constructor( + public readonly code: BranchAdjudicationErrorCode, + message: string, + public readonly details?: Record, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "BranchAdjudicationError"; + } +} + +export interface BranchAdjudicationRowSpec { + id: number; + repoId: number; + name: string; + action: "reclassify-local"; + expectedIsRemote?: number | boolean; + expectedLastCommitSha: string; + expectedRepoRevision?: string; + evidenceRepoPath: string; + /** + * Exact ref that proves the stored commit is present. Defaults to + * `refs/heads/`. Legacy-source rows may intentionally use a preserved + * evidence ref such as `refs/heads/legacy-preserved/` while the row + * itself remains named `` for relocation's preservation logic. + */ + evidenceRef?: string; +} + +export interface BranchAdjudicationRequest { + rows: BranchAdjudicationRowSpec[]; + actor: string; + idempotencyKey: string; + apply?: boolean; + expectedPlanHash?: string; + databasePath?: string; +} + +export interface BranchAdjudicationPlannedRow { + id: number; + repo_id: number; + name: string; + action: "reclassify-local"; + before: { + is_remote: number; + last_commit_sha: string; + ahead: number; + behind: number; + }; + after: { + is_remote: 0; + last_commit_sha: string; + ahead: 0; + behind: 0; + }; + repo: { + path: string; + updated_at: string; + }; + evidence: { + repo_path: string; + ref: string; + expected_commit: string; + actual_commit: string; + }; +} + +export interface BranchAdjudicationReceipt { + schema: typeof AUDIT_SCHEMA; + id: string; + idempotency_key: string; + request_hash: string; + plan_hash: string; + operation: typeof OPERATION; + actor: string; + row_count: number; + before: BranchAdjudicationPlannedRow[]; + after: BranchAdjudicationPlannedRow[]; + rows: BranchAdjudicationPlannedRow[]; + created_at: string; +} + +export interface BranchAdjudicationResult { + schema: typeof SCHEMA; + ok: true; + applied: boolean; + replayed: boolean; + plan: { + request_hash: string; + plan_hash: string; + can_apply: boolean; + rows: BranchAdjudicationPlannedRow[]; + }; + receipt: BranchAdjudicationReceipt | null; +} + +interface BranchRow { + id: number; + repo_id: number; + name: string; + is_remote: number; + last_commit_sha: string | null; + ahead: number; + behind: number; +} + +interface RepoRow { + id: number; + path: string; + updated_at: string; +} + +interface AuditRow { + id: string; + idempotency_key: string; + request_hash: string; + plan_hash: string; + operation: string; + actor: string; + row_count: number; + before_json: string; + after_json: string; + rows_json: string; + created_at: string; +} + +interface ValidatedRowSpec { + id: number; + repoId: number; + name: string; + action: "reclassify-local"; + expectedIsRemote: 1; + expectedLastCommitSha: string; + expectedRepoRevision?: string; + evidenceRepoPath: string; + evidenceRef: string; +} + +interface ValidatedRequest { + rows: ValidatedRowSpec[]; + actor: string; + idempotencyKey: string; + apply: boolean; + expectedPlanHash?: string; + requestHash: string; + databasePath?: string; +} + +interface BranchAdjudicationPlan { + requestHash: string; + planHash: string; + rows: BranchAdjudicationPlannedRow[]; +} + +let adjudicationDbContext: Database | null = null; + +function adjudicationDb(): Database { + return adjudicationDbContext ?? getDb(); +} + +function fail( + code: BranchAdjudicationErrorCode, + message: string, + details?: Record, + cause?: unknown, +): never { + throw new BranchAdjudicationError( + code, + message, + details, + cause === undefined ? undefined : { cause }, + ); +} + +function stable(value: unknown): string { + if (value === null || typeof value !== "object") return JSON.stringify(value); + if (Array.isArray(value)) return `[${value.map(stable).join(",")}]`; + const object = value as Record; + return `{${Object.keys(object).sort().map((key) => `${JSON.stringify(key)}:${stable(object[key])}`).join(",")}}`; +} + +function hash(value: unknown): string { + return createHash("sha256").update(stable(value)).digest("hex"); +} + +function normalizeKey(value: unknown, label: string): string { + if (typeof value !== "string" || !SAFE_KEY_PATTERN.test(value)) { + fail("INVALID_REQUEST", `${label} must be a stable non-empty key`); + } + return value; +} + +function normalizeSha(value: unknown, label: string): string { + if (typeof value !== "string" || !ABBREVIATED_SHA_PATTERN.test(value.toLowerCase())) { + fail("INVALID_REQUEST", `${label} must be a lowercase hexadecimal commit prefix`); + } + return value.toLowerCase(); +} + +function normalizeAbsolutePath(value: unknown, label: string): string { + if (typeof value !== "string" || !value || value.includes("\0") || !isAbsolute(value)) { + fail("INVALID_REQUEST", `${label} must be a non-empty absolute path`); + } + return value; +} + +function normalizeRef(value: unknown, label: string): string { + if (typeof value !== "string" || !value || value.includes("\0") || value.includes("\n") || value.includes("\r")) { + fail("INVALID_REQUEST", `${label} must be a non-empty ref`); + } + if (!value.startsWith("refs/heads/")) { + fail("INVALID_REQUEST", `${label} must be an exact refs/heads/* ref`); + } + return value; +} + +function normalizeName(value: unknown): string { + if (typeof value !== "string" || !value || value.includes("\0") || value.includes("\n") || value.includes("\r")) { + fail("INVALID_REQUEST", "branch name must be non-empty and single-line"); + } + return value; +} + +function parseBooleanRemote(value: unknown): 1 { + if (value === undefined || value === true || value === 1) return 1; + fail("INVALID_REQUEST", "reclassify-local requires expected_is_remote=1"); +} + +function normalizePositiveInteger(value: unknown, label: string): number { + if (typeof value !== "number" || !Number.isInteger(value) || value < 1) { + fail("INVALID_REQUEST", `${label} must be a positive integer`); + } + return value; +} + +function validateRequest(request: BranchAdjudicationRequest): ValidatedRequest { + const actor = normalizeKey(request.actor, "actor"); + const idempotencyKey = normalizeKey(request.idempotencyKey, "idempotency key"); + if (!Array.isArray(request.rows) || request.rows.length === 0 || request.rows.length > MAX_ROWS) { + fail("INVALID_REQUEST", `rows must contain 1-${MAX_ROWS} entries`); + } + const seen = new Set(); + const rows = request.rows.map((row, index): ValidatedRowSpec => { + if (!row || typeof row !== "object") fail("INVALID_REQUEST", `row ${index} must be an object`); + if (row.action !== "reclassify-local") fail("INVALID_REQUEST", `row ${index} action must be reclassify-local`); + const id = normalizePositiveInteger(row.id, `row ${index} id`); + if (seen.has(id)) fail("INVALID_REQUEST", `row ${id} is duplicated`); + seen.add(id); + const name = normalizeName(row.name); + const expectedRepoRevision = row.expectedRepoRevision; + if (expectedRepoRevision !== undefined && (typeof expectedRepoRevision !== "string" || expectedRepoRevision.length === 0)) { + fail("INVALID_REQUEST", `row ${id} expected repo revision must be non-empty when supplied`); + } + return { + id, + repoId: normalizePositiveInteger(row.repoId, `row ${index} repo id`), + name, + action: "reclassify-local", + expectedIsRemote: parseBooleanRemote(row.expectedIsRemote), + expectedLastCommitSha: normalizeSha(row.expectedLastCommitSha, `row ${id} expected last commit sha`), + expectedRepoRevision, + evidenceRepoPath: normalizeAbsolutePath(row.evidenceRepoPath, `row ${id} evidence repo path`), + evidenceRef: normalizeRef(row.evidenceRef ?? `refs/heads/${name}`, `row ${id} evidence ref`), + }; + }).sort((left, right) => left.id - right.id); + const requestEnvelope = { + operation: OPERATION, + actor, + idempotency_key: idempotencyKey, + rows: rows.map((row) => ({ + id: row.id, + repo_id: row.repoId, + name: row.name, + action: row.action, + expected_is_remote: row.expectedIsRemote, + expected_last_commit_sha: row.expectedLastCommitSha, + expected_repo_revision: row.expectedRepoRevision ?? null, + evidence_repo_path: row.evidenceRepoPath, + evidence_ref: row.evidenceRef, + })), + }; + if (request.apply) { + if (!request.expectedPlanHash) fail("PLAN_HASH_REQUIRED", "apply requires the exact reviewed dry-run plan hash"); + if (!HASH_PATTERN.test(request.expectedPlanHash)) fail("INVALID_REQUEST", "expected plan hash must be a lowercase sha256 digest"); + } + return { + rows, + actor, + idempotencyKey, + apply: Boolean(request.apply), + expectedPlanHash: request.expectedPlanHash, + requestHash: hash(requestEnvelope), + databasePath: request.databasePath, + }; +} + +function runGit( + path: string, + args: string[], + code: BranchAdjudicationErrorCode = "TARGET_NOT_GIT_CHECKOUT", + message = "evidence repo path is not a readable Git checkout", + details?: Record, +): string { + try { + return execFileSync("git", ["-C", path, ...args], { + encoding: "utf8", + stdio: ["ignore", "pipe", "pipe"], + timeout: 10_000, + env: { + ...process.env, + GIT_CONFIG_NOSYSTEM: "1", + GIT_CONFIG_GLOBAL: process.platform === "win32" ? "NUL" : "/dev/null", + GIT_TERMINAL_PROMPT: "0", + }, + }).trim(); + } catch (cause) { + fail(code, message, details, cause); + } +} + +function resolveEvidenceCommit(spec: ValidatedRowSpec): string { + runGit( + spec.evidenceRepoPath, + ["check-ref-format", spec.evidenceRef], + "EVIDENCE_REF_MISMATCH", + "evidence ref is not an exact Git ref name", + { row_id: spec.id, evidence_ref: spec.evidenceRef }, + ); + const actual = runGit( + spec.evidenceRepoPath, + ["show-ref", "--verify", "--hash", spec.evidenceRef], + "EVIDENCE_REF_MISMATCH", + "evidence ref does not exist as an exact ref", + { row_id: spec.id, evidence_ref: spec.evidenceRef }, + ).toLowerCase(); + const objectType = runGit( + spec.evidenceRepoPath, + ["cat-file", "-t", actual], + "EVIDENCE_REF_MISMATCH", + "evidence ref object is unreadable", + { row_id: spec.id, evidence_ref: spec.evidenceRef, actual_commit: actual }, + ); + if (objectType !== "commit") { + fail("EVIDENCE_REF_MISMATCH", "evidence ref does not point at a commit", { + row_id: spec.id, + evidence_ref: spec.evidenceRef, + actual_commit: actual, + object_type: objectType, + }); + } + const candidates = runGit( + spec.evidenceRepoPath, + ["rev-parse", `--disambiguate=${spec.expectedLastCommitSha}`], + "EVIDENCE_REF_MISMATCH", + "stored commit prefix cannot be resolved", + { row_id: spec.id, expected_last_commit_sha: spec.expectedLastCommitSha }, + ).split(/\r?\n/).map((line) => line.trim().toLowerCase()).filter(Boolean); + if (candidates.length !== 1 || candidates[0] !== actual) { + fail("EVIDENCE_REF_MISMATCH", "evidence ref does not resolve to the expected stored commit", { + row_id: spec.id, + expected_last_commit_sha: spec.expectedLastCommitSha, + evidence_ref: spec.evidenceRef, + actual_commit: actual, + matching_objects: candidates.length, + }); + } + return actual; +} + +function getBranchRow(id: number): BranchRow | null { + return adjudicationDb().query("SELECT * FROM branches WHERE id = ?").get(id) as BranchRow | null; +} + +function getRepoRow(id: number): RepoRow | null { + return adjudicationDb().query("SELECT id, path, updated_at FROM repos WHERE id = ?").get(id) as RepoRow | null; +} + +function duplicateLocalRow(spec: ValidatedRowSpec): BranchRow | null { + return adjudicationDb().query(` + SELECT * FROM branches + WHERE repo_id = ? AND name = ? AND is_remote = 0 AND id != ? + `).get(spec.repoId, spec.name, spec.id) as BranchRow | null; +} + +function buildPlan(request: ValidatedRequest): BranchAdjudicationPlan { + const plannedRows: BranchAdjudicationPlannedRow[] = []; + for (const spec of request.rows) { + const repo = getRepoRow(spec.repoId); + if (!repo) fail("REPO_NOT_FOUND", "repo row not found", { repo_id: spec.repoId, row_id: spec.id }); + if (spec.expectedRepoRevision !== undefined && repo.updated_at !== spec.expectedRepoRevision) { + fail("STALE_REPO_ROW", "repo row revision changed", { + repo_id: spec.repoId, + row_id: spec.id, + expected_repo_revision: spec.expectedRepoRevision, + actual_repo_revision: repo.updated_at, + }); + } + const row = getBranchRow(spec.id); + if (!row) fail("ROW_NOT_FOUND", "branch row not found", { row_id: spec.id }); + const actualSha = String(row.last_commit_sha ?? "").toLowerCase(); + const mismatches: Record = {}; + if (row.repo_id !== spec.repoId) mismatches["repo_id"] = row.repo_id; + if (row.name !== spec.name) mismatches["name"] = row.name; + if (row.is_remote !== spec.expectedIsRemote) mismatches["is_remote"] = row.is_remote; + if (actualSha !== spec.expectedLastCommitSha) mismatches["last_commit_sha"] = actualSha; + if (Object.keys(mismatches).length > 0) { + fail("ROW_MISMATCH", "branch row does not match exact guarded input", { row_id: spec.id, mismatches }); + } + const duplicate = duplicateLocalRow(spec); + if (duplicate) { + fail("DUPLICATE_LOCAL_ROW", "a local branch row already exists for this repo/name", { + row_id: spec.id, + duplicate_row_id: duplicate.id, + repo_id: spec.repoId, + name: spec.name, + }); + } + const evidenceCommit = resolveEvidenceCommit(spec); + plannedRows.push({ + id: row.id, + repo_id: row.repo_id, + name: row.name, + action: "reclassify-local", + before: { is_remote: row.is_remote, last_commit_sha: actualSha, ahead: row.ahead, behind: row.behind }, + after: { is_remote: 0, last_commit_sha: actualSha, ahead: 0, behind: 0 }, + repo: { path: repo.path, updated_at: repo.updated_at }, + evidence: { + repo_path: spec.evidenceRepoPath, + ref: spec.evidenceRef, + expected_commit: spec.expectedLastCommitSha, + actual_commit: evidenceCommit, + }, + }); + } + return { + requestHash: request.requestHash, + planHash: hash({ operation: OPERATION, request_hash: request.requestHash, rows: plannedRows }), + rows: plannedRows, + }; +} + +function receiptFromAudit(row: AuditRow): BranchAdjudicationReceipt { + return { + schema: AUDIT_SCHEMA, + id: row.id, + idempotency_key: row.idempotency_key, + request_hash: row.request_hash, + plan_hash: row.plan_hash, + operation: OPERATION, + actor: row.actor, + row_count: row.row_count, + before: JSON.parse(row.before_json) as BranchAdjudicationPlannedRow[], + after: JSON.parse(row.after_json) as BranchAdjudicationPlannedRow[], + rows: JSON.parse(row.rows_json) as BranchAdjudicationPlannedRow[], + created_at: row.created_at, + }; +} + +function existingAudit(idempotencyKey: string): AuditRow | null { + const hasTable = adjudicationDb().query( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'branch_adjudication_audit'", + ).get(); + if (!hasTable) return null; + return adjudicationDb().query("SELECT * FROM branch_adjudication_audit WHERE idempotency_key = ?").get(idempotencyKey) as AuditRow | null; +} + +function existingIdempotentResult(request: ValidatedRequest): BranchAdjudicationResult | null { + const row = existingAudit(request.idempotencyKey); + if (!row) return null; + if (row.request_hash !== request.requestHash || row.plan_hash !== request.expectedPlanHash) { + fail("IDEMPOTENCY_CONFLICT", "idempotency key was already used for a different branch adjudication", { + idempotency_key: request.idempotencyKey, + }); + } + const receipt = receiptFromAudit(row); + return { + schema: SCHEMA, + ok: true, + applied: true, + replayed: true, + plan: { request_hash: row.request_hash, plan_hash: row.plan_hash, can_apply: true, rows: receipt.rows }, + receipt, + }; +} + +export function adjudicateBranches(request: BranchAdjudicationRequest): BranchAdjudicationResult { + const validated = validateRequest(request); + if (validated.apply) { + const db = getDb(validated.databasePath); + adjudicationDbContext = db; + try { + const replay = existingIdempotentResult(validated); + if (replay) return replay; + } finally { + adjudicationDbContext = null; + } + } + + const planningContext = validated.apply ? null : openNonMigratingDb(validated.databasePath); + if (planningContext) adjudicationDbContext = planningContext.db; + let plan: BranchAdjudicationPlan; + try { + plan = buildPlan(validated); + } finally { + adjudicationDbContext = null; + planningContext?.close(); + } + + if (!validated.apply) { + return { + schema: SCHEMA, + ok: true, + applied: false, + replayed: false, + plan: { request_hash: plan.requestHash, plan_hash: plan.planHash, can_apply: true, rows: plan.rows }, + receipt: null, + }; + } + + if (validated.expectedPlanHash !== plan.planHash) { + fail("PLAN_HASH_MISMATCH", "live branch adjudication plan differs from the reviewed dry-run plan", { + expected_plan_hash: validated.expectedPlanHash, + actual_plan_hash: plan.planHash, + }); + } + + const db = getDb(validated.databasePath); + adjudicationDbContext = db; + let began = false; + try { + db.exec("BEGIN IMMEDIATE"); + began = true; + const replay = existingIdempotentResult(validated); + if (replay) { + db.exec("COMMIT"); + return replay; + } + const currentPlan = buildPlan(validated); + if (currentPlan.planHash !== plan.planHash || currentPlan.planHash !== validated.expectedPlanHash) { + fail("PLAN_HASH_MISMATCH", "registry state changed after dry-run review", { + expected_plan_hash: validated.expectedPlanHash, + actual_plan_hash: currentPlan.planHash, + }); + } + const update = db.query(` + UPDATE branches + SET is_remote = 0, ahead = 0, behind = 0 + WHERE id = ? AND repo_id = ? AND name = ? AND is_remote = 1 AND last_commit_sha = ? + `); + for (const row of currentPlan.rows) { + const result = update.run(row.id, row.repo_id, row.name, row.before.last_commit_sha); + if (result.changes !== 1) fail("TRANSACTION_CONFLICT", "exact branch row update failed", { row_id: row.id }); + } + const afterRows = currentPlan.rows.map((row) => { + const updated = getBranchRow(row.id); + if ( + !updated + || updated.repo_id !== row.repo_id + || updated.name !== row.name + || updated.is_remote !== 0 + || String(updated.last_commit_sha ?? "").toLowerCase() !== row.before.last_commit_sha + || updated.ahead !== 0 + || updated.behind !== 0 + ) { + fail("TRANSACTION_CONFLICT", "exact branch row verification failed", { row_id: row.id }); + } + return { + ...row, + after: { + is_remote: 0 as const, + last_commit_sha: String(updated.last_commit_sha ?? "").toLowerCase(), + ahead: 0 as const, + behind: 0 as const, + }, + }; + }); + const createdAt = new Date().toISOString(); + const receipt: BranchAdjudicationReceipt = { + schema: AUDIT_SCHEMA, + id: randomUUID(), + idempotency_key: validated.idempotencyKey, + request_hash: validated.requestHash, + plan_hash: currentPlan.planHash, + operation: OPERATION, + actor: validated.actor, + row_count: currentPlan.rows.length, + before: currentPlan.rows, + after: afterRows, + rows: currentPlan.rows, + created_at: createdAt, + }; + db.query(` + INSERT INTO branch_adjudication_audit ( + id, idempotency_key, request_hash, plan_hash, operation, actor, + row_count, before_json, after_json, rows_json, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `).run( + receipt.id, + receipt.idempotency_key, + receipt.request_hash, + receipt.plan_hash, + receipt.operation, + receipt.actor, + receipt.row_count, + JSON.stringify(receipt.before), + JSON.stringify(receipt.after), + JSON.stringify(receipt.rows), + receipt.created_at, + ); + db.exec("COMMIT"); + began = false; + return { + schema: SCHEMA, + ok: true, + applied: true, + replayed: false, + plan: { request_hash: currentPlan.requestHash, plan_hash: currentPlan.planHash, can_apply: true, rows: currentPlan.rows }, + receipt, + }; + } catch (error) { + if (began) { + try { db.exec("ROLLBACK"); } catch { /* preserve original error */ } + } + if (error instanceof BranchAdjudicationError) throw error; + fail("TRANSACTION_CONFLICT", "branch adjudication failed", undefined, error); + } finally { + adjudicationDbContext = null; + } +} diff --git a/src/db/database.test.ts b/src/db/database.test.ts index 89dab46..cd90f38 100644 --- a/src/db/database.test.ts +++ b/src/db/database.test.ts @@ -112,7 +112,7 @@ describe("database", () => { const migrated = getDb(path); expect(migrated).toBe(raw); expect(migrated.query("SELECT version FROM migrations ORDER BY version").all()) - .toEqual([1, 2, 3, 4, 6, 7, 8, 9, 10].map((version) => ({ version }))); + .toEqual([1, 2, 3, 4, 6, 7, 8, 9, 10, 11].map((version) => ({ version }))); expect(getDb(path)).toBe(migrated); expect(migrated.query("SELECT count(*) AS count FROM migrations WHERE version = 9").get()) .toEqual({ count: 1 }); @@ -179,6 +179,12 @@ describe("database", () => { expect(table).toBeTruthy(); }); + it("should create the durable branch adjudication audit table", () => { + const db = getDb(":memory:"); + const table = db.query("SELECT name FROM sqlite_master WHERE type='table' AND name='branch_adjudication_audit'").get(); + expect(table).toBeTruthy(); + }); + it("should create FTS5 tables", () => { const db = getDb(":memory:"); const ftsRepos = db.query("SELECT name FROM sqlite_master WHERE name='fts_repos'").get(); @@ -193,7 +199,7 @@ describe("database", () => { const db = getDb(":memory:"); const migrations = db.query("SELECT version FROM migrations ORDER BY version").all() as { version: number }[]; expect(migrations.length).toBeGreaterThanOrEqual(5); - expect(migrations.map((row) => row.version)).toEqual([1, 2, 3, 4, 6, 7, 8, 9, 10]); + expect(migrations.map((row) => row.version)).toEqual([1, 2, 3, 4, 6, 7, 8, 9, 10, 11]); }); it("migrates existing branch uniqueness to include remote classification", () => { @@ -297,7 +303,7 @@ describe("database", () => { expect(db.query("SELECT name FROM sqlite_master WHERE type='table' AND name='worktree_leases'").get()).toBeTruthy(); expect(db.query("SELECT name FROM sqlite_master WHERE type='table' AND name='repo_relocation_audit'").get()).toBeTruthy(); expect((db.query("SELECT version FROM migrations ORDER BY version").all() as { version: number }[]) - .map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]); + .map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]); expect(db.query("PRAGMA foreign_key_check").all()).toEqual([]); } finally { closeDb(); @@ -814,7 +820,7 @@ describe("database", () => { const results = await Promise.all(children.map(({ completed }) => completed)); expect(results).toEqual(Array.from({ length: 8 }, () => ({ code: 0, - stdout: JSON.stringify([1, 2, 3, 4, 6, 7, 8, 9, 10].map((version) => ({ version }))), + stdout: JSON.stringify([1, 2, 3, 4, 6, 7, 8, 9, 10, 11].map((version) => ({ version }))), stderr: "", }))); } finally { diff --git a/src/db/database.ts b/src/db/database.ts index e607a98..ba0e575 100644 --- a/src/db/database.ts +++ b/src/db/database.ts @@ -758,4 +758,25 @@ const MIGRATIONS: Migration[] = [ `); }, }, + { + version: 11, + sql: ` + CREATE TABLE IF NOT EXISTS branch_adjudication_audit ( + id TEXT PRIMARY KEY, + idempotency_key TEXT NOT NULL UNIQUE, + request_hash TEXT NOT NULL, + plan_hash TEXT NOT NULL, + operation TEXT NOT NULL, + actor TEXT NOT NULL, + row_count INTEGER NOT NULL, + before_json TEXT NOT NULL, + after_json TEXT NOT NULL, + rows_json TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + CREATE INDEX IF NOT EXISTS idx_branch_adjudication_audit_created + ON branch_adjudication_audit(created_at); + `, + }, ]; diff --git a/src/index.ts b/src/index.ts index 484539c..57a7237 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,6 +4,18 @@ export type { DatabaseOpenOptions, NonMigratingDatabaseContext } from "./db/data export { applyPostgresMigrations } from "./db/pg-migrations.js"; export type { PostgresMigrationClient } from "./db/pg-migrations.js"; export { sanitizeRemoteIdentity, sanitizeRemoteOutput } from "./lib/remote-identity.js"; +export { + BranchAdjudicationError, + adjudicateBranches, +} from "./db/branch-adjudication.js"; +export type { + BranchAdjudicationErrorCode, + BranchAdjudicationPlannedRow, + BranchAdjudicationReceipt, + BranchAdjudicationRequest, + BranchAdjudicationResult, + BranchAdjudicationRowSpec, +} from "./db/branch-adjudication.js"; export { PrimaryRelocationError, relocatePrimaryRepo,