#!/usr/bin/env python3 """Machine-readable Ty gate for the Task26 delivery-capability surface.""" from __future__ import annotations import argparse import ast import hashlib import json import sys from pathlib import Path from typing import Mapping, cast SCHEMA = "task26-ty-capability-surface-receipt-v3" EXECUTION_SCHEMA = "task26-hermetic-ty-execution-v2" TARGET_PYTHON_VERSION = "3.12" TY_COMMAND_ARGV = [ "sealed-ty", "check", "gateway/platforms/nutrition_coaching.py", "--output-format", "gitlab", "--python-version", TARGET_PYTHON_VERSION, ] PASS_STATUS = "PASS_CHANGED_SURFACE_WITH_PREEXISTING_OUTSIDE_SCOPE_DIAGNOSTICS" FAIL_STATUS = "FAIL_CHANGED_SURFACE_DIAGNOSTICS" TARGET_SYMBOLS = ( "DeliveryLaunchAuthorization", "DraftDeliveryCapability", "DeliveryDispatchClaim", "validate_delivery_transport", "claim_delivery_transport", "authorize_delivery_provider_transport", "record_customer_surface_receipt", "issue_delivery_capability", "_prepare_delivery_locked", ) def canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def symbol_spans(source: Path) -> dict[str, dict[str, int]]: tree = ast.parse(source.read_text(encoding="utf-8"), filename=str(source)) found: dict[str, dict[str, int]] = {} for node in ast.walk(tree): if ( isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in TARGET_SYMBOLS ): if node.name in found or node.end_lineno is None: raise ValueError(f"capability symbol cardinality is invalid: {node.name}") found[node.name] = { "start_line": node.lineno, "end_line": node.end_lineno, } if set(found) != set(TARGET_SYMBOLS): missing = sorted(set(TARGET_SYMBOLS) - set(found)) raise ValueError(f"capability symbols are missing: {missing}") return {name: found[name] for name in TARGET_SYMBOLS} def diagnostic_line(diagnostic: dict[str, object]) -> int: location = diagnostic.get("location") if not isinstance(location, dict): raise ValueError("Ty diagnostic location is invalid") location_object = cast(dict[str, object], location) positions = location_object.get("positions") if not isinstance(positions, dict): raise ValueError("Ty diagnostic positions are invalid") positions_object = cast(dict[str, object], positions) begin = positions_object.get("begin") if not isinstance(begin, dict): raise ValueError("Ty diagnostic begin position is invalid") begin_object = cast(dict[str, object], begin) line = begin_object.get("line") if type(line) is not int or line < 1: raise ValueError("Ty diagnostic line is invalid") return line def read_diagnostics_bytes(data: bytes) -> list[dict[str, object]]: value = json.loads(data) if ( not isinstance(value, list) or not value or any(not isinstance(row, dict) for row in value) ): raise ValueError("Ty GitLab report is invalid") return [cast(dict[str, object], row) for row in value] def build_receipt( source: Path, diagnostics: list[dict[str, object]], *, raw_diagnostics: bytes, command: list[str], execution_provenance: Mapping[str, object], ) -> dict[str, object]: spans = symbol_spans(source) capability: list[dict[str, object]] = [] outside: list[dict[str, object]] = [] for diagnostic in diagnostics: line = diagnostic_line(diagnostic) if any( span["start_line"] <= line <= span["end_line"] for span in spans.values() ): capability.append(diagnostic) else: outside.append(diagnostic) outside_projection = sorted( outside, key=lambda row: canonical(row), ) interpreter = execution_provenance.get("qualification_interpreter") if not isinstance(interpreter, dict): raise ValueError("Ty qualification interpreter is invalid") interpreter_value = cast(dict[str, object], interpreter) interpreter_version = interpreter_value.get("version") digest_fields = ( "wheel_sha256", "installed_executable_sha256", "runner_tool_sha256", "gate_tool_sha256", ) if ( execution_provenance.get("schema") != EXECUTION_SCHEMA or execution_provenance.get("target_python_version") != TARGET_PYTHON_VERSION or execution_provenance.get("ty_command_argv") != TY_COMMAND_ARGV or TY_COMMAND_ARGV.count("--python-version") != 1 or any( not isinstance(execution_provenance.get(field), str) or len(cast(str, execution_provenance[field])) != 64 or any( character not in "0123456789abcdef" for character in cast(str, execution_provenance[field]) ) for field in digest_fields ) or interpreter_value.get("executable_role") != "selected_qualification_python" or not isinstance(interpreter_version, list) or interpreter_version[:2] != [3, 12] ): raise ValueError("Ty semantic target provenance is invalid") semantic_projection = sorted(diagnostics, key=lambda row: canonical(row)) receipt: dict[str, object] = { "schema": SCHEMA, "status": PASS_STATUS if not capability else FAIL_STATUS, "source_path": "gateway/platforms/nutrition_coaching.py", "source_sha256": sha256_file(source), "diagnostics_format": "gitlab", "raw_diagnostics_sha256": hashlib.sha256(raw_diagnostics).hexdigest(), "raw_diagnostics_size": len(raw_diagnostics), "raw_diagnostics_count": len(diagnostics), "target_python_version": TARGET_PYTHON_VERSION, "qualification_interpreter": interpreter_value, "ty_command_argv": command, "ty_wheel_sha256": execution_provenance.get("wheel_sha256"), "ty_executable_sha256": execution_provenance.get( "installed_executable_sha256" ), "runner_tool_sha256": execution_provenance.get("runner_tool_sha256"), "gate_tool_sha256": execution_provenance.get("gate_tool_sha256"), "diagnostics_semantic_fingerprint_sha256": hashlib.sha256( canonical(semantic_projection) ).hexdigest(), "ty_execution_provenance": dict(execution_provenance), "target_symbol_spans": spans, "total_diagnostic_count": len(diagnostics), "capability_diagnostic_count": len(capability), "outside_surface_diagnostic_count": len(outside), "outside_surface_fingerprint_sha256": hashlib.sha256( canonical(outside_projection) ).hexdigest(), "capability_diagnostics": capability, "history": { "v13_nutrition_coaching_scanned_or_sealed": False, "v14_first_exposed_dirty_monolith_ty_debt": True, "outside_scope_diagnostics_fixed_or_claimed_clean": False, }, } receipt["receipt_sha256"] = hashlib.sha256(canonical(receipt)).hexdigest() return receipt def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--source", type=Path, required=True) parser.add_argument("--diagnostics-json", type=Path, required=True) parser.add_argument("--execution-provenance", type=Path, required=True) parser.add_argument("--receipt", type=Path) args = parser.parse_args() source = args.source.absolute() provenance_value = json.loads(args.execution_provenance.read_bytes()) if ( not isinstance(provenance_value, dict) or provenance_value.get("schema") != EXECUTION_SCHEMA ): raise ValueError("Ty execution provenance is invalid") execution_provenance = cast(dict[str, object], provenance_value) raw_diagnostics = args.diagnostics_json.read_bytes() diagnostics = read_diagnostics_bytes(raw_diagnostics) command_value = execution_provenance.get("ty_command_argv") if not isinstance(command_value, list) or any( not isinstance(item, str) for item in command_value ): raise ValueError("Ty command provenance is invalid") command = cast(list[str], command_value) receipt = build_receipt( source, diagnostics, raw_diagnostics=raw_diagnostics, command=command, execution_provenance=execution_provenance, ) output = canonical(receipt) + b"\n" if args.receipt is not None: args.receipt.write_bytes(output) sys.stdout.buffer.write(output) return 0 if receipt["capability_diagnostic_count"] == 0 else 1 if __name__ == "__main__": raise SystemExit(main())