from __future__ import annotations import argparse from collections import OrderedDict import csv from datetime import datetime import hashlib import io import json import math import os from pathlib import Path from typing import Any, Mapping, Sequence from .archive import sha256_file, validate_pdf from .items import ITEM_KEYS from .manifests import (MANIFEST_COLUMNS, canonical_json_bytes, write_bytes_create_new, write_json_create_new) from .models import ContractError, ErrorCode, EXPECTED_REPORT_KEYS, HASH_RE, TaskSpec from .quota import QUOTA_COLUMNS, QuotaLedger from .terminal import canonical_terminal_bytes PLAN_KEYS = ( "schema_version", "plan_id", "created_at_utc", "quota_date", "quota_ledger", "historical_quota_date", "historical_quota_ledger", "trigger_cap", "entries", ) PLAN_ENTRY_KEYS = ( "slot_id", "mode", "batch_id", "item_order", "query", "expected_report", "discovery_sha256", "external_exclusion_allowed", "selected_from_readonly_discovery", ) EVIDENCE_REFERENCE_KEYS = ( "slot_id", "task_path", "discovery_path", "terminal_path", "timing_path", "manifest_path", "item_id", ) ITEM_WINDOW_KEYS = ( "item_index", "started_monotonic_ns", "work_deadline_monotonic_ns", "close_deadline_monotonic_ns", "terminal_monotonic_ns", "elapsed_ms", "terminal_delta_from_previous_ms", "deadline_met", ) SAMPLE_KEYS = ( "slot_id", "report_identity", "status", "elapsed_ms", "terminal_delta_from_previous_ms", "triggered", "pdf_magic_pass", "byte_match_pass", "hash_match_pass", "openability_pass", "page_count_pass", "manifest_pass", "quota_pass", "terminal_sha256", "timing_sha256", ) SUMMARY_KEYS = ( "schema_version", "plan_id", "started_at_utc", "ended_at_utc", "primary_total", "included_total", "success_total", "trigger_total", "confirmed_total", "uncertain_total", "active_total", "median_ms", "p90_ms", "batch_first_ms", "batch_increment_ms", "magic_pass", "byte_match_pass", "hash_match_pass", "openability_pass", "page_count_pass", "manifest_pass", "quota_pass", "result", "blockers", "evidence_manifest_path", "evidence_manifest_bytes", "evidence_manifest_sha256", "samples", ) PERFORMANCE_MANIFEST_COLUMNS = ("slot_id", "evidence_type", "path", "bytes", "sha256") PREPLAN_KEYS = ( "schema_version", "task_id", "design_plan_id", "execution_started_at_utc", "ended_at_utc", "phase", "status", "stop_code", "exit_code", "evidence_root", "plan_path", "plan_presence", "plan_bytes", "plan_sha256", "discovery_completed", "qualified_candidate_total", "reserve_count", "trigger_count", "quota_write_count", "blockers", ) PREPLAN_STOP_CODES = ( "PRECHECK_FAILED", "ACCESS_CONTROL", "QUOTA_INSUFFICIENT", "DISCOVERY_FAILED", "CANDIDATES_INSUFFICIENT", "PLAN_CREATE_FAILED", "STATE_UNCERTAIN", ) PREPLAN_EVIDENCE_REFERENCE_KEYS = ("evidence_type", "path", "status") PREPLAN_MANIFEST_COLUMNS = ("evidence_type", "path", "bytes", "sha256", "status") EXECUTION_CONTEXT_KEYS = ( "schema_version", "task_id", "design_plan_id", "evidence_root", "execution_started_at_utc", "timezone", "trigger_cap", "subjects", ) DISCOVERY_KEYS = ( "schema_version", "task_id", "handoff_id", "performance_plan_id", "performance_slot_id", "query", "observed_at_utc", "completed_at_utc", "started_monotonic_ns", "ended_monotonic_ns", "screens_scanned", "candidate_count", "candidates", "quota_ledger_touched", "trigger_attempted", ) DISCOVERY_SLOTS = ( "PERF-S01", "PERF-S02", "PERF-S03", "PERF-S04", "PERF-S05", "PERF-S06", "PERF-B01", ) DISCOVERY_EVIDENCE_TYPES = tuple(f"discovery_{slot}" for slot in DISCOVERY_SLOTS) PREPLAN_OPTIONAL_EVIDENCE_TYPES = ( "quota_ledger", "performance_plan_request", "performance_plan", ) PREPLAN_EVIDENCE_STATUSES = ("VALID", "INVALID", "UNCERTAIN") PERFORMANCE_TASK_ID = "DEV-ANA-HIBOR-FAST-COLLECTION-20260729-001" PERFORMANCE_DESIGN_PLAN_ID = "PERF-TEST-PLAN-ANA-HIBOR-FAST-COLLECTION-V003" PERFORMANCE_EVIDENCE_ROOT = Path( r"E:\mb-ms-doc\project-info\ana-data\tmp\HIBOR-FAST-PERFORMANCE-20260730-004" ) PERFORMANCE_SUBJECTS = ("三环集团", "国瓷材料", "MLCC") PERFORMANCE_SUBJECT_CODEPOINTS = ( (0x4E09, 0x73AF, 0x96C6, 0x56E2), (0x56FD, 0x74F7, 0x6750, 0x6599), (0x004D, 0x004C, 0x0043, 0x0043), ) PDFINFO_EXECUTABLE = Path( r"C:\Users\Cai\.cache\codex-runtimes\codex-primary-runtime\dependencies\native\poppler\Library\bin\pdfinfo.exe" ) PDFINFO_BYTES = 65_536 PDFINFO_SHA256 = "bc2c0f980c9a2a29cd1e06aacd8d1c7b67a5304e9d1d6f75190bdeb9c81a4365" def _timestamp(value: Any, field: str) -> str: if not isinstance(value, str) or not value.endswith("Z"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "UTC Z timestamp required") try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "invalid timestamp") from exc if parsed.tzinfo is None: raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "timezone required") return value def _expected_report(value: Any, field: str) -> dict[str, Any]: if not isinstance(value, Mapping) or tuple(value.keys()) != EXPECTED_REPORT_KEYS: raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "exact report keys required") report = dict(value) if not isinstance(report["report_identity"], str) or not HASH_RE.fullmatch(report["report_identity"]): raise ContractError(ErrorCode.TASK_SPEC_INVALID, f"{field}.report_identity", "SHA-256 required") for key in ("title", "institution", "report_date"): if not isinstance(report[key], str) or not report[key]: raise ContractError(ErrorCode.TASK_SPEC_INVALID, f"{field}.{key}", "non-empty string") if type(report["page_count"]) is not int or report["page_count"] < 1: raise ContractError(ErrorCode.TASK_SPEC_INVALID, f"{field}.page_count", "positive INT") preimage = "|".join((report["title"], report["institution"], report["report_date"])) if hashlib.sha256(preimage.encode("utf-8")).hexdigest() != report["report_identity"]: raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, field, "identity preimage mismatch") return report def validate_performance_plan(value: Mapping[str, Any]) -> dict[str, Any]: if tuple(value.keys()) != PLAN_KEYS or value.get("schema_version") != "HIBOR_FAST_PERFORMANCE_PLAN_V002": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "performance_plan", "schema/key order") plan = dict(value) if not isinstance(plan["plan_id"], str) or not plan["plan_id"]: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_id", "non-empty") _timestamp(plan["created_at_utc"], "created_at_utc") for date_key, path_key in (("quota_date", "quota_ledger"), ("historical_quota_date", "historical_quota_ledger")): date_value = plan[date_key] if not isinstance(date_value, str) or not isinstance(plan[path_key], str): raise ContractError(ErrorCode.TASK_SPEC_INVALID, date_key, "date/path required") if Path(plan[path_key]).name != f"daily_quota_{date_value}.csv": raise ContractError(ErrorCode.QUOTA_DATE_UNCERTAIN, path_key, "filename/date mismatch") if plan["historical_quota_date"] != "2026-07-29": raise ContractError(ErrorCode.QUOTA_DATE_UNCERTAIN, "historical_quota_date", "known baseline date") if plan["trigger_cap"] != 10: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "trigger_cap", "exactly 10") entries = plan["entries"] if not isinstance(entries, list) or len(entries) != 10: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "entries", "exactly 10 primary rows") expected_queries = ("三环集团", "三环集团", "国瓷材料", "国瓷材料", "MLCC", "MLCC", "MLCC", "MLCC", "MLCC", "MLCC") batch_ids: set[str] = set() slots: set[str] = set() identities: set[str] = set() for index, raw in enumerate(entries): if not isinstance(raw, Mapping) or tuple(raw.keys()) != PLAN_ENTRY_KEYS: raise ContractError(ErrorCode.TASK_SPEC_INVALID, f"entries[{index}]", "exact key order") if not isinstance(raw["slot_id"], str) or not raw["slot_id"] or raw["slot_id"] in slots: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "slot_id", "unique non-empty") slots.add(raw["slot_id"]) if raw["query"] != expected_queries[index]: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "query", "fixed subject/order") if index < 6: if (raw["mode"], raw["batch_id"], raw["item_order"]) != ("collect-one", None, 1): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "entries", "six collect-one rows first") else: if raw["mode"] != "batch" or not isinstance(raw["batch_id"], str) or raw["item_order"] != index - 5: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "entries", "one ordered four-item batch last") batch_ids.add(raw["batch_id"]) report = _expected_report(raw["expected_report"], f"entries[{index}].expected_report") if report["report_identity"] in identities: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "report_identity", "duplicate") identities.add(report["report_identity"]) if not isinstance(raw["discovery_sha256"], str) or not HASH_RE.fullmatch(raw["discovery_sha256"]): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery_sha256", "SHA-256 required") if raw["external_exclusion_allowed"] is not True or raw["selected_from_readonly_discovery"] is not True: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery", "true/true required") if len(batch_ids) != 1: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "batch_id", "one shared batch id") return plan def write_performance_plan(path: Path, value: Mapping[str, Any]) -> tuple[int, str]: return write_json_create_new(path, validate_performance_plan(value)) def _file(root: Path, value: Any, field: str) -> Path: if not isinstance(value, str) or not value: raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "path required") candidate = Path(value) if not candidate.is_absolute(): candidate = root / candidate try: resolved = candidate.resolve(strict=True) root_resolved = root.resolve(strict=True) except OSError as exc: raise ContractError(ErrorCode.PERSIST_LATE, field, "path absent/unresolvable") from exc if resolved != root_resolved and root_resolved not in resolved.parents: raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "outside evidence root") if not resolved.is_file() or resolved.is_symlink(): raise ContractError(ErrorCode.PERSIST_LATE, field, "ordinary file required") return resolved def _quota_file(root: Path, value: Any, field: str, quota_date: str) -> Path: """Accept a run-local fixture or the exact shared daily quota ledger.""" if not isinstance(value, str) or not value: raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "path required") candidate = Path(value) if not candidate.is_absolute(): candidate = root / candidate try: resolved = candidate.resolve(strict=True) root_resolved = root.resolve(strict=True) except OSError as exc: raise ContractError(ErrorCode.PERSIST_LATE, field, "path absent/unresolvable") from exc inside_root = resolved == root_resolved or root_resolved in resolved.parents if not inside_root: try: shared = (root.parent / "report-collection-control" / f"daily_quota_{quota_date}.csv").resolve(strict=True) except OSError as exc: raise ContractError(ErrorCode.PERSIST_LATE, field, "shared quota path absent/unresolvable") from exc if resolved != shared: raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "unapproved quota path") if not resolved.is_file() or resolved.is_symlink(): raise ContractError(ErrorCode.PERSIST_LATE, field, "ordinary file required") return resolved def _runtime_pdfinfo(value: Any) -> Path: """Validate the reviewed external runtime tool without treating it as case data.""" if not isinstance(value, str) or not value: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "pdfinfo_executable", "path required") try: candidate = Path(value).resolve(strict=True) expected = PDFINFO_EXECUTABLE.resolve(strict=True) except OSError as exc: raise ContractError(ErrorCode.PERSIST_LATE, "pdfinfo_executable", "runtime absent") from exc if candidate != expected: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "pdfinfo_executable", "unreviewed runtime path") if not candidate.is_file() or candidate.is_symlink(): raise ContractError(ErrorCode.PERSIST_LATE, "pdfinfo_executable", "ordinary file required") try: size = candidate.stat().st_size digest = sha256_file(candidate) except OSError as exc: raise ContractError(ErrorCode.PERSIST_LATE, "pdfinfo_executable", "runtime unreadable") from exc if size != PDFINFO_BYTES or digest != PDFINFO_SHA256: raise ContractError(ErrorCode.HASH_MISMATCH, "pdfinfo_executable", "runtime snapshot drift") return candidate def _json(path: Path, field: str, *, canonical: bool) -> tuple[dict[str, Any], bytes]: try: raw = path.read_bytes() value = json.loads(raw.decode("utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise ContractError(ErrorCode.PERSIST_LATE, field, "strict JSON required") from exc if not isinstance(value, dict): raise ContractError(ErrorCode.PERSIST_LATE, field, "JSON object required") if canonical and canonical_json_bytes(value) != raw: raise ContractError(ErrorCode.PERSIST_LATE, field, "canonical bytes required") return value, raw def _quota_rows(path: Path, quota_date: str) -> list[dict[str, str]]: try: with path.open("r", encoding="utf-8", newline="") as stream: reader = csv.DictReader(stream) if tuple(reader.fieldnames or ()) != QUOTA_COLUMNS: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "column mismatch") rows = list(reader) except (OSError, UnicodeError, csv.Error) as exc: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "read failed") from exc if any(row["quota_date"] != quota_date for row in rows): raise ContractError(ErrorCode.QUOTA_DATE_UNCERTAIN, "quota", "row date drift") QuotaLedger(path, quota_date=quota_date).snapshot() return rows def _manifest_rows(path: Path) -> list[dict[str, str]]: try: with path.open("r", encoding="utf-8", newline="") as stream: reader = csv.DictReader(stream) if tuple(reader.fieldnames or ()) != MANIFEST_COLUMNS: raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "column mismatch") return list(reader) except (OSError, UnicodeError, csv.Error) as exc: raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "read failed") from exc def _run_id(spec: TaskSpec) -> str: canonical = json.dumps(spec.raw, ensure_ascii=False, separators=(",", ":"), sort_keys=False).encode("utf-8") token = hashlib.sha256(b"HIBOR-RUN-V001\x00" + canonical).hexdigest()[:24] import re task = re.sub(r"[^A-Za-z0-9_.-]", "-", spec.task_id)[:80] return f"RUN-{task}-{token}" def _evidence_row(slot_id: str, evidence_type: str, path: Path) -> dict[str, str]: return { "slot_id": slot_id, "evidence_type": evidence_type, "path": str(path), "bytes": str(path.stat().st_size), "sha256": sha256_file(path), } def _write_evidence_manifest(path: Path, rows: Sequence[Mapping[str, str]]) -> tuple[int, str]: buffer = io.StringIO(newline="") writer = csv.DictWriter(buffer, fieldnames=PERFORMANCE_MANIFEST_COLUMNS, lineterminator="\r\n") writer.writeheader() for row in rows: if tuple(row.keys()) != PERFORMANCE_MANIFEST_COLUMNS: raise ContractError(ErrorCode.MANIFEST_INVALID, "performance_manifest", "key order") writer.writerow(row) data = buffer.getvalue().encode("utf-8") return write_bytes_create_new(path, data) def validate_execution_context_value(value: Mapping[str, Any], root: Path, expected_started_at_utc: str) -> None: if (tuple(value.keys()) != EXECUTION_CONTEXT_KEYS or value.get("schema_version") != "HIBOR_FAST_PERFORMANCE_EXECUTION_CONTEXT_V001"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "schema/key order") if ((value["task_id"], value["design_plan_id"]) != (PERFORMANCE_TASK_ID, PERFORMANCE_DESIGN_PLAN_ID)): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "task/design mismatch") if value["evidence_root"] != str(root): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "root mismatch") _timestamp(value["execution_started_at_utc"], "execution_started_at_utc") if value["execution_started_at_utc"] != expected_started_at_utc: raise ContractError(ErrorCode.CLOCK_INVALID, "execution_context", "start mismatch") if (value["timezone"], value["trigger_cap"], value["subjects"]) != ( "Asia/Shanghai", 10, list(PERFORMANCE_SUBJECTS)): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "fixed values") actual_codepoints = tuple(tuple(ord(char) for char in subject) for subject in value["subjects"]) if actual_codepoints != PERFORMANCE_SUBJECT_CODEPOINTS: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "subject codepoints") def validate_execution_context_bytes(data: bytes, root: Path, expected_started_at_utc: str) -> OrderedDict[str, Any]: if (data.startswith(b"\xef\xbb\xbf") or b"\r" in data or b"\x00" in data or data.endswith(b"\n")): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "byte profile") try: text = data.decode("utf-8", "strict") value = json.loads(text, object_pairs_hook=OrderedDict) except (UnicodeError, json.JSONDecodeError) as exc: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "UTF-8/JSON") from exc if not isinstance(value, Mapping): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "mapping required") validate_execution_context_value(value, root, expected_started_at_utc) if canonical_json_bytes(value) != data: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "canonical roundtrip") return OrderedDict(value) def build_execution_context(evidence_root: Path, execution_started_at_utc: str) -> tuple[OrderedDict[str, Any], bytes]: # abspath is lexical and deliberately does not follow junctions/symlinks. # The mutating coordinator validates every existing lexical ancestor before # it is permitted to resolve or create this path. root = Path(os.path.abspath(os.fspath(evidence_root))) if not root.is_absolute(): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_root", "absolute path required") try: str(root).encode("ascii", "strict") except UnicodeError as exc: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_root", "ASCII path required") from exc values: dict[str, Any] = { "schema_version": "HIBOR_FAST_PERFORMANCE_EXECUTION_CONTEXT_V001", "task_id": PERFORMANCE_TASK_ID, "design_plan_id": PERFORMANCE_DESIGN_PLAN_ID, "evidence_root": str(root), "execution_started_at_utc": execution_started_at_utc, "timezone": "Asia/Shanghai", "trigger_cap": 10, "subjects": list(PERFORMANCE_SUBJECTS), } context = OrderedDict((key, values[key]) for key in EXECUTION_CONTEXT_KEYS) validate_execution_context_value(context, root, execution_started_at_utc) data = canonical_json_bytes(context) validated = validate_execution_context_bytes(data, root, execution_started_at_utc) if validated != context: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "memory roundtrip") return context, data def _validate_execution_context(value: Mapping[str, Any], root: Path, terminal: Mapping[str, Any]) -> None: validate_execution_context_value(value, root, terminal["execution_started_at_utc"]) def validate_preplan_terminal(value: Mapping[str, Any], evidence_root: Path) -> dict[str, Any]: if (tuple(value.keys()) != PREPLAN_KEYS or value.get("schema_version") != "HIBOR_FAST_PREPLAN_TERMINAL_V001"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "preplan_terminal", "schema/key order") terminal = dict(value) root = Path(evidence_root).resolve() if ((terminal["task_id"], terminal["design_plan_id"]) != (PERFORMANCE_TASK_ID, PERFORMANCE_DESIGN_PLAN_ID)): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "preplan_terminal", "task/design mismatch") start = datetime.fromisoformat(_timestamp( terminal["execution_started_at_utc"], "execution_started_at_utc").replace("Z", "+00:00")) end = datetime.fromisoformat(_timestamp( terminal["ended_at_utc"], "ended_at_utc").replace("Z", "+00:00")) if end < start: raise ContractError(ErrorCode.CLOCK_INVALID, "preplan_terminal", "end before start") if (terminal["phase"], terminal["status"]) != ("PREPLAN", "STOPPED"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "preplan_terminal", "phase/status") if terminal["stop_code"] not in PREPLAN_STOP_CODES: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "stop_code", "enum") expected_exit = 27 if terminal["stop_code"] == "STATE_UNCERTAIN" else 10 if terminal["exit_code"] != expected_exit: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "exit_code", "stop projection") if terminal["evidence_root"] != str(root): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_root", "root mismatch") expected_plan_path = root / "control" / "performance_plan.json" try: actual_plan_path = Path(terminal["plan_path"]).resolve(strict=False) except (OSError, TypeError) as exc: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_path", "invalid") from exc if actual_plan_path != expected_plan_path: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_path", "fixed path mismatch") presence = terminal["plan_presence"] if presence not in ("N", "A", "I", "U"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "preplan enum") if presence == "U" and terminal["stop_code"] != "STATE_UNCERTAIN": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "U requires STATE_UNCERTAIN") if presence in ("N", "A", "U"): if terminal["plan_bytes"] is not None or terminal["plan_sha256"] is not None: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "null bytes/hash required") else: if (type(terminal["plan_bytes"]) is not int or terminal["plan_bytes"] < 0 or not isinstance(terminal["plan_sha256"], str) or not HASH_RE.fullmatch(terminal["plan_sha256"])): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "known invalid receipt") for field, upper in (("discovery_completed", 7), ("qualified_candidate_total", None), ("quota_write_count", None)): current = terminal[field] if type(current) is not int or current < 0 or (upper is not None and current > upper): raise ContractError(ErrorCode.TASK_SPEC_INVALID, field, "bounded nonnegative INT") if terminal["reserve_count"] != 0 or terminal["trigger_count"] != 0: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "preplan_terminal", "reserve/trigger must be zero") blockers = terminal["blockers"] if not isinstance(blockers, list) or not blockers or blockers[0] != terminal["stop_code"] or not all( isinstance(item, str) and item for item in blockers): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "blockers", "first blocker must be stop") if presence in ("N", "A") and (expected_plan_path.exists() or expected_plan_path.is_symlink()): raise ContractError(ErrorCode.PERSIST_LATE, "plan_path", "unexpected object") if presence == "I": if not expected_plan_path.is_file() or expected_plan_path.is_symlink(): raise ContractError(ErrorCode.PERSIST_LATE, "plan_path", "known invalid file absent") if (expected_plan_path.stat().st_size != terminal["plan_bytes"] or sha256_file(expected_plan_path) != terminal["plan_sha256"]): raise ContractError(ErrorCode.HASH_MISMATCH, "plan_path", "known invalid receipt drift") return terminal def _write_preplan_manifest(path: Path, rows: Sequence[Mapping[str, str]]) -> tuple[int, str]: buffer = io.StringIO(newline="") writer = csv.DictWriter(buffer, fieldnames=PREPLAN_MANIFEST_COLUMNS, lineterminator="\r\n") writer.writeheader() for row in rows: if tuple(row.keys()) != PREPLAN_MANIFEST_COLUMNS: raise ContractError(ErrorCode.MANIFEST_INVALID, "preplan_manifest", "key order") writer.writerow(row) return write_bytes_create_new(path, buffer.getvalue().encode("utf-8")) def write_preplan_package(*, terminal_path: Path, manifest_path: Path, terminal_value: Mapping[str, Any], evidence_root: Path, evidence_paths: Sequence[Mapping[str, Any]]) -> tuple[tuple[int, str], tuple[int, str]]: root = Path(evidence_root).resolve(strict=True) if not root.is_dir() or root.is_symlink(): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_root", "ordinary directory") expected_terminal = root / "preplan" / "preplan_terminal.json" expected_manifest = root / "preplan" / "preplan_manifest.csv" if (Path(terminal_path).resolve(strict=False), Path(manifest_path).resolve(strict=False)) != ( expected_terminal, expected_manifest): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "preplan_output", "fixed paths required") terminal = validate_preplan_terminal(terminal_value, root) if not isinstance(evidence_paths, Sequence) or not evidence_paths: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_paths", "non-empty rows") rows: list[dict[str, str]] = [] seen_types: set[str] = set() observed_types: list[str] = [] qualified_identities: set[str] = set() observed_quota_writes = 0 observed_performance_plan_id: str | None = None plan_evidence_path: Path | None = None for index, raw in enumerate(evidence_paths): if not isinstance(raw, Mapping) or tuple(raw.keys()) != PREPLAN_EVIDENCE_REFERENCE_KEYS: raise ContractError(ErrorCode.TASK_SPEC_INVALID, f"evidence_paths[{index}]", "exact keys") if (not isinstance(raw["evidence_type"], str) or not raw["evidence_type"] or raw["evidence_type"] in seen_types or raw["status"] not in PREPLAN_EVIDENCE_STATUSES): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_paths", "unique type/status") seen_types.add(raw["evidence_type"]) observed_types.append(raw["evidence_type"]) if (raw["evidence_type"] != "execution_context" and raw["evidence_type"] not in DISCOVERY_EVIDENCE_TYPES and raw["evidence_type"] not in PREPLAN_OPTIONAL_EVIDENCE_TYPES): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_paths", "unknown evidence type") evidence_path = _file(root, raw["path"], "evidence_path") if index == 0: if (raw["evidence_type"] != "execution_context" or evidence_path != root / "control" / "execution_context.json"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_paths", "context first") context, _ = _json(evidence_path, "execution_context", canonical=True) _validate_execution_context(context, root, terminal) if raw["status"] != "VALID": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "execution_context", "status") elif raw["evidence_type"] in DISCOVERY_EVIDENCE_TYPES: slot = raw["evidence_type"].removeprefix("discovery_") expected_path = root / "discovery" / f"{slot}.json" if evidence_path != expected_path: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery", "fixed path mismatch") discovery, _ = _json(evidence_path, "discovery", canonical=True) if (tuple(discovery.keys()) != DISCOVERY_KEYS or discovery.get("schema_version") != "HIBOR_FAST_DISCOVERY_V001" or discovery.get("performance_slot_id") != slot or discovery.get("quota_ledger_touched") is not False or discovery.get("trigger_attempted") is not False): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery", "schema/slot/side effect") expected_queries = { "PERF-S01": "三环集团", "PERF-S02": "三环集团", "PERF-S03": "国瓷材料", "PERF-S04": "国瓷材料", "PERF-S05": "MLCC", "PERF-S06": "MLCC", "PERF-B01": "MLCC", } if (discovery.get("query") != expected_queries[slot] or any(not isinstance(discovery.get(field), str) or not discovery[field] for field in ("task_id", "handoff_id", "performance_plan_id"))): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery", "identity/query") if observed_performance_plan_id is None: observed_performance_plan_id = discovery["performance_plan_id"] elif observed_performance_plan_id != discovery["performance_plan_id"]: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery", "plan id drift") candidates = discovery.get("candidates") if (not isinstance(candidates, list) or discovery.get("candidate_count") != len(candidates) or any(type(discovery.get(field)) is not int or discovery[field] < 0 for field in ("started_monotonic_ns", "ended_monotonic_ns", "screens_scanned", "candidate_count")) or discovery["ended_monotonic_ns"] < discovery["started_monotonic_ns"]): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery", "counter mismatch") observed_at = datetime.fromisoformat(_timestamp( discovery.get("observed_at_utc"), "discovery.observed_at_utc", ).replace("Z", "+00:00")) completed_at = datetime.fromisoformat(_timestamp( discovery.get("completed_at_utc"), "discovery.completed_at_utc", ).replace("Z", "+00:00")) if completed_at < observed_at: raise ContractError(ErrorCode.CLOCK_INVALID, "discovery", "completed before observed") for candidate_index, candidate in enumerate(candidates): report = _expected_report(candidate, f"discovery.candidates[{candidate_index}]") qualified_identities.add(report["report_identity"]) if raw["status"] != "VALID": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery", "status") elif raw["evidence_type"] == "quota_ledger": import re match = re.fullmatch(r"daily_quota_(\d{4}-\d{2}-\d{2})\.csv", evidence_path.name) if not match: raise ContractError(ErrorCode.QUOTA_DATE_UNCERTAIN, "quota_ledger", "filename") quota_rows = _quota_rows(evidence_path, match.group(1)) if any(row["event_family"] in {"RESERVATION", "QUOTA_TERMINAL", "ARTIFACT_TERMINAL"} for row in quota_rows): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota_ledger", "post-reserve event") observed_quota_writes = sum(row["event_type"] == "APP_RECONCILE" for row in quota_rows) if raw["status"] != "VALID": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "quota_ledger", "status") elif raw["evidence_type"] == "performance_plan_request": if evidence_path != root / "control" / "performance_plan_request.json": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "performance_plan_request", "fixed path") plan_request, _ = _json( evidence_path, "performance_plan_request", canonical=True, ) request_valid = False if (tuple(plan_request.keys()) == ("kind", "value") and plan_request.get("kind") == "PLAN" and isinstance(plan_request.get("value"), Mapping)): try: validate_performance_plan(plan_request["value"]) except ContractError: pass else: request_valid = True expected_status = "VALID" if request_valid else "INVALID" if raw["status"] != expected_status: raise ContractError( ErrorCode.TASK_SPEC_INVALID, "performance_plan_request", "status", ) elif raw["evidence_type"] == "performance_plan": if evidence_path != root / "control" / "performance_plan.json": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "performance_plan", "fixed path") if raw["status"] != "INVALID": raise ContractError(ErrorCode.TASK_SPEC_INVALID, "performance_plan", "status") plan_evidence_path = evidence_path row = _evidence_row("PREPLAN", raw["evidence_type"], evidence_path) rows.append(dict(zip(PREPLAN_MANIFEST_COLUMNS, ( row["evidence_type"], row["path"], row["bytes"], row["sha256"], raw["status"], )))) expected_discovery_types = list(DISCOVERY_EVIDENCE_TYPES[:terminal["discovery_completed"]]) actual_discovery_types = [item for item in observed_types if item in DISCOVERY_EVIDENCE_TYPES] if actual_discovery_types != expected_discovery_types: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "discovery_completed", "physical prefix mismatch") expected_order = (["execution_context"] + expected_discovery_types + [item for item in PREPLAN_OPTIONAL_EVIDENCE_TYPES if item in observed_types]) if observed_types != expected_order: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_paths", "fixed evidence order") if terminal["qualified_candidate_total"] != len(qualified_identities): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "qualified_candidate_total", "physical mismatch") if terminal["quota_write_count"] != observed_quota_writes: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota_write_count", "physical mismatch") if terminal["plan_presence"] == "I": if plan_evidence_path is None: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "invalid plan evidence absent") try: candidate_plan, _ = _json(plan_evidence_path, "performance_plan", canonical=True) validate_performance_plan(candidate_plan) except ContractError: pass else: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "valid plan cannot PREPLAN_STOP") elif plan_evidence_path is not None: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "unexpected plan evidence") elif terminal["plan_presence"] == "U": # U is reserved for a genuinely unclassifiable filesystem state. It # cannot be used to hide an ordinary, readable and valid plan merely # by omitting that file from the caller-supplied evidence list. expected_plan_path = root / "control" / "performance_plan.json" valid_plan_observed = False if expected_plan_path.is_file() and not expected_plan_path.is_symlink(): try: candidate_plan, _ = _json(expected_plan_path, "performance_plan", canonical=True) validate_performance_plan(candidate_plan) except ContractError: pass else: valid_plan_observed = True if valid_plan_observed: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "plan_presence", "valid plan cannot be U") terminal_receipt = write_json_create_new(expected_terminal, terminal) terminal_row = _evidence_row("PREPLAN", "preplan_terminal", expected_terminal) rows.append(dict(zip(PREPLAN_MANIFEST_COLUMNS, ( terminal_row["evidence_type"], terminal_row["path"], terminal_row["bytes"], terminal_row["sha256"], "STOPPED", )))) manifest_receipt = _write_preplan_manifest(expected_manifest, rows) return terminal_receipt, manifest_receipt def build_performance_summary_from_evidence(*, plan_path: Path, plan_bytes: int, plan_sha256: str, evidence_root: Path, evidence_references: Sequence[Mapping[str, Any]], started_at_utc: str, ended_at_utc: str, synthetic: bool = False) -> tuple[dict[str, Any], list[dict[str, str]]]: _timestamp(started_at_utc, "started_at_utc") _timestamp(ended_at_utc, "ended_at_utc") root = Path(evidence_root) if not root.is_dir() or root.is_symlink(): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_root", "ordinary directory") physical_plan_path = _file(root, str(plan_path), "plan_path") plan_raw, physical_plan_bytes = _json(physical_plan_path, "performance_plan", canonical=True) if (type(plan_bytes) is not int or plan_bytes < 1 or not isinstance(plan_sha256, str) or not HASH_RE.fullmatch(plan_sha256)): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "performance_plan", "authorized bytes/hash required") if (len(physical_plan_bytes) != plan_bytes or hashlib.sha256(physical_plan_bytes).hexdigest() != plan_sha256): raise ContractError(ErrorCode.HASH_MISMATCH, "performance_plan", "authorized snapshot drift") plan_value = validate_performance_plan(plan_raw) if not isinstance(evidence_references, Sequence) or len(evidence_references) > 10: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_references", "0..10 rows") plan_by_slot = {row["slot_id"]: row for row in plan_value["entries"]} seen_slots: set[str] = set() samples: list[dict[str, Any]] = [] evidence_manifest: list[dict[str, str]] = [ _evidence_row("GLOBAL", "performance_plan", physical_plan_path) ] runtime_pdfinfo_path: Path | None = None quota_path = _quota_file(root, plan_value["quota_ledger"], "quota_ledger", plan_value["quota_date"]) historical_path = _quota_file(root, plan_value["historical_quota_ledger"], "historical_quota_ledger", plan_value["historical_quota_date"]) quota_rows = _quota_rows(quota_path, plan_value["quota_date"]) historical_rows = _quota_rows(historical_path, plan_value["historical_quota_date"]) if not any(row["event_family"] == "BASELINE" and int(row["external_baseline_floor"] or 0) >= 3 for row in historical_rows): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "historical_quota", "baseline floor 3 absent") quota_snapshot = QuotaLedger(quota_path, quota_date=plan_value["quota_date"]).snapshot() evidence_manifest.extend((_evidence_row("GLOBAL", "quota_ledger", quota_path), _evidence_row("GLOBAL", "historical_quota_ledger", historical_path))) event_by_id = {row["event_id"]: row for row in quota_rows} for ref_index, raw_ref in enumerate(evidence_references): if not isinstance(raw_ref, Mapping) or tuple(raw_ref.keys()) != EVIDENCE_REFERENCE_KEYS: raise ContractError(ErrorCode.TASK_SPEC_INVALID, f"evidence_references[{ref_index}]", "exact keys") slot_id = raw_ref["slot_id"] if slot_id not in plan_by_slot or slot_id in seen_slots: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "slot_id", "unknown/duplicate") seen_slots.add(slot_id) plan_entry = plan_by_slot[slot_id] expected = plan_entry["expected_report"] task_path = _file(root, raw_ref["task_path"], "task_path") discovery_path = _file(root, raw_ref["discovery_path"], "discovery_path") terminal_path = _file(root, raw_ref["terminal_path"], "terminal_path") timing_path = _file(root, raw_ref["timing_path"], "timing_path") manifest_path = _file(root, raw_ref["manifest_path"], "manifest_path") task_value, _ = _json(task_path, "task", canonical=False) spec = TaskSpec.from_mapping(task_value) if not synthetic: verified_pdfinfo = _runtime_pdfinfo(spec.pdfinfo_executable) if runtime_pdfinfo_path is None: runtime_pdfinfo_path = verified_pdfinfo evidence_manifest.append(_evidence_row("GLOBAL", "runtime_pdfinfo", verified_pdfinfo)) elif verified_pdfinfo != runtime_pdfinfo_path: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "pdfinfo_executable", "runtime changed") if spec.mode != plan_entry["mode"] or spec.performance_plan_id != plan_value["plan_id"]: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "task", "mode/plan mismatch") if ((spec.performance_slot_id != slot_id if spec.mode == "collect-one" else spec.performance_slot_id != plan_entry["batch_id"])): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "task", "slot/batch mismatch") item_order = plan_entry["item_order"] if len(spec.expected_reports) < item_order or spec.expected_reports[item_order - 1] != expected: raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "task.expected_reports", "plan mismatch") discovery, discovery_raw = _json(discovery_path, "discovery", canonical=False) if (hashlib.sha256(discovery_raw).hexdigest() != plan_entry["discovery_sha256"] or discovery.get("schema_version") != "HIBOR_FAST_DISCOVERY_V001" or discovery.get("quota_ledger_touched") is not False or discovery.get("trigger_attempted") is not False or expected not in discovery.get("candidates", [])): raise ContractError(ErrorCode.DETAIL_RESULT_MISMATCH, "discovery", "unverified expected identity") terminal, terminal_raw = _json(terminal_path, "terminal", canonical=True) if canonical_terminal_bytes(terminal) != terminal_raw: raise ContractError(ErrorCode.PERSIST_LATE, "terminal", "canonical mismatch") expected_run_id = _run_id(spec) if ((terminal.get("task_id"), terminal.get("handoff_id"), terminal.get("run_id")) != (spec.task_id, spec.handoff_id, expected_run_id) or terminal.get("terminal_presence") != "V"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "terminal", "identity/presence mismatch") item_id = raw_ref["item_id"] expected_item_id = f"ITEM-{item_order:03d}" if item_id != expected_item_id: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "item_id", "order mismatch") matching_items = [row for row in terminal.get("items", []) if row.get("item_id") == item_id] if len(matching_items) != 1 or tuple(matching_items[0].keys()) != ITEM_KEYS: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "terminal.items", "unique fixed item required") item = matching_items[0] if (item["report_identity"] != expected["report_identity"] or item["trigger_attempted"] is not True or item["reused_without_new_trigger"] is not False or item["quota_reservation_id"] is None): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "terminal.item", "reserved real item required") sample_success = (terminal.get("status") == "SUCCESS" and terminal.get("exit_code") == 0 and item["status"] == "SUCCESS" and item["triggered"] is True and item["quota_state"] == "CONFIRMED" and item["stop_code"] is None) timing, timing_raw = _json(timing_path, "timing", canonical=True) if (timing.get("schema_version") != "HIBOR_FAST_TIMING_V002" or timing.get("run_id") != expected_run_id): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "timing", "schema/run mismatch") windows = [row for row in timing.get("item_windows", []) if isinstance(row, Mapping) and row.get("item_index") == item_order] if len(windows) != 1 or tuple(windows[0].keys()) != ITEM_WINDOW_KEYS: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "timing.item_windows", "unique fixed window") window = dict(windows[0]) calculated_deadline_met = window["terminal_monotonic_ns"] < window["close_deadline_monotonic_ns"] if (window["terminal_monotonic_ns"] < window["started_monotonic_ns"] or window["elapsed_ms"] != (window["terminal_monotonic_ns"] - window["started_monotonic_ns"]) // 1_000_000 or type(window["deadline_met"]) is not bool or window["deadline_met"] != calculated_deadline_met): raise ContractError(ErrorCode.CLOCK_INVALID, "timing.item_window", "cross-field mismatch") if item_order == 1 and window["terminal_delta_from_previous_ms"] is not None: raise ContractError(ErrorCode.CLOCK_INVALID, "timing.delta", "first must be null") if item_order > 1 and (type(window["terminal_delta_from_previous_ms"]) is not int or window["terminal_delta_from_previous_ms"] < 0): raise ContractError(ErrorCode.CLOCK_INVALID, "timing.delta", "later item INT required") manifest_rows = _manifest_rows(manifest_path) matching_manifest = [row for row in manifest_rows if row["item_id"] == item_id and row["report_identity"] == expected["report_identity"]] if len(matching_manifest) != 1: raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "unique item row") manifest_row = matching_manifest[0] if (manifest_row["status"] != manifest_row["download_status"] or manifest_row["reused_without_new_trigger"] != "false" or (manifest_row["quota_terminal_event_id"] or None) != item["quota_terminal_event_id"] or (manifest_row["quota_artifact_event_id"] or None) != item["quota_artifact_event_id"]): raise ContractError(ErrorCode.MANIFEST_INVALID, "manifest", "status/quota mismatch") pdf_path: Path | None = None pdf_magic = byte_match = hash_match = openable = page_match = False if item["final_path"] is not None: pdf_path = _file(root, item["final_path"], "final_path") pdfinfo = None if synthetic else runtime_pdfinfo_path pdf = validate_pdf(pdf_path, pdfinfo_executable=pdfinfo) try: byte_match = (pdf.bytes == item["bytes"] == int(manifest_row["bytes"]) == int(manifest_row["remote_bytes"]) == int(manifest_row["local_bytes"])) hash_match = (pdf.sha256 == item["sha256"] == manifest_row["sha256"] == manifest_row["remote_sha256"] == manifest_row["local_sha256"]) page_match = (item["page_count"] == int(manifest_row["page_count"]) == expected["page_count"] and (pdf.page_count is None or pdf.page_count == expected["page_count"])) except (TypeError, ValueError): byte_match = hash_match = page_match = False pdf_magic, openable = pdf.pdf_magic_valid, pdf.openable if sample_success and not (pdf_magic and openable and byte_match and hash_match and page_match): raise ContractError(ErrorCode.HASH_MISMATCH, "pdf", "success evidence mismatch") reservation = event_by_id.get(item["quota_reservation_id"]) quota_terminal = event_by_id.get(item["quota_terminal_event_id"]) artifact_terminal = event_by_id.get(item["quota_artifact_event_id"]) if not reservation or not quota_terminal: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "referenced event absent") common = (spec.task_id, spec.handoff_id, expected["report_identity"]) if ((reservation["task_id"], reservation["handoff_id"], reservation["report_identity"]) != common or reservation["event_type"] != "RESERVE" or (quota_terminal["task_id"], quota_terminal["handoff_id"], quota_terminal["report_identity"]) != common or quota_terminal["ref_event_id"] != item["quota_reservation_id"]): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "event chain mismatch") quota_projection = {"CONFIRMED": "CONSUME_CONFIRMED", "UNCERTAIN": "CONSUME_UNCERTAIN", "RELEASED": "RELEASE"} if quota_projection.get(item["quota_state"]) != quota_terminal["event_type"]: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "terminal projection mismatch") if quota_terminal["event_type"] == "RELEASE": if artifact_terminal is not None: raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "released artifact forbidden") elif (not artifact_terminal or artifact_terminal["ref_event_id"] != item["quota_terminal_event_id"] or artifact_terminal["report_identity"] != expected["report_identity"]): raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "artifact chain mismatch") if sample_success and artifact_terminal["event_type"] != "ARTIFACT_SUCCESS": raise ContractError(ErrorCode.QUOTA_LEDGER_INVALID, "quota", "success artifact required") sample = dict(zip(SAMPLE_KEYS, ( slot_id, expected["report_identity"], "SUCCESS" if sample_success else "FAILED", window["elapsed_ms"], window["terminal_delta_from_previous_ms"], item["triggered"], pdf_magic, byte_match, hash_match, openable, page_match, True, True, hashlib.sha256(terminal_raw).hexdigest(), hashlib.sha256(timing_raw).hexdigest(), ))) samples.append(sample) evidence_paths = [("task", task_path), ("discovery", discovery_path), ("terminal", terminal_path), ("timing", timing_path), ("manifest", manifest_path)] if pdf_path is not None: evidence_paths.append(("pdf", pdf_path)) for evidence_type, path in evidence_paths: evidence_manifest.append(_evidence_row(slot_id, evidence_type, path)) ordered_slots = [row["slot_id"] for row in plan_value["entries"]] if [row["slot_id"] for row in samples] != ordered_slots[:len(samples)]: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "evidence_references", "must be ordered plan prefix") durations = sorted(row["elapsed_ms"] for row in samples) median = ((durations[(len(durations) - 1) // 2] + durations[len(durations) // 2]) / 2 if durations else None) p90 = durations[max(0, math.ceil(0.9 * len(durations)) - 1)] if durations else None batch_samples = [(entry, sample) for entry, sample in zip(plan_value["entries"], samples) if entry["mode"] == "batch"] batch_first = next((sample["elapsed_ms"] for entry, sample in batch_samples if entry["item_order"] == 1), None) batch_increments = [sample["terminal_delta_from_previous_ms"] for entry, sample in batch_samples if entry["item_order"] > 1] checks = { "magic_pass": all(row["pdf_magic_pass"] for row in samples), "byte_match_pass": all(row["byte_match_pass"] for row in samples), "hash_match_pass": all(row["hash_match_pass"] for row in samples), "openability_pass": all(row["openability_pass"] for row in samples), "page_count_pass": all(row["page_count_pass"] for row in samples), "manifest_pass": all(row["manifest_pass"] for row in samples), "quota_pass": (all(row["quota_pass"] for row in samples) and quota_snapshot.active == 0 and quota_snapshot.effective_consumed <= 27), } blockers: list[str] = [] if len(samples) < 10: blockers.append("INSUFFICIENT_PERFORMANCE_SAMPLES") if any(row["status"] != "SUCCESS" for row in samples): blockers.append("SAMPLE_NOT_SUCCESS") if len(samples) == 10 and len(batch_increments) != 3: blockers.append("BATCH_STRUCTURE_INCOMPLETE") if sum(row["triggered"] for row in samples) > 10: blockers.append("TRIGGER_CAP_EXCEEDED") if median is None or median > 480_000: blockers.append("MEDIAN_SLA_FAILED") if p90 is None or p90 > 600_000: blockers.append("P90_SLA_FAILED") if batch_first is None or batch_first > 600_000: blockers.append("BATCH_FIRST_SLA_FAILED") if len(batch_increments) != 3 or any(value is None or value > 240_000 for value in batch_increments): blockers.append("BATCH_INCREMENT_SLA_FAILED") for key, passed in checks.items(): if not passed: blockers.append(key.upper()) result = ("PERFORMANCE_PASS" if not blockers else "INSUFFICIENT_PERFORMANCE_SAMPLES" if "INSUFFICIENT_PERFORMANCE_SAMPLES" in blockers else "PERFORMANCE_FAILED") summary = dict(zip(SUMMARY_KEYS, ( "HIBOR_FAST_PERFORMANCE_SUMMARY_V003", plan_value["plan_id"], started_at_utc, ended_at_utc, 10, len(samples), sum(row["status"] == "SUCCESS" for row in samples), sum(row["triggered"] for row in samples), quota_snapshot.confirmed, quota_snapshot.uncertain, quota_snapshot.active, median, p90, batch_first, batch_increments, checks["magic_pass"], checks["byte_match_pass"], checks["hash_match_pass"], checks["openability_pass"], checks["page_count_pass"], checks["manifest_pass"], checks["quota_pass"], result, blockers, None, None, None, samples, ))) return summary, evidence_manifest def write_performance_package(*, summary_path: Path, evidence_manifest_path: Path, plan_path: Path, plan_bytes: int, plan_sha256: str, evidence_root: Path, evidence_references: Sequence[Mapping[str, Any]], started_at_utc: str, ended_at_utc: str, synthetic: bool = False) -> tuple[tuple[int, str], tuple[int, str]]: summary, manifest_rows = build_performance_summary_from_evidence( plan_path=plan_path, plan_bytes=plan_bytes, plan_sha256=plan_sha256, evidence_root=evidence_root, evidence_references=evidence_references, started_at_utc=started_at_utc, ended_at_utc=ended_at_utc, synthetic=synthetic, ) manifest_receipt = _write_evidence_manifest(evidence_manifest_path, manifest_rows) summary["evidence_manifest_path"] = str(evidence_manifest_path) summary["evidence_manifest_bytes"] = manifest_receipt[0] summary["evidence_manifest_sha256"] = manifest_receipt[1] summary_receipt = write_json_create_new(summary_path, summary) return summary_receipt, manifest_receipt def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="hibor-fast-performance-package") parser.add_argument("--request", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--manifest-output", type=Path) args = parser.parse_args(argv) try: request = json.loads(args.request.read_bytes().decode("utf-8")) if not isinstance(request, Mapping): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "request", "mapping required") kind = request.get("kind") formal_root = PERFORMANCE_EVIDENCE_ROOT.resolve(strict=False) if kind == "PLAN" and tuple(request.keys()) == ("kind", "value"): if (args.request.resolve(strict=False) != formal_root / "control" / "performance_plan_request.json" or args.output.resolve(strict=False) != formal_root / "control" / "performance_plan.json" or args.manifest_output is not None): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "request", "fixed PLAN paths") count, digest = write_performance_plan(args.output, request["value"]) receipt: Any = {"path": str(args.output), "bytes": count, "sha256": digest} elif kind == "PREPLAN" and tuple(request.keys()) == ( "kind", "terminal_value", "evidence_root", "evidence_paths") and args.manifest_output is not None: if (Path(request["evidence_root"]).resolve(strict=False) != formal_root or args.request.resolve(strict=False) != formal_root / "control" / "preplan_stop_request.json" or args.output.resolve(strict=False) != formal_root / "preplan" / "preplan_terminal.json" or args.manifest_output.resolve(strict=False) != formal_root / "preplan" / "preplan_manifest.csv"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "request", "fixed PREPLAN paths") terminal_receipt, manifest_receipt = write_preplan_package( terminal_path=args.output, manifest_path=args.manifest_output, terminal_value=request["terminal_value"], evidence_root=Path(request["evidence_root"]), evidence_paths=request["evidence_paths"], ) receipt = { "terminal": {"path": str(args.output), "bytes": terminal_receipt[0], "sha256": terminal_receipt[1]}, "manifest": {"path": str(args.manifest_output), "bytes": manifest_receipt[0], "sha256": manifest_receipt[1]}, } elif kind == "SUMMARY" and tuple(request.keys()) == ( "kind", "plan_path", "plan_bytes", "plan_sha256", "evidence_root", "evidence_references", "started_at_utc", "ended_at_utc") and args.manifest_output is not None: if (Path(request["evidence_root"]).resolve(strict=False) != formal_root or args.request.resolve(strict=False) != formal_root / "control" / "performance_summary_request.json" or Path(request["plan_path"]).resolve(strict=False) != formal_root / "control" / "performance_plan.json" or args.output.resolve(strict=False) != formal_root / "package" / "performance_summary.json" or args.manifest_output.resolve(strict=False) != formal_root / "package" / "performance_manifest.csv"): raise ContractError(ErrorCode.TASK_SPEC_INVALID, "request", "fixed SUMMARY paths") summary_receipt, manifest_receipt = write_performance_package( summary_path=args.output, evidence_manifest_path=args.manifest_output, plan_path=Path(request["plan_path"]), plan_bytes=request["plan_bytes"], plan_sha256=request["plan_sha256"], evidence_root=Path(request["evidence_root"]), evidence_references=request["evidence_references"], started_at_utc=request["started_at_utc"], ended_at_utc=request["ended_at_utc"], synthetic=False, ) receipt = { "summary": {"path": str(args.output), "bytes": summary_receipt[0], "sha256": summary_receipt[1]}, "manifest": {"path": str(args.manifest_output), "bytes": manifest_receipt[0], "sha256": manifest_receipt[1]}, } else: raise ContractError(ErrorCode.TASK_SPEC_INVALID, "request", "exact PLAN/PREPLAN/SUMMARY request") print(canonical_json_bytes(receipt).decode("utf-8")) return 0 except (OSError, UnicodeError, ValueError, json.JSONDecodeError, ContractError): return 12 if __name__ == "__main__": raise SystemExit(main())