#!/usr/bin/env python3 """Select and run the smallest valid Blacksite verification gates.""" from __future__ import annotations import argparse import hashlib import json import os import subprocess import sys import tempfile import time from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable, Sequence try: import tomllib except ModuleNotFoundError as error: # pragma: no cover raise SystemExit("verify.py requires Python 3.11 or newer") from error @dataclass(frozen=True) class Gate: name: str command: tuple[str, ...] reason: str lane: str | None = None compile_gate: bool = False native_required: bool = False @dataclass(frozen=True) class LocalPackage: name: str root: str dependencies: frozenset[str] DIGEST_SCHEMA_VERSION = 2 RUST_GATE_CONFIG_INPUTS = ( "Cargo.toml", "Cargo.lock", ".cargo/config.toml", ".codex/workflow.toml", "rust-toolchain", "rust-toolchain.toml", ".rustfmt.toml", "rustfmt.toml", "clippy.toml", "scripts/codex/cargo_lane.py", ) RUST_GATE_ENVIRONMENT = ( "CARGO_BUILD_RUSTC_WRAPPER", "CARGO_BUILD_TARGET", "CARGO_ENCODED_RUSTFLAGS", "CARGO_HOME", "CC", "CFLAGS", "CXX", "CXXFLAGS", "HOST", "LD", "RUSTC", "RUSTC_WRAPPER", "RUSTDOCFLAGS", "RUSTFLAGS", "RUSTUP_TOOLCHAIN", "TARGET", ) IGNORED_SCOPE_DIRECTORIES = { ".git", ".import-cache", ".codex", "node_modules", "target", } def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def repository_root() -> Path: override = os.environ.get("BLACKSITE_WORKSPACE_ROOT") configured = None config_path = Path.cwd() / ".codex/config.toml" if not override and config_path.is_file(): with config_path.open("rb") as handle: configured = tomllib.load(handle).get("workspace_root") candidate = Path(override or configured).expanduser() if (override or configured) else Path(__file__).parents[2] candidate = candidate.absolute() for current in (candidate, *candidate.parents): if (current / "Cargo.toml").is_file() and (current / ".git").exists(): return current raise SystemExit(f"could not locate Blacksite repository from {candidate}") def load_workflow(root: Path) -> dict[str, Any]: path = root / ".codex" / "workflow.toml" if not path.is_file(): raise SystemExit(f"missing workflow configuration: {path}") with path.open("rb") as handle: document = tomllib.load(handle) if document.get("version") != 1: raise SystemExit(f"unsupported workflow version in {path}") return document def run_text(root: Path, command: Sequence[str]) -> str: completed = subprocess.run( command, cwd=root, check=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, ) if completed.returncode: raise SystemExit( f"command failed ({completed.returncode}): {' '.join(command)}\n" + completed.stdout[-4000:] ) return completed.stdout def changed_paths(root: Path) -> list[str]: output = subprocess.run( ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], cwd=root, check=True, stdout=subprocess.PIPE, ).stdout fields = output.split(b"\0") paths: list[str] = [] index = 0 while index < len(fields): field = fields[index] index += 1 if not field: continue text = field.decode("utf-8", "surrogateescape") status = text[:2] path = text[3:] if status[:1] in {"R", "C"} and index < len(fields): target = fields[index].decode("utf-8", "surrogateescape") index += 1 paths.extend((path, target)) else: paths.append(path) return sorted(set(paths)) def normalize_paths(values: Iterable[str]) -> list[str]: return sorted({value.replace("\\", "/").removeprefix("./") for value in values}) def classify(paths: Sequence[str]) -> set[str]: areas: set[str] = set() for path in paths: if ( path == "AGENTS.md" or path.endswith("/AGENTS.md") or path.startswith((".agents/", ".codex/", "scripts/codex/", ".cursor/rules/")) or path in {".gitignore", ".cargo/config.toml"} ): areas.add("workflow") if path == "README.md" or path.endswith(".md") or path.startswith("docs/"): areas.add("docs") if path.startswith("crates/content_pipeline/"): areas.add("content_pipeline") if path.startswith("crates/shared/"): areas.add("shared") if path.startswith("crates/editor/"): areas.add("editor") if path.startswith("crates/editor/src/ui/"): areas.add("editor_ui") if path.startswith("crates/blacksite_surface/") or path.endswith((".wgsl", ".shader.ron")): areas.add("surface") if path.startswith("xtask/"): areas.add("xtask") if path.startswith("crates/scene/"): areas.update(("scene", "migration")) if path.startswith("crates/settings/"): areas.add("settings") if path.startswith("crates/game/"): areas.add("game") if path.startswith("assets/"): areas.add("assets") if any(token in path for token in ("migrat", "upgrade", "schema")) and not path.endswith(".md"): areas.add("migration") if path.endswith(".rs") or path in {"Cargo.toml", "Cargo.lock", ".cargo/config.toml"}: areas.add("rust") return areas def cargo_gate(name: str, lane: str, reason: str, *cargo_args: str) -> Gate: return Gate( name=name, command=("cargo", *cargo_args), lane=lane, reason=reason, compile_gate=any( item in cargo_args for item in ("check", "test", "clippy", "build", "run") ), ) def command_gate(name: str, reason: str, *command: str) -> Gate: return Gate(name=name, command=tuple(command), reason=reason) def add_unique(gates: list[Gate], gate: Gate) -> None: if not any(existing.name == gate.name for existing in gates): gates.append(gate) def select_gates(tier: str, areas: set[str], root: Path) -> tuple[list[Gate], list[str]]: gates: list[Gate] = [] skipped: list[str] = [] python = sys.executable if not areas: return [], ["No changed inputs; all gates skipped."] if "workflow" in areas: for script in ( "state.py", "summarize_command.py", "cargo_lane.py", "build_storage.py", "architecture_audit.py", ): if (root / "scripts" / "codex" / script).is_file(): add_unique( gates, command_gate( f"workflow-{script.removesuffix('.py')}", "Workflow implementation changed.", python, f"scripts/codex/{script}", "self-test", ), ) add_unique( gates, command_gate( "workflow-verify-plan", "Verification selection must remain deterministic.", python, "scripts/codex/verify.py", "plan", "--paths", "docs/README.md", ), ) if "docs" in areas or "workflow" in areas: add_unique( gates, command_gate( "docs-audit-changed", "Documentation or workflow authority changed.", python, "scripts/codex/docs_audit.py", "--changed", ), ) rust_areas = areas.intersection( {"content_pipeline", "shared", "editor", "surface", "xtask", "scene", "settings", "game"} ) if rust_areas or "workflow" in areas: add_unique( gates, command_gate( "architecture-audit", "Production architecture inputs changed.", python, "scripts/codex/architecture_audit.py", "check", ), ) if rust_areas: add_unique( gates, cargo_gate("rustfmt", "dev", "Rust source changed.", "fmt", "--all", "--", "--check"), ) package_specs = { "content_pipeline": ("content_pipeline", "lib"), "shared": ("shared", "lib"), "editor": ("editor", "lib"), "surface": ("blacksite_surface", "lib"), "scene": ("scene", "lib"), "settings": ("settings", "lib"), "game": ("game", "lib"), "xtask": ("xtask", "bins"), } for area, (package, target_kind) in package_specs.items(): if area not in areas: continue target_flag = "--bins" if target_kind == "bins" else "--lib" add_unique( gates, cargo_gate( f"check-{package}", "dev", f"{area} owns changed Rust inputs.", "check", "-p", package, target_flag, ), ) if area not in {"editor", "xtask"}: add_unique( gates, cargo_gate( f"test-{package}", "dev", f"Focused {area} invariants are lightweight enough for this tier.", "test", "-p", package, target_flag, ), ) elif tier == "fast": skipped.append(f"Skipped heavy {package} test binary in the fast loop.") if tier == "slice": for area, (package, target_kind) in package_specs.items(): if area not in areas: continue target_flag = "--bins" if target_kind == "bins" else "--lib" add_unique( gates, cargo_gate( f"clippy-{package}", "dev", f"Stable {area} slice requires strict affected-package lint.", "clippy", "-p", package, target_flag, "--", "-D", "warnings", ), ) if "editor_ui" in areas: gates.append( Gate( name="native-editor-ui", command=("bash", "scripts/codex/native_qa.sh", "plan", "material-slot-live-edit"), reason="Editor UI changed; native interaction is a slice requirement.", native_required=True, ) ) if "assets" in areas or "content_pipeline" in areas: add_unique( gates, cargo_gate( "process-assets-check", "package", "Asset processing behavior changed.", "process-assets", "--project", ".", "--check", ), ) if "migration" in areas: skipped.append("Migration dry-run is required at slice review and needs an explicit fixture/project.") if tier == "candidate": gates = [ command_gate( "candidate-architecture", "Candidate modules must satisfy the architecture debt ratchet.", python, "scripts/codex/architecture_audit.py", "check", ), cargo_gate( "candidate-tests", "candidate", "Nominated candidate requires full all-feature tests once.", "test", "--workspace", "--all-features", ), cargo_gate( "candidate-clippy", "candidate", "Nominated candidate requires distinct strict lint evidence.", "clippy", "--workspace", "--all-features", "--all-targets", "--", "-D", "warnings", ), cargo_gate( "candidate-process-assets", "package", "Candidate content catalog must validate deterministically.", "process-assets", "--project", ".", "--check", ), cargo_gate( "candidate-levels", "package", "Candidate levels require headless validation.", "validate-levels", ), cargo_gate( "candidate-samples", "package", "Candidate samples require headless validation.", "validate-samples", ), cargo_gate( "candidate-package", "package", "Candidate package output must be generated and validated.", "package-project", "--project", ".", "--profile", "qa", ), command_gate( "candidate-docs", "Candidate documentation authority must be coherent.", python, "scripts/codex/docs_audit.py", ), ] skipped.append("No redundant workspace check precedes candidate workspace tests.") return gates, skipped def state_is_candidate_ready(root: Path, workflow: dict[str, Any]) -> bool: state_path = root / workflow["session"]["state_file"] if not state_path.is_file(): return False return "Current state: Candidate-ready" in state_path.read_text(encoding="utf-8") def ledger_path(root: Path, workflow: dict[str, Any]) -> Path: return root / workflow["session"]["verification_file"] def read_ledger(path: Path) -> dict[str, Any]: if not path.is_file(): return {"version": 1, "gates": {}, "invalidations": []} try: document = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {"version": 1, "gates": {}, "invalidations": []} return document if isinstance(document, dict) else {"version": 1, "gates": {}} def atomic_json(path: Path, document: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(document, handle, indent=2, sort_keys=True) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) finally: try: os.unlink(temporary) except FileNotFoundError: pass def _dependency_tables(document: dict[str, Any]) -> Iterable[dict[str, Any]]: for key in ("dependencies", "dev-dependencies", "build-dependencies"): table = document.get(key) if isinstance(table, dict): yield table targets = document.get("target") if isinstance(targets, dict): for target in targets.values(): if not isinstance(target, dict): continue for key in ("dependencies", "dev-dependencies", "build-dependencies"): table = target.get(key) if isinstance(table, dict): yield table def workspace_packages(root: Path) -> dict[str, LocalPackage]: workspace_manifest = root / "Cargo.toml" if not workspace_manifest.is_file(): return {} try: with workspace_manifest.open("rb") as handle: workspace = tomllib.load(handle) except (OSError, tomllib.TOMLDecodeError): return {} members = workspace.get("workspace", {}).get("members", []) package_documents: list[tuple[str, str, dict[str, Any]]] = [] for member in members if isinstance(members, list) else []: if not isinstance(member, str): continue for candidate in sorted(root.glob(member)): manifest = candidate / "Cargo.toml" if candidate.is_dir() else candidate if not manifest.is_file(): continue try: with manifest.open("rb") as handle: document = tomllib.load(handle) except (OSError, tomllib.TOMLDecodeError): continue package = document.get("package") if not isinstance(package, dict) or not isinstance(package.get("name"), str): continue package_documents.append( ( package["name"], manifest.parent.relative_to(root).as_posix(), document, ) ) local_names = {name for name, _, _ in package_documents} packages: dict[str, LocalPackage] = {} for name, relative_root, document in package_documents: dependencies: set[str] = set() for table in _dependency_tables(document): for dependency_name, specification in table.items(): resolved_name = dependency_name if isinstance(specification, dict) and isinstance(specification.get("package"), str): resolved_name = specification["package"] if resolved_name in local_names: dependencies.add(resolved_name) packages[name] = LocalPackage(name, relative_root, frozenset(dependencies)) return packages def local_dependency_closure(packages: dict[str, LocalPackage], names: Iterable[str]) -> set[str]: closure: set[str] = set() pending = list(names) while pending: name = pending.pop() if name in closure or name not in packages: continue closure.add(name) pending.extend(packages[name].dependencies) return closure def cargo_scope(root: Path, gate: Gate) -> tuple[set[str], set[str], bool]: """Return local package names, extra roots, and whether only Rust source matters.""" packages = workspace_packages(root) arguments = list(gate.command[1:]) if gate.command[:1] == ("cargo",) else list(gate.command) package_name: str | None = None for index, argument in enumerate(arguments): if argument in {"-p", "--package"} and index + 1 < len(arguments): package_name = arguments[index + 1] break if argument.startswith("--package="): package_name = argument.partition("=")[2] break subcommand = arguments[0] if arguments else "" rust_only = subcommand == "fmt" if package_name is not None: selected = local_dependency_closure(packages, (package_name,)) elif subcommand in {"process-assets", "validate-levels", "validate-samples", "package-project"}: selected = local_dependency_closure(packages, ("xtask",)) else: selected = set(packages) extra_roots: set[str] = set() if subcommand in {"process-assets", "validate-levels", "validate-samples", "package-project"}: extra_roots.add("assets") return selected, extra_roots, rust_only def _under_scope(relative: str, roots: Iterable[str]) -> bool: return any( root in {"", "."} or relative == root or relative.startswith(f"{root}/") for root in roots ) def _is_documentation(relative: str) -> bool: path = relative.lower() return path == "readme.md" or path == "agents.md" or path.endswith(".md") or path.startswith("docs/") def _walk_files(root: Path, roots: Iterable[str]) -> set[str]: found: set[str] = set() for relative_root in roots: start = root if relative_root in {"", "."} else root / relative_root if start.is_file(): found.add(start.relative_to(root).as_posix()) continue if not start.is_dir(): continue for directory, directory_names, file_names in os.walk(start): directory_names[:] = [ name for name in directory_names if name not in IGNORED_SCOPE_DIRECTORIES ] base = Path(directory) for file_name in file_names: found.add((base / file_name).relative_to(root).as_posix()) return found def _tracked_scope_files(root: Path, roots: Iterable[str]) -> set[str]: normalized_roots = sorted(set(roots)) if not normalized_roots: return set() if (root / ".git").exists(): completed = subprocess.run( ["git", "ls-files", "-z", "--", *normalized_roots], cwd=root, check=False, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, ) if completed.returncode == 0: return { item.decode("utf-8", "surrogateescape") for item in completed.stdout.split(b"\0") if item } return _walk_files(root, normalized_roots) def gate_static_inputs(root: Path, gate: Gate) -> set[str]: inputs: set[str] = set() if gate.lane is not None: inputs.update(RUST_GATE_CONFIG_INPUTS) packages = workspace_packages(root) selected, _, _ = cargo_scope(root, gate) inputs.update(f"{packages[name].root}/Cargo.toml" for name in selected if name in packages) return inputs for argument in gate.command: normalized = argument.replace("\\", "/").removeprefix("./") if ( normalized and not normalized.startswith("-") and not Path(normalized).is_absolute() and (root / normalized).is_file() ): inputs.add(normalized) if gate.name.startswith("workflow-"): inputs.add(".codex/workflow.toml") if gate.name.startswith("docs-") or gate.name == "candidate-docs": inputs.add("docs/authority.toml") if gate.native_required and gate.command: scenario = gate.command[-1] inputs.add( f".agents/skills/blacksite-native-qa/references/scenarios/{scenario}.yaml" ) return inputs def gate_input_paths(root: Path, gate: Gate, changed: Sequence[str]) -> list[str]: inputs = gate_static_inputs(root, gate) normalized_changes = normalize_paths(changed) if gate.lane is not None: packages = workspace_packages(root) selected, extra_roots, rust_only = cargo_scope(root, gate) roots = {packages[name].root for name in selected if name in packages} | extra_roots scoped = _tracked_scope_files(root, roots) def relevant(relative: str) -> bool: if relative in inputs: return True if not _under_scope(relative, roots) or _is_documentation(relative): return False return relative.endswith(".rs") if rust_only else True inputs.update(relative for relative in scoped if relevant(relative)) # Add relevant untracked and deleted paths supplied by Git status or --paths. inputs.update(relative for relative in normalized_changes if relevant(relative)) elif gate.name == "candidate-docs": inputs.update( relative for relative in _tracked_scope_files(root, ("docs",)) if _is_documentation(relative) ) inputs.update(relative for relative in normalized_changes if _is_documentation(relative)) elif gate.name.startswith("docs-"): inputs.update(relative for relative in normalized_changes if _is_documentation(relative)) elif gate.name.startswith("workflow-"): inputs.update( relative for relative in normalized_changes if relative in inputs or relative == ".codex/workflow.toml" ) elif gate.native_required: inputs.update( relative for relative in normalized_changes if "editor_ui" in classify((relative,)) ) else: inputs.update(normalized_changes) return sorted(inputs) def _version_output(root: Path, command: Sequence[str]) -> str: completed = subprocess.run( command, cwd=root, check=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, ) if completed.returncode: return f"unavailable(exit={completed.returncode})" return completed.stdout.strip() def gate_tool_versions(root: Path, gate: Gate) -> dict[str, str]: if gate.lane is not None: versions = { "cargo": _version_output(root, ("cargo", "--version", "--verbose")), "rustc": _version_output(root, ("rustc", "--version", "--verbose")), } if gate.command[1:2] == ("fmt",): versions["rustfmt"] = _version_output(root, ("rustfmt", "--version")) if gate.command[1:2] == ("clippy",): versions["clippy-driver"] = _version_output(root, ("clippy-driver", "--version")) return versions executable = Path(gate.command[0]).name if gate.command else "" if executable.startswith("python") or any(argument.endswith(".py") for argument in gate.command): return {"python": f"{sys.executable}\n{sys.version}"} if executable == "bash": return {"bash": _version_output(root, ("bash", "--version")).splitlines()[0]} return {"executable": executable} def gate_environment(gate: Gate) -> dict[str, str | None]: if gate.lane is None: return {} return {name: os.environ.get(name) for name in RUST_GATE_ENVIRONMENT} def gate_digest_details( root: Path, gate: Gate, paths: Sequence[str], *, tool_versions: dict[str, str] | None = None, environment: dict[str, str | None] | None = None, ) -> tuple[str, list[str], dict[str, str], dict[str, str | None]]: inputs = gate_input_paths(root, gate, paths) versions = tool_versions if tool_versions is not None else gate_tool_versions(root, gate) environment_values = environment if environment is not None else gate_environment(gate) digest = hashlib.sha256() digest.update( json.dumps( { "schema": DIGEST_SCHEMA_VERSION, "gate": asdict(gate), "tool_versions": versions, "environment": environment_values, }, sort_keys=True, separators=(",", ":"), ).encode() ) for relative in inputs: digest.update(b"\0path\0") digest.update(relative.encode("utf-8", "surrogateescape")) candidate = root / relative if not candidate.is_file(): digest.update(b"\0missing") continue digest.update(b"\0file\0") digest.update(candidate.read_bytes()) return digest.hexdigest(), inputs, versions, environment_values def gate_digest( root: Path, gate: Gate, paths: Sequence[str], *, tool_versions: dict[str, str] | None = None, environment: dict[str, str | None] | None = None, ) -> str: return gate_digest_details( root, gate, paths, tool_versions=tool_versions, environment=environment, )[0] def gate_command(root: Path, gate: Gate) -> list[str]: if gate.lane is None: return list(gate.command) return [ sys.executable, "scripts/codex/cargo_lane.py", "exec", gate.lane, "--", *gate.command, ] def run_storage(root: Path, phase: str) -> int: script = root / "scripts" / "codex" / "build_storage.py" if not script.is_file(): return 0 return subprocess.run( [sys.executable, str(script), "enforce", "--phase", phase], cwd=root ).returncode def execute_gate( root: Path, workflow: dict[str, Any], gate: Gate, digest: str, ) -> dict[str, Any]: log_dir = root / workflow["output"]["log_dir"] log_dir.mkdir(parents=True, exist_ok=True) result_path = log_dir / f"{gate.name}-{digest[:12]}.json" log_path = log_dir / f"{gate.name}-{digest[:12]}.log" command = gate_command(root, gate) if gate.native_required: return { "status": "NOT_RUN", "exit_code": 0, "duration_seconds": 0.0, "log": None, "note": "Native QA remains user-controlled until explicitly delegated.", } if gate.compile_gate and run_storage(root, "pre") != 0: return { "status": "BLOCKED", "exit_code": 2, "duration_seconds": 0.0, "log": None, "note": "Build-storage preflight blocked this compile gate.", } summarizer = [ sys.executable, "scripts/codex/summarize_command.py", "--gate", gate.name, "--log", str(log_path), "--json-result", str(result_path), "--", *command, ] started = time.monotonic() exit_code = subprocess.run(summarizer, cwd=root).returncode duration = time.monotonic() - started if gate.compile_gate: post = run_storage(root, "post") if exit_code == 0 and post != 0: exit_code = post if result_path.is_file(): result = json.loads(result_path.read_text(encoding="utf-8")) else: result = { "status": "PASS" if exit_code == 0 else "FAIL", "exit_code": exit_code, "duration_seconds": duration, "log": str(log_path.relative_to(root)), } result["exit_code"] = exit_code return result def print_plan(tier: str, paths: Sequence[str], areas: set[str], gates: Sequence[Gate], skipped: Sequence[str]) -> None: print(f"Verification tier: {tier}") print("Changed areas: " + (", ".join(sorted(areas)) if areas else "none")) for gate in gates: lane = f" [{gate.lane}]" if gate.lane else "" print(f"SELECT {gate.name}{lane} — {gate.reason}") for reason in skipped: print(f"SKIP — {reason}") if paths: print(f"Inputs: {len(paths)} changed path(s)") def diagnose_build(root: Path) -> int: commands = [ [sys.executable, "scripts/codex/cargo_lane.py", "env", "dev", "--json"], [sys.executable, "scripts/codex/build_storage.py", "status", "--json"], ] for command in commands: completed = subprocess.run(command, cwd=root) if completed.returncode: return completed.returncode return 0 def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description=__doc__) result.add_argument("action", choices=("plan", "fast", "slice", "candidate", "status", "invalidate", "diagnose-build", "self-test")) result.add_argument("--paths", nargs="*", help="override Git changes for planning/self-tests") result.add_argument( "--tier", choices=("fast", "slice", "candidate"), default="fast", help="tier to preview with `plan` (default: fast)", ) result.add_argument("--reason", help="invalidation reason") return result def self_test(root: Path) -> int: cases = { "docs-only": (["docs/editor/material-system.md"], {"docs"}), "pipeline": (["crates/content_pipeline/src/lib.rs"], {"content_pipeline", "rust"}), "editor-ui": (["crates/editor/src/ui/inspector.rs"], {"editor", "editor_ui", "rust"}), } for name, (paths, expected) in cases.items(): found = classify(paths) if not expected.issubset(found): raise AssertionError(f"{name}: expected {expected}, found {found}") gates, _ = select_gates("fast", found, root) names = {gate.name for gate in gates} if name == "docs-only" and any(gate.lane for gate in gates): raise AssertionError("docs-only plan selected a Cargo lane") if name == "pipeline" and "check-content_pipeline" not in names: raise AssertionError("pipeline plan missed focused check") if name == "editor-ui" and "check-editor" not in names: raise AssertionError("editor UI plan missed editor library check") if any(gate.name.startswith("candidate-") for gate in gates): raise AssertionError(f"{name}: fast plan selected candidate gate") candidate_gates, _ = select_gates("candidate", {"rust"}, root) candidate_package = next( gate for gate in candidate_gates if gate.name == "candidate-package" ) if candidate_package.command[-2:] != ("--profile", "qa"): raise AssertionError("candidate package gate must select assets/build_profiles/qa.ron") with tempfile.TemporaryDirectory(prefix="blacksite-verify-digest-") as temporary: fixture = Path(temporary) def write(relative: str, content: str) -> None: path = fixture / relative path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") write( "Cargo.toml", '[workspace]\nmembers = ["crates/content_pipeline", "crates/shared", "crates/editor"]\n', ) write("Cargo.lock", "version = 4\n") write(".cargo/config.toml", "[build]\nincremental = true\n") write(".codex/workflow.toml", "version = 1\n") write("scripts/codex/cargo_lane.py", "# fixture lane wrapper\n") write( "crates/content_pipeline/Cargo.toml", '[package]\nname = "content_pipeline"\nversion = "0.0.0"\n' '[dependencies]\nshared = { path = "../shared" }\n', ) write("crates/content_pipeline/src/lib.rs", "pub fn pipeline() {}\n") write("crates/content_pipeline/README.md", "pipeline notes\n") write( "crates/shared/Cargo.toml", '[package]\nname = "shared"\nversion = "0.0.0"\n', ) write("crates/shared/src/lib.rs", "pub struct Shared;\n") write( "crates/editor/Cargo.toml", '[package]\nname = "editor"\nversion = "0.0.0"\n', ) write("crates/editor/src/lib.rs", "pub fn editor() {}\n") write("docs/note.md", "first docs revision\n") pipeline_gate = cargo_gate( "check-content_pipeline", "dev", "digest fixture", "check", "-p", "content_pipeline", "--lib", ) fake_tools = {"cargo": "cargo fixture", "rustc": "rustc fixture"} empty_environment: dict[str, str | None] = {} changed = ["crates/content_pipeline/src/lib.rs"] baseline, scoped_inputs, _, _ = gate_digest_details( fixture, pipeline_gate, changed, tool_versions=fake_tools, environment=empty_environment, ) if "crates/shared/src/lib.rs" not in scoped_inputs: raise AssertionError("package digest omitted a local dependency input") if "docs/note.md" in scoped_inputs or "crates/content_pipeline/README.md" in scoped_inputs: raise AssertionError("package digest included documentation") if "crates/editor/src/lib.rs" in scoped_inputs: raise AssertionError("package digest included an unrelated package") write("docs/note.md", "second docs revision\n") write("crates/content_pipeline/README.md", "updated pipeline notes\n") docs_changed = gate_digest( fixture, pipeline_gate, [*changed, "docs/note.md", "crates/content_pipeline/README.md"], tool_versions=fake_tools, environment=empty_environment, ) if docs_changed != baseline: raise AssertionError("unrelated documentation invalidated Rust evidence") write("crates/editor/src/lib.rs", "pub fn unrelated_editor_change() {}\n") unrelated_package_changed = gate_digest( fixture, pipeline_gate, [*changed, "crates/editor/src/lib.rs"], tool_versions=fake_tools, environment=empty_environment, ) if unrelated_package_changed != baseline: raise AssertionError("unrelated package input invalidated a focused Rust gate") write("crates/shared/src/lib.rs", "pub struct ChangedShared;\n") dependency_changed = gate_digest( fixture, pipeline_gate, [*changed, "crates/shared/src/lib.rs"], tool_versions=fake_tools, environment=empty_environment, ) if dependency_changed == baseline: raise AssertionError("local dependency change did not invalidate Rust evidence") write("crates/shared/src/lib.rs", "pub struct Shared;\n") write(".cargo/config.toml", "[build]\nincremental = false\n") config_changed = gate_digest( fixture, pipeline_gate, changed, tool_versions=fake_tools, environment=empty_environment, ) if config_changed == baseline: raise AssertionError("relevant Cargo configuration did not invalidate Rust evidence") write(".cargo/config.toml", "[build]\nincremental = true\n") tool_changed = gate_digest( fixture, pipeline_gate, changed, tool_versions={"cargo": "cargo fixture", "rustc": "rustc changed"}, environment=empty_environment, ) if tool_changed == baseline: raise AssertionError("toolchain change did not invalidate Rust evidence") print( "PASS verify-self-test — selection and gate digests are scoped; " "unrelated docs/packages reuse evidence and relevant dependency/config/tool changes invalidate it" ) return 0 def main(argv: Sequence[str] | None = None) -> int: args = parser().parse_args(argv) root = repository_root() workflow = load_workflow(root) ledger_file = ledger_path(root, workflow) ledger = read_ledger(ledger_file) if args.action == "self-test": return self_test(root) if args.action == "status": gates = ledger.get("gates", {}) if not gates: print("Verification ledger is empty.") return 0 for name, record in sorted(gates.items()): print(f"{record.get('status', 'UNKNOWN')} {name} — {record.get('checked_at', 'unknown')} — {str(record.get('digest', ''))[:12]}") return 0 if args.action == "invalidate": if not args.reason: raise SystemExit("invalidate requires --reason") for record in ledger.setdefault("gates", {}).values(): record["valid"] = False ledger.setdefault("invalidations", []).append({"at": utc_now(), "reason": args.reason}) atomic_json(ledger_file, ledger) print(f"Invalidated verification evidence: {args.reason}") return 0 if args.action == "diagnose-build": return diagnose_build(root) paths = normalize_paths(args.paths if args.paths is not None else changed_paths(root)) areas = classify(paths) tier = args.tier if args.action == "plan" else args.action gates, skipped = select_gates(tier, areas, root) print_plan(tier, paths, areas, gates, skipped) if args.action == "plan": return 0 if args.action == "candidate" and workflow["verification"].get("candidate_requires_explicit_state", True): if not state_is_candidate_ready(root, workflow): print("candidate gate refused: session state is not Candidate-ready", file=sys.stderr) return 2 failed = False for gate in gates: digest, input_paths, tool_versions, environment = gate_digest_details( root, gate, paths ) previous = ledger.setdefault("gates", {}).get(gate.name, {}) if ( workflow["verification"].get("reuse_by_input_digest", True) and previous.get("valid", True) and previous.get("status") == "PASS" and previous.get("digest") == digest and previous.get("command") == gate_command(root, gate) ): print(f"PASS {gate.name} — reused {digest[:12]}") continue result = execute_gate(root, workflow, gate, digest) record = { **result, "valid": result.get("status") == "PASS", "digest": digest, "digest_schema": DIGEST_SCHEMA_VERSION, "input_paths": input_paths, "tool_versions": tool_versions, "environment": environment, "command": gate_command(root, gate), "lane": gate.lane, "checked_at": utc_now(), } ledger["gates"][gate.name] = record atomic_json(ledger_file, ledger) if result.get("status") not in {"PASS", "NOT_RUN"}: failed = True break return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(main())