#!/usr/bin/env python3 """Build a fail-closed C-001 teaching receipt without executing a workload. The default path is deliberately architecture-only. Capture mode verifies and joins artifacts that a separate, real workload run already produced; it never launches Hermes, GLM-5.2, a serving engine, CUDA, or a profiler. """ from __future__ import annotations import argparse import hashlib import json import math import sys from copy import deepcopy from pathlib import Path from typing import Any, Iterable ROOT = Path(__file__).resolve().parent DEFAULT_CONFIG = ROOT / "reference-config.json" RUN_RECEIPT_SCHEMA = "touchdown.hbm-run-receipt.v1" ARCHITECTURE_PLAN_SCHEMA = "touchdown.hbm-c001-architecture-plan.v1" CAPTURE_MANIFEST_SCHEMA = "touchdown.hbm-c001-capture-artifacts.v1" EVENT_ORDER = ( "request", "context", "route", "prefix_lookup", "prefill", "attention_moe", "lowering", "gpu_hbm", "kv_placement", "decode", "tool_call", "retry_verifier", "outcome_bill", ) REQUIRED_ARTIFACT_KINDS = ( "request", "context", "route", "prefix_cache", "engine_trace", "operator_trace", "compiler_artifacts", "kernel_profile", "memory_placement", "fabric", "tool", "verifier", "power", "facility", "cost", ) EVENT_ARTIFACT_KINDS = { "request": ("request",), "context": ("context",), "route": ("route",), "prefix_lookup": ("prefix_cache",), "prefill": ("engine_trace", "operator_trace"), "attention_moe": ("operator_trace", "fabric"), "lowering": ("compiler_artifacts",), "gpu_hbm": ("kernel_profile", "memory_placement"), "kv_placement": ("memory_placement", "fabric"), "decode": ("engine_trace", "operator_trace"), "tool_call": ("tool",), "retry_verifier": ("verifier",), "outcome_bill": ("verifier", "power", "facility", "cost"), } EVIDENCE_STATES = frozenset( { "measured", "measured_parent_allocated", "vendor_rated", "derived", "modeled", "source_backed_architecture", "illustrative", "unknown", "invalid_comparison", } ) COVERAGE_STATES = ( "not_started", "parser_only", "fixture_backed", "live_validated", "release_ready", "not_applicable", ) OBSERVATION_STATES = ( "supported", "configured", "initialized", "dispatched", "measured", "verified", "accepted", ) RESOURCE_ARTIFACT_KINDS = frozenset({"power", "facility", "cost"}) ENERGY_LEDGER_IDS = frozenset({"it_energy", "generated_heat", "cooling_electricity"}) WATER_LEDGER_ROLES = { "coolant_circulation": "circulation_not_consumption", "site_water_consumption": "site_consumption", "electricity_generation_water": "upstream_generation", } class HarnessError(ValueError): """Raised when a receipt would overstate or mix evidence.""" def _read_json(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise HarnessError(f"cannot read JSON {path}: {exc}") from exc if not isinstance(value, dict): raise HarnessError(f"JSON root must be an object: {path}") return value def _require_nonempty_string(value: Any, label: str) -> str: if not isinstance(value, str) or not value.strip(): raise HarnessError(f"{label} must be a non-empty string") return value def _require_enum(value: Any, allowed: Iterable[str], label: str) -> str: normalized = _require_nonempty_string(value, label) if normalized not in allowed: raise HarnessError(f"{label} has unsupported value: {normalized}") return normalized def _require_finite_nonnegative_number(value: Any, label: str) -> int | float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise HarnessError(f"{label} must be numeric") if not math.isfinite(float(value)) or value < 0: raise HarnessError(f"{label} must be finite and non-negative") return value def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _safe_artifact_path(base: Path, relative: Any, label: str) -> Path: raw = Path(_require_nonempty_string(relative, label)) if raw.is_absolute(): raise HarnessError(f"{label} must be relative to --artifacts-dir") base_resolved = base.resolve(strict=True) candidate = (base_resolved / raw).resolve(strict=True) try: candidate.relative_to(base_resolved) except ValueError as exc: raise HarnessError(f"{label} escapes --artifacts-dir") from exc if not candidate.is_file(): raise HarnessError(f"{label} is not a regular file") return candidate def load_reference_config(path: Path = DEFAULT_CONFIG) -> dict[str, Any]: config = _read_json(path) if config.get("schema_version") != "touchdown.hbm-c001-reference-config.v1": raise HarnessError("reference config schema is not supported") if config.get("fixture_id") != "C-001": raise HarnessError("reference config must describe fixture C-001") if tuple(config.get("event_order", ())) != EVENT_ORDER: raise HarnessError("reference config event order does not match the 13-phase contract") events = config.get("events") if not isinstance(events, dict) or tuple(events) != EVENT_ORDER: raise HarnessError("reference config must define exactly the 13 ordered events") if tuple(config.get("artifact_slots", ())) != REQUIRED_ARTIFACT_KINDS: raise HarnessError("reference config artifact slots do not match the capture contract") return config def architecture_receipt(config: dict[str, Any]) -> dict[str, Any]: events = [] for sequence, event_id in enumerate(EVENT_ORDER, start=1): definition = config["events"][event_id] events.append( { "event_id": event_id, "type": event_id, "sequence": sequence, "run_id": None, "phase": event_id, "lanes": { "user_business": definition["user_business"], "software_gpu": definition["software_gpu"], "physics_economics": definition["physics_economics"], }, "components": list(definition["components"]), "state_objects": list(definition["state_objects"]), "artifact_refs": [], "evidence_state": "source_backed_architecture", "coverage_state": "fixture_backed", "observation_state": "supported", } ) artifact_slots = { kind: { "kind": kind, "run_id": None, "path": None, "sha256": None, "present": False, "evidence_state": "unknown", "coverage_state": "not_started", "observation_state": "supported", } for kind in REQUIRED_ARTIFACT_KINDS } memory_model = [] for level in config["memory_model"]: item = deepcopy(level) item.update( { "run_id": None, "evidence_state": "source_backed_architecture", "coverage_state": "fixture_backed", "observation_state": "supported", "artifact_ref": None, } ) memory_model.append(item) resource_ledger = [] for metric in config["resource_ledger"]: item = deepcopy(metric) item.update( { "run_id": None, "evidence_state": "unknown", "coverage_state": "not_started", "observation_state": "supported", "artifact_ref": None, "accounting_role": None, "water_role": None, } ) resource_ledger.append(item) return { "schema": ARCHITECTURE_PLAN_SCHEMA, "schema_version": ARCHITECTURE_PLAN_SCHEMA, "fixture_id": "C-001", "comparison_id": config["comparison_id"], "comparison": { "primary_fixture_id": "C-001", "contrast_fixture_id": "V-001", "comparison_id": config["comparison_id"], "shared_run_id_forbidden": True, }, "run_id": None, "capture_kind": "architecture_only", "banner": "ARCHITECTURE ONLY / RUN NOT CAPTURED", "captured_at": None, "workload": deepcopy(config["workload"]), "reference_platform": deepcopy(config["reference_platform"]), "events": events, "artifact_slots": artifact_slots, "memory_model": memory_model, "kv_placements": [], "fabric_movements": [], "resource_ledger": resource_ledger, "outcome": { "run_id": None, "status": "pending", "accepted": False, "tests_passed": None, "acceptance_boundary": config["workload"]["accepted_boundary"], "verifier_artifact": None, "evidence_state": "unknown", "coverage_state": "not_started", "observation_state": "supported", }, "guardrails": { "network_requests": False, "workload_executed": False, "gpu_execution_claimed": False, "single_run_identity_required": True, "energy_ledgers_are_non_overlapping": None, "water_boundaries_are_separate": True, }, } def _validate_clock(manifest: dict[str, Any]) -> None: clock = manifest.get("clock") if not isinstance(clock, dict): raise HarnessError("capture manifest must include a clock object") if clock.get("status") != "synchronized": raise HarnessError("capture clock status must be synchronized") _require_nonempty_string(clock.get("source"), "clock.source") _require_finite_nonnegative_number(clock.get("max_skew_ms"), "clock.max_skew_ms") if clock.get("artifact_timestamps_checked") is not True: raise HarnessError("clock.artifact_timestamps_checked must be true") def _validate_cache_identity(payload: dict[str, Any]) -> None: cache_status = payload.get("cache_status") if cache_status is None: return _require_enum(cache_status, ("hit", "miss", "partial"), "prefix_cache.cache_status") identity = payload.get("cache_identity") if not isinstance(identity, dict): raise HarnessError("cache claims require prefix_cache.cache_identity") for key in ( "model_revision", "tokenizer_revision", "engine_revision", "prefix_sha256", "kv_layout_id", "precision", ): _require_nonempty_string(identity.get(key), f"prefix_cache.cache_identity.{key}") prefix_sha = identity["prefix_sha256"] if len(prefix_sha) != 64 or any(char not in "0123456789abcdef" for char in prefix_sha.lower()): raise HarnessError("prefix_cache.cache_identity.prefix_sha256 must be a SHA-256") def _validate_verifier(payload: dict[str, Any], slot: dict[str, Any]) -> dict[str, Any]: status = _require_enum( payload.get("status"), ("pending", "rejected", "verified", "accepted"), "verifier.status", ) accepted = payload.get("accepted") tests_passed = payload.get("tests_passed") if not isinstance(accepted, bool): raise HarnessError("verifier.accepted must be boolean") if tests_passed is not None and not isinstance(tests_passed, bool): raise HarnessError("verifier.tests_passed must be boolean or null") if accepted or status == "accepted": if not (accepted is True and status == "accepted" and tests_passed is True): raise HarnessError("accepted-task claim requires accepted status and passing tests") _require_nonempty_string(payload.get("acceptance_boundary"), "verifier.acceptance_boundary") checks = payload.get("checks") if not isinstance(checks, list) or not checks: raise HarnessError("accepted-task claim requires non-empty verifier.checks") if slot["evidence_state"] != "measured": raise HarnessError("accepted-task verifier evidence must be measured") if slot["coverage_state"] not in ("live_validated", "release_ready"): raise HarnessError("accepted-task verifier must be live_validated or release_ready") if slot["observation_state"] != "accepted": raise HarnessError("accepted-task verifier observation must be accepted") elif status == "accepted": raise HarnessError("accepted status cannot have accepted=false") return { "run_id": slot["run_id"], "status": status, "accepted": accepted, "tests_passed": tests_passed, "acceptance_boundary": payload.get("acceptance_boundary"), "verifier_artifact": "verifier", "evidence_state": slot["evidence_state"], "coverage_state": slot["coverage_state"], "observation_state": slot["observation_state"], } def _load_artifact_payload(path: Path, run_id: str, kind: str) -> dict[str, Any] | None: if path.suffix.lower() != ".json": return None payload = _read_json(path) if payload.get("run_id") != run_id: raise HarnessError(f"{kind} JSON artifact must carry the capture run_id") if payload.get("kind") not in (None, kind): raise HarnessError(f"{kind} JSON artifact declares a different kind") return payload def _join_memory_model( receipt: dict[str, Any], payload: dict[str, Any] | None, run_id: str ) -> None: if payload is None or "memory_levels" not in payload: return levels = payload["memory_levels"] if not isinstance(levels, list): raise HarnessError("memory_placement.memory_levels must be an array") by_id = {item["id"]: item for item in receipt["memory_model"]} seen: set[str] = set() for item in levels: if not isinstance(item, dict): raise HarnessError("memory level must be an object") level_id = _require_nonempty_string(item.get("id"), "memory level id") if level_id not in by_id or level_id in seen: raise HarnessError(f"unknown or duplicate memory level: {level_id}") seen.add(level_id) target = by_id[level_id] for field in ("capacity_bytes", "measured_traffic_bytes"): if item.get(field) is not None: target[field] = _require_finite_nonnegative_number( item[field], f"memory level {level_id}.{field}" ) target.update( { "run_id": run_id, "evidence_state": _require_enum( item.get("evidence_state"), EVIDENCE_STATES, "memory evidence_state" ), "coverage_state": _require_enum( item.get("coverage_state"), COVERAGE_STATES, "memory coverage_state" ), "observation_state": _require_enum( item.get("observation_state"), OBSERVATION_STATES, "memory observation_state", ), "artifact_ref": "memory_placement", } ) kv = payload.get("kv_placements", []) if not isinstance(kv, list): raise HarnessError("memory_placement.kv_placements must be an array") for index, item in enumerate(kv): if not isinstance(item, dict): raise HarnessError("KV placement must be an object") copied = deepcopy(item) copied["run_id"] = run_id copied["object_id"] = _require_nonempty_string( item.get("object_id"), f"kv_placements[{index}].object_id" ) copied["tier"] = _require_nonempty_string(item.get("tier"), f"kv_placements[{index}].tier") copied["bytes"] = _require_finite_nonnegative_number( item.get("bytes"), f"kv_placements[{index}].bytes" ) copied["evidence_state"] = _require_enum( item.get("evidence_state"), EVIDENCE_STATES, "KV evidence_state" ) copied["coverage_state"] = _require_enum( item.get("coverage_state"), COVERAGE_STATES, "KV coverage_state" ) copied["observation_state"] = _require_enum( item.get("observation_state"), OBSERVATION_STATES, "KV observation_state" ) receipt["kv_placements"].append(copied) def _join_fabric_movements( receipt: dict[str, Any], payload: dict[str, Any] | None, run_id: str ) -> None: if payload is None or "movements" not in payload: return movements = payload["movements"] if not isinstance(movements, list): raise HarnessError("fabric.movements must be an array") for index, item in enumerate(movements): if not isinstance(item, dict): raise HarnessError("fabric movement must be an object") copied = deepcopy(item) copied["run_id"] = run_id for field in ("source", "target", "transport"): copied[field] = _require_nonempty_string( item.get(field), f"fabric.movements[{index}].{field}" ) copied["bytes"] = _require_finite_nonnegative_number( item.get("bytes"), f"fabric.movements[{index}].bytes" ) copied["evidence_state"] = _require_enum( item.get("evidence_state"), EVIDENCE_STATES, "fabric evidence_state" ) copied["coverage_state"] = _require_enum( item.get("coverage_state"), COVERAGE_STATES, "fabric coverage_state" ) copied["observation_state"] = _require_enum( item.get("observation_state"), OBSERVATION_STATES, "fabric observation_state" ) receipt["fabric_movements"].append(copied) def _join_resource_measurements( receipt: dict[str, Any], artifact_payloads: dict[str, dict[str, Any] | None], manifest: dict[str, Any], run_id: str, ) -> None: ledger_by_id = {item["id"]: item for item in receipt["resource_ledger"]} measurements: dict[str, tuple[str, dict[str, Any]]] = {} for kind in RESOURCE_ARTIFACT_KINDS: payload = artifact_payloads.get(kind) if payload is None: continue declared = payload.get("measurements", []) if not isinstance(declared, list): raise HarnessError(f"{kind}.measurements must be an array") for item in declared: if not isinstance(item, dict): raise HarnessError(f"{kind} measurement must be an object") metric_id = _require_nonempty_string(item.get("id"), f"{kind} measurement id") if metric_id not in ledger_by_id: raise HarnessError(f"unknown resource metric: {metric_id}") if metric_id in measurements: raise HarnessError(f"resource metric appears in multiple ledgers: {metric_id}") measurements[metric_id] = (kind, item) if measurements: accounting = manifest.get("accounting") if not isinstance(accounting, dict): raise HarnessError("numeric resource claims require manifest.accounting") if accounting.get("energy_double_counting_check") != "passed": raise HarnessError("energy_double_counting_check must be passed") if accounting.get("water_boundaries_separated") is not True: raise HarnessError("water_boundaries_separated must be true") accepted = receipt["outcome"]["accepted"] for metric_id, (kind, item) in measurements.items(): target = ledger_by_id[metric_id] if item.get("unit") != target["unit"]: raise HarnessError(f"unit mismatch for resource metric {metric_id}") value = _require_finite_nonnegative_number(item.get("value"), f"{metric_id}.value") boundary = _require_nonempty_string(item.get("boundary"), f"{metric_id}.boundary") accounting_role = item.get("accounting_role") if metric_id in ENERGY_LEDGER_IDS: accounting_role = _require_enum( accounting_role, ("exclusive", "parent_allocated", "diagnostic_not_summed"), f"{metric_id}.accounting_role", ) if metric_id == "cost_per_accepted_task" and not accepted: raise HarnessError("cost_per_accepted_task requires an accepted verifier outcome") water_role = item.get("water_role") if metric_id in WATER_LEDGER_ROLES: water_role = _require_enum( water_role, (WATER_LEDGER_ROLES[metric_id],), f"{metric_id}.water_role", ) target.update( { "run_id": run_id, "value": value, "boundary": boundary, "evidence_state": _require_enum( item.get("evidence_state"), EVIDENCE_STATES, "resource evidence_state" ), "coverage_state": _require_enum( item.get("coverage_state"), COVERAGE_STATES, "resource coverage_state" ), "observation_state": _require_enum( item.get("observation_state"), OBSERVATION_STATES, "resource observation_state", ), "artifact_ref": kind, "accounting_role": accounting_role, "water_role": water_role, } ) receipt["guardrails"]["energy_ledgers_are_non_overlapping"] = ( True if measurements else None ) def captured_receipt( config: dict[str, Any], manifest_path: Path, artifacts_dir: Path, run_id: str ) -> dict[str, Any]: manifest = _read_json(manifest_path) if manifest.get("schema_version") != CAPTURE_MANIFEST_SCHEMA: raise HarnessError("capture manifest schema is not supported") if manifest.get("fixture_id") != "C-001": raise HarnessError("capture manifest must describe fixture C-001") if manifest.get("comparison_id") != config["comparison_id"]: raise HarnessError("capture comparison_id does not match the C-001 contract") if manifest.get("capture_scope") != "real_workload": raise HarnessError("capture_scope must be real_workload") run_id = _require_nonempty_string(run_id, "--run-id") if manifest.get("run_id") != run_id: raise HarnessError("capture manifest and CLI run IDs do not match") captured_at = _require_nonempty_string(manifest.get("captured_at"), "captured_at") _validate_clock(manifest) entries = manifest.get("artifacts") if not isinstance(entries, list): raise HarnessError("capture manifest artifacts must be an array") by_kind: dict[str, dict[str, Any]] = {} artifact_payloads: dict[str, dict[str, Any] | None] = {} for index, entry in enumerate(entries): if not isinstance(entry, dict): raise HarnessError(f"artifacts[{index}] must be an object") kind = _require_nonempty_string(entry.get("kind"), f"artifacts[{index}].kind") if kind not in REQUIRED_ARTIFACT_KINDS or kind in by_kind: raise HarnessError(f"unknown or duplicate artifact kind: {kind}") if entry.get("run_id") != run_id: raise HarnessError(f"artifact {kind} does not carry the capture run_id") path = _safe_artifact_path(artifacts_dir, entry.get("path"), f"artifact {kind}.path") expected_sha = _require_nonempty_string(entry.get("sha256"), f"artifact {kind}.sha256") if len(expected_sha) != 64 or any( char not in "0123456789abcdef" for char in expected_sha.lower() ): raise HarnessError(f"artifact {kind}.sha256 must be a SHA-256") actual_sha = _sha256(path) if actual_sha != expected_sha.lower(): raise HarnessError(f"artifact {kind} SHA-256 mismatch") slot = { "kind": kind, "run_id": run_id, "path": str(Path(entry["path"])), "sha256": actual_sha, "present": True, "evidence_state": _require_enum( entry.get("evidence_state"), EVIDENCE_STATES, f"artifact {kind}.evidence_state" ), "coverage_state": _require_enum( entry.get("coverage_state"), COVERAGE_STATES, f"artifact {kind}.coverage_state" ), "observation_state": _require_enum( entry.get("observation_state"), OBSERVATION_STATES, f"artifact {kind}.observation_state", ), } by_kind[kind] = slot artifact_payloads[kind] = _load_artifact_payload(path, run_id, kind) missing = sorted(set(REQUIRED_ARTIFACT_KINDS) - set(by_kind)) if missing: raise HarnessError(f"capture manifest is missing artifact kinds: {', '.join(missing)}") prefix_payload = artifact_payloads["prefix_cache"] if prefix_payload is not None: _validate_cache_identity(prefix_payload) verifier_payload = artifact_payloads["verifier"] if verifier_payload is None: raise HarnessError("verifier artifact must be JSON") receipt = architecture_receipt(config) receipt.update( { "schema": RUN_RECEIPT_SCHEMA, "schema_version": RUN_RECEIPT_SCHEMA, "run_id": run_id, "capture_kind": "measured_run", "banner": "CAPTURED ARTIFACTS / OUTCOME PENDING", "captured_at": captured_at, "artifact_slots": by_kind, } ) receipt["outcome"] = _validate_verifier(verifier_payload, by_kind["verifier"]) receipt["banner"] = f"CAPTURED ARTIFACTS / OUTCOME {receipt['outcome']['status'].upper()}" receipt["guardrails"].update( { "workload_executed": True, "gpu_execution_claimed": by_kind["kernel_profile"]["observation_state"] in ("dispatched", "measured", "verified", "accepted"), } ) for event in receipt["events"]: event["run_id"] = run_id event["artifact_refs"] = list(EVENT_ARTIFACT_KINDS[event["event_id"]]) related = [by_kind[kind] for kind in event["artifact_refs"]] event["coverage_state"] = min( (slot["coverage_state"] for slot in related), key=COVERAGE_STATES.index, ) event["observation_state"] = min( (slot["observation_state"] for slot in related), key=OBSERVATION_STATES.index, ) event["evidence_state"] = ( related[0]["evidence_state"] if len({slot["evidence_state"] for slot in related}) == 1 else "unknown" ) _join_memory_model(receipt, artifact_payloads["memory_placement"], run_id) _join_fabric_movements(receipt, artifact_payloads["fabric"], run_id) _join_resource_measurements(receipt, artifact_payloads, manifest, run_id) return receipt def receipt_summary(receipt: dict[str, Any]) -> dict[str, Any]: return { "schema_version": receipt["schema_version"], "fixture_id": receipt["fixture_id"], "comparison_id": receipt["comparison_id"], "run_id": receipt["run_id"], "capture_kind": receipt["capture_kind"], "banner": receipt["banner"], "event_order": [event["event_id"] for event in receipt["events"]], "artifact_paths_all_null": all( slot["path"] is None for slot in receipt["artifact_slots"].values() ), "resource_values_all_null": all( metric["value"] is None for metric in receipt["resource_ledger"] ), "outcome_status": receipt["outcome"]["status"], } def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG) parser.add_argument("--capture-manifest", type=Path) parser.add_argument("--artifacts-dir", type=Path) parser.add_argument("--run-id") parser.add_argument("--output", type=Path) parser.add_argument("--pretty", action="store_true") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) capture_fields = (args.capture_manifest, args.artifacts_dir, args.run_id) if any(value is not None for value in capture_fields) and not all( value is not None for value in capture_fields ): raise HarnessError( "capture mode requires --capture-manifest, --artifacts-dir, and --run-id together" ) config = load_reference_config(args.config) if all(value is not None for value in capture_fields): receipt = captured_receipt( config, args.capture_manifest, args.artifacts_dir, args.run_id ) else: receipt = architecture_receipt(config) text = json.dumps(receipt, indent=2 if args.pretty else None, sort_keys=True) if args.output: args.output.write_text(text + "\n", encoding="utf-8") else: print(text) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except HarnessError as exc: print(f"c001_reference_harness: {exc}", file=sys.stderr) raise SystemExit(2) from exc