Add receipt check utility

Makefile

@@ -91,6 +91,7 @@ validate-cli:
 	$(PYCLI) steer preflight --sourceset gpt2-small.res-jb --pretty >/tmp/agent-machine-pycli-steer-preflight.json
 	$(BOOTSTRAP_CLI) steer preflight --sourceset gpt2-small.res-jb --pretty >/tmp/agent-machine-bootstrap-steer-preflight.json
 	$(PYCLI) steer resolve-artifacts --sourceset gpt2-small.res-jb --local-dir /tmp/agent-machine-steering-artifacts --receipt-out /tmp/agent-machine-steering-artifact-receipt.json --dry-run --pretty >/tmp/agent-machine-pycli-artifact-receipt.json
+	$(PYTHON) scripts/verify-steering-receipt.py examples/steering-artifact-receipts/gpt2-small-res-jb.missing.steering-artifact-receipt.json --expect-status not_configured --pretty >/tmp/agent-machine-steering-load-preflight.json
 	$(PYCLI) version
 	$(PYCLI) paths --format json
 	$(PYCLI) doctor --format json

docs/index.md

@@ -16,6 +16,7 @@ Agent Machine is a bootstrap runtime-control substrate for SourceOS agent worklo
 | [Steering sourceset registry](steering-sourcesets.md) | Registered model/SAE sourceset records for local steering work. |
 | [Steering artifact receipts](steering-artifact-receipts.md) | Artifact-resolution receipt contract for model and SAE files. |
 | [Steering artifact resolution](steering-artifact-resolution.md) | Operator command for resolving model/SAE files and emitting a complete receipt. |
+| [Steering receipt loader](steering-loader.md) | Fail-closed receipt path and digest verification before runtime loading. |
 | [GPT-2 Small steering activation path](steering-activation-path.md) | Fail-closed real-path entrypoint and remaining blockers for controlled activation. |
 
 ## Architecture

docs/steering-loader.md

@@ -0,0 +1,43 @@
+# Steering Receipt Loader
+
+Status: receipt verification tranche for local steering work.
+
+## Purpose
+
+Before any local steering runtime may load model or SAE files, Agent Machine must verify that every artifact referenced by a `SteeringArtifactReceipt` exists locally and matches the receipt's SHA-256 digest.
+
+This document describes the fail-closed loader preflight. It does not claim applied steering.
+
+## Verification command
+
+```bash
+scripts/verify-steering-receipt.py \
+  examples/steering-artifact-receipts/gpt2-small-res-jb.missing.steering-artifact-receipt.json \
+  --expect-status not_configured \
+  --pretty
+```
+
+The fixture paths intentionally do not exist. The expected result is `status: not_configured`, with missing-file diagnostics for each absent artifact.
+
+## Runtime rule
+
+A future runtime loader must not attempt to load GPT-2 Small or the residual-stream SAE until:
+
+- the receipt validates against `contracts/steering-artifact-receipt.schema.json`
+- each referenced local path exists
+- each referenced path is a file
+- each file's SHA-256 digest matches the receipt
+
+If any check fails, the runtime must fail closed and return a non-applied posture.
+
+## Boundary
+
+This tranche verifies receipt integrity only. It does not:
+
+- load GPT-2 Small into memory
+- load the SAE into memory
+- run inference
+- inject activations
+- return `status: applied`
+
+The next implementation tranche may add optional runtime loading after this digest gate succeeds.

examples/steering-artifact-receipts/gpt2-small-res-jb.missing.steering-artifact-receipt.json

@@ -0,0 +1,63 @@
+{
+  "specVersion": "0.1.0",
+  "id": "urn:srcos:agent-machine:steering-artifact-receipt:gpt2-small.res-jb.missing-fixture",
+  "kind": "SteeringArtifactReceipt",
+  "sourcesetId": "gpt2-small.res-jb",
+  "status": "complete",
+  "generatedAt": "1970-01-01T00:00:00Z",
+  "activationIssue": "active-steering-work",
+  "artifactRecords": [
+    {
+      "role": "model-config",
+      "source": {
+        "type": "huggingface",
+        "repo": "openai-community/gpt2",
+        "filePath": "config.json",
+        "resolvedRevision": "0000000000000000000000000000000000000000",
+        "url": "https://huggingface.co/openai-community/gpt2/blob/0000000000000000000000000000000000000000/config.json"
+      },
+      "storage": {
+        "localPath": "/tmp/agent-machine-nonexistent/gpt2/config.json",
+        "sizeBytes": 0,
+        "storageReceiptRef": null
+      },
+      "digest": {
+        "algorithm": "sha256",
+        "sha256": "0000000000000000000000000000000000000000000000000000000000000000",
+        "verified": true
+      }
+    },
+    {
+      "role": "sae-artifact",
+      "source": {
+        "type": "huggingface",
+        "repo": "jbloom/GPT2-Small-SAEs-Reformatted",
+        "filePath": "blocks.6.hook_resid_pre/sae_weights.safetensors",
+        "resolvedRevision": "0000000000000000000000000000000000000000",
+        "url": "https://huggingface.co/jbloom/GPT2-Small-SAEs-Reformatted/blob/0000000000000000000000000000000000000000/blocks.6.hook_resid_pre/sae_weights.safetensors"
+      },
+      "storage": {
+        "localPath": "/tmp/agent-machine-nonexistent/gpt2-sae/sae_weights.safetensors",
+        "sizeBytes": 0,
+        "storageReceiptRef": null
+      },
+      "digest": {
+        "algorithm": "sha256",
+        "sha256": "1111111111111111111111111111111111111111111111111111111111111111",
+        "verified": true
+      }
+    }
+  ],
+  "missing": [],
+  "storageReceiptRefs": [],
+  "policyRefs": [],
+  "agentRegistryGrantRefs": [],
+  "receiptSafety": {
+    "includeRawArtifacts": false,
+    "includeAuthMaterial": false
+  },
+  "notes": [
+    "Fixture for fail-closed loader validation only.",
+    "Paths intentionally do not exist. The loader preflight must report not_configured rather than using artifacts."
+  ]
+}

scripts/verify-steering-receipt.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+"""Verify steering artifact receipt file paths and SHA-256 digests."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from agent_machine.steering_loader import verify_receipt_files  # noqa: E402
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Verify steering artifact receipt paths and digests")
+    parser.add_argument("receipt", type=Path)
+    parser.add_argument("--expect-status", choices=["available", "not_configured"])
+    parser.add_argument("--pretty", action="store_true")
+    args = parser.parse_args()
+
+    result = verify_receipt_files(args.receipt)
+    if args.pretty:
+        print(json.dumps(result, indent=2, sort_keys=True))
+    else:
+        print(json.dumps(result, sort_keys=True, separators=(",", ":")))
+
+    if args.expect_status and result.get("status") != args.expect_status:
+        print(f"expected status {args.expect_status}, got {result.get('status')}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/agent_machine/steering_loader.py

@@ -0,0 +1,85 @@
+"""Receipt-backed local artifact verification for steering runtime.
+
+This module verifies a SteeringArtifactReceipt before any runtime may use the
+referenced files. It is deliberately fail-closed: absent files or digest mismatch
+produce a not_configured result rather than a runtime claim.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Any
+
+from agent_machine.contracts import load_json, validate_by_kind
+from agent_machine.paths import repo_root_from_file
+
+REPO_ROOT = repo_root_from_file(__file__)
+
+
+def verify_receipt_files(receipt_path: Path) -> dict[str, Any]:
+    """Verify receipt local paths and SHA-256 digests without loading artifacts."""
+    receipt_path = Path(receipt_path)
+    validate_by_kind(receipt_path, REPO_ROOT)
+    receipt = load_json(receipt_path)
+    records = receipt.get("artifactRecords", [])
+    checks = [verify_artifact_record(record) for record in records if isinstance(record, dict)]
+    missing = [item for check in checks for item in check.get("missing", [])]
+    digest_mismatches = [item for check in checks for item in check.get("digestMismatches", [])]
+    verified = [check for check in checks if check.get("verified")]
+    ready = bool(records) and len(verified) == len(records) and not missing and not digest_mismatches
+    return {
+        "ok": True,
+        "status": "available" if ready else "not_configured",
+        "receiptPath": str(receipt_path),
+        "sourcesetId": receipt.get("sourcesetId"),
+        "receiptStatus": receipt.get("status"),
+        "artifactCount": len(records),
+        "verifiedArtifactCount": len(verified),
+        "readyForRuntimeUse": ready,
+        "missing": missing,
+        "digestMismatches": digest_mismatches,
+        "checks": checks,
+    }
+
+
+def verify_artifact_record(record: dict[str, Any]) -> dict[str, Any]:
+    source = record.get("source", {}) if isinstance(record.get("source"), dict) else {}
+    storage = record.get("storage", {}) if isinstance(record.get("storage"), dict) else {}
+    digest = record.get("digest", {}) if isinstance(record.get("digest"), dict) else {}
+    local_path = Path(str(storage.get("localPath", "")))
+    expected_sha = str(digest.get("sha256", ""))
+    result: dict[str, Any] = {
+        "role": record.get("role"),
+        "repo": source.get("repo"),
+        "filePath": source.get("filePath"),
+        "resolvedRevision": source.get("resolvedRevision"),
+        "localPath": str(local_path),
+        "expectedSha256": expected_sha,
+        "actualSha256": None,
+        "exists": local_path.exists(),
+        "verified": False,
+        "missing": [],
+        "digestMismatches": [],
+    }
+    if not local_path.exists():
+        result["missing"].append(f"artifact file missing: {local_path}")
+        return result
+    if not local_path.is_file():
+        result["missing"].append(f"artifact path is not a file: {local_path}")
+        return result
+    actual_sha = sha256_file(local_path)
+    result["actualSha256"] = actual_sha
+    if actual_sha != expected_sha:
+        result["digestMismatches"].append(f"sha256 mismatch for {local_path}: expected {expected_sha}, got {actual_sha}")
+        return result
+    result["verified"] = True
+    return result
+
+
+def sha256_file(path: Path) -> str:
+    hasher = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            hasher.update(chunk)
+    return hasher.hexdigest()