Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/guides/memory-plane-offline-verification.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
35 changes: 33 additions & 2 deletions scripts/evidence_bundle.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"})


Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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}")
Expand Down
34 changes: 31 additions & 3 deletions scripts/memory_offline_smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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")
Expand Down
57 changes: 56 additions & 1 deletion tests/test_evidence_bundle.py
Original file line number Diff line number Diff line change
@@ -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:
Expand Down Expand Up @@ -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"]
36 changes: 36 additions & 0 deletions tests/test_memory_offline_smoke.py
Original file line number Diff line number Diff line change
@@ -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"