1168 lines
41 KiB
Python
Executable File
1168 lines
41 KiB
Python
Executable File
#!/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())
|