#!/usr/bin/env python3 """Generate deterministic ownership assignments for unassigned manifest seeds. This report is deliberately additive. Manifest v1 has no owner field, so the assignment is a reviewable workstream proposal and does not mutate the locked manifest or context. Release eligibility remains blocked until the proposal is reviewed and encoded in a new, explicitly versioned context. """ from __future__ import annotations import argparse import hashlib import json from pathlib import Path from typing import Any CONTEXT = Path("context-lock.json") SOURCE_INDEX = Path("coverage/source-index.jsonl") MANIFEST = Path("coverage/manifest.jsonl") GATE = Path("coverage/release-candidate-gate.report.json") INTAKE = Path("coverage/release-remediation-intake.report.json") OUTPUT = Path("coverage/manifest-ownership-remediation.report.json") OWNER_POLICIES = { "ws-0-baseline-owner": { "workstream": "WS-0", "review_owner": "release-closure-owner", "basis": "baseline, source inventory, or documentation path", }, "ws-1-core-runtime-owner": { "workstream": "WS-1", "review_owner": "native-wasm-reviewer", "basis": "core interpreter, task, TP, posemath, motion, or kinematics path", }, "ws-2-worker-runtime-owner": { "workstream": "WS-2", "review_owner": "worker-runtime-reviewer", "basis": "HAL, IO, virtual-device, or Worker path", }, "ws-3-workspace-owner": { "workstream": "WS-3", "review_owner": "workspace-recovery-reviewer", "basis": "workspace, journal, migration, or recovery path", }, "ws-4-python-compat-owner": { "workstream": "WS-4", "review_owner": "python-security-reviewer", "basis": "Python, remap, filter, or VCP compatibility path", }, "ws-5-frontend-owner": { "workstream": "WS-5", "review_owner": "frontend-compatibility-reviewer", "basis": "frontend or browser workflow path", }, "ws-6-reference-test-owner": { "workstream": "WS-6", "review_owner": "reference-test-reviewer", "basis": "test or unit-test path", }, "ws-7-release-owner": { "workstream": "WS-7", "review_owner": "release-gate-owner", "basis": "release-only evidence path", }, } 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 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 = [] with (root / path).open(encoding="utf-8") as stream: for line_number, line in enumerate(stream, 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 classify(source: str, kind: str) -> tuple[str, str]: path = source.split(":", 1)[1] if ":" in source else source if source.startswith(("build:", "doc:")): return "ws-0-baseline-owner", "build/doc source prefix" if source.startswith(("test:", "unit_test:")) or kind == "test": return "ws-6-reference-test-owner", "test source prefix" if path.startswith("wasm-core/"): return "ws-1-core-runtime-owner", "wasm-core source prefix" if path.startswith("web/worker/"): return "ws-2-worker-runtime-owner", "web/worker source prefix" if path.startswith("web/workspace/"): return "ws-3-workspace-owner", "web/workspace source prefix" if path.startswith("web/"): return "ws-5-frontend-owner", "web source prefix" if "/python/" in path or "/remap/" in path or "/vcp/" in path or path.startswith("linuxcnc/src/emc/python/"): return "ws-4-python-compat-owner", "Python/remap/VCP path" if ( path.startswith("linuxcnc/src/hal/") or "/iotask/" in path or "/io/" in path or "/motion/" in path or "/motioncontrol/" in path ): return "ws-2-worker-runtime-owner", "HAL/IO/motion virtual-device path" if path.startswith("linuxcnc/src/emc/") or path.startswith("linuxcnc/src/libnml/"): return "ws-1-core-runtime-owner", "LinuxCNC core path" if path.startswith("linuxcnc/src/"): return "ws-1-core-runtime-owner", "LinuxCNC source path" if path.startswith("linuxcnc/" ): return "ws-0-baseline-owner", "LinuxCNC non-runtime path" return "ws-7-release-owner", "unmatched release evidence path" 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 generate(root: Path, output: Path = OUTPUT) -> dict[str, Any]: root = root.resolve() lock, lock_error = read_json(root, CONTEXT) gate, gate_error = read_json(root, GATE) intake, intake_error = read_json(root, INTAKE) source_rows, source_error = read_jsonl(root, SOURCE_INDEX) manifest_rows, manifest_error = read_jsonl(root, MANIFEST) errors = { path.as_posix(): error for path, error in ( (CONTEXT, lock_error), (GATE, gate_error), (INTAKE, intake_error), (SOURCE_INDEX, source_error), (MANIFEST, manifest_error), ) if error } source_by_id = {str(row.get("source_id")): row for row in source_rows} rows: list[dict[str, Any]] = [] owner_counts: dict[str, int] = {} all_valid = not errors and len(source_rows) == len(manifest_rows) for manifest in manifest_rows: if manifest.get("mapping_status") != "needs_owner": continue source = str(manifest.get("source", "")) owner, rule = classify(source, str(manifest.get("kind", ""))) policy = OWNER_POLICIES[owner] source_row = source_by_id.get(source) source_match = bool(source_row) and source_row.get("sha256") == manifest.get("source_sha") valid = bool(manifest.get("id")) and bool(source_row) and source_match row = { "id": manifest.get("id"), "source": source, "source_sha": manifest.get("source_sha"), "kind": manifest.get("kind"), "owner": owner, "owner_type": "workstream_role", "workstream": policy["workstream"], "review_owner": policy["review_owner"], "assignment_rule": rule, "assignment_basis": policy["basis"], "original_mapping_status": manifest.get("mapping_status"), "assignment_status": "proposed_blocked_review" if valid else "blocked_invalid_source", "release_status": "blocked", "evidence_status": "source_hash_verified" if valid else "blocked", } rows.append(row) owner_counts[owner] = owner_counts.get(owner, 0) + 1 all_valid = all_valid and valid rows.sort(key=lambda row: str(row["id"])) expected_ids = sorted(str(row.get("id")) for row in manifest_rows if row.get("mapping_status") == "needs_owner") observed_ids = sorted(str(row.get("id")) for row in rows) coverage_ok = expected_ids == observed_ids and len(observed_ids) == len(set(observed_ids)) stop_ok = ( gate.get("release_action") == "STOP_RELEASE" and gate.get("candidate_eligible") is False and gate.get("blocked_count") == 6 and intake.get("release_action") == "STOP_RELEASE" and intake.get("candidate_eligible") is False ) checks = { "manifest_source_coverage": {"status": "verified" if coverage_ok else "blocked", "expected": len(expected_ids), "observed": len(observed_ids)}, "source_hashes": {"status": "verified" if all_valid else "blocked", "verified": sum(row["evidence_status"] == "source_hash_verified" for row in rows)}, "deterministic_owner_policy": {"status": "verified" if all(row["owner"] in OWNER_POLICIES for row in rows) else "blocked", "owners": owner_counts}, "context_consistency": {"status": "verified" if lock.get("context_id") and intake.get("context_id") == lock.get("context_id") else "blocked", "context_id": lock.get("context_id")}, "release_stop_preserved": {"status": "verified" if stop_ok else "blocked", "release_action": "STOP_RELEASE" if stop_ok else None}, "manifest_unchanged": {"status": "verified", "reason": "manifest v1 has no owner field; proposal is additive and requires review before encoding"}, } checks_ok = not errors and all(value["status"] == "verified" for value in checks.values()) report: dict[str, Any] = { "schema": "manifest-ownership-remediation.v1", "context_id": lock.get("context_id"), "task_id": "K8-REL-08", "status": "complete" if checks_ok else "blocked", "remediation_status": "blocked_pending_manifest_review", "release_decision": "blocked_not_release_candidate", "release_action": "STOP_RELEASE", "candidate_eligible": False, "source_baseline": {"linuxcnc_version": lock.get("linuxcnc_version"), "linuxcnc_source_sha": lock.get("linuxcnc_source_sha")}, "inputs": [ artifact(root, CONTEXT, "context_lock"), artifact(root, SOURCE_INDEX, "source_index"), artifact(root, MANIFEST, "manifest_seed"), artifact(root, GATE, "release_candidate_gate"), artifact(root, INTAKE, "remediation_intake"), ], "counts": {"manifest_total": len(manifest_rows), "needs_owner": len(rows), "mapped": len(rows), "accepted_by_this_artifact": 0}, "owner_counts": dict(sorted(owner_counts.items())), "checks": checks, "assignments": rows, "review_policy": { "owner_semantics": "workstream role proposal; not a named human approval", "required_before_manifest_update": ["feature owner confirmation", "feature_id/test_id/oracle evidence", "new context_id if manifest hash changes"], "forbidden": ["promote_to_accepted", "delete_unassigned_rows", "reclassify_as_REAL-EXCLUDED", "overwrite_context_lock_in_place"], }, "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, args.output) print("manifest_ownership_remediation=" + report["status"] + " release=" + report["release_action"] + " mapped=" + str(report["counts"]["mapped"])) return 0 if __name__ == "__main__": raise SystemExit(main())