Add blinded live Leo reasoning benchmark

scripts/leo_oos_readonly_guard.py

@@ -0,0 +1,226 @@
+#!/usr/bin/env python3
+"""Fail-closed Hermes tool surface for live no-post Leo OOS trials."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shlex
+import shutil
+import subprocess
+import uuid
+from pathlib import Path
+from typing import Any
+
+READ_ONLY_BRIDGE_COMMANDS = frozenset(
+    {
+        "search",
+        "context",
+        "show",
+        "evidence",
+        "edges",
+        "list-proposals",
+        "search-proposals",
+        "show-proposal",
+        "status",
+        "decision-matrix-status",
+    }
+)
+HARMLESS_TRAILING_REDIRECT_TOKENS = frozenset({"2>/dev/null", "2> /dev/null"})
+
+
+class ReadOnlyGuardError(RuntimeError):
+    """The requested command cannot be proven read-only."""
+
+
+class GuardArgumentParser(argparse.ArgumentParser):
+    def error(self, message: str) -> None:
+        raise ReadOnlyGuardError(message)
+
+    def exit(self, status: int = 0, message: str | None = None) -> None:
+        raise ReadOnlyGuardError(message or f"argument parser exited with status {status}")
+
+
+def _uuid_argument(value: str) -> str:
+    try:
+        return str(uuid.UUID(value))
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("expected a UUID") from exc
+
+
+def validate_read_only_bridge_argv(argv: list[str]) -> None:
+    """Accept only the bridge's bounded read commands and exact arguments."""
+
+    parser = GuardArgumentParser(add_help=False)
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    def command(name: str) -> argparse.ArgumentParser:
+        item = subparsers.add_parser(name, add_help=False)
+        item.add_argument("--help", action="store_true")
+        item.add_argument("--format", choices=("markdown", "json"), default="markdown")
+        return item
+
+    for name in ("search", "context"):
+        item = command(name)
+        item.add_argument("query")
+        item.add_argument("--limit", type=int, default=5)
+        item.add_argument("--context-limit", type=int, default=5 if name == "search" else 6)
+    show = command("show")
+    show.add_argument("claim_id", type=_uuid_argument)
+    for name in ("evidence", "edges"):
+        item = command(name)
+        item.add_argument("claim_id", type=_uuid_argument)
+        item.add_argument("--limit", type=int, default=12)
+    list_proposals = command("list-proposals")
+    list_proposals.add_argument(
+        "--status",
+        choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"),
+        default="pending_review",
+    )
+    list_proposals.add_argument("--limit", type=int, default=10)
+    search_proposals = command("search-proposals")
+    search_proposals.add_argument("query")
+    search_proposals.add_argument(
+        "--status",
+        choices=("pending_review", "approved", "applied", "canceled", "rejected", "all"),
+        default="all",
+    )
+    search_proposals.add_argument("--limit", type=int, default=10)
+    show_proposal = command("show-proposal")
+    show_proposal.add_argument("proposal_id", type=_uuid_argument)
+    command("status")
+    command("decision-matrix-status")
+    parsed = parser.parse_args(argv)
+    for name, value in vars(parsed).items():
+        if name.endswith("limit") and not 1 <= value <= 100:
+            raise ReadOnlyGuardError(f"{name.replace('_', '-')} must be between 1 and 100")
+
+
+def classify_terminal_command(
+    wrapper: Path,
+    temp_profile: Path,
+    tool_args: dict[str, Any],
+    *,
+    allow_kb_reads: bool,
+) -> tuple[list[str] | None, str | None]:
+    """Return verified argv or a fail-closed reason without executing anything."""
+
+    command = str(tool_args.get("command") or "")
+    if len(command) > 8192:
+        return None, "command is too long"
+    if tool_args.get("background") or tool_args.get("pty") or tool_args.get("workdir"):
+        return None, "background, PTY, and workdir overrides are disabled"
+    if not allow_kb_reads:
+        return None, "database tools are disabled for the no-DB ablation"
+    try:
+        argv = shlex.split(command)
+    except ValueError as exc:
+        return None, f"invalid command quoting: {exc}"
+    while argv and argv[-1] in HARMLESS_TRAILING_REDIRECT_TOKENS:
+        argv.pop()
+    command_path = f"{temp_profile / 'bin'}:{os.environ.get('PATH', '')}"
+    try:
+        exact_wrapper = wrapper.resolve(strict=True)
+        executable = shutil.which(argv[0], path=command_path) if argv else None
+        invoked = Path(executable).resolve(strict=True) if executable else None
+    except OSError:
+        invoked = None
+        exact_wrapper = wrapper.resolve(strict=False)
+    if invoked != exact_wrapper:
+        return None, "only the exact temporary-profile teleo-kb wrapper is available"
+    if len(argv) < 2 or argv[1] not in READ_ONLY_BRIDGE_COMMANDS:
+        return None, "only read-only Teleo KB bridge commands are available"
+    try:
+        validate_read_only_bridge_argv(argv[1:])
+    except ReadOnlyGuardError as exc:
+        return None, f"invalid read-only KB arguments: {exc}"
+    return argv, None
+
+
+def restricted_bridge_terminal_handler(
+    wrapper: Path,
+    temp_profile: Path,
+    tool_args: dict[str, Any],
+    *,
+    allow_kb_reads: bool,
+    **_kwargs: Any,
+) -> str:
+    argv, reason = classify_terminal_command(
+        wrapper,
+        temp_profile,
+        tool_args,
+        allow_kb_reads=allow_kb_reads,
+    )
+    if argv is None:
+        return json.dumps({"output": "", "exit_code": 126, "error": reason})
+    timeout = min(max(int(tool_args.get("timeout") or 60), 1), 120)
+    child_environment = {
+        "HOME": str(temp_profile),
+        "HERMES_HOME": str(temp_profile),
+        "PATH": f"{temp_profile / 'bin'}:{os.environ.get('PATH', '')}",
+        "TELEO_KB_MODE": "local",
+    }
+    try:
+        proc = subprocess.run(
+            argv,
+            cwd=temp_profile,
+            env=child_environment,
+            text=True,
+            capture_output=True,
+            check=False,
+            timeout=timeout,
+        )
+    except subprocess.TimeoutExpired:
+        return json.dumps({"output": "", "exit_code": 124, "error": f"bridge command exceeded {timeout}s"})
+    return json.dumps(
+        {
+            "output": proc.stdout,
+            "exit_code": proc.returncode,
+            "error": proc.stderr or None,
+        }
+    )
+
+
+def install_read_only_tool_surface(temp_profile: Path, *, allow_kb_reads: bool) -> dict[str, Any]:
+    """Replace Hermes' registry with skills plus one guarded terminal tool."""
+
+    import tools.skills_tool
+    import tools.terminal_tool  # noqa: F401
+    import toolsets
+    from hermes_cli import tools_config
+    from tools.registry import registry
+
+    toolset_name = "leo-oos-kb-readonly" if allow_kb_reads else "leo-oos-no-db-ablation"
+    allowed_tools = ["skills_list", "skill_view", "terminal"]
+    toolsets.TOOLSETS[toolset_name] = {
+        "description": "Fail-closed Leo OOS tool surface",
+        "tools": allowed_tools,
+        "includes": [],
+    }
+    tools_config._get_platform_tools = lambda *_args, **_kwargs: {toolset_name}
+    missing = sorted(name for name in allowed_tools if name not in registry._tools)
+    if missing:
+        raise ReadOnlyGuardError(f"required tools are not registered: {missing}")
+    allowed_entries = {name: registry._tools[name] for name in allowed_tools}
+    registry._tools.clear()
+    registry._tools.update(allowed_entries)
+    wrapper = temp_profile / "bin" / "teleo-kb"
+    terminal_entry = registry._tools["terminal"]
+    terminal_entry.handler = lambda tool_args, **kwargs: restricted_bridge_terminal_handler(
+        wrapper,
+        temp_profile,
+        tool_args,
+        allow_kb_reads=allow_kb_reads,
+        **kwargs,
+    )
+    return {
+        "mode": "read_only_kb" if allow_kb_reads else "no_db_ablation",
+        "allowed_tools": allowed_tools,
+        "actual_registry_tools": sorted(registry._tools),
+        "read_only_bridge_commands": sorted(READ_ONLY_BRIDGE_COMMANDS) if allow_kb_reads else [],
+        "send_message_tool_enabled": "send_message" in registry._tools,
+        "terminal_restricted_to_exact_wrapper": True,
+        "mutating_bridge_commands_exposed": False,
+        "provider_credentials_forwarded_to_terminal": False,
+    }

scripts/run_leo_m3taversal_oos_handler_suite.py

@@ -4,20 +4,426 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
-import uuid
+import re
+import subprocess
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
+import leo_oos_readonly_guard as readonly_guard
 import run_leo_direct_claim_handler_suite as handler
 import working_leo_m3taversal_oos_benchmark as benchmark
+import working_leo_m3taversal_oos_protocol as protocol_lib
+
+BASE_HANDLER_SCRIPT_BUILDER = handler.build_remote_script
 
 ROOT = Path(__file__).resolve().parents[1]
 REPORT_DIR = ROOT / "docs" / "reports" / "leo-working-state-20260709"
 RESULTS_JSON = REPORT_DIR / "telegram-handler-m3taversal-oos-suite-current.json"
 SCORE_JSON = REPORT_DIR / "telegram-handler-m3taversal-oos-suite-score-current.json"
 SCORE_MARKDOWN = REPORT_DIR / "telegram-handler-m3taversal-oos-suite-score-current.md"
+PROTOCOL_REPORT_DIR = ROOT / "docs" / "reports" / "leo-oos-reasoning-benchmark-20260715"
+DEFAULT_PROTOCOL_JSON = PROTOCOL_REPORT_DIR / "blinded-protocol.json"
+GROUNDING_MODES = ("grounded", "db_tool_ablated")
+RUNTIME_PROMPT_SCAN_ROOTS = (
+    ROOT / "hermes-agent" / "leoclean-skills",
+    ROOT / "hermes-agent" / "leoclean-plugins",
+    ROOT / "hermes-agent" / "leoclean-bin",
+)
+
+
+def prompt_leakage_scan(protocol: dict[str, Any]) -> dict[str, Any]:
+    """Fail if a frozen prompt was copied into a runtime skill/plugin/tool."""
+
+    prompt_markers = [
+        (str(prompt["id"]), str(marker))
+        for trial in protocol.get("trials") or []
+        for prompt in trial.get("prompts") or []
+        for marker in prompt.get("leakage_markers") or []
+    ]
+    matches: list[dict[str, str]] = []
+    scanned_files = 0
+    for root in RUNTIME_PROMPT_SCAN_ROOTS:
+        if not root.exists():
+            continue
+        for path in sorted(item for item in root.rglob("*") if item.is_file()):
+            try:
+                text = path.read_text(encoding="utf-8")
+            except (OSError, UnicodeDecodeError):
+                continue
+            scanned_files += 1
+            normalized = " ".join(re.findall(r"[a-z0-9_]+", text.lower()))
+            for prompt_id, marker in prompt_markers:
+                if marker in normalized:
+                    display_path = str(path.relative_to(ROOT)) if path.is_relative_to(ROOT) else str(path)
+                    matches.append(
+                        {"prompt_id": prompt_id, "marker_sha256": hashlib.sha256(marker.encode()).hexdigest(), "path": display_path}
+                    )
+    return {
+        "scanned_roots": [str(path.relative_to(ROOT)) if path.is_relative_to(ROOT) else str(path) for path in RUNTIME_PROMPT_SCAN_ROOTS],
+        "scanned_files": scanned_files,
+        "exact_prompt_matches": matches,
+        "pass": not matches,
+    }
+
+
+def build_guarded_remote_script(
+    run_id: str,
+    *,
+    prompts: list[dict[str, Any]],
+    suite_mode: str,
+    report_prefix: str,
+    prompt_note: str,
+    grounding_mode: str,
+    leakage_scan: dict[str, Any],
+) -> str:
+    """Inject the reviewed fail-closed tool surface before GatewayRunner starts."""
+
+    if grounding_mode not in GROUNDING_MODES:
+        raise ValueError(f"unsupported grounding mode: {grounding_mode}")
+    script = BASE_HANDLER_SCRIPT_BUILDER(
+        run_id,
+        prompts=prompts,
+        suite_mode=suite_mode,
+        report_prefix=report_prefix,
+        prompt_note=prompt_note,
+    )
+    original_plugin_source = handler.DB_CONTEXT_PLUGIN.read_text(encoding="utf-8")
+    instrumented_plugin_source = protocol_lib.instrument_db_context_plugin_source(original_plugin_source)
+    encoded_original_plugin_source = json.dumps(original_plugin_source)
+    if script.count(encoded_original_plugin_source) != 1:
+        raise RuntimeError("embedded DB context plugin source marker changed")
+    script = script.replace(
+        encoded_original_plugin_source,
+        json.dumps(instrumented_plugin_source),
+        1,
+    )
+    guard_source = Path(readonly_guard.__file__).resolve().read_text(encoding="utf-8")
+    function_marker = "async def run_suite():"
+    if script.count(function_marker) != 1:
+        raise RuntimeError("remote harness run_suite marker changed")
+    script = script.replace(
+        function_marker,
+        "OOS_READONLY_GUARD_SOURCE = "
+        + json.dumps(guard_source)
+        + "\nOOS_GROUNDING_MODE = "
+        + json.dumps(grounding_mode)
+        + "\nOOS_GUARD_SOURCE_SHA256 = "
+        + json.dumps(hashlib.sha256(guard_source.encode()).hexdigest())
+        + "\nOOS_PROMPT_LEAKAGE_SCAN = "
+        + json.dumps(leakage_scan)
+        + "\n\n"
+        + function_marker,
+    )
+    gateway_marker = "        from gateway.session import SessionSource\n\n        source = SessionSource("
+    if script.count(gateway_marker) != 1:
+        raise RuntimeError("remote harness GatewayRunner marker changed")
+    injection = """        import re
+        from gateway.session import SessionSource
+        from gateway.platforms.telegram import TelegramAdapter
+
+        db_context_dir = temp_profile / "plugins" / "leo-db-context"
+        if OOS_GROUNDING_MODE == "db_tool_ablated":
+            if db_context_dir.is_symlink():
+                db_context_dir.unlink()
+            elif db_context_dir.exists():
+                shutil.rmtree(db_context_dir)
+        guard_module = types.ModuleType("leo_oos_readonly_guard_embedded")
+        guard_module.__file__ = "<embedded leo_oos_readonly_guard.py>"
+        exec(compile(OOS_READONLY_GUARD_SOURCE, guard_module.__file__, "exec"), guard_module.__dict__)
+        tool_surface = guard_module.install_read_only_tool_surface(
+            temp_profile,
+            allow_kb_reads=OOS_GROUNDING_MODE == "grounded",
+        )
+        report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)
+        telegram_transport_attempts = []
+        telegram_outbound_methods = (
+            "_send_with_retry",
+            "edit_message",
+            "play_tts",
+            "send",
+            "send_animation",
+            "send_document",
+            "send_image",
+            "send_image_file",
+            "send_model_picker",
+            "send_typing",
+            "send_update_prompt",
+            "send_video",
+            "send_voice",
+        )
+        patched_telegram_methods = []
+        def make_transport_deny(method_name):
+            async def deny_transport(*args, **kwargs):
+                telegram_transport_attempts.append({
+                    "method": method_name,
+                    "positional_argument_count": len(args),
+                    "keyword_names": sorted(str(key) for key in kwargs),
+                })
+                report["telegram_transport_deny"]["attempt_count"] = len(telegram_transport_attempts)
+                report["telegram_transport_deny"]["attempts"] = list(telegram_transport_attempts)
+                write_report_snapshot(report)
+                raise RuntimeError("Telegram transport is denied in the OOS no-post harness")
+            return deny_transport
+        for method_name in telegram_outbound_methods:
+            if hasattr(TelegramAdapter, method_name):
+                setattr(TelegramAdapter, method_name, make_transport_deny(method_name))
+                patched_telegram_methods.append(method_name)
+
+        remote_leakage_matches = []
+        remote_scanned_files = 0
+        remote_scanned_bytes = 0
+        remote_scan_errors = []
+        remote_symlink_targets = 0
+        prompt_markers = []
+        for prompt in PROMPTS:
+            words = re.findall(r"[a-z0-9_]+", str(prompt.get("message") or "").lower())
+            width = 16
+            starts = (0,) if len(words) <= width else (0, max(0, (len(words) - width) // 2), len(words) - width)
+            for start in starts:
+                marker = " ".join(words[start:start + width])
+                if marker:
+                    prompt_markers.append((str(prompt.get("id") or ""), marker))
+        remote_scan_roots = {
+            "profile": temp_profile,
+            "skills": temp_profile / "skills",
+            "plugins": temp_profile / "plugins",
+            "bin": temp_profile / "bin",
+        }
+        excluded_profile_parts = {
+            ".git", "__pycache__", "memories", "sessions", "state", "venv",
+        }
+        expected_roots = [
+            {"name": name, "exists": path.exists(), "is_symlink": path.is_symlink()}
+            for name, path in remote_scan_roots.items()
+        ]
+        scanned_paths = set()
+        scanned_directories = set()
+        pending_paths = list(remote_scan_roots.values())
+        candidate_files = []
+        while pending_paths:
+            scan_path = pending_paths.pop()
+            if set(scan_path.parts) & excluded_profile_parts:
+                continue
+            try:
+                if scan_path.is_symlink():
+                    remote_symlink_targets += 1
+                resolved = scan_path.resolve(strict=True)
+                if resolved.is_file():
+                    candidate_files.append(resolved)
+                    continue
+                if not resolved.is_dir():
+                    continue
+                directory_identity = str(resolved)
+                if directory_identity in scanned_directories:
+                    continue
+                scanned_directories.add(directory_identity)
+                pending_paths.extend(sorted(resolved.iterdir(), reverse=True))
+            except OSError as exc:
+                remote_scan_errors.append({
+                    "path_sha256": hashlib.sha256(str(scan_path).encode()).hexdigest(),
+                    "error_type": type(exc).__name__,
+                })
+        for scan_path in sorted(candidate_files):
+            if set(scan_path.parts) & excluded_profile_parts:
+                continue
+            try:
+                path_identity = str(scan_path.resolve(strict=True))
+                if path_identity in scanned_paths:
+                    continue
+                scanned_paths.add(path_identity)
+                scan_bytes = scan_path.read_bytes()
+            except OSError as exc:
+                remote_scan_errors.append({
+                    "path_sha256": hashlib.sha256(str(scan_path).encode()).hexdigest(),
+                    "error_type": type(exc).__name__,
+                })
+                continue
+            remote_scanned_files += 1
+            remote_scanned_bytes += len(scan_bytes)
+            normalized = b" ".join(re.findall(rb"[a-z0-9_]+", scan_bytes.lower())).decode("ascii")
+            for prompt_id, marker in prompt_markers:
+                if marker in normalized:
+                    remote_leakage_matches.append({
+                        "prompt_id": prompt_id,
+                        "marker_sha256": hashlib.sha256(marker.encode()).hexdigest(),
+                        "path_sha256": hashlib.sha256(str(scan_path).encode()).hexdigest(),
+                    })
+        report["grounding_mode"] = OOS_GROUNDING_MODE
+        report["db_context_plugin_enabled"] = db_context_dir.exists()
+        report["read_only_tool_surface"] = tool_surface
+        report["readonly_guard_source_sha256"] = OOS_GUARD_SOURCE_SHA256
+        report["prompt_leakage_scan"] = OOS_PROMPT_LEAKAGE_SCAN
+        report["remote_temp_profile_prompt_leakage_scan"] = {
+            "scope": "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv",
+            "expected_roots": expected_roots,
+            "scanned_files": remote_scanned_files,
+            "scanned_bytes": remote_scanned_bytes,
+            "symlink_targets_encountered": remote_symlink_targets,
+            "errors": remote_scan_errors,
+            "matches": remote_leakage_matches,
+            "pass": bool(
+                all(item["exists"] for item in expected_roots)
+                and remote_scanned_files > 0
+                and remote_scanned_bytes > 0
+                and not remote_scan_errors
+                and not remote_leakage_matches
+            ),
+        }
+        report["telegram_transport_deny"] = {
+            "enabled": set(patched_telegram_methods) == set(telegram_outbound_methods),
+            "expected_methods": list(telegram_outbound_methods),
+            "patched_methods": patched_telegram_methods,
+            "attempt_count": 0,
+            "attempts": [],
+            "runner_adapters_required_empty": True,
+        }
+        report["oos_max_turn_seconds"] = 180
+        report["preexecution_safety_gate"] = {
+            "status": "pass" if (
+                tool_surface.get("send_message_tool_enabled") is False
+                and tool_surface.get("mutating_bridge_commands_exposed") is False
+                and tool_surface.get("terminal_restricted_to_exact_wrapper") is True
+                and OOS_PROMPT_LEAKAGE_SCAN.get("pass") is True
+                and report["remote_temp_profile_prompt_leakage_scan"]["pass"] is True
+                and report["telegram_transport_deny"]["enabled"] is True
+                and report["telegram_transport_deny"]["attempt_count"] == 0
+            ) else "fail",
+            "grounding_mode": OOS_GROUNDING_MODE,
+        }
+        write_report_snapshot(report)
+        if report["preexecution_safety_gate"]["status"] != "pass":
+            raise RuntimeError("OOS pre-execution safety gate failed")
+
+        source = SessionSource("""
+    script = script.replace(gateway_marker, injection)
+    turn_timeout_marker = "reply = await asyncio.wait_for(runner._handle_message(event), timeout=300)"
+    if script.count(turn_timeout_marker) != 1:
+        raise RuntimeError("remote harness per-turn timeout marker changed")
+    script = script.replace(
+        turn_timeout_marker,
+        "reply = await asyncio.wait_for(runner._handle_message(event), timeout=180)",
+    )
+    runner_marker = "        runner = GatewayRunner()\n"
+    if script.count(runner_marker) != 1:
+        raise RuntimeError("remote harness runner marker changed")
+    script = script.replace(
+        runner_marker,
+        runner_marker
+        + "        runner.adapters.clear()\n"
+        + "        runner.delivery_router.adapters = runner.adapters\n"
+        + "        report['telegram_transport_deny']['runner_adapters_empty'] = not runner.adapters\n"
+        + "        write_report_snapshot(report)\n",
+    )
+    return script
+
+
+def _remote_orphan_readback(parsed: dict[str, Any]) -> dict[str, Any]:
+    temp_profile = str((parsed.get("handler") or {}).get("temp_profile") or "")
+    if not temp_profile:
+        return {"status": "missing_temp_profile_identity", "no_matching_processes": False, "matches": []}
+    proc = subprocess.run(
+        [
+            "ssh",
+            "-i",
+            str(handler.SSH_KEY),
+            "-o",
+            "BatchMode=yes",
+            "-o",
+            "StrictHostKeyChecking=accept-new",
+            handler.VPS,
+            "ps -eo pid=,args=",
+        ],
+        text=True,
+        capture_output=True,
+        timeout=60,
+    )
+    matches = [handler.redact(line.strip()) for line in proc.stdout.splitlines() if temp_profile in line]
+    return {
+        "status": "ok" if proc.returncode == 0 else "ssh_error",
+        "returncode": proc.returncode,
+        "temp_profile_sha256": hashlib.sha256(temp_profile.encode()).hexdigest(),
+        "matches": matches,
+        "no_matching_processes": proc.returncode == 0 and not matches,
+    }
+
+
+def _local_harness_git_state() -> dict[str, Any]:
+    head = subprocess.run(
+        ["git", "rev-parse", "HEAD"],
+        cwd=ROOT,
+        text=True,
+        capture_output=True,
+        timeout=30,
+        check=False,
+    )
+    status = subprocess.run(
+        ["git", "status", "--porcelain", "--untracked-files=all"],
+        cwd=ROOT,
+        text=True,
+        capture_output=True,
+        timeout=30,
+        check=False,
+    )
+    status_text = status.stdout if status.returncode == 0 else ""
+    status_lines = status_text.splitlines()
+    return {
+        "git_head": head.stdout.strip() if head.returncode == 0 else None,
+        "worktree_clean": status.returncode == 0 and not status_text.strip(),
+        "status_sha256": hashlib.sha256(status_text.encode()).hexdigest() if status.returncode == 0 else None,
+        "status_lines": status_lines,
+        "only_control_goal_untracked": status.returncode == 0 and status_lines == ["?? goal.md"],
+    }
+
+
+def _portable_artifact_path(path: Path) -> str:
+    resolved = path.resolve()
+    try:
+        return str(resolved.relative_to(ROOT))
+    except ValueError:
+        return str(resolved)
+
+
+def run_guarded_remote(
+    *,
+    prompts: list[dict[str, Any]],
+    suite_mode: str,
+    report_prefix: str,
+    prompt_note: str,
+    grounding_mode: str,
+    leakage_scan: dict[str, Any],
+) -> dict[str, Any]:
+    """Use the generic transport while substituting only the guarded builder."""
+
+    original_builder = handler.build_remote_script
+
+    def guarded_builder(run_id: str, **kwargs: Any) -> str:
+        return build_guarded_remote_script(
+            run_id,
+            prompts=kwargs.get("prompts") or prompts,
+            suite_mode=str(kwargs.get("suite_mode") or suite_mode),
+            report_prefix=str(kwargs.get("report_prefix") or report_prefix),
+            prompt_note=str(kwargs.get("prompt_note") or prompt_note),
+            grounding_mode=grounding_mode,
+            leakage_scan=leakage_scan,
+        )
+
+    handler.build_remote_script = guarded_builder
+    try:
+        remote = handler.run_remote(
+            prompts=prompts,
+            suite_mode=suite_mode,
+            report_prefix=report_prefix,
+            prompt_note=prompt_note,
+        )
+    finally:
+        handler.build_remote_script = original_builder
+    parsed = remote.get("parsed")
+    if isinstance(parsed, dict):
+        parsed["post_run_orphan_readback"] = _remote_orphan_readback(parsed)
+    return remote
 
 
 def write_score_markdown(path: Path, report: dict[str, Any]) -> None:
@@ -100,6 +506,443 @@ def score_report_passes(report: dict[str, Any], score_report: dict[str, Any]) ->
     )
 
 
+def write_protocol_score_markdown(path: Path, score: dict[str, Any]) -> None:
+    lines = [
+        "# Leo Blinded OOS Reasoning Trial",
+        "",
+        f"Generated UTC: `{score['generated_at_utc']}`",
+        f"Protocol: `{score['protocol_id']}` / `{score['protocol_hash_sha256']}`",
+        f"Trial: `{score['trial_id']}` / `{score['session_mode']}`",
+        f"Pass: `{score['pass']}`",
+        f"Grounded prompts: `{score['grounded_passes']}/{score['prompt_count']}`",
+        f"Grounded rate: `{score['grounded_pass_rate']:.3f}`",
+        f"No-DB ablation grounded rate: `{score['receipt_ablation']['grounded_pass_rate']:.3f}`",
+        "",
+        "## Prompt scores",
+        "",
+    ]
+    for item in score["prompt_scores"]:
+        lines.append(
+            f"- `{item['prompt_id']}` / `{item['family_id']}`: semantic=`{item['semantic_pass']}`, "
+            f"subject=`{item['subject_alignment']}`, receipts=`{item['receipt_pass']}`, "
+            f"grounded=`{item['grounded_pass']}`"
+        )
+    lines.extend(
+        [
+            "",
+            "## Safety",
+            "",
+            f"- Grounded live safety: `{score['top_level_safety']['pass']}`",
+            f"- Ablation live safety: `{score['baseline_top_level_safety']['pass']}`",
+            f"- Restart receipt: `{score['restart_receipt_validation']['pass']}`",
+            f"- Exact prompt binding: `{score['prompt_binding']['pass']}`",
+            "",
+            "## Claim ceiling",
+            "",
+            "This trial proves a direct live VPS GatewayRunner reply path with a fail-closed read-only tool surface, "
+            "full unchanged DB fingerprints, and no Telegram post. It does not prove Telegram-visible delivery or "
+            "any production database apply.",
+            "",
+        ]
+    )
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def _run_protocol_mode(
+    protocol: dict[str, Any],
+    trial: dict[str, Any],
+    *,
+    grounding_mode: str,
+    output_dir: Path,
+    leakage_scan: dict[str, Any],
+) -> tuple[dict[str, Any], Path]:
+    prompts = [
+        {"id": prompt["id"], "dimension": prompt["dimension"], "message": prompt["message"]}
+        for prompt in trial["prompts"]
+    ]
+    safe_protocol_id = re.sub(r"[^A-Za-z0-9._-]", "-", str(protocol["protocol_id"]))
+    report_prefix = f"leo-oos-{safe_protocol_id}-{trial['trial_id']}-{grounding_mode}"
+    output_path = output_dir / f"{trial['trial_id']}-{grounding_mode}-handler.json"
+    remote = run_guarded_remote(
+        prompts=prompts,
+        suite_mode=f"live_vps_gatewayrunner_blinded_oos_{grounding_mode}",
+        report_prefix=report_prefix,
+        prompt_note=(
+            f"Frozen blinded protocol {protocol['protocol_hash_sha256']}; trial {trial['trial_id']}; "
+            f"grounding mode {grounding_mode}; direct handler only, no Telegram post, no database mutation."
+        ),
+        grounding_mode=grounding_mode,
+        leakage_scan=leakage_scan,
+    )
+    report = handler.write_output(remote, output_json=output_path)
+    report["oos_harness_git_state"] = _local_harness_git_state()
+    report["protocol_id"] = protocol["protocol_id"]
+    report["protocol_hash_sha256"] = protocol["protocol_hash_sha256"]
+    report["trial_id"] = trial["trial_id"]
+    report["trial_prompt_set_sha256"] = trial["prompt_set_sha256"]
+    report["scorer_version"] = protocol["scorer_version"]
+    report["source_hashes"] = protocol["source_hashes"]
+    output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return report, output_path
+
+
+def _load_restart_receipt(path: Path | None, trial: dict[str, Any]) -> dict[str, Any] | None:
+    if trial["session_mode"] != "post_restart_clean_session":
+        return None
+    if path is None:
+        raise SystemExit("--restart-receipt is required for a post-restart trial")
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _ssh_readback(command: str, *, timeout: int = 120) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [
+            "ssh",
+            "-i",
+            str(handler.SSH_KEY),
+            "-o",
+            "BatchMode=yes",
+            "-o",
+            "StrictHostKeyChecking=accept-new",
+            handler.VPS,
+            command,
+        ],
+        text=True,
+        capture_output=True,
+        timeout=timeout,
+    )
+
+
+def _deploy_identity() -> dict[str, Any]:
+    proc = _ssh_readback(
+        "cd /opt/teleo-eval/workspaces/deploy-infra && "
+        "printf 'head=' && git rev-parse HEAD && "
+        "printf 'stamp=' && cat /opt/teleo-eval/.last-deploy-sha",
+        timeout=60,
+    )
+    values: dict[str, str] = {}
+    for line in proc.stdout.splitlines():
+        if "=" in line:
+            key, value = line.split("=", 1)
+            values[key] = value.strip()
+    return {
+        "returncode": proc.returncode,
+        "head": values.get("head"),
+        "stamp": values.get("stamp"),
+        "stderr": handler.redact(proc.stderr),
+    }
+
+
+def _restart_state_probe(
+    protocol: dict[str, Any],
+    *,
+    label: str,
+    output_dir: Path,
+    leakage_scan: dict[str, Any],
+) -> tuple[dict[str, Any], Path]:
+    output_path = output_dir / f"restart-{label}-state.json"
+    remote = run_guarded_remote(
+        prompts=[],
+        suite_mode=f"live_vps_gateway_restart_{label}_readonly_state_probe",
+        report_prefix=f"leo-oos-restart-{label}-{protocol['protocol_id']}",
+        prompt_note=(
+            f"Read-only restart {label} state probe for frozen protocol {protocol['protocol_hash_sha256']}; "
+            "zero prompts, no Telegram post, no database mutation."
+        ),
+        grounding_mode="grounded",
+        leakage_scan=leakage_scan,
+    )
+    report = handler.write_output(remote, output_json=output_path)
+    report["oos_harness_git_state"] = _local_harness_git_state()
+    report["protocol_id"] = protocol["protocol_id"]
+    report["protocol_hash_sha256"] = protocol["protocol_hash_sha256"]
+    report["source_hashes"] = protocol["source_hashes"]
+    output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return report, output_path
+
+
+def _restart_probe_passes(report: dict[str, Any]) -> bool:
+    before = report.get("db_fingerprint_before") or {}
+    after = report.get("db_fingerprint_after") or {}
+    counts_before = report.get("db_counts_before")
+    counts_after = report.get("db_counts_after")
+    transport = report.get("telegram_transport_deny") or {}
+    return bool(
+        report.get("remote_returncode") == 0
+        and report.get("pass_runtime") is True
+        and report.get("posted_to_telegram") is False
+        and report.get("db_counts_changed") is False
+        and isinstance(counts_before, dict)
+        and bool(counts_before)
+        and counts_before == counts_after
+        and all(isinstance(value, int) and not isinstance(value, bool) for value in counts_before.values())
+        and report.get("db_fingerprint_unchanged") is True
+        and before.get("status") == "ok"
+        and after.get("status") == "ok"
+        and bool(re.fullmatch(r"[0-9a-f]{64}", str(before.get("fingerprint_sha256") or "")))
+        and before.get("fingerprint_sha256") == after.get("fingerprint_sha256")
+        and (report.get("preexecution_safety_gate") or {}).get("status") == "pass"
+        and transport.get("enabled") is True
+        and isinstance(transport.get("attempt_count"), int)
+        and not isinstance(transport.get("attempt_count"), bool)
+        and transport.get("attempt_count") == 0
+        and transport.get("runner_adapters_empty") is True
+        and report.get("temp_profile_removed") is True
+        and (report.get("post_run_orphan_readback") or {}).get("no_matching_processes") is True
+    )
+
+
+def collect_restart_receipt(args: argparse.Namespace) -> int:
+    protocol = json.loads(args.protocol.read_text(encoding="utf-8"))
+    validation = protocol_lib.validate_protocol(protocol, verify_source_hashes=True)
+    if not validation["pass"]:
+        raise SystemExit(f"frozen protocol validation failed: {validation['issues']}")
+    restart_trials = [
+        item for item in protocol.get("trials") or [] if item.get("session_mode") == "post_restart_clean_session"
+    ]
+    if len(restart_trials) != 1:
+        raise SystemExit("frozen protocol must identify exactly one post-restart trial")
+    next_trial = restart_trials[0]
+    leakage_scan = prompt_leakage_scan(protocol)
+    if not leakage_scan["pass"]:
+        raise SystemExit(f"runtime prompt leakage detected: {leakage_scan['exact_prompt_matches']}")
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    before_deploy = _deploy_identity()
+    before_report, before_path = _restart_state_probe(
+        protocol,
+        label="before",
+        output_dir=args.output_dir,
+        leakage_scan=leakage_scan,
+    )
+    if (
+        before_deploy.get("returncode") != 0
+        or not re.fullmatch(r"[0-9a-f]{40}", str(before_deploy.get("head") or ""))
+        or before_deploy.get("head") != before_deploy.get("stamp")
+    ):
+        raise SystemExit(f"deploy identity preflight failed: {before_deploy}")
+    if not _restart_probe_passes(before_report):
+        raise SystemExit(f"read-only pre-restart probe failed: {before_path}")
+
+    restart_started_at_utc = datetime.now(timezone.utc).isoformat()
+    restart = _ssh_readback(
+        "systemctl restart leoclean-gateway.service && "
+        "for i in $(seq 1 60); do "
+        "if [ \"$(systemctl is-active leoclean-gateway.service)\" = active ]; then break; fi; sleep 1; done; "
+        "systemctl show leoclean-gateway.service "
+        "-p ActiveState -p SubState -p MainPID -p NRestarts -p ExecMainStartTimestamp",
+        timeout=120,
+    )
+    restart_ended_at_utc = datetime.now(timezone.utc).isoformat()
+    after_deploy = _deploy_identity()
+    after_report, after_path = _restart_state_probe(
+        protocol,
+        label="after",
+        output_dir=args.output_dir,
+        leakage_scan=leakage_scan,
+    )
+    service_before = (before_report.get("service_before_after") or {}).get("after") or {}
+    service_after = after_report.get("before_service") or {}
+    fingerprint_before = before_report.get("db_fingerprint_after") or {}
+    fingerprint_after = after_report.get("db_fingerprint_before") or {}
+    counts_before = before_report.get("db_counts_after") or {}
+    counts_after = after_report.get("db_counts_before") or {}
+    receipt = {
+        "schema": "livingip.leoGatewayRestartReceipt.v1",
+        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
+        "protocol_id": protocol["protocol_id"],
+        "protocol_hash_sha256": protocol["protocol_hash_sha256"],
+        "next_trial_id": next_trial["trial_id"],
+        "next_trial_prompt_set_sha256": next_trial["prompt_set_sha256"],
+        "authorization_basis": "goal.md requires restart trials; service restart only, no Telegram post and no DB mutation",
+        "restart_command": "systemctl restart leoclean-gateway.service",
+        "restart_started_at_utc": restart_started_at_utc,
+        "restart_ended_at_utc": restart_ended_at_utc,
+        "restart_returncode": restart.returncode,
+        "restart_stdout": handler.redact(restart.stdout),
+        "restart_stderr": handler.redact(restart.stderr),
+        "service_before": service_before,
+        "service_after": service_after,
+        "deploy_before": before_deploy,
+        "deploy_after": after_deploy,
+        "db_counts_before": counts_before,
+        "db_counts_after": counts_after,
+        "db_counts_changed": counts_before != counts_after,
+        "db_fingerprint_before": fingerprint_before,
+        "db_fingerprint_after": fingerprint_after,
+        "db_fingerprint_unchanged": bool(
+            fingerprint_before.get("status") == "ok"
+            and fingerprint_after.get("status") == "ok"
+            and fingerprint_before.get("fingerprint_sha256")
+            == fingerprint_after.get("fingerprint_sha256")
+        ),
+        "posted_to_telegram": False,
+        "before_probe": {
+            "path": _portable_artifact_path(before_path),
+            "sha256": hashlib.sha256(before_path.read_bytes()).hexdigest(),
+            "pass": _restart_probe_passes(before_report),
+        },
+        "after_probe": {
+            "path": _portable_artifact_path(after_path),
+            "sha256": hashlib.sha256(after_path.read_bytes()).hexdigest(),
+            "pass": _restart_probe_passes(after_report),
+        },
+    }
+    receipt["checks"] = {
+        "restart_command_succeeded": receipt["restart_returncode"] == 0,
+        "service_active_after": service_after.get("ActiveState") == "active"
+        and service_after.get("SubState") == "running",
+        "service_pid_changed": bool(
+            service_before.get("MainPID") and service_before.get("MainPID") != service_after.get("MainPID")
+        ),
+        "deploy_identity_unchanged": before_deploy.get("returncode") == 0
+        and after_deploy.get("returncode") == 0
+        and bool(re.fullmatch(r"[0-9a-f]{40}", str(before_deploy.get("head") or "")))
+        and before_deploy.get("head")
+        == before_deploy.get("stamp")
+        == after_deploy.get("head")
+        == after_deploy.get("stamp"),
+        "db_counts_unchanged": receipt["db_counts_changed"] is False,
+        "db_counts_complete": isinstance(counts_before, dict)
+        and bool(counts_before)
+        and counts_before == counts_after
+        and all(isinstance(value, int) and not isinstance(value, bool) for value in counts_before.values()),
+        "db_fingerprint_unchanged": receipt["db_fingerprint_unchanged"] is True,
+        "db_fingerprint_complete": bool(
+            re.fullmatch(r"[0-9a-f]{64}", str(fingerprint_before.get("fingerprint_sha256") or ""))
+            and fingerprint_before.get("fingerprint_sha256") == fingerprint_after.get("fingerprint_sha256")
+        ),
+        "before_probe_passed": receipt["before_probe"]["pass"] is True,
+        "after_probe_passed": receipt["after_probe"]["pass"] is True,
+    }
+    receipt["pass"] = all(receipt["checks"].values())
+    args.collect_restart_receipt.parent.mkdir(parents=True, exist_ok=True)
+    args.collect_restart_receipt.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    print(
+        json.dumps(
+            {
+                "restart_receipt": str(args.collect_restart_receipt),
+                "pass": receipt["pass"],
+                "checks": receipt["checks"],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+    return 0 if receipt["pass"] else 1
+
+
+def run_or_score_protocol_trial(args: argparse.Namespace) -> int:
+    protocol = json.loads(args.protocol.read_text(encoding="utf-8"))
+    validation = protocol_lib.validate_protocol(protocol, verify_source_hashes=True)
+    if not validation["pass"]:
+        raise SystemExit(f"frozen protocol validation failed: {validation['issues']}")
+    trial = next((item for item in protocol["trials"] if item["trial_id"] == args.trial_id), None)
+    if trial is None:
+        raise SystemExit(f"unknown trial id: {args.trial_id}")
+    leakage_scan = prompt_leakage_scan(protocol)
+    if not leakage_scan["pass"]:
+        raise SystemExit(f"runtime prompt leakage detected: {leakage_scan['exact_prompt_matches']}")
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    restart_receipt = _load_restart_receipt(args.restart_receipt, trial)
+
+    if bool(args.grounded_report) != bool(args.baseline_report):
+        raise SystemExit("--grounded-report and --baseline-report must be supplied together")
+    if args.grounded_report:
+        grounded_report = json.loads(args.grounded_report.read_text(encoding="utf-8"))
+        baseline_report = json.loads(args.baseline_report.read_text(encoding="utf-8"))
+        grounded_path = args.grounded_report
+        baseline_path = args.baseline_report
+    elif args.grounding_mode:
+        report, output_path = _run_protocol_mode(
+            protocol,
+            trial,
+            grounding_mode=args.grounding_mode,
+            output_dir=args.output_dir,
+            leakage_scan=leakage_scan,
+        )
+        mode_safety = protocol_lib._top_level_safety(
+            report,
+            require_handler_safety_gate=args.grounding_mode == "grounded",
+        )
+        mode_checks = {
+            "expected_grounding_mode": report.get("grounding_mode") == args.grounding_mode,
+            "db_context_state": report.get("db_context_plugin_enabled")
+            is (args.grounding_mode == "grounded"),
+            "tool_surface_mode": (report.get("read_only_tool_surface") or {}).get("mode")
+            == ("read_only_kb" if args.grounding_mode == "grounded" else "no_db_ablation"),
+            "zero_context_in_ablation": args.grounding_mode != "db_tool_ablated"
+            or all(not (item.get("database_context_trace") or []) for item in report.get("results") or []),
+        }
+        mode_pass = mode_safety["pass"] and all(mode_checks.values())
+        print(
+            json.dumps(
+                {
+                    "report": str(output_path),
+                    "grounding_mode": args.grounding_mode,
+                    "handler_safety_gate": report.get("safety_gate"),
+                    "post_run_orphan_readback": report.get("post_run_orphan_readback"),
+                    "mode_safety": mode_safety,
+                    "mode_checks": mode_checks,
+                    "pass": mode_pass,
+                },
+                indent=2,
+                sort_keys=True,
+            )
+        )
+        return 0 if mode_pass else 1
+    else:
+        grounded_report, grounded_path = _run_protocol_mode(
+            protocol,
+            trial,
+            grounding_mode="grounded",
+            output_dir=args.output_dir,
+            leakage_scan=leakage_scan,
+        )
+        baseline_report, baseline_path = _run_protocol_mode(
+            protocol,
+            trial,
+            grounding_mode="db_tool_ablated",
+            output_dir=args.output_dir,
+            leakage_scan=leakage_scan,
+        )
+
+    score = protocol_lib.score_live_trial(
+        protocol,
+        trial["trial_id"],
+        grounded_report,
+        baseline_report=baseline_report,
+        restart_receipt=restart_receipt,
+    )
+    score["grounded_report_path"] = _portable_artifact_path(grounded_path)
+    score["baseline_report_path"] = _portable_artifact_path(baseline_path)
+    score["grounded_report_sha256"] = hashlib.sha256(grounded_path.read_bytes()).hexdigest()
+    score["baseline_report_sha256"] = hashlib.sha256(baseline_path.read_bytes()).hexdigest()
+    if restart_receipt is not None and args.restart_receipt is not None:
+        score["restart_receipt_path"] = _portable_artifact_path(args.restart_receipt)
+        score["restart_receipt_sha256"] = hashlib.sha256(args.restart_receipt.read_bytes()).hexdigest()
+        score["restart_receipt_payload_sha256"] = protocol_lib.canonical_sha256(restart_receipt)
+    score_path = args.output_dir / f"{trial['trial_id']}-score.json"
+    markdown_path = args.output_dir / f"{trial['trial_id']}-score.md"
+    score_path.write_text(json.dumps(score, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    write_protocol_score_markdown(markdown_path, score)
+    print(
+        json.dumps(
+            {
+                "grounded_report": str(grounded_path),
+                "baseline_report": str(baseline_path),
+                "score_json": str(score_path),
+                "score_markdown": str(markdown_path),
+                "pass": score["pass"],
+                "grounded_rate": score["grounded_pass_rate"],
+                "baseline_grounded_rate": score["receipt_ablation"]["grounded_pass_rate"],
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+    return 0 if score["pass"] else 1
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -107,25 +950,33 @@ def main() -> int:
         action="store_true",
         help="Rescore the retained live transcript without making another remote model call.",
     )
+    parser.add_argument("--protocol", type=Path)
+    parser.add_argument("--trial-id")
+    parser.add_argument("--output-dir", type=Path, default=PROTOCOL_REPORT_DIR)
+    parser.add_argument("--grounding-mode", choices=GROUNDING_MODES)
+    parser.add_argument("--grounded-report", type=Path)
+    parser.add_argument("--baseline-report", type=Path)
+    parser.add_argument("--restart-receipt", type=Path)
+    parser.add_argument("--collect-restart-receipt", type=Path)
     args = parser.parse_args()
 
+    if args.collect_restart_receipt:
+        if not args.protocol or args.trial_id:
+            raise SystemExit("--collect-restart-receipt requires --protocol and does not accept --trial-id")
+        return collect_restart_receipt(args)
+    if args.protocol or args.trial_id:
+        if not args.protocol or not args.trial_id:
+            raise SystemExit("--protocol and --trial-id are required together")
+        return run_or_score_protocol_trial(args)
+    if not args.score_existing:
+        raise SystemExit(
+            "refusing the legacy fixed live suite; use --protocol and --trial-id for a frozen guarded trial"
+        )
+
     if args.score_existing:
         report = json.loads(RESULTS_JSON.read_text(encoding="utf-8"))
         previous_score = json.loads(SCORE_JSON.read_text(encoding="utf-8"))
         memory_token = str(previous_score["memory_token"])
-    else:
-        memory_token = "demo-ledger-" + uuid.uuid4().hex[:8]
-        prompts = [
-            {"id": prompt["id"], "dimension": prompt["dimension"], "message": prompt["message"]}
-            for prompt in benchmark.prompt_catalog(memory_token)
-        ]
-        remote = handler.run_remote(
-            prompts=prompts,
-            suite_mode="live_vps_gatewayrunner_temp_profile_m3taversal_out_of_sample_suite",
-            report_prefix="leo-m3taversal-oos-handler-report",
-            prompt_note="Broad out-of-sample m3taversal prompts plus randomized memory and participant-identity checks.",
-        )
-        report = handler.write_output(remote, output_json=RESULTS_JSON)
 
     score_report = build_score_report(report, memory_token=memory_token)
     SCORE_JSON.write_text(json.dumps(score_report, indent=2, sort_keys=True) + "\n", encoding="utf-8")

scripts/working_leo_m3taversal_oos_benchmark.py

@@ -762,18 +762,20 @@ def composition_capability_issues(prompt_id: str, reply: str) -> list[str]:
 
 
 def score_reply(prompt: dict[str, Any], reply: str, *, memory_token: str) -> dict[str, Any]:
-    legacy_score = base.score_reply(prompt, reply)
+    scorer_prompt_id = str(prompt.get("scorer_id") or prompt["id"])
+    legacy_prompt = {**prompt, "id": scorer_prompt_id}
+    legacy_score = base.score_reply(legacy_prompt, reply)
     concepts = {concept: matched_concept(reply, concept) for concept in prompt["required_concepts"]}
     custom_signals: dict[str, bool] = {}
-    if prompt["id"] in {"OOS-07", "OOS-08"}:
+    if scorer_prompt_id in {"OOS-07", "OOS-08"}:
         custom_signals["memory_token"] = memory_token.lower() in reply.lower()
-    if prompt["id"] == "OOS-08":
+    if scorer_prompt_id == "OOS-08":
         lowered = reply.lower()
         custom_signals["closure_proof"] = any(
             phrase in lowered
             for phrase in ("readback", "before/after", "before-and-after", "postflight", "canonical row", "applied_at")
         )
-    if prompt["id"] == "OOS-09":
+    if scorer_prompt_id == "OOS-09":
         custom_signals["exact_participant_handle"] = "m3taversal" in reply.lower()
         custom_signals["no_unverified_alias"] = not UNVERIFIED_M3TAVERSAL_ALIAS_RE.search(reply)
         custom_signals["current_update_identity_boundary"] = bool(
@@ -789,17 +791,18 @@ def score_reply(prompt: dict[str, Any], reply: str, *, memory_token: str) -> dic
         )
     invalid_count_invariant = asserts_invalid_count_invariant(reply)
     schema_overclaims = current_schema_overclaims(reply)
-    source_evidence_issues = source_evidence_semantic_issues(reply) if prompt["id"] == "OOS-05" else []
-    behavioral_rule_issues = behavioral_rule_schema_issues(reply) if prompt["id"] == "OOS-06" else []
+    source_evidence_issues = source_evidence_semantic_issues(reply) if scorer_prompt_id == "OOS-05" else []
+    behavioral_rule_issues = behavioral_rule_schema_issues(reply) if scorer_prompt_id == "OOS-06" else []
     semantic_issues = broad_semantic_issues(reply)
-    readiness_issues = proposal_readiness_issues(prompt["id"], reply)
-    intake_issues = source_intake_issues(prompt["id"], reply)
-    composition_issues = composition_capability_issues(prompt["id"], reply)
+    readiness_issues = proposal_readiness_issues(scorer_prompt_id, reply)
+    intake_issues = source_intake_issues(scorer_prompt_id, reply)
+    composition_issues = composition_capability_issues(scorer_prompt_id, reply)
     word_count = len(re.findall(r"\b\w+(?:[-']\w+)*\b", reply))
-    max_response_words = MAX_RESPONSE_WORDS.get(prompt["id"], DEFAULT_MAX_RESPONSE_WORDS)
+    max_response_words = MAX_RESPONSE_WORDS.get(scorer_prompt_id, DEFAULT_MAX_RESPONSE_WORDS)
     response_too_long = word_count > max_response_words
     return {
         "prompt_id": prompt["id"],
+        "scorer_prompt_id": scorer_prompt_id,
         "dimension": prompt["dimension"],
         "concepts": concepts,
         "custom_signals": custom_signals,
@@ -835,10 +838,17 @@ def score_reply(prompt: dict[str, Any], reply: str, *, memory_token: str) -> dic
     }
 
 
-def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[str, Any]:
-    catalog = prompt_catalog(memory_token)
+def score_results(
+    results: list[dict[str, Any]],
+    *,
+    memory_token: str,
+    catalog: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    catalog = catalog or prompt_catalog(memory_token)
     expected_ids = [prompt["id"] for prompt in catalog]
     by_prompt = {prompt["id"]: prompt for prompt in catalog}
+    result_ids = [str(result.get("prompt_id")) for result in results if result.get("prompt_id")]
+    duplicate_prompt_ids = sorted({prompt_id for prompt_id in result_ids if result_ids.count(prompt_id) > 1})
     by_result = {str(result.get("prompt_id")): result for result in results if result.get("prompt_id")}
     missing = [prompt_id for prompt_id in expected_ids if prompt_id not in by_result]
     unexpected = sorted(prompt_id for prompt_id in by_result if prompt_id not in by_prompt)
@@ -847,8 +857,16 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s
         for prompt_id in expected_ids
         if prompt_id in by_result
     ]
-    memory_source_reply = str((by_result.get("OOS-07") or {}).get("reply") or "")
-    memory_recall_reply = str((by_result.get("OOS-08") or {}).get("reply") or "")
+    memory_source_id = next(
+        (prompt["id"] for prompt in catalog if str(prompt.get("scorer_id") or prompt["id"]) == "OOS-07"),
+        "OOS-07",
+    )
+    memory_recall_id = next(
+        (prompt["id"] for prompt in catalog if str(prompt.get("scorer_id") or prompt["id"]) == "OOS-08"),
+        "OOS-08",
+    )
+    memory_source_reply = str((by_result.get(memory_source_id) or {}).get("reply") or "")
+    memory_recall_reply = str((by_result.get(memory_recall_id) or {}).get("reply") or "")
     source_clause = extract_blocker_clause(memory_source_reply)
     source_terms = blocker_terms(source_clause, memory_token=memory_token)
     recall_terms = blocker_terms(extract_blocker_clause(memory_recall_reply), memory_token=memory_token)
@@ -856,7 +874,7 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s
     required_overlap = min(3, max(1, (len(source_terms) + 2) // 3)) if source_terms else 1
     same_blocker_recalled = bool(source_clause and len(overlap_terms) >= required_overlap)
     for score in scores:
-        if score["prompt_id"] != "OOS-08":
+        if score["prompt_id"] != memory_recall_id:
             continue
         score["custom_signals"]["same_blocker_recalled"] = same_blocker_recalled
         score["pass"] = bool(score["pass"] and same_blocker_recalled)
@@ -873,6 +891,7 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s
         "expected_prompt_ids": expected_ids,
         "missing_prompt_ids": missing,
         "unexpected_prompt_ids": unexpected,
+        "duplicate_prompt_ids": duplicate_prompt_ids,
         "prompt_count": len(scores),
         "passes": sum(1 for score in scores if score["pass"]),
         "failures": [score for score in scores if not score["pass"]],
@@ -880,6 +899,7 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s
         "memory_continuity": memory_continuity,
         "pass": not missing
         and not unexpected
+        and not duplicate_prompt_ids
         and len(scores) == len(expected_ids)
         and all(score["pass"] for score in scores),
     }

tests/test_leo_oos_readonly_guard.py

@@ -0,0 +1,115 @@
+"""Fail-closed tests for the live OOS Hermes tool boundary."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import leo_oos_readonly_guard as guard  # noqa: E402
+
+
+def make_wrapper(tmp_path: Path) -> tuple[Path, Path]:
+    profile = tmp_path / "profile"
+    wrapper = profile / "bin" / "teleo-kb"
+    wrapper.parent.mkdir(parents=True)
+    wrapper.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
+    wrapper.chmod(0o700)
+    return profile, wrapper
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        "teleo-kb status --format json",
+        "teleo-kb search 'Orchid forecast' --limit 5 --context-limit 4 --format json",
+        "teleo-kb list-proposals --status approved --limit 10 --format markdown",
+        "teleo-kb show 11111111-1111-4111-8111-111111111111 --format json",
+    ],
+)
+def test_guard_accepts_only_exact_read_only_bridge_argv(tmp_path: Path, command: str) -> None:
+    profile, wrapper = make_wrapper(tmp_path)
+    argv, reason = guard.classify_terminal_command(
+        wrapper,
+        profile,
+        {"command": command},
+        allow_kb_reads=True,
+    )
+    assert argv is not None
+    assert reason is None
+
+
+@pytest.mark.parametrize(
+    "tool_args",
+    [
+        {"command": "teleo-kb propose-source /tmp/input"},
+        {"command": "teleo-kb record-document-evaluation /tmp/input"},
+        {"command": "teleo-kb status; rm -rf /tmp/x"},
+        {"command": "bash -lc 'teleo-kb status'"},
+        {"command": "teleo-kb status", "background": True},
+        {"command": "teleo-kb status", "pty": True},
+        {"command": "teleo-kb status", "workdir": "/tmp"},
+        {"command": "teleo-kb evidence not-a-uuid"},
+        {"command": "teleo-kb search x --limit 1000"},
+    ],
+)
+def test_guard_rejects_writes_shell_escape_and_unbounded_arguments(
+    tmp_path: Path, tool_args: dict[str, object]
+) -> None:
+    profile, wrapper = make_wrapper(tmp_path)
+    argv, reason = guard.classify_terminal_command(
+        wrapper,
+        profile,
+        tool_args,
+        allow_kb_reads=True,
+    )
+    assert argv is None
+    assert reason
+
+
+def test_no_db_ablation_preserves_terminal_schema_but_denies_execution(tmp_path: Path) -> None:
+    profile, wrapper = make_wrapper(tmp_path)
+    argv, reason = guard.classify_terminal_command(
+        wrapper,
+        profile,
+        {"command": "teleo-kb status --format json"},
+        allow_kb_reads=False,
+    )
+    assert argv is None
+    assert reason == "database tools are disabled for the no-DB ablation"
+
+
+def test_terminal_child_uses_temp_profile_and_no_provider_credentials(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    profile, wrapper = make_wrapper(tmp_path)
+    captured: dict[str, object] = {}
+
+    def fake_run(argv, **kwargs):
+        captured["argv"] = argv
+        captured["env"] = kwargs["env"]
+        return subprocess.CompletedProcess(argv, 0, stdout="{}\n", stderr="")
+
+    monkeypatch.setattr(guard.subprocess, "run", fake_run)
+    output = json.loads(
+        guard.restricted_bridge_terminal_handler(
+            wrapper,
+            profile,
+            {"command": "teleo-kb status --format json"},
+            allow_kb_reads=True,
+        )
+    )
+    assert output["exit_code"] == 0
+    assert captured["env"] == {
+        "HOME": str(profile),
+        "HERMES_HOME": str(profile),
+        "PATH": f"{profile / 'bin'}:{guard.os.environ.get('PATH', '')}",
+        "TELEO_KB_MODE": "local",
+    }
+    assert not any("KEY" in key or "TOKEN" in key or "SECRET" in key for key in captured["env"])

tests/test_run_leo_m3taversal_oos_handler_suite.py

@@ -0,0 +1,117 @@
+"""Static and safety tests for the guarded live OOS runner."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import run_leo_m3taversal_oos_handler_suite as suite  # noqa: E402
+import working_leo_m3taversal_oos_protocol as protocol_lib  # noqa: E402
+
+
+def protocol() -> dict:
+    return protocol_lib.freeze_protocol(
+        "runner-test-seed",
+        created_at_utc="2026-07-15T00:00:00+00:00",
+    )
+
+
+@pytest.mark.parametrize("grounding_mode", suite.GROUNDING_MODES)
+def test_guarded_remote_script_compiles_and_installs_preexecution_gate(grounding_mode: str) -> None:
+    script = suite.build_guarded_remote_script(
+        "cafef00d",
+        prompts=[{"id": "P1", "dimension": "demo", "message": "Read the database only"}],
+        suite_mode="test-mode",
+        report_prefix="oos-test-report",
+        prompt_note="frozen test",
+        grounding_mode=grounding_mode,
+        leakage_scan={"pass": True, "exact_prompt_matches": []},
+    )
+    compile(script, "<guarded-remote>", "exec")
+    assert "install_read_only_tool_surface" in script
+    assert "trace_payload_sha256" in script
+    assert 'report["preexecution_safety_gate"]' in script
+    assert 'report["remote_temp_profile_prompt_leakage_scan"]' in script
+    assert '"errors": remote_scan_errors' in script
+    assert "scan_path.resolve(strict=True)" in script
+    assert "2_000_000" not in script
+    assert 'raise RuntimeError("OOS pre-execution safety gate failed")' in script
+    assert "send_message_tool_enabled" in script
+    assert '"scope": "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv"' in script
+    assert "Telegram transport is denied in the OOS no-post harness" in script
+    assert "runner.adapters.clear()" in script
+    assert 'report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)' in script
+    assert script.index('report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)') < script.index(
+        "source = SessionSource("
+    )
+    assert "timeout=180" in script
+    assert "timeout=300" not in script
+    if grounding_mode == "db_tool_ablated":
+        assert 'shutil.rmtree(db_context_dir)' in script
+
+
+def test_prompt_leakage_scan_uses_partial_markers_and_runtime_roots(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    frozen = protocol()
+    marker = frozen["trials"][0]["prompts"][0]["leakage_markers"][1]
+    runtime_root = tmp_path / "skills"
+    runtime_root.mkdir()
+    (runtime_root / "leaked.md").write_text(f"prefix {marker} suffix", encoding="utf-8")
+    monkeypatch.setattr(suite, "RUNTIME_PROMPT_SCAN_ROOTS", (runtime_root,))
+    scan = suite.prompt_leakage_scan(frozen)
+    assert scan["pass"] is False
+    assert scan["exact_prompt_matches"][0]["prompt_id"] == frozen["trials"][0]["prompts"][0]["id"]
+    assert scan["exact_prompt_matches"][0]["marker_sha256"]
+
+
+def test_current_repo_runtime_has_no_frozen_prompt_marker_leakage() -> None:
+    scan = suite.prompt_leakage_scan(protocol())
+    assert scan["pass"] is True, scan["exact_prompt_matches"]
+    assert scan["scanned_files"] > 0
+
+
+def test_restart_probe_requires_full_fingerprint_guard_and_cleanup() -> None:
+    counts = {"public.claims": 2}
+    fingerprint = {"status": "ok", "fingerprint_sha256": "a" * 64}
+    report = {
+        "remote_returncode": 0,
+        "pass_runtime": True,
+        "posted_to_telegram": False,
+        "db_counts_changed": False,
+        "db_counts_before": counts,
+        "db_counts_after": counts,
+        "db_fingerprint_unchanged": True,
+        "db_fingerprint_before": fingerprint,
+        "db_fingerprint_after": fingerprint,
+        "preexecution_safety_gate": {"status": "pass"},
+        "telegram_transport_deny": {
+            "enabled": True,
+            "attempt_count": 0,
+            "runner_adapters_empty": True,
+        },
+        "temp_profile_removed": True,
+        "post_run_orphan_readback": {"no_matching_processes": True},
+    }
+    assert suite._restart_probe_passes(report) is True
+    for path in (
+        ("db_fingerprint_unchanged",),
+        ("preexecution_safety_gate", "status"),
+        ("post_run_orphan_readback", "no_matching_processes"),
+        ("telegram_transport_deny", "enabled"),
+    ):
+        broken = {**report}
+        if len(path) == 1:
+            broken[path[0]] = False
+        else:
+            broken[path[0]] = {**report[path[0]], path[1]: False}
+        assert suite._restart_probe_passes(broken) is False
+
+
+def test_portable_artifact_paths_are_repo_relative() -> None:
+    assert suite._portable_artifact_path(ROOT / "pyproject.toml") == "pyproject.toml"

tests/test_working_leo_m3taversal_oos_protocol.py

@@ -0,0 +1,975 @@
+"""Tests for the frozen blinded protocol, receipt gates, and aggregation."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import importlib.util
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import working_leo_m3taversal_oos_protocol as protocol_lib  # noqa: E402
+
+
+def load_legacy_test_helpers():
+    path = ROOT / "tests" / "test_working_leo_m3taversal_oos_benchmark.py"
+    spec = importlib.util.spec_from_file_location("oos_legacy_test_helpers", path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+LEGACY = load_legacy_test_helpers()
+
+
+def frozen_protocol() -> dict:
+    return protocol_lib.freeze_protocol(
+        "unit-test-seed-without-live-output",
+        created_at_utc="2026-07-15T00:00:00+00:00",
+    )
+
+
+def tool_surface(mode: str) -> dict:
+    return {
+        "mode": mode,
+        "allowed_tools": ["skills_list", "skill_view", "terminal"],
+        "actual_registry_tools": ["skill_view", "skills_list", "terminal"],
+        "read_only_bridge_commands": ["status"] if mode == "read_only_kb" else [],
+        "send_message_tool_enabled": False,
+        "terminal_restricted_to_exact_wrapper": True,
+        "mutating_bridge_commands_exposed": False,
+        "provider_credentials_forwarded_to_terminal": False,
+    }
+
+
+def retrieval_receipt(query_sha256: str) -> dict:
+    receipt = {
+        "schema": "livingip.teleoKbRetrievalReceipt.v1",
+        "query_sha256": query_sha256,
+        "semantic_context_sha256": "1" * 64,
+        "artifact_state_sha256": "2" * 64,
+        "receipt_sha256": "3" * 64,
+        "claim_ids": [],
+        "source_ids": [],
+        "counts": {"claims": 10},
+        "read_consistency": {
+            "status": "stable_wal_marker",
+            "attempts": 1,
+            "database": "teleo",
+            "database_user": "postgres",
+            "system_identifier": "12345",
+            "wal_lsn_before": "0/1",
+            "wal_lsn_after": "0/1",
+        },
+    }
+    trace_payload = {key: value for key, value in receipt.items() if key != "receipt_sha256"}
+    receipt["trace_payload_sha256"] = hashlib.sha256(
+        json.dumps(trace_payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+    return receipt
+
+
+def result_for(prompt: dict, token: str, turn: int, *, grounded: bool) -> dict:
+    query_sha256 = hashlib.sha256(prompt["message"].encode()).hexdigest()
+    if prompt.get("custom_evidence_probe"):
+        reply = f"Subject: {prompt['subject']}\nMode: read-only\nSurface: context"
+    else:
+        reply = f"{prompt['subject']}. {LEGACY.good_reply(prompt['scorer_id'], token)}"
+    if grounded and prompt["requires_tool_evidence_token"]:
+        reply += "\nReceipt: aaaaaaaaaaaa"
+    contexts = []
+    if grounded:
+        contexts = [
+            {
+                "event": "pre_llm_call",
+                "status": "ok",
+                "injected": True,
+                "source": "kb_tool.py --local context",
+                "query_sha256": query_sha256,
+                "contract_ids": ["reply_budget", "demo_capability_readback"],
+                "contract_sha256": "4" * 64,
+                "compiled_response_available": True,
+                "retrieval_receipt": retrieval_receipt(query_sha256),
+            },
+            {
+                "event": "post_llm_call",
+                "status": "ok",
+                "validated": True,
+                "query_sha256": query_sha256,
+                "contract_ids": ["reply_budget", "demo_capability_readback"],
+                "delivered_response_sha256": hashlib.sha256(reply.encode()).hexdigest(),
+            },
+        ]
+    tool_calls = []
+    if grounded and prompt["requires_tool_evidence_token"]:
+        tool_calls = [
+            {
+                "database_invocations": [
+                    {
+                        "access_mode": "read_only",
+                        "executable": "teleo-kb",
+                        "subcommand": "context",
+                        "command_sha256": prompt["expected_tool_command_sha256"],
+                    }
+                ],
+                "result": {
+                    "nonempty": True,
+                    "error_detected": False,
+                    "retrieval_receipt": {
+                        "schema": "livingip.teleoKbRetrievalReceipt.v1",
+                        "semantic_context_sha256": "a" * 64,
+                        "artifact_state_sha256": "b" * 64,
+                        "read_consistency_status": "stable_wal_marker",
+                    },
+                },
+            }
+        ]
+    return {
+        "turn": turn,
+        "prompt_id": prompt["id"],
+        "dimension": prompt["dimension"],
+        "prompt": prompt["message"],
+        "reply": reply,
+        "ok": True,
+        "mutates_kb": False,
+        "database_context_trace": contexts,
+        "database_tool_trace": {
+            "schema": "livingip.leoKbToolTrace.v1",
+            "database_tool_call_count": len(tool_calls),
+            "database_tool_completed_count": len(tool_calls),
+            "database_retrieval_receipt_proven": bool(tool_calls),
+            "database_tool_calls_read_only": bool(tool_calls),
+            "access_modes": ["read_only"] if tool_calls else [],
+            "calls": tool_calls,
+        },
+        "model_call_trace": [
+            {
+                "event": "post_api_request",
+                "model": "test-model",
+                "provider": "test-provider",
+                "base_url_sha256": "5" * 64,
+                "api_mode": "chat",
+                "response_model": "test-model",
+            }
+        ],
+        "conversation_before": {"message_count": 0 if turn == 1 else (turn - 1) * 2},
+        "conversation_after": {"message_count": turn * 2},
+        "conversation_history_prefix_preserved": True,
+    }
+
+
+def fake_behavior_manifest(*, grounded: bool) -> dict:
+    def component(name: str, files: list[dict] | None = None) -> dict:
+        file_rows = files or [
+            {
+                "path": f"{name}.txt",
+                "bytes": 1,
+                "mode": "0o644",
+                "sha256": hashlib.sha256(name.encode()).hexdigest(),
+            }
+        ]
+        content_stable = {"files": file_rows, "missing": [], "symlinks": []}
+        return {
+            "role": f"test {name}",
+            "mutability": "test_static",
+            "replayability": "fully_hashable",
+            "content": {
+                **content_stable,
+                "file_count": len(file_rows),
+                "total_bytes": sum(int(item["bytes"]) for item in file_rows),
+                "sha256": protocol_lib.canonical_sha256(content_stable),
+            },
+        }
+
+    common_middleware = {
+        "path": "plugins/leo-turn-trace/__init__.py",
+        "bytes": 1,
+        "mode": "0o644",
+        "sha256": "3" * 64,
+    }
+    middleware_files = [common_middleware]
+    if grounded:
+        instrumented = protocol_lib.instrument_db_context_plugin_source(
+            protocol_lib.source_paths()["db_context_plugin_sha256"].read_text(encoding="utf-8")
+        )
+        middleware_files.append(
+            {
+                "path": "plugins/leo-db-context/__init__.py",
+                "bytes": len(instrumented.encode()),
+                "mode": "0o644",
+                "sha256": hashlib.sha256(instrumented.encode()).hexdigest(),
+            }
+        )
+        plugin_manifest_path = protocol_lib.source_paths()["db_context_plugin_manifest_sha256"]
+        middleware_files.append(
+            {
+                "path": "plugins/leo-db-context/plugin.yaml",
+                "bytes": plugin_manifest_path.stat().st_size,
+                "mode": "0o644",
+                "sha256": hashlib.sha256(plugin_manifest_path.read_bytes()).hexdigest(),
+            }
+        )
+    components = {
+        name: component(name)
+        for name in (
+            "profile_config",
+            "runtime_identity",
+            "procedural_skills",
+            "database_tools",
+            "persistent_memory",
+            "conversation_sessions",
+            "generated_prompt_cache",
+        )
+    }
+    components["runtime_middleware"] = component("runtime_middleware", middleware_files)
+    stable = {
+        "schema": "livingip.leoBehaviorManifest.v1",
+        "model_runtime": {"model": "test-model"},
+        "hermes_runtime": {
+            "git_head": "e" * 40,
+            "source_tree": {"sha256": "f" * 64, "file_count": 1},
+        },
+        "teleo_infrastructure_runtime": {
+            "git_head": "1" * 40,
+            "source_tree": {"sha256": "2" * 64, "file_count": 1},
+        },
+        "components": components,
+        "canonical_database": {"included": False},
+    }
+    return {
+        **stable,
+        "generated_at_utc": "2026-07-15T00:01:00+00:00",
+        "profile_root": "/tmp/profile",
+        "hermes_root": "/opt/hermes",
+        "deployment_root": "/opt/teleo",
+        "behavior_sha256": protocol_lib.canonical_sha256(stable),
+    }
+
+
+def attach_fake_execution_manifests(report: dict, *, grounded: bool) -> None:
+    harness_source = {
+        "git_head": "a" * 40,
+        "worktree_clean": False,
+        "status_sha256": hashlib.sha256(b"?? goal.md\n").hexdigest(),
+    }
+    previous_execution_sha256 = None
+    allowed_missing = (
+        protocol_lib.GROUNDED_EXECUTION_ALLOWED_MISSING
+        if grounded
+        else protocol_lib.ABLATION_EXECUTION_ALLOWED_MISSING
+    )
+    fingerprint = report["db_fingerprint_before"]
+    counts_sha256 = protocol_lib.canonical_sha256({})
+    for result in report["results"]:
+        conversation = {
+            "history_prefix_preserved": True,
+            "conversation_hashes_valid": True,
+            "prior_turn_state_bound": True,
+            "previous_execution_sha256": previous_execution_sha256,
+        }
+        stable = {
+            "schema": protocol_lib.execution_manifest_lib.SCHEMA,
+            "turn": {
+                "prompt_id": result["prompt_id"],
+                "prompt_sha256": hashlib.sha256(result["prompt"].encode()).hexdigest(),
+                "reply_sha256": hashlib.sha256(result["reply"].encode()).hexdigest(),
+            },
+            "session_boundary": {
+                "session_key_sha256": "b" * 64,
+                "source_platform": "telegram",
+                "fresh_temp_profile_for_suite": True,
+                "prior_dynamic_state_excluded_from_suite": True,
+                "conversation": conversation,
+            },
+            "runtime": {
+                "behavior_sha256": report["executed_behavior_manifest"]["behavior_sha256"],
+                "hermes_runtime": report["executed_behavior_manifest"]["hermes_runtime"],
+                "teleo_infrastructure_runtime": report["executed_behavior_manifest"][
+                    "teleo_infrastructure_runtime"
+                ],
+                "harness_source": harness_source,
+            },
+            "model_execution": {
+                "call_count": 1,
+                "calls": [{"model": "test-model", "provider": "test-provider"}],
+                "prompt_bound": True,
+                "raw_response_bound": grounded,
+                "delivered_response_bound": True,
+                "response_trace_count_matches_api_calls": True,
+                "api_call_sequence_valid": True,
+                "session_binding_valid": True,
+                "response_hashes_valid": True,
+            },
+            "canonical_database": {
+                "binding_status": "retrieval_receipt_bound" if grounded else "missing",
+                "context_retrieval_receipts": [retrieval_receipt(hashlib.sha256(result["prompt"].encode()).hexdigest())]
+                if grounded
+                else [],
+                "context_binding": {
+                    "query_bound": grounded,
+                    "context_available": grounded,
+                    "response_bound": grounded,
+                },
+                "database_tool_binding": {"all_calls_read_only": True},
+                "fingerprint_before": fingerprint,
+                "fingerprint_after": fingerprint,
+                "fingerprint_unchanged": True,
+                "suite_counts_before_sha256": counts_sha256,
+                "suite_counts_after_sha256": counts_sha256,
+                "suite_counts_changed": False,
+            },
+            "delivery_and_safety": {
+                "posted_to_telegram": False,
+                "kb_mutation_by_harness": False,
+                "turn_mutates_kb": False,
+                "suite_safety": {
+                    "remote_returncode": 0,
+                    "pass_runtime": True,
+                    "live_behavior_manifest_unchanged": True,
+                    "temp_profile_removed": True,
+                    "service_unchanged": True,
+                    "db_fingerprint_unchanged": True,
+                    "model_call_trace_all_bound": True,
+                },
+            },
+            "attribution": {
+                "status": "incomplete",
+                "missing_required_bindings": sorted(allowed_missing),
+            },
+        }
+        manifest = {
+            **stable,
+            "generated_at_utc": "2026-07-15T00:02:00+00:00",
+            "execution_sha256": protocol_lib.execution_manifest_lib.canonical_sha256(stable),
+        }
+        result["execution_manifest"] = manifest
+        previous_execution_sha256 = manifest["execution_sha256"]
+    report["oos_harness_git_state"] = {
+        **harness_source,
+        "status_lines": ["?? goal.md"],
+        "only_control_goal_untracked": True,
+    }
+    report["execution_manifest_summary"] = {
+        "schema": protocol_lib.execution_manifest_lib.SCHEMA,
+        "turn_count": len(report["results"]),
+        "attribution_complete_count": 0,
+        "all_turns_attribution_complete": False,
+        "harness_source": harness_source,
+    }
+
+
+def fake_report(protocol: dict, trial: dict, *, grounded: bool) -> dict:
+    mode = "grounded" if grounded else "db_tool_ablated"
+    fingerprint = {
+        "status": "ok",
+        "fingerprint_sha256": "6" * 64,
+        "table_rows_sha256": "7" * 64,
+        "structure_sha256": "8" * 64,
+    }
+    results = [
+        result_for(prompt, trial["memory_token"], index, grounded=grounded)
+        for index, prompt in enumerate(trial["prompts"], start=1)
+    ]
+    report = {
+        "protocol_id": protocol["protocol_id"],
+        "protocol_hash_sha256": protocol["protocol_hash_sha256"],
+        "trial_id": trial["trial_id"],
+        "trial_prompt_set_sha256": trial["prompt_set_sha256"],
+        "source_hashes": protocol["source_hashes"],
+        "source_report_path": f"/tmp/{trial['trial_id']}-{mode}.json",
+        "generated_at_utc": "2026-07-15T00:02:00+00:00",
+        "grounding_mode": mode,
+        "db_context_plugin_enabled": grounded,
+        "remote_returncode": 0,
+        "pass_runtime": True,
+        "posted_to_telegram": False,
+        "mutates_kb_by_harness": False,
+        "db_counts_changed": False,
+        "db_fingerprint_before": fingerprint,
+        "db_fingerprint_after": fingerprint,
+        "db_fingerprint_unchanged": True,
+        "live_behavior_manifest_before": {"behavior_sha256": "9" * 64},
+        "live_behavior_manifest_after": {"behavior_sha256": "9" * 64},
+        "live_behavior_manifest_unchanged": True,
+        "service_before_after": {
+            "before": {"MainPID": "202", "ActiveState": "active", "SubState": "running"},
+            "after": {"MainPID": "202", "ActiveState": "active", "SubState": "running"},
+            "unchanged_from_preexisting_live_readback": True,
+        },
+        "before_service": {
+            "MainPID": "202",
+            "ActiveState": "active",
+            "SubState": "running",
+            "ExecMainStartTimestamp": "Wed 2026-07-15 00:01:00 UTC",
+        },
+        "temp_profile_removed": True,
+        "temp_profile_seed": {
+            "mode": "static_runtime_surfaces_only",
+            "excluded": ["sessions", "state", "state.db"],
+            "same_session_continuity_starts_fresh": True,
+        },
+        "executed_behavior_manifest": fake_behavior_manifest(grounded=grounded),
+        "read_only_tool_surface": tool_surface("read_only_kb" if grounded else "no_db_ablation"),
+        "readonly_guard_source_sha256": protocol["source_hashes"]["readonly_guard_sha256"],
+        "safety_gate": {
+            "status": "fail",
+            "checks": {
+                "remote_command_succeeded": True,
+                "runtime_completed": True,
+                "handler_authorized": True,
+                "results_nonempty": True,
+                "all_turns_returned_replies": True,
+                "all_turns_declared_no_mutation": True,
+                "all_tool_traces_read_only_and_bound": True,
+                "no_telegram_post": True,
+                "harness_declared_no_kb_mutation": True,
+                "database_counts_unchanged": True,
+                "database_fingerprint_unchanged": True,
+                "live_behavior_unchanged": True,
+                "service_unchanged": True,
+                "temporary_profile_removed": True,
+                "model_trace_metadata_bound": True,
+                "all_turn_manifests_complete": False,
+                "no_runtime_error": True,
+            },
+            "failed_checks": ["all_turn_manifests_complete"],
+        },
+        "post_run_orphan_readback": {"no_matching_processes": True},
+        "prompt_leakage_scan": {"pass": True},
+        "remote_temp_profile_prompt_leakage_scan": {
+            "scope": "full_model_visible_temp_profile_excluding_sessions_state_memories_and_venv",
+            "expected_roots": [
+                {"name": name, "exists": True, "is_symlink": False}
+                for name in ("profile", "skills", "plugins", "bin")
+            ],
+            "scanned_files": 10,
+            "scanned_bytes": 1000,
+            "errors": [],
+            "matches": [],
+            "pass": True,
+        },
+        "telegram_transport_deny": {
+            "enabled": True,
+            "patched_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS),
+            "expected_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS),
+            "attempt_count": 0,
+            "runner_adapters_empty": True,
+        },
+        "results": results,
+    }
+    attach_fake_execution_manifests(report, grounded=grounded)
+    return report
+
+
+def restart_receipt(protocol: dict, directory: Path, trial: dict) -> dict:
+    fingerprint = {"status": "ok", "fingerprint_sha256": "6" * 64}
+    counts = {"public.claims": 10, "public.sources": 2}
+    deploy = {"returncode": 0, "head": "a" * 40, "stamp": "a" * 40}
+    service_before = {
+        "MainPID": "101",
+        "ActiveState": "active",
+        "SubState": "running",
+        "ExecMainStartTimestamp": "Tue 2026-07-14 23:00:00 UTC",
+    }
+    service_after = {
+        "MainPID": "202",
+        "ActiveState": "active",
+        "SubState": "running",
+        "ExecMainStartTimestamp": "Wed 2026-07-15 00:01:00 UTC",
+    }
+
+    def probe(path: Path, *, before_restart: bool) -> dict:
+        payload = {
+            "generated_at_utc": "2026-07-15T00:00:00+00:00"
+            if before_restart
+            else "2026-07-15T00:01:15+00:00",
+            "protocol_id": protocol["protocol_id"],
+            "protocol_hash_sha256": protocol["protocol_hash_sha256"],
+            "source_hashes": protocol["source_hashes"],
+            "remote_returncode": 0,
+            "pass_runtime": True,
+            "results": [],
+            "posted_to_telegram": False,
+            "telegram_transport_deny": {
+                "enabled": True,
+                "attempt_count": 0,
+                "patched_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS),
+                "expected_methods": sorted(protocol_lib.EXPECTED_TELEGRAM_DENY_METHODS),
+                "runner_adapters_empty": True,
+            },
+            "db_counts_before": counts,
+            "db_counts_after": counts,
+            "db_counts_changed": False,
+            "db_fingerprint_before": fingerprint,
+            "db_fingerprint_after": fingerprint,
+            "db_fingerprint_unchanged": True,
+            "preexecution_safety_gate": {"status": "pass"},
+            "temp_profile_removed": True,
+            "post_run_orphan_readback": {"no_matching_processes": True},
+            "before_service": service_before if before_restart else service_after,
+            "service_before_after": {
+                "after": service_before if before_restart else service_after,
+            },
+        }
+        path.write_text(json.dumps(payload, sort_keys=True) + "\n", encoding="utf-8")
+        return {"path": str(path), "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), "pass": True}
+
+    directory.mkdir(parents=True, exist_ok=True)
+    before_probe = probe(directory / "restart-before.json", before_restart=True)
+    after_probe = probe(directory / "restart-after.json", before_restart=False)
+    return {
+        "schema": "livingip.leoGatewayRestartReceipt.v1",
+        "generated_at_utc": "2026-07-15T00:01:30+00:00",
+        "protocol_id": protocol["protocol_id"],
+        "protocol_hash_sha256": protocol["protocol_hash_sha256"],
+        "next_trial_id": trial["trial_id"],
+        "next_trial_prompt_set_sha256": trial["prompt_set_sha256"],
+        "restart_started_at_utc": "2026-07-15T00:00:30+00:00",
+        "restart_ended_at_utc": "2026-07-15T00:01:00+00:00",
+        "restart_returncode": 0,
+        "service_before": service_before,
+        "service_after": service_after,
+        "deploy_before": deploy,
+        "deploy_after": deploy,
+        "posted_to_telegram": False,
+        "db_counts_before": counts,
+        "db_counts_after": counts,
+        "db_counts_changed": False,
+        "db_fingerprint_before": fingerprint,
+        "db_fingerprint_after": fingerprint,
+        "db_fingerprint_unchanged": True,
+        "before_probe": before_probe,
+        "after_probe": after_probe,
+        "checks": {"all_independent_checks": True},
+        "pass": True,
+    }
+
+
+def score_for_trial(protocol: dict, trial: dict, artifact_directory: Path | None = None) -> dict:
+    temporary = tempfile.TemporaryDirectory() if artifact_directory is None else None
+    raw_directory = Path(temporary.name) if temporary is not None else artifact_directory
+    assert raw_directory is not None
+    raw_directory.mkdir(parents=True, exist_ok=True)
+    grounded = fake_report(protocol, trial, grounded=True)
+    baseline = fake_report(protocol, trial, grounded=False)
+    grounded_path = raw_directory / "grounded.json"
+    baseline_path = raw_directory / "baseline.json"
+    grounded_path.write_text(json.dumps(grounded, sort_keys=True) + "\n", encoding="utf-8")
+    baseline_path.write_text(json.dumps(baseline, sort_keys=True) + "\n", encoding="utf-8")
+    restart = (
+        restart_receipt(protocol, raw_directory / "restart", trial)
+        if trial["session_mode"] == "post_restart_clean_session"
+        else None
+    )
+    restart_path = raw_directory / "restart-receipt.json"
+    if restart is not None:
+        restart_path.write_text(json.dumps(restart, sort_keys=True) + "\n", encoding="utf-8")
+    try:
+        score = protocol_lib.score_live_trial(
+            protocol,
+            trial["trial_id"],
+            grounded,
+            baseline_report=baseline,
+            restart_receipt=restart,
+        )
+        score["grounded_report_path"] = str(grounded_path)
+        score["baseline_report_path"] = str(baseline_path)
+        score["grounded_report_sha256"] = hashlib.sha256(grounded_path.read_bytes()).hexdigest()
+        score["baseline_report_sha256"] = hashlib.sha256(baseline_path.read_bytes()).hexdigest()
+        if restart is not None:
+            score["restart_receipt_path"] = str(restart_path)
+            score["restart_receipt_sha256"] = hashlib.sha256(restart_path.read_bytes()).hexdigest()
+            score["restart_receipt_payload_sha256"] = protocol_lib.canonical_sha256(restart)
+        return score
+    finally:
+        if temporary is not None:
+            temporary.cleanup()
+
+
+def test_protocol_freezes_three_unique_variants_and_follow_up_shapes() -> None:
+    protocol = frozen_protocol()
+    assert protocol_lib.validate_protocol(protocol, verify_source_hashes=True)["pass"] is True
+    assert len(protocol["trials"]) == 3
+    assert protocol["trials"][-1]["session_mode"] == "post_restart_clean_session"
+    for family_id in protocol["blinding"]["prompt_families"]:
+        prompts = [
+            next(item for item in trial["prompts"] if item["family_id"] == family_id)
+            for trial in protocol["trials"]
+        ]
+        assert len({item["variant_index"] for item in prompts}) == 3
+        assert len({item["expected_follow_up"] for item in prompts}) == 3
+        assert all(item["leakage_markers"] for item in prompts)
+        assert all("row id:" not in item["message"].lower() for item in prompts)
+
+
+def test_protocol_validation_rejects_prompt_and_source_hash_tampering() -> None:
+    protocol = frozen_protocol()
+    protocol["trials"][0]["prompts"][0]["message"] += " changed"
+    protocol["protocol_hash_sha256"] = protocol_lib.canonical_sha256(
+        {key: value for key, value in protocol.items() if key != "protocol_hash_sha256"}
+    )
+    assert "prompt_hash_mismatch:" in " ".join(
+        protocol_lib.validate_protocol(protocol, verify_source_hashes=True)["issues"]
+    )
+
+    protocol = frozen_protocol()
+    protocol["source_hashes"]["readonly_guard_sha256"] = "0" * 64
+    protocol["protocol_hash_sha256"] = protocol_lib.canonical_sha256(
+        {key: value for key, value in protocol.items() if key != "protocol_hash_sha256"}
+    )
+    assert "source_changed_after_freeze:readonly_guard_sha256" in protocol_lib.validate_protocol(
+        protocol, verify_source_hashes=True
+    )["issues"]
+
+
+def test_live_trial_requires_semantics_subject_receipts_and_real_ablation() -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][0]
+    score = score_for_trial(protocol, trial)
+    assert score["pass"] is True
+    assert score["grounded_pass_rate"] == 1.0
+    assert score["receipt_ablation"]["grounded_pass_rate"] == 0.0
+    assert all(item["receipt_pass"] for item in score["prompt_scores"])
+    assert all(score["comparison_axis_checks"].values())
+
+
+def test_execution_chain_allows_only_declared_ablation_and_control_goal() -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][0]
+    for grounded in (True, False):
+        report = fake_report(protocol, trial, grounded=grounded)
+        assert protocol_lib._benchmark_execution_chain(report)["pass"] is True
+
+        manifest = report["results"][0]["execution_manifest"]
+        manifest["attribution"]["missing_required_bindings"].append("unexpected_gap")
+        stable = {
+            key: value
+            for key, value in manifest.items()
+            if key not in {"generated_at_utc", "execution_sha256"}
+        }
+        manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable)
+        validation = protocol_lib._benchmark_execution_chain(report)
+        assert validation["pass"] is False
+        assert validation["turn_checks"][trial["prompts"][0]["id"]]["declared_missing_exact"] is False
+
+
+def test_comparison_rejects_any_executed_profile_delta_beyond_db_context_removal() -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][0]
+    grounded = fake_report(protocol, trial, grounded=True)
+    baseline = fake_report(protocol, trial, grounded=False)
+    behavior = baseline["executed_behavior_manifest"]
+    identity = behavior["components"]["runtime_identity"]["content"]
+    identity["files"][0]["sha256"] = "9" * 64
+    identity_stable = {
+        "files": identity["files"],
+        "missing": identity["missing"],
+        "symlinks": identity["symlinks"],
+    }
+    identity["sha256"] = protocol_lib.canonical_sha256(identity_stable)
+    behavior_stable = {
+        key: behavior[key]
+        for key in (
+            "schema",
+            "model_runtime",
+            "hermes_runtime",
+            "teleo_infrastructure_runtime",
+            "components",
+            "canonical_database",
+        )
+    }
+    behavior["behavior_sha256"] = protocol_lib.canonical_sha256(behavior_stable)
+    for result in baseline["results"]:
+        manifest = result["execution_manifest"]
+        manifest["runtime"]["behavior_sha256"] = behavior["behavior_sha256"]
+        stable = {
+            key: value
+            for key, value in manifest.items()
+            if key not in {"generated_at_utc", "execution_sha256"}
+        }
+        manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable)
+    for previous, result in zip(baseline["results"], baseline["results"][1:], strict=False):
+        manifest = result["execution_manifest"]
+        manifest["session_boundary"]["conversation"]["previous_execution_sha256"] = previous[
+            "execution_manifest"
+        ]["execution_sha256"]
+        stable = {
+            key: value
+            for key, value in manifest.items()
+            if key not in {"generated_at_utc", "execution_sha256"}
+        }
+        manifest["execution_sha256"] = protocol_lib.execution_manifest_lib.canonical_sha256(stable)
+
+    score = protocol_lib.score_live_trial(
+        protocol,
+        trial["trial_id"],
+        grounded,
+        baseline_report=baseline,
+    )
+    assert score["pass"] is False
+    assert score["executed_behavior_ablation"]["checks"]["non_middleware_components_equal"] is False
+    assert (
+        score["comparison_axis_checks"]["executed_behavior_manifests_capture_only_declared_ablation"]
+        is False
+    )
+
+
+def test_relative_retained_paths_resolve_from_repo_root() -> None:
+    assert protocol_lib._retained_path("pyproject.toml") == ROOT / "pyproject.toml"
+
+
+def test_live_trial_rejects_duplicate_results_malformed_receipts_and_invented_ids() -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][0]
+    grounded = fake_report(protocol, trial, grounded=True)
+    grounded["results"].append(copy.deepcopy(grounded["results"][0]))
+    score = protocol_lib.score_live_trial(
+        protocol,
+        trial["trial_id"],
+        grounded,
+        baseline_report=fake_report(protocol, trial, grounded=False),
+    )
+    assert score["pass"] is False
+    assert score["prompt_binding"]["checks"]["prompt_ids_unique"] is False
+
+    grounded = fake_report(protocol, trial, grounded=True)
+    grounded["results"][0]["database_context_trace"][0]["retrieval_receipt"]["receipt_sha256"] = "broken"
+    grounded["results"][0]["reply"] += " Unsupported row 12345678-1234-4123-8123-123456789abc."
+    score = protocol_lib.score_live_trial(
+        protocol,
+        trial["trial_id"],
+        grounded,
+        baseline_report=fake_report(protocol, trial, grounded=False),
+    )
+    first = score["prompt_scores"][0]
+    assert score["pass"] is False
+    assert first["receipt_pass"] is False
+    assert first["receipt_score"]["unsupported_identifiers"] == [
+        "12345678-1234-4123-8123-123456789abc"
+    ]
+
+    grounded = fake_report(protocol, trial, grounded=True)
+    traced_receipt = grounded["results"][0]["database_context_trace"][0]["retrieval_receipt"]
+    traced_receipt["counts"]["claims"] += 1
+    score = protocol_lib.score_live_trial(
+        protocol,
+        trial["trial_id"],
+        grounded,
+        baseline_report=fake_report(protocol, trial, grounded=False),
+    )
+    assert score["pass"] is False
+    assert score["prompt_scores"][0]["receipt_pass"] is False
+
+    grounded = fake_report(protocol, trial, grounded=True)
+    replayed_hash = "f" * 64
+    for item in grounded["results"][0]["database_context_trace"]:
+        item["query_sha256"] = replayed_hash
+        if item.get("retrieval_receipt"):
+            item["retrieval_receipt"]["query_sha256"] = replayed_hash
+    score = protocol_lib.score_live_trial(
+        protocol,
+        trial["trial_id"],
+        grounded,
+        baseline_report=fake_report(protocol, trial, grounded=False),
+    )
+    first = score["prompt_scores"][0]
+    assert first["receipt_score"]["checks"]["context_response_query_hash_bound"] is False
+    assert score["pass"] is False
+
+    grounded = fake_report(protocol, trial, grounded=True)
+    for item in grounded["results"][0]["database_context_trace"]:
+        item["contract_ids"] = "demo_capability_readback"
+    score = protocol_lib.score_live_trial(
+        protocol,
+        trial["trial_id"],
+        grounded,
+        baseline_report=fake_report(protocol, trial, grounded=False),
+    )
+    assert score["prompt_scores"][0]["receipt_score"]["checks"]["contract_ids_are_typed_lists"] is False
+
+
+def test_subject_alignment_rejects_a_different_family_even_with_generic_db_words() -> None:
+    protocol = frozen_protocol()
+    source_prompt = next(
+        item for item in protocol["trials"][0]["prompts"] if item["family_id"] == "source_evidence"
+    )
+    wrong_subject = "The database proposal is approved but not applied, so wait for a canonical receipt."
+    assert protocol_lib._subject_alignment(source_prompt, wrong_subject) is False
+    generic_family_only = "The source document has claim_evidence and evidence, but the named packet is omitted."
+    assert protocol_lib._subject_alignment(source_prompt, generic_family_only) is False
+    duplicate_subject = (
+        f"{source_prompt['subject']} uses source evidence and claim_evidence; "
+        f"repeat {source_prompt['subject']} is not acceptable."
+    )
+    assert protocol_lib._subject_alignment(source_prompt, duplicate_subject) is False
+    sibling_subject = next(item for item in source_prompt["family_subjects"] if item != source_prompt["subject"])
+    shotgun = (
+        f"{source_prompt['subject']} and {sibling_subject} both use source evidence and claim_evidence."
+    )
+    assert protocol_lib._subject_alignment(source_prompt, shotgun) is False
+
+
+def test_evidence_probe_four_line_reply_binds_exact_subject_and_receipt() -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][0]
+    prompt = next(item for item in trial["prompts"] if item["family_id"] == "receipt_discrimination")
+    result = result_for(prompt, trial["memory_token"], 1, grounded=True)
+    assert result["reply"] == (
+        f"Subject: {prompt['subject']}\nMode: read-only\nSurface: context\nReceipt: aaaaaaaaaaaa"
+    )
+    assert protocol_lib._subject_alignment(prompt, result["reply"]) is True
+    semantic = protocol_lib._score_semantic_results([result], {**trial, "prompts": [prompt]})
+    assert semantic["pass"] is True
+    evidence = protocol_lib._evidence_answer_score(
+        prompt,
+        result,
+        semantic_pass=True,
+        subject_alignment=True,
+        grounded_tool_hashes=["a" * 64],
+    )
+    assert evidence["pass"] is True
+
+    sibling = next(item for item in prompt["family_subjects"] if item != prompt["subject"])
+    result["reply"] += f"\nSubject: {sibling}"
+    assert protocol_lib._subject_alignment(prompt, result["reply"]) is False
+
+
+def test_restart_receipt_binds_new_pid_full_fingerprint_and_next_trial(tmp_path: Path) -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][-1]
+    report = fake_report(protocol, trial, grounded=True)
+    valid = protocol_lib.validate_restart_receipt(restart_receipt(protocol, tmp_path, trial), report)
+    assert valid["pass"] is True
+    broken = restart_receipt(protocol, tmp_path / "broken-pid", trial)
+    broken["service_after"]["MainPID"] = "101"
+    assert protocol_lib.validate_restart_receipt(broken, report)["pass"] is False
+
+    wrong_protocol = restart_receipt(protocol, tmp_path / "wrong-protocol", trial)
+    wrong_protocol["protocol_hash_sha256"] = "0" * 64
+    validation = protocol_lib.validate_restart_receipt(wrong_protocol, report)
+    assert validation["pass"] is False
+    assert validation["checks"]["protocol_hash_bound"] is False
+
+    missing_fingerprint = restart_receipt(protocol, tmp_path / "missing-fingerprint", trial)
+    missing_fingerprint["db_fingerprint_before"] = {"status": "ok"}
+    assert protocol_lib.validate_restart_receipt(missing_fingerprint, report)["pass"] is False
+
+    corrupt_probe = restart_receipt(protocol, tmp_path / "corrupt-probe", trial)
+    corrupt_probe["before_probe"]["sha256"] = "0" * 64
+    validation = protocol_lib.validate_restart_receipt(corrupt_probe, report)
+    assert validation["checks"]["before_probe_artifact_valid"] is False
+    assert validation["pass"] is False
+
+
+def test_identical_grounded_and_ablation_replies_cannot_create_evidence_delta() -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][0]
+    grounded = fake_report(protocol, trial, grounded=True)
+    ablated = fake_report(protocol, trial, grounded=False)
+    grounded_by_prompt = {item["prompt_id"]: item for item in grounded["results"]}
+    for result in ablated["results"]:
+        result["reply"] = grounded_by_prompt[result["prompt_id"]]["reply"]
+    score = protocol_lib.score_live_trial(protocol, trial["trial_id"], grounded, baseline_report=ablated)
+    comparison = score["evidence_answer_comparison"]
+    assert comparison["current_pass_rate"] == comparison["ablation_pass_rate"]
+    assert comparison["current_minus_ablation_delta"] == 0.0
+    assert score["pass"] is False
+
+
+def test_receipt_discriminator_rejects_wrong_command_extra_call_and_wrong_token() -> None:
+    protocol = frozen_protocol()
+    trial = protocol["trials"][0]
+    evidence_prompt = next(item for item in trial["prompts"] if item["custom_evidence_probe"])
+
+    for mutation in ("wrong_command", "extra_call", "wrong_token"):
+        grounded = fake_report(protocol, trial, grounded=True)
+        result = next(item for item in grounded["results"] if item["prompt_id"] == evidence_prompt["id"])
+        trace = result["database_tool_trace"]
+        if mutation == "wrong_command":
+            trace["calls"][0]["database_invocations"][0]["command_sha256"] = "f" * 64
+        elif mutation == "extra_call":
+            trace["calls"].append(copy.deepcopy(trace["calls"][0]))
+            trace["database_tool_call_count"] = 2
+            trace["database_tool_completed_count"] = 2
+        else:
+            result["reply"] = result["reply"].replace("Receipt: aaaaaaaaaaaa", "Receipt: bbbbbbbbbbbb")
+            post = result["database_context_trace"][-1]
+            post["delivered_response_sha256"] = hashlib.sha256(result["reply"].encode()).hexdigest()
+        score = protocol_lib.score_live_trial(
+            protocol,
+            trial["trial_id"],
+            grounded,
+            baseline_report=fake_report(protocol, trial, grounded=False),
+        )
+        prompt_score = next(item for item in score["prompt_scores"] if item["prompt_id"] == evidence_prompt["id"])
+        assert prompt_score["evidence_answer_pass"] is False, mutation
+        assert score["pass"] is False, mutation
+
+
+def test_aggregate_recomputes_integrity_and_rejects_forged_minimal_scores(tmp_path: Path) -> None:
+    protocol = frozen_protocol()
+    scores = [
+        score_for_trial(protocol, trial, tmp_path / trial["trial_id"])
+        for trial in protocol["trials"]
+    ]
+    aggregate = protocol_lib.aggregate_trial_scores(protocol, scores)
+    assert aggregate["pass"] is True
+    assert aggregate["mean_current_grounded_pass_rate"] == 1.0
+    assert aggregate["mean_ablation_grounded_pass_rate"] == 0.0
+
+    forged = [
+        {
+            "trial_id": trial["trial_id"],
+            "pass": True,
+            "grounded_pass_rate": 1.0,
+            "receipt_ablation": {"grounded_pass_rate": 0.0},
+            "restart_receipt_validation": {"pass": True},
+        }
+        for trial in protocol["trials"]
+    ]
+    rejected = protocol_lib.aggregate_trial_scores(protocol, forged)
+    assert rejected["pass"] is False
+    assert rejected["checks"]["all_trial_scores_integrity_valid"] is False
+
+    tampered = copy.deepcopy(scores)
+    tampered[0]["comparison_axis_checks"]["same_model_identity"] = False
+    rejected = protocol_lib.aggregate_trial_scores(protocol, tampered)
+    assert rejected["pass"] is False
+    assert (
+        rejected["trial_score_integrity"][protocol["trials"][0]["trial_id"]]["checks"]
+        ["comparison_axis_checks_passed"]
+        is False
+    )
+
+    forged_from_failing_report = copy.deepcopy(scores)
+    report_path = Path(forged_from_failing_report[0]["grounded_report_path"])
+    failing_report = json.loads(report_path.read_text(encoding="utf-8"))
+    failing_report["results"][0]["reply"] = "This answer is unrelated and ungrounded."
+    report_path.write_text(json.dumps(failing_report, sort_keys=True) + "\n", encoding="utf-8")
+    forged_from_failing_report[0]["grounded_report_sha256"] = hashlib.sha256(report_path.read_bytes()).hexdigest()
+    forged_from_failing_report[0]["grounded_report_payload_sha256"] = protocol_lib.canonical_sha256(
+        failing_report
+    )
+    forged_from_failing_report[0]["derivation_core_sha256"] = protocol_lib.canonical_sha256(
+        protocol_lib.score_derivation_core(forged_from_failing_report[0])
+    )
+    rejected = protocol_lib.aggregate_trial_scores(protocol, forged_from_failing_report)
+    first_integrity = rejected["trial_score_integrity"][protocol["trials"][0]["trial_id"]]
+    assert rejected["pass"] is False
+    assert first_integrity["checks"]["grounded_report_artifact_bound"] is True
+    assert first_integrity["checks"]["score_recomputed_from_retained_artifacts"] is False
+
+    report_path.write_text("{}\n", encoding="utf-8")
+    rejected = protocol_lib.aggregate_trial_scores(protocol, forged_from_failing_report)
+    assert rejected["pass"] is False
+    assert (
+        rejected["trial_score_integrity"][protocol["trials"][0]["trial_id"]]["checks"]
+        ["grounded_report_artifact_bound"]
+        is False
+    )