Implement configuration, runtime, frontend, and task governance closure
This commit is contained in:
253
scripts/generate_configuration_startup_closure.py
Normal file
253
scripts/generate_configuration_startup_closure.py
Normal file
@@ -0,0 +1,253 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate deterministic configuration startup closure evidence.
|
||||
|
||||
The resolver intentionally has a narrow search policy. Ordinary relative
|
||||
references resolve from their entrypoint directory, repository-qualified
|
||||
references resolve from the repository root, and only an explicit ``LIB:``
|
||||
prefix searches the locked LinuxCNC library roots. Bare basenames are never
|
||||
silently replaced by a same-named candidate elsewhere in the checkout.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import shlex
|
||||
from collections import Counter
|
||||
from pathlib import Path, PurePosixPath
|
||||
from typing import Any
|
||||
|
||||
|
||||
CONTEXT = Path("context-lock.json")
|
||||
RESOURCES = Path("coverage/config-resources.jsonl")
|
||||
REGRESSION = Path("coverage/configuration-regression.report.json")
|
||||
GRAPH = Path("coverage/kinematics-config-graph.report.json")
|
||||
GATE = Path("coverage/release-candidate-gate.report.json")
|
||||
OUTPUT = Path("coverage/configuration-startup-closure.report.json")
|
||||
SCHEMA = "configuration-startup-closure.v1"
|
||||
LIBRARY_ROOTS = ("linuxcnc/lib/hallib", "linuxcnc/lib")
|
||||
HOST_PREFIXES = ("~", "$", "${")
|
||||
|
||||
|
||||
def canonical(value: Any) -> str:
|
||||
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def digest(value: Any) -> str:
|
||||
return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def file_digest(path: Path) -> str | None:
|
||||
if not path.is_file():
|
||||
return None
|
||||
handle = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
for block in iter(lambda: stream.read(1024 * 1024), b""):
|
||||
handle.update(block)
|
||||
return handle.hexdigest()
|
||||
|
||||
|
||||
def artifact(root: Path, path: Path, role: str) -> dict[str, Any]:
|
||||
full = root / path
|
||||
sha = file_digest(full)
|
||||
if sha is None:
|
||||
return {"path": path.as_posix(), "role": role, "status": "missing"}
|
||||
return {"path": path.as_posix(), "role": role, "status": "present", "bytes": full.stat().st_size, "sha256": sha}
|
||||
|
||||
|
||||
def read_json(root: Path, path: Path) -> tuple[dict[str, Any], str | None]:
|
||||
try:
|
||||
value = json.loads((root / path).read_text(encoding="utf-8"))
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{path} must contain an object")
|
||||
return value, None
|
||||
except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
|
||||
return {}, str(exc)
|
||||
|
||||
|
||||
def read_jsonl(root: Path, path: Path) -> tuple[list[dict[str, Any]], str | None]:
|
||||
try:
|
||||
rows: list[dict[str, Any]] = []
|
||||
for line_number, line in enumerate((root / path).read_text(encoding="utf-8").splitlines(), 1):
|
||||
if not line.strip():
|
||||
continue
|
||||
value = json.loads(line)
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{path}:{line_number} must contain an object")
|
||||
rows.append(value)
|
||||
return rows, None
|
||||
except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc:
|
||||
return [], str(exc)
|
||||
|
||||
|
||||
def normalize(value: str) -> tuple[str | None, str | None]:
|
||||
if not isinstance(value, str) or not value:
|
||||
return None, "invalid_reference"
|
||||
if "\x00" in value:
|
||||
return None, "invalid_reference"
|
||||
parts: list[str] = []
|
||||
for part in value.replace("\\", "/").split("/"):
|
||||
if not part or part == ".":
|
||||
continue
|
||||
if part == "..":
|
||||
if not parts:
|
||||
return None, "path_escape"
|
||||
parts.pop()
|
||||
continue
|
||||
parts.append(part)
|
||||
return "/".join(parts), None
|
||||
|
||||
|
||||
def first_token(raw: Any) -> tuple[str | None, list[str], str | None]:
|
||||
if not isinstance(raw, str):
|
||||
return None, [], "invalid_reference"
|
||||
value = raw.strip().strip("\"'")
|
||||
if not value:
|
||||
return None, [], "empty_reference"
|
||||
try:
|
||||
tokens = shlex.split(value, posix=True)
|
||||
except ValueError:
|
||||
tokens = re.findall(r"[^\s]+", value)
|
||||
if not tokens:
|
||||
return None, [], "empty_reference"
|
||||
return tokens[0], tokens[1:], None
|
||||
|
||||
|
||||
def candidates(raw_value: Any, entrypoint: str) -> tuple[list[str], str, str | None, list[str]]:
|
||||
value, args, error = first_token(raw_value)
|
||||
if error:
|
||||
return [], "blocked", error, args
|
||||
assert value is not None
|
||||
if value.startswith(HOST_PREFIXES):
|
||||
return [], "blocked", "host_path", args
|
||||
entry, error = normalize(entrypoint)
|
||||
if error or entry is None:
|
||||
return [], "blocked", error or "invalid_reference", args
|
||||
if value.startswith("LIB:"):
|
||||
name = value[4:]
|
||||
if not name or name.startswith("/") or ".." in PurePosixPath(name).parts:
|
||||
return [], "library", "invalid_reference", args
|
||||
paths = []
|
||||
for root in LIBRARY_ROOTS:
|
||||
path, path_error = normalize(f"{root}/{name}")
|
||||
if path_error is None and path is not None:
|
||||
paths.append(path)
|
||||
return paths, "library", None, args
|
||||
if value.startswith("/"):
|
||||
return [], "blocked", "outside_repo", args
|
||||
if value.startswith("linuxcnc/"):
|
||||
path, path_error = normalize(value)
|
||||
return ([path] if path else []), "repo_relative", path_error, args
|
||||
base = PurePosixPath(entry).parent.as_posix()
|
||||
path, path_error = normalize(f"{base}/{value}")
|
||||
if path_error:
|
||||
return [], "relative", path_error, args
|
||||
if path is not None and not (path == "linuxcnc" or path.startswith("linuxcnc/")):
|
||||
return [], "relative", "outside_repo", args
|
||||
return ([path] if path else []), "relative", None, args
|
||||
|
||||
|
||||
def available_paths(root: Path) -> set[str]:
|
||||
return {path.relative_to(root).as_posix() for path in root.rglob("*") if path.is_file() or path.is_dir()}
|
||||
|
||||
|
||||
def resolve_row(row: dict[str, Any], available: set[str]) -> dict[str, Any]:
|
||||
original_path = str(row.get("path", ""))
|
||||
original_status = str(row.get("parse_status", ""))
|
||||
raw_value = row.get("raw_value", original_path)
|
||||
if original_status in {"resolved_file", "resolved_directory"} and original_path in available:
|
||||
return {
|
||||
"resource_source_id": row.get("resource_source_id"),
|
||||
"entrypoint": row.get("entrypoint"),
|
||||
"original_path": original_path,
|
||||
"original_status": original_status,
|
||||
"raw_value": raw_value,
|
||||
"status": "resolved_existing",
|
||||
"code": None,
|
||||
"path": original_path,
|
||||
"args": [],
|
||||
"attempted_paths": [original_path],
|
||||
"resolution": "indexed",
|
||||
}
|
||||
paths, resolution, error, args = candidates(raw_value, str(row.get("entrypoint", "")))
|
||||
existing = [path for path in paths if path in available]
|
||||
if error:
|
||||
status = f"blocked_{error}"
|
||||
return {"resource_source_id": row.get("resource_source_id"), "entrypoint": row.get("entrypoint"), "original_path": original_path, "original_status": original_status, "raw_value": raw_value, "status": status, "code": error, "path": None, "args": args, "attempted_paths": paths, "resolution": resolution}
|
||||
if len(existing) > 1:
|
||||
return {"resource_source_id": row.get("resource_source_id"), "entrypoint": row.get("entrypoint"), "original_path": original_path, "original_status": original_status, "raw_value": raw_value, "status": "blocked_ambiguous", "code": "ambiguous", "path": None, "args": args, "attempted_paths": paths, "resolution": resolution}
|
||||
if len(existing) == 1:
|
||||
return {"resource_source_id": row.get("resource_source_id"), "entrypoint": row.get("entrypoint"), "original_path": original_path, "original_status": original_status, "raw_value": raw_value, "status": f"resolved_{resolution}", "code": None, "path": existing[0], "args": args, "attempted_paths": paths, "resolution": resolution}
|
||||
return {"resource_source_id": row.get("resource_source_id"), "entrypoint": row.get("entrypoint"), "original_path": original_path, "original_status": original_status, "raw_value": raw_value, "status": "blocked_missing", "code": "missing", "path": None, "args": args, "attempted_paths": paths, "resolution": resolution}
|
||||
|
||||
|
||||
def generate(root: Path, output: Path = OUTPUT) -> dict[str, Any]:
|
||||
root = root.resolve()
|
||||
lock, lock_error = read_json(root, CONTEXT)
|
||||
regression, regression_error = read_json(root, REGRESSION)
|
||||
graph, graph_error = read_json(root, GRAPH)
|
||||
gate, gate_error = read_json(root, GATE)
|
||||
rows, rows_error = read_jsonl(root, RESOURCES)
|
||||
errors = {path.as_posix(): error for path, error in ((CONTEXT, lock_error), (REGRESSION, regression_error), (GRAPH, graph_error), (GATE, gate_error), (RESOURCES, rows_error)) if error}
|
||||
available = available_paths(root)
|
||||
records = [resolve_row(row, available) for row in rows]
|
||||
records.sort(key=lambda row: str(row.get("resource_source_id", "")))
|
||||
occurrence_counter: Counter[str] = Counter()
|
||||
for record in records:
|
||||
source_id = str(record.get("resource_source_id", ""))
|
||||
occurrence_counter[source_id] += 1
|
||||
record["resource_record_id"] = f"{source_id}#occurrence:{occurrence_counter[source_id]}"
|
||||
statuses = Counter(str(row["status"]) for row in records)
|
||||
codes = Counter(str(row["code"]) for row in records if row.get("code"))
|
||||
duplicate_source_ids = sorted(source_id for source_id, count in occurrence_counter.items() if count > 1)
|
||||
entrypoints = sorted({str(row.get("entrypoint")) for row in records})
|
||||
ids = [str(row.get("resource_source_id")) for row in records]
|
||||
config_entries = regression.get("configuration_entries", []) if isinstance(regression.get("configuration_entries"), list) else []
|
||||
context_id = lock.get("context_id")
|
||||
stop_ok = gate.get("release_action") == "STOP_RELEASE" and gate.get("candidate_eligible") is False
|
||||
checks = {
|
||||
"input_integrity": {"status": "verified" if not errors and context_id else "blocked", "context_id": context_id},
|
||||
"configuration_coverage": {"status": "verified" if len(config_entries) == 264 and len(entrypoints) == 233 else "blocked", "registered_entries": len(config_entries), "entrypoints": len(entrypoints)},
|
||||
"resource_identity": {"status": "verified" if len({str(row.get('resource_record_id')) for row in records}) == len(records) and ids == sorted(ids) else "blocked", "rows": len(ids), "duplicate_source_ids": duplicate_source_ids, "duplicate_policy": "source_id_preserved_with_deterministic_occurrence_suffix"},
|
||||
"explicit_library_policy": {"status": "verified" if all(row.get("resolution") in {"indexed", "relative", "repo_relative", "library", "blocked"} for row in records) else "blocked", "library_references": sum(1 for row in records if str(row.get("raw_value", "")).lstrip().startswith("LIB:")), "library_resolved": sum(1 for row in records if row.get("status") == "resolved_library")},
|
||||
"no_basename_substitution": {"status": "verified" if all(row.get("resolution") != "basename" for row in records) else "blocked"},
|
||||
"release_stop_preserved": {"status": "verified" if stop_ok else "blocked", "release_action": "STOP_RELEASE" if stop_ok else None},
|
||||
}
|
||||
blocked_count = sum(value for key, value in statuses.items() if key.startswith("blocked_"))
|
||||
report: dict[str, Any] = {
|
||||
"schema": SCHEMA,
|
||||
"context_id": context_id,
|
||||
"task_id": "M3-CFG-01",
|
||||
"status": "verified" if not errors and blocked_count == 0 else "partial" if not errors else "blocked",
|
||||
"implementation_status": "verified" if not errors else "blocked",
|
||||
"candidate_eligible": False,
|
||||
"release_action": "STOP_RELEASE",
|
||||
"inputs": [artifact(root, CONTEXT, "context_lock"), artifact(root, RESOURCES, "config_resources"), artifact(root, REGRESSION, "configuration_regression"), artifact(root, GRAPH, "kinematics_config_graph"), artifact(root, GATE, "release_candidate_gate")],
|
||||
"counts": {"resource_rows": len(records), "entrypoints": len(entrypoints), "registered_configuration_entries": len(config_entries), "resolved": len(records) - blocked_count, "blocked": blocked_count, "duplicate_source_ids": len(duplicate_source_ids), "statuses": dict(sorted(statuses.items())), "blocked_codes": dict(sorted(codes.items()))},
|
||||
"checks": checks,
|
||||
"resources": records,
|
||||
"next_transition": "execute M3-RT-02 HAL component manifest and executable configuration graph",
|
||||
"errors": errors,
|
||||
"report_sha256": None,
|
||||
}
|
||||
report["report_sha256"] = digest({key: value for key, value in report.items() if key != "report_sha256"})
|
||||
destination = output if output.is_absolute() else root / output
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
destination.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8")
|
||||
return report
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
|
||||
parser.add_argument("--output", type=Path, default=OUTPUT)
|
||||
args = parser.parse_args()
|
||||
report = generate(args.repo_root.resolve(), args.output)
|
||||
print(f"configuration_startup_closure={report['status']} resolved={report['counts']['resolved']} blocked={report['counts']['blocked']}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user