diff --git a/docs/guides/memory-plane-offline-verification.md b/docs/guides/memory-plane-offline-verification.md index 4194257..0adac47 100644 --- a/docs/guides/memory-plane-offline-verification.md +++ b/docs/guides/memory-plane-offline-verification.md @@ -11,6 +11,12 @@ The report proves only local invariants: SQLite persistence, deterministic FTS search, deterministic export, redaction, and durable gate-event count. It does not claim semantic retrieval quality or external-adapter availability. +The report also includes `schema_version`, backend identity, a redacted record +schema/provenance shape, and a digest of that shape. Content, metadata values, +temporary database paths, prompts, and credentials are deliberately omitted. +The output is stable across runs, so it can be copied into a release evidence +directory without turning local paths or memory contents into public evidence. + The canonical record contract includes: | Field group | Required evidence | diff --git a/scripts/evidence_bundle.py b/scripts/evidence_bundle.py index 25e3a82..e4fc1e5 100755 --- a/scripts/evidence_bundle.py +++ b/scripts/evidence_bundle.py @@ -6,16 +6,18 @@ import argparse import gzip import hashlib +import importlib.util import io import json import tarfile -from collections.abc import Iterable +from collections.abc import Callable, Iterable from pathlib import Path, PurePosixPath -from typing import Any +from typing import Any, cast EVIDENCE_DIR = Path(__file__).parent.parent / "evidence" BUNDLE_DIR = Path(__file__).parent.parent / "evidence_bundle" MANIFEST_NAME = "evidence_manifest.json" +MEMORY_SMOKE_NAME = "memory_offline_smoke.json" _ALLOWED_SUFFIXES = frozenset({".json", ".md", ".txt", ".log", ".xml"}) @@ -77,6 +79,34 @@ def collect_evidence(evidence_dir: Path = EVIDENCE_DIR) -> dict[str, Any]: return {"schema_version": 1, "artifacts": artifacts} +def collect_memory_offline_smoke(evidence_dir: Path = EVIDENCE_DIR) -> Path: + """Write the deterministic, redacted MemoryPlane smoke report.""" + + smoke_run: Callable[[], dict[str, object]] + if __package__: + from scripts.memory_offline_smoke import run as smoke_run + else: + # ``python scripts/evidence_bundle.py`` puts ``scripts/`` (rather than + # the repository root) first on ``sys.path``. Load the sibling by its + # resolved path so an unrelated installed ``scripts`` package cannot + # silently provide stale evidence code. + smoke_path = Path(__file__).with_name("memory_offline_smoke.py") + spec = importlib.util.spec_from_file_location("verdict_memory_offline_smoke", smoke_path) + if spec is None or spec.loader is None: + raise ValueError(f"cannot load MemoryPlane smoke script: {smoke_path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + smoke_run = cast(Callable[[], dict[str, object]], module.run) + + evidence_dir.mkdir(parents=True, exist_ok=True) + output = evidence_dir / MEMORY_SMOKE_NAME + content = ( + json.dumps(smoke_run(), ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n" + ).encode("utf-8") + output.write_bytes(content) + return output + + def _validated_manifest(manifest: Any) -> dict[str, Any]: if not isinstance(manifest, dict) or manifest.get("schema_version") != 1: raise ValueError("unsupported or malformed evidence manifest") @@ -252,6 +282,7 @@ def main(argv: list[str] | None = None) -> int: manifest = verify_bundle(args.evidence_dir, args.bundle) print(f"verified {len(manifest['artifacts'])} evidence artifacts") return 0 + collect_memory_offline_smoke(args.evidence_dir) evidence = collect_evidence(args.evidence_dir) bundle_path = create_bundle(evidence, args.evidence_dir, args.output_dir) print(f"created deterministic evidence bundle: {bundle_path}") diff --git a/scripts/memory_offline_smoke.py b/scripts/memory_offline_smoke.py index 0f277f8..8ddc474 100644 --- a/scripts/memory_offline_smoke.py +++ b/scripts/memory_offline_smoke.py @@ -3,14 +3,32 @@ from __future__ import annotations import argparse +import hashlib import json import tempfile from pathlib import Path +from typing import Any from verdict.memory_gate import MemoryGate, MemoryWriteRequest from verdict.memory_plane import MemoryPlane +def _redacted_record_shape(record: dict[str, Any]) -> dict[str, Any]: + """Expose schema/provenance shape without exporting memory content.""" + + provenance = record.get("provenance") + return { + "fields": sorted(record), + "schema_version": record.get("schema_version"), + "content_hash_present": isinstance(record.get("content_hash"), str), + "provenance_fields": sorted(provenance) if isinstance(provenance, dict) else [], + "content_field_present": "content" in record, + "content_values_redacted": True, + "metadata_field_present": "metadata" in record, + "metadata_values_redacted": True, + } + + def run() -> dict[str, object]: with tempfile.TemporaryDirectory(prefix="verdict-memory-smoke-") as directory: path = Path(directory) / "memory.db" @@ -39,22 +57,32 @@ def run() -> dict[str, object]: search = [record.record_id for record in plane.search("safe", scope="smoke")] export_a = plane.export_records(scope="smoke") export_b = plane.export_records(scope="smoke") + schema_shapes = sorted( + (_redacted_record_shape(record) for record in export_a), + key=lambda shape: json.dumps(shape, sort_keys=True, separators=(",", ":")), + ) return { + "schema_version": 1, + "backend": plane.health()["backend"], "status": "ready" if first.allowed and duplicate.allowed else "blocked", "network": "disabled", "provider": "not_required", "records": len(export_a), - "search_record_ids": search, + "search_hit_count": len(search), "deterministic_export": export_a == export_b, "redaction_proven": "not-persisted" not in json.dumps(export_a, sort_keys=True), "gate_event_count": len(gate.get_write_history()), + "record_schema": schema_shapes, + "provenance_shape_digest": hashlib.sha256( + json.dumps(schema_shapes, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest(), } -def main() -> None: +def main(argv: list[str] | None = None) -> None: parser = argparse.ArgumentParser() parser.add_argument("--output", type=Path) - args = parser.parse_args() + args = parser.parse_args(argv) report = json.dumps(run(), indent=2, sort_keys=True) + "\n" if args.output: args.output.write_text(report, encoding="utf-8") diff --git a/tests/test_evidence_bundle.py b/tests/test_evidence_bundle.py index f69aa29..1dba490 100644 --- a/tests/test_evidence_bundle.py +++ b/tests/test_evidence_bundle.py @@ -1,11 +1,21 @@ """Tests for deterministic, content-addressed release evidence bundles.""" import json +import os +import subprocess +import sys from pathlib import Path import pytest -from scripts.evidence_bundle import MANIFEST_NAME, collect_evidence, create_bundle, verify_bundle +from scripts.evidence_bundle import ( + MANIFEST_NAME, + MEMORY_SMOKE_NAME, + collect_evidence, + collect_memory_offline_smoke, + create_bundle, + verify_bundle, +) def _write_fixtures(root: Path) -> None: @@ -79,3 +89,48 @@ def test_manifest_rejects_duplicate_and_traversal_paths(tmp_path: Path) -> None: ) with pytest.raises(ValueError, match="duplicate"): verify_bundle(evidence_dir) + + +def test_memory_smoke_is_included_and_tamper_evident(tmp_path: Path) -> None: + evidence_dir = tmp_path / "evidence" + bundle_dir = tmp_path / "bundles" + collect_memory_offline_smoke(evidence_dir) + + first = collect_evidence(evidence_dir) + bundle = create_bundle(first, evidence_dir, bundle_dir) + assert MEMORY_SMOKE_NAME in {item["path"] for item in first["artifacts"]} + verify_bundle(evidence_dir, bundle) + + (evidence_dir / MEMORY_SMOKE_NAME).write_text("tampered\n", encoding="utf-8") + with pytest.raises(ValueError, match="digest mismatch"): + verify_bundle(evidence_dir, bundle) + + +def test_evidence_bundle_cli_generates_current_memory_smoke_report(tmp_path: Path) -> None: + evidence_dir = tmp_path / "evidence" + bundle_dir = tmp_path / "bundles" + script = Path(__file__).parents[1] / "scripts" / "evidence_bundle.py" + environment = os.environ.copy() + environment.pop("PYTHONPATH", None) + + result = subprocess.run( + [ + sys.executable, + str(script), + "--evidence-dir", + str(evidence_dir), + "--output-dir", + str(bundle_dir), + ], + check=True, + capture_output=True, + text=True, + cwd=tmp_path, + env=environment, + ) + + assert "created deterministic evidence bundle" in result.stdout + report = json.loads((evidence_dir / MEMORY_SMOKE_NAME).read_text(encoding="utf-8")) + assert report["search_hit_count"] == 1 + assert "search_record_ids" not in report + assert report["record_schema"] diff --git a/tests/test_memory_offline_smoke.py b/tests/test_memory_offline_smoke.py new file mode 100644 index 0000000..b640fff --- /dev/null +++ b/tests/test_memory_offline_smoke.py @@ -0,0 +1,36 @@ +"""Tests for the credential-free deterministic MemoryPlane smoke report.""" + +import json +from pathlib import Path + +from scripts.memory_offline_smoke import main, run + + +def test_smoke_report_is_deterministic_and_redacted() -> None: + first = run() + second = run() + + assert first == second + assert first["schema_version"] == 1 + assert first["backend"] == "sqlite" + assert first["network"] == "disabled" + assert first["provider"] == "not_required" + assert first["status"] == "ready" + assert first["redaction_proven"] is True + assert first["search_hit_count"] == 1 + assert first["record_schema"] + assert all(item["content_field_present"] is True for item in first["record_schema"]) + assert all(item["metadata_field_present"] is True for item in first["record_schema"]) + assert all(item["content_values_redacted"] is True for item in first["record_schema"]) + assert all(item["metadata_values_redacted"] is True for item in first["record_schema"]) + assert "verdict-memory-smoke-" not in json.dumps(first) + + +def test_smoke_cli_writes_canonical_json(tmp_path: Path, capsys) -> None: + output = tmp_path / "memory_offline_smoke.json" + + main(["--output", str(output)]) + + captured = capsys.readouterr().out + assert json.loads(captured) == json.loads(output.read_text(encoding="utf-8")) + assert output.read_text(encoding="utf-8") == json.dumps(run(), indent=2, sort_keys=True) + "\n"