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())
|
