Add direct WP-0024 load driver handoff

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa

scripts/wp0024-t02-driver-candidate.py

@@ -0,0 +1,247 @@
+#!/usr/bin/env python3
+"""Register or inspect a revision-pinned T02 synthetic-load driver candidate.
+
+Registration reads and hashes committed source.  It never executes the driver,
+reads a bearer, or authorizes the live database-lease exercise.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_CONTRACT = ROOT / "interfaces" / "RAILIANCE-WP-0024-T02-synthetic-load-driver.json"
+DEFAULT_AUDIT_REPO = ROOT.parent / "audit-core"
+DEFAULT_API_BASE = os.environ.get("STATE_HUB_URL", "http://127.0.0.1:8000")
+PREFIX = "WP0024-T02-DRIVER-CANDIDATE"
+
+
+class CandidateError(RuntimeError):
+    pass
+
+
+def canonical(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+
+
+def digest(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def load_contract(path: Path = DEFAULT_CONTRACT) -> dict[str, Any]:
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise CandidateError("synthetic-load contract is unavailable or invalid") from exc
+    if (
+        not isinstance(value, dict)
+        or value.get("interface") != "railiance.synthetic-load-driver"
+        or value.get("version") != 1
+        or value.get("task_id") != "RAILIANCE-WP-0024-T02"
+        or value.get("owner") != "audit-core"
+    ):
+        raise CandidateError("unsupported synthetic-load contract")
+    phases = value.get("phases")
+    if not isinstance(phases, dict) or set(phases) != {
+        "baseline", "expect-unavailable", "expect-recovered", "cleanup"
+    }:
+        raise CandidateError("synthetic-load contract has the wrong phase set")
+    for phase, record in phases.items():
+        if not isinstance(record, dict) or not record.get("keys") or not record.get("status"):
+            raise CandidateError(f"synthetic-load phase is incomplete: {phase}")
+    return value
+
+
+def contract_digest(contract: dict[str, Any]) -> str:
+    return digest(canonical(contract).encode("utf-8"))
+
+
+def run_git(repo: Path, args: list[str]) -> str:
+    completed = subprocess.run(
+        ["git", "-C", str(repo), *args], text=True, capture_output=True, check=False
+    )
+    if completed.returncode != 0:
+        raise CandidateError("cannot resolve committed driver source")
+    return completed.stdout.strip()
+
+
+def inspect_driver(repo: Path, source_path: str) -> dict[str, str]:
+    repo = repo.resolve()
+    target = (repo / source_path).resolve()
+    try:
+        relative = target.relative_to(repo).as_posix()
+    except ValueError as exc:
+        raise CandidateError("driver path escapes the audit-core repository") from exc
+    if not target.is_file() or not os.access(target, os.X_OK):
+        raise CandidateError("driver candidate must be an executable file")
+    if run_git(repo, ["status", "--porcelain", "--", relative]):
+        raise CandidateError("driver candidate must be committed with no path-local changes")
+    tracked = run_git(repo, ["ls-files", "--error-unmatch", "--", relative])
+    if tracked != relative:
+        raise CandidateError("driver candidate is not tracked at the requested path")
+    revision = run_git(repo, ["log", "-1", "--format=%H", "--", relative])
+    if not re.fullmatch(r"[0-9a-f]{40}", revision):
+        raise CandidateError("driver source revision is not a full Git commit")
+    return {
+        "source_repo": "audit-core",
+        "source_revision": revision,
+        "source_path": relative,
+        "driver_revision": "sha256:" + digest(target.read_bytes()),
+    }
+
+
+def reviewer(value: str) -> str:
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:@/+\-]{1,127}", value):
+        raise CandidateError("reviewer must be a stable 2-128 character identifier")
+    return value
+
+
+def build_receipt(
+    contract: dict[str, Any], source: dict[str, str], reviewer_id: str,
+    *, now: datetime | None = None,
+) -> dict[str, Any]:
+    return {
+        "interface": "railiance.synthetic-load-driver-candidate",
+        "version": 1,
+        "workplan_id": contract["workplan_id"],
+        "task_id": contract["task_id"],
+        "contract_id": contract["contract_id"],
+        "contract_digest": contract_digest(contract),
+        "owner": "audit-core",
+        "reviewer": reviewer(reviewer_id),
+        **source,
+        "created_at": (now or datetime.now(UTC)).astimezone(UTC).isoformat().replace("+00:00", "Z"),
+        "registration_executed_driver": False,
+        "secret_values_observed": False,
+    }
+
+
+def subject(receipt: dict[str, Any]) -> str:
+    return "/".join((PREFIX, "v1", receipt["contract_digest"], receipt["driver_revision"][7:]))
+
+
+def http_json(method: str, url: str, payload: dict[str, Any] | None = None) -> Any:
+    data = canonical(payload).encode("utf-8") if payload is not None else None
+    request = urllib.request.Request(
+        url, data=data, method=method,
+        headers={"Content-Type": "application/json"} if data else {},
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=10) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except (OSError, urllib.error.URLError, json.JSONDecodeError) as exc:
+        raise CandidateError(f"State Hub request failed: {method} {url}") from exc
+
+
+def post(receipt: dict[str, Any], api_base: str) -> dict[str, Any]:
+    value = http_json(
+        "POST",
+        f"{api_base.rstrip('/')}/messages/",
+        {
+            "from_agent": "audit-core",
+            "to_agent": "railiance-platform",
+            "subject": subject(receipt),
+            "body": canonical(receipt),
+        },
+    )
+    if not isinstance(value, dict):
+        raise CandidateError("State Hub returned an invalid candidate response")
+    return value
+
+
+def parse_message(message: dict[str, Any], contract: dict[str, Any]) -> dict[str, Any] | None:
+    if message.get("from_agent") != "audit-core":
+        return None
+    try:
+        receipt = json.loads(message["body"])
+    except (KeyError, TypeError, json.JSONDecodeError):
+        return None
+    if not isinstance(receipt, dict):
+        return None
+    try:
+        valid = (
+            message.get("subject") == subject(receipt)
+            and receipt.get("interface") == "railiance.synthetic-load-driver-candidate"
+            and receipt.get("version") == 1
+            and receipt.get("task_id") == contract["task_id"]
+            and receipt.get("contract_id") == contract["contract_id"]
+            and receipt.get("contract_digest") == contract_digest(contract)
+            and receipt.get("owner") == "audit-core"
+            and receipt.get("source_repo") == "audit-core"
+            and re.fullmatch(r"[0-9a-f]{40}", receipt.get("source_revision", ""))
+            and re.fullmatch(r"sha256:[0-9a-f]{64}", receipt.get("driver_revision", ""))
+            and receipt.get("registration_executed_driver") is False
+            and receipt.get("secret_values_observed") is False
+        )
+    except (KeyError, TypeError):
+        return None
+    return receipt if valid else None
+
+
+def status(contract: dict[str, Any], api_base: str) -> dict[str, Any]:
+    query = urllib.parse.urlencode({"to_agent": "railiance-platform", "limit": 500})
+    value = http_json("GET", f"{api_base.rstrip('/')}/messages/?{query}")
+    if not isinstance(value, list):
+        raise CandidateError("State Hub messages response is not a list")
+    candidates = [item for message in value if (item := parse_message(message, contract))]
+    latest = max(candidates, key=lambda item: item.get("created_at", ""), default=None)
+    return {
+        "interface": contract["interface"],
+        "version": contract["version"],
+        "task_id": contract["task_id"],
+        "contract_id": contract["contract_id"],
+        "contract_digest": contract_digest(contract),
+        "candidate_registered": latest is not None,
+        "candidate": latest,
+    }
+
+
+def parser() -> argparse.ArgumentParser:
+    result = argparse.ArgumentParser(description=__doc__)
+    result.add_argument("--contract", type=Path, default=DEFAULT_CONTRACT)
+    result.add_argument("--api-base", default=DEFAULT_API_BASE)
+    sub = result.add_subparsers(dest="command", required=True)
+    sub.add_parser("show")
+    register = sub.add_parser("register")
+    register.add_argument("--repo", type=Path, default=DEFAULT_AUDIT_REPO)
+    register.add_argument("--driver", required=True, help="path relative to the audit-core repo")
+    register.add_argument("--reviewer", required=True)
+    sub.add_parser("status")
+    return result
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parser().parse_args(argv)
+    try:
+        contract = load_contract(args.contract)
+        if args.command == "show":
+            output = {**contract, "contract_digest": contract_digest(contract)}
+        elif args.command == "register":
+            source = inspect_driver(args.repo, args.driver)
+            receipt = build_receipt(contract, source, args.reviewer)
+            response = post(receipt, args.api_base)
+            output = {"submitted": True, "message_id": response.get("id"), "receipt": receipt}
+        else:
+            output = status(contract, args.api_base)
+    except CandidateError as exc:
+        print(canonical({"ok": False, "error": str(exc)}), file=sys.stderr)
+        return 2
+    print(json.dumps(output, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())