#!/usr/bin/env python3
"""Run one isolated challenger-agent versus Leo exchange without sending or writing."""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import os
import re
import subprocess
import tempfile
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

try:
    import leo_agent_challenger_protocol as protocol
    import run_leo_direct_claim_handler_suite as handler_harness
except ImportError:  # pragma: no cover - imported as scripts.* in tests
    from scripts import leo_agent_challenger_protocol as protocol
    from scripts import run_leo_direct_claim_handler_suite as handler_harness

ROOT = Path(__file__).resolve().parents[1]
REPORT_DIR = ROOT / "docs" / "reports" / "leo-working-state-20260709"
OUTPUT_JSON = REPORT_DIR / "leo-agent-challenger-loop-source-current.json"
RECEIPT_JSON = REPORT_DIR / "leo-agent-challenger-loop-receipt-current.json"
NEGATIVE_JSON = REPORT_DIR / "leo-agent-challenger-loop-negative-controls-current.json"
LEDGER_JSONL = REPORT_DIR / "leo-agent-challenger-loop-ledger.jsonl"
DEFAULT_MODEL = "google/gemini-2.5-flash"
DEFAULT_MAX_TOKENS = 768
DEFAULT_REPORT_PREFIX = "leo-agent-challenger-loop"
REMOTE_REPORT_FETCH_ATTEMPTS = 3
REMOTE_REPORT_FETCH_TIMEOUT_SECONDS = 45
LOCAL_RECOVERY_DIR = Path(tempfile.gettempdir()) / "teleo-agent-challenger-loop-recovery"
RETENTION_PROJECTION_SCHEMA = "livingip.leoAgentChallengerRetentionProjection.v1"
BEHAVIOR_RETENTION_SCHEMA = "livingip.leoBehaviorManifestRetentionProjection.v1"
RETENTION_FORBIDDEN_PATTERNS = (
    (
        "absolute runtime path",
        re.compile(r"(?<![A-Za-z0-9])/(?:Users|private|home|root|opt|tmp|var|etc)/[^\s\"']+"),
    ),
    ("PEM private key", re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----")),
    ("credentialed URL", re.compile(r"https?://[^/\s:@]+:[^/\s@]+@")),
    ("JWT", re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b")),
    (
        "known token form",
        re.compile(
            r"\b(?:sk-[A-Za-z0-9_-]{20,}|AIza[0-9A-Za-z_-]{35}|gh[pousr]_[A-Za-z0-9]{20,}|"
            r"xox[baprs]-[A-Za-z0-9-]{20,})\b"
        ),
    ),
    ("Bearer credential", re.compile(r"\bBearer\s+[A-Za-z0-9._~+/=-]{20,}\b", re.IGNORECASE)),
)

LEO_CHALLENGER_SOUL_OVERLAY = """

## Isolated challenger-loop contract

This temporary profile is running a no-write adversarial reasoning canary. The challenger message and the public
conversation are the task. Read-only database context may supply evidence, but it must never replace the requested
answer with an unrelated capability, source-intake, Telegram, or deployment response. If the context is unrelated,
ignore it. Follow the challenger's requested labels exactly. When asked for alternatives, return Candidate A: and
Candidate B: and say they are review-only. When answering an objection, return Revised candidate:, Addresses
objection:, Source:, and Status: pending_review; apply not authorized or executed. Never self-merge or write.
"""

DB_CONTEXT_PLUGIN = ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
KB_TOOL = ROOT / "hermes-agent" / "leoclean-bin" / "kb_tool.py"
BEHAVIOR_MANIFEST = ROOT / "scripts" / "leo_behavior_manifest.py"
POSTGRES_MANIFEST = ROOT / "ops" / "postgres_parity_manifest.sql"

READ_ONLY_COMMANDS = (
    "status",
    "search",
    "context",
    "show",
    "evidence",
    "edges",
    "list-proposals",
    "search-proposals",
    "show-proposal",
    "decision-matrix-status",
)

TURN_TRACE_PLUGIN_SOURCE = r"""import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path


def _sha(value):
    return hashlib.sha256(str(value or "").encode("utf-8")).hexdigest()


def _safe_usage(value):
    if not isinstance(value, dict):
        return {}
    return {
        key: item
        for key, item in value.items()
        if key in {
            "input_tokens",
            "output_tokens",
            "prompt_tokens",
            "completion_tokens",
            "total_tokens",
            "cache_read_tokens",
            "cache_write_tokens",
            "reasoning_tokens",
        }
        and isinstance(item, (int, float))
        and not isinstance(item, bool)
    }


def _write(record):
    raw_path = os.getenv("LEO_TURN_API_TRACE_PATH", "").strip()
    if not raw_path:
        return
    path = Path(raw_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\n")
    path.chmod(0o600)


def _pre_llm_call(**kwargs):
    _write({
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "event": "turn_pre_llm_call",
        "session_id_sha256": _sha(kwargs.get("session_id")),
        "prompt_sha256": _sha(kwargs.get("user_message")),
    })


def _post_api_request(**kwargs):
    _write({
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "event": "post_api_request",
        "session_id_sha256": _sha(kwargs.get("session_id")),
        "model": kwargs.get("model"),
        "provider": kwargs.get("provider"),
        "response_model": kwargs.get("response_model"),
        "api_call_count": kwargs.get("api_call_count"),
        "finish_reason": kwargs.get("finish_reason"),
        "usage": _safe_usage(kwargs.get("usage")),
    })


def _post_llm_call(**kwargs):
    _write({
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "event": "turn_post_llm_call",
        "session_id_sha256": _sha(kwargs.get("session_id")),
        "prompt_sha256": _sha(kwargs.get("user_message")),
        "raw_assistant_response_sha256": _sha(kwargs.get("assistant_response")),
    })


def register(ctx):
    ctx.register_hook("pre_llm_call", _pre_llm_call)
    ctx.register_hook("post_api_request", _post_api_request)
    ctx.register_hook("post_llm_call", _post_llm_call)
"""

TURN_TRACE_PLUGIN_MANIFEST = """name: leo-turn-execution-trace
version: "1.0.0"
description: Secret-safe model-call trace for the isolated challenger loop.
provides_hooks:
  - pre_llm_call
  - post_api_request
  - post_llm_call
"""

READ_ONLY_KB_WRAPPER = r'''#!/usr/bin/env python3
"""Fail-closed launcher for the no-write Leo canary."""
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

ALLOWED = __READ_ONLY_COMMANDS__
KNOWN = ALLOWED | {
    "propose-edge",
    "propose-attachment-evaluation",
    "propose-source",
    "prepare-source",
    "record-document-evaluation",
    "propose-core-change",
    "approve",
    "apply",
}
args = sys.argv[1:]
command = next((item for item in args if item in KNOWN), None)
trace_path = os.getenv("LEO_KB_GUARD_TRACE_PATH", "").strip()
record = {
    "generated_at_utc": datetime.now(timezone.utc).isoformat(),
    "command": command,
    "allowed": command in ALLOWED,
    "argv_sha256": __import__("hashlib").sha256(json.dumps(args).encode()).hexdigest(),
}
if trace_path:
    path = Path(trace_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\n")
if command not in ALLOWED:
    print("blocked by isolated challenger-loop read-only guard", file=sys.stderr)
    raise SystemExit(77)
implementation = Path(__file__).with_name("_kb_tool_readonly_impl.py")
os.execv(sys.executable, [sys.executable, str(implementation), *args])
'''


REMOTE_SCRIPT = r'''
import asyncio
import copy
import hashlib
import json
import multiprocessing
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
import traceback
import types
import uuid
from datetime import datetime, timezone
from pathlib import Path

import aiohttp
import yaml

LIVE_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean")
AGENT_ROOT = Path("/home/teleo/.hermes/hermes-agent")
DEPLOY_ROOT = Path("/opt/teleo-eval/workspaces/deploy-infra")
OPENROUTER_KEY = Path("/opt/teleo-eval/secrets/openrouter-key")
OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions"
CHAT_ID = "__CHAT_ID__"
USER_ID = "__USER_ID__"
RUN_ID = "__RUN_ID__"
REPORT_PREFIX = __REPORT_PREFIX_JSON__
REPORT_PATH = Path(f"/tmp/{REPORT_PREFIX}-{RUN_ID}.json")
OBJECTIVE = __OBJECTIVE_JSON__
CHALLENGER_MODEL = __CHALLENGER_MODEL_JSON__
CHALLENGER_MAX_TOKENS = __CHALLENGER_MAX_TOKENS__
PROTOCOL_SOURCE = __PROTOCOL_SOURCE_JSON__
DB_CONTEXT_PLUGIN_SOURCE = __DB_CONTEXT_PLUGIN_SOURCE_JSON__
KB_TOOL_SOURCE = __KB_TOOL_SOURCE_JSON__
READ_ONLY_KB_WRAPPER_SOURCE = __READ_ONLY_KB_WRAPPER_SOURCE_JSON__
BEHAVIOR_MANIFEST_SOURCE = __BEHAVIOR_MANIFEST_SOURCE_JSON__
POSTGRES_MANIFEST_SQL = __POSTGRES_MANIFEST_SQL_JSON__
TURN_TRACE_PLUGIN_SOURCE = __TURN_TRACE_PLUGIN_SOURCE_JSON__
TURN_TRACE_PLUGIN_MANIFEST = __TURN_TRACE_PLUGIN_MANIFEST_JSON__
LEO_CHALLENGER_SOUL_OVERLAY = __LEO_CHALLENGER_SOUL_OVERLAY_JSON__


def canonical_sha256(value):
    return hashlib.sha256(
        json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str).encode("utf-8")
    ).hexdigest()


def text_sha256(value):
    return hashlib.sha256(str(value or "").encode("utf-8")).hexdigest()


def process_identity():
    pid = os.getpid()
    try:
        start_ticks = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()[21]
    except Exception:
        start_ticks = "unavailable"
    return {
        "process_pid": pid,
        "process_start_ticks": start_ticks,
        "process_identity_sha256": text_sha256(f"{RUN_ID}:{pid}:{start_ticks}"),
    }


def service_state():
    raw = subprocess.check_output(
        [
            "systemctl",
            "show",
            "leoclean-gateway.service",
            "-p",
            "ActiveState",
            "-p",
            "SubState",
            "-p",
            "MainPID",
            "-p",
            "NRestarts",
            "-p",
            "ExecMainStartTimestamp",
        ],
        text=True,
    )
    return dict(line.split("=", 1) for line in raw.splitlines() if "=" in line)


def db_read_marker():
    sql = """
select jsonb_build_object(
  'database', current_database(),
  'database_user', current_user,
  'system_identifier', (select system_identifier::text from pg_control_system()),
  'wal_lsn', pg_current_wal_lsn()::text
)::text;
"""
    raw = subprocess.check_output(
        ["docker", "exec", "-i", "teleo-pg", "psql", "-U", "postgres", "-d", "teleo", "-At", "-q"],
        input=sql,
        text=True,
        stderr=subprocess.STDOUT,
    )
    return json.loads(raw.strip())


def db_fingerprint():
    marker_before = db_read_marker()
    raw = subprocess.check_output(
        ["docker", "exec", "-i", "teleo-pg", "psql", "-U", "postgres", "-d", "teleo", "-At", "-q"],
        input=POSTGRES_MANIFEST_SQL,
        text=True,
        stderr=subprocess.STDOUT,
        timeout=180,
    )
    marker_after = db_read_marker()
    rows = []
    for line in raw.splitlines():
        candidate = line.strip()
        if not candidate or candidate in {"BEGIN", "SET", "ROLLBACK"}:
            continue
        value = json.loads(candidate)
        if isinstance(value, dict):
            rows.append(value)
    tables = {
        f"{row['schema']}.{row['table']}": {
            "row_count": int(row["row_count"]),
            "rowset_md5": row["rowset_md5"],
        }
        for row in rows
        if row.get("kind") == "table"
    }
    singleton = {
        str(row["kind"]): row.get("items", [])
        for row in rows
        if row.get("kind")
        in {
            "schemas",
            "extensions",
            "application_roles",
            "columns",
            "constraints",
            "indexes",
            "sequences",
            "views",
            "functions",
            "triggers",
            "types",
            "policies",
        }
    }
    identities = [row for row in rows if row.get("kind") == "identity"]
    if len(identities) != 1 or not tables:
        raise RuntimeError("canonical database fingerprint manifest is incomplete")
    identity = identities[0]
    stable = {
        "identity": {
            key: identity.get(key)
            for key in (
                "database",
                "server_version_num",
                "server_encoding",
                "database_collation",
                "database_ctype",
                "transaction_read_only",
            )
        },
        "singleton_sha256": {
            key: canonical_sha256(value) for key, value in sorted(singleton.items())
        },
        "tables": tables,
    }
    return {
        "schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
        "status": "ok",
        "fingerprint_sha256": canonical_sha256(stable),
        "table_count": len(tables),
        "total_rows": sum(item["row_count"] for item in tables.values()),
        "table_rows_sha256": canonical_sha256(tables),
        "structure_sha256": canonical_sha256(stable["singleton_sha256"]),
        "read_consistency": {
            "status": "stable_wal_marker" if marker_before == marker_after else "wal_changed_during_fingerprint",
            "before": marker_before,
            "after": marker_after,
        },
    }


def safe_db_fingerprint():
    try:
        return db_fingerprint()
    except Exception as exc:
        return {
            "schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
            "status": "error",
            "error": type(exc).__name__,
        }


def write_report(report):
    try:
        REPORT_PATH.write_text(json.dumps(report, sort_keys=True, default=str), encoding="utf-8")
        REPORT_PATH.chmod(0o600)
    except Exception:
        pass


def read_jsonl(path):
    if not path or not path.exists():
        return []
    values = []
    for line in path.read_text(encoding="utf-8").splitlines():
        try:
            value = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(value, dict):
            values.append(value)
    return values


def safe_file_manifest(profile):
    included = []
    for relative in (
        "config.yaml",
        "SOUL.md",
        "bin/kb_tool.py",
        "bin/_kb_tool_readonly_impl.py",
        "plugins/leo-db-context/__init__.py",
        "plugins/leo-db-context/plugin.yaml",
        "plugins/leo-turn-execution-trace/__init__.py",
        "plugins/leo-turn-execution-trace/plugin.yaml",
    ):
        path = profile / relative
        included.append(
            {
                "path": relative,
                "sha256": hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None,
                "is_symlink": path.is_symlink(),
            }
        )
    state_dirs = {}
    for name in ("memories", "sessions", "state", "skills", "workspace"):
        path = profile / name
        state_dirs[name] = sorted(item.name for item in path.iterdir()) if path.is_dir() else None
    safe = {
        "included_files": included,
        "state_directories_at_seed": state_dirs,
        "secret_files_copied_but_not_hashed_or_retained": [".env", "auth.json"],
        "symlink_count": sum(path.is_symlink() for path in profile.rglob("*")),
    }
    return {**safe, "manifest_sha256": canonical_sha256(safe)}


def build_leo_profile(tmp_root):
    profile = tmp_root / "leo-profile"
    profile.mkdir(mode=0o700)
    for name in (".env", "auth.json", "config.yaml", "SOUL.md"):
        shutil.copy2(LIVE_PROFILE / name, profile / name)
    with (profile / "SOUL.md").open("a", encoding="utf-8") as handle:
        handle.write(LEO_CHALLENGER_SOUL_OVERLAY)
    if (LIVE_PROFILE / "models_dev_cache.json").is_file():
        shutil.copy2(LIVE_PROFILE / "models_dev_cache.json", profile / "models_dev_cache.json")
    config = yaml.safe_load((profile / "config.yaml").read_text(encoding="utf-8")) or {}
    memory = config.setdefault("memory", {})
    memory["enabled"] = False
    memory["memory_enabled"] = False
    memory["user_profile_enabled"] = False
    config["tools"] = {key: False for key in (config.get("tools") or {})}
    config["toolsets"] = []
    config.setdefault("skills", {})["external_dirs"] = []
    config.setdefault("checkpoints", {})["enabled"] = False
    config.setdefault("compression", {})["enabled"] = False
    config.setdefault("gateway", {}).setdefault("telegram", {})["enabled"] = False
    (profile / "config.yaml").write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8")
    for name in ("memories", "sessions", "state", "plugins", "skills", "workspace", "bin", "cache", "logs"):
        (profile / name).mkdir(mode=0o700, exist_ok=True)

    db_plugin = profile / "plugins" / "leo-db-context"
    db_plugin.mkdir(parents=True)
    db_plugin_source = DB_CONTEXT_PLUGIN_SOURCE.replace(
        '"--limit",\n        "0",', '"--limit",\n        "4",'
    ).replace('"--context-limit",\n        "0",', '"--context-limit",\n        "6",')
    (db_plugin / "__init__.py").write_text(db_plugin_source, encoding="utf-8")
    (db_plugin / "plugin.yaml").write_text(
        'name: leo-db-context\nversion: "1.0.0"\ndescription: Read-only current database context.\n'
        'provides_hooks:\n  - pre_llm_call\n  - post_llm_call\n',
        encoding="utf-8",
    )

    trace_plugin = profile / "plugins" / "leo-turn-execution-trace"
    trace_plugin.mkdir(parents=True)
    (trace_plugin / "__init__.py").write_text(TURN_TRACE_PLUGIN_SOURCE, encoding="utf-8")
    (trace_plugin / "plugin.yaml").write_text(TURN_TRACE_PLUGIN_MANIFEST, encoding="utf-8")

    implementation = profile / "bin" / "_kb_tool_readonly_impl.py"
    implementation.write_text(KB_TOOL_SOURCE, encoding="utf-8")
    implementation.chmod(0o700)
    wrapper = profile / "bin" / "kb_tool.py"
    wrapper.write_text(READ_ONLY_KB_WRAPPER_SOURCE, encoding="utf-8")
    wrapper.chmod(0o700)

    manifest = safe_file_manifest(profile)
    contract = {
        "profile_realpath": str(profile.resolve()),
        "profile_realpath_sha256": text_sha256(str(profile.resolve())),
        "profile_manifest": manifest,
        "memory_seed_empty": manifest["state_directories_at_seed"]["memories"] == [],
        "session_seed_empty": manifest["state_directories_at_seed"]["sessions"] == [],
        "state_seed_empty": manifest["state_directories_at_seed"]["state"] == [],
        "tools_enabled": any(bool(value) for value in config.get("tools", {}).values())
        or bool(config.get("toolsets")),
        "kb_plugin_present": (db_plugin / "__init__.py").is_file(),
        "soul_sha256": hashlib.sha256((profile / "SOUL.md").read_bytes()).hexdigest(),
    }
    return profile, contract


def conversation_rows(messages):
    rows = []
    for message in messages:
        content = message.get("content")
        text = content if isinstance(content, str) else json.dumps(content, sort_keys=True, default=str)
        tool_calls = message.get("tool_calls") or []
        rows.append(
            {
                "role": message.get("role"),
                "content_sha256": text_sha256(text),
                "content_bytes": len(text.encode("utf-8")),
                "tool_calls_sha256": canonical_sha256(tool_calls),
                "tool_call_count": len(tool_calls),
            }
        )
    return rows


def conversation_trace(messages):
    rows = conversation_rows(messages)
    return {
        "message_count": len(rows),
        "messages_sha256": canonical_sha256(rows),
        "last_message_role": rows[-1]["role"] if rows else None,
        "last_message_content_sha256": rows[-1]["content_sha256"] if rows else None,
    }


def current_agent_messages(runner, session_key):
    cache = getattr(runner, "_agent_cache", None) or {}
    cache_lock = getattr(runner, "_agent_cache_lock", None)
    if cache_lock is not None:
        with cache_lock:
            cached = cache.get(session_key)
    else:
        cached = cache.get(session_key)
    if not cached:
        return []
    messages = getattr(cached[0], "_session_messages", None) or []
    return [copy.deepcopy(message) for message in messages if isinstance(message, dict)]


def model_response_trace(messages):
    return [row for message, row in zip(messages, conversation_rows(messages)) if message.get("role") == "assistant"]


def tool_audit(messages):
    calls = []
    forbidden_pattern = re.compile(
        r"(?:propose(?:-|_)|record-document|approve_proposal|apply_proposal|stage_normalized|"
        r"\b(?:insert|update|delete|alter|drop|truncate|create)\b)",
        re.IGNORECASE,
    )
    for message in messages:
        if not isinstance(message, dict):
            continue
        for call in message.get("tool_calls") or []:
            function = call.get("function") if isinstance(call, dict) else {}
            name = str((function or {}).get("name") or (call or {}).get("name") or "")
            arguments = (function or {}).get("arguments") if isinstance(function, dict) else None
            raw = arguments if isinstance(arguments, str) else json.dumps(arguments, sort_keys=True, default=str)
            calls.append(
                {
                    "tool_name": name,
                    "arguments_sha256": text_sha256(raw),
                    "forbidden_mutation": bool(forbidden_pattern.search(f"{name} {raw}")),
                }
            )
    return {
        "tool_call_count": len(calls),
        "calls": calls,
        "forbidden_mutation_detected": any(call["forbidden_mutation"] for call in calls),
        "unknown_tool_call_detected": bool(calls),
    }


def database_tool_trace(context_events):
    calls = []
    for event in context_events:
        receipt = event.get("retrieval_receipt") if isinstance(event, dict) else None
        if event.get("event", "pre_llm_call") != "pre_llm_call" or not isinstance(receipt, dict):
            continue
        row_ids = [str(item) for item in receipt.get("claim_ids") or []]
        source_ids = [str(item) for item in receipt.get("source_ids") or []]
        calls.append(
            {
                "database_invocations": [
                    {"executable": "kb_tool.py", "subcommand": "context", "access_mode": "read_only"}
                ],
                "result": {
                    "row_ids": row_ids,
                    "source_ids": source_ids,
                    "content_sha256": receipt.get("receipt_sha256"),
                    "nonempty": bool(row_ids or source_ids),
                    "error_detected": False,
                },
            }
        )
    return {
        "database_tool_call_count": len(calls),
        "database_tool_completed_count": len(calls),
        "database_tool_call_proven": bool(calls),
        "database_retrieval_receipt_proven": bool(calls),
        "database_tool_calls_read_only": bool(calls),
        "calls": calls,
    }


def leo_worker(profile, connection):
    identity = process_identity()
    try:
        os.environ["HERMES_HOME"] = str(profile)
        os.environ["HOME"] = "/home/teleo"
        context_trace_path = profile / "state" / "leo-db-context-trace.jsonl"
        model_trace_path = profile / "state" / "leo-turn-api-trace.jsonl"
        guard_trace_path = profile / "state" / "leo-kb-guard-trace.jsonl"
        os.environ["LEO_DB_CONTEXT_TRACE_PATH"] = str(context_trace_path)
        os.environ["LEO_TURN_API_TRACE_PATH"] = str(model_trace_path)
        os.environ["LEO_KB_GUARD_TRACE_PATH"] = str(guard_trace_path)
        os.environ["LEO_CHALLENGER_CONTEXT_ONLY"] = "1"
        sys.path.insert(0, str(AGENT_ROOT))
        from gateway.config import Platform
        from gateway.platforms.base import MessageEvent, MessageType
        from gateway.run import GatewayRunner
        from gateway.session import SessionSource

        source = SessionSource(
            platform=Platform.TELEGRAM,
            chat_id=CHAT_ID,
            chat_name="Leo isolated no-send canary",
            chat_type="group",
            user_id=USER_ID,
            user_name="isolated challenger-loop operator",
        )
        runner = GatewayRunner()
        session_key = runner._session_key_for_source(source)
        ready = {
            "event": "ready",
            **identity,
            "authorized": runner._is_user_authorized(source),
            "session_key_sha256": text_sha256(session_key),
        }
        connection.send(ready)
        if not ready["authorized"]:
            return

        async def loop():
            while True:
                command = await asyncio.to_thread(connection.recv)
                if command.get("action") == "stop":
                    connection.send({"event": "stopped", **identity})
                    return
                prompt = str(command.get("prompt") or "")
                turn_id = str(command.get("turn_id") or "")
                if command.get("use_database_context") is False:
                    os.environ["LEO_CHALLENGER_SKIP_DB_CONTEXT"] = "1"
                else:
                    os.environ.pop("LEO_CHALLENGER_SKIP_DB_CONTEXT", None)
                context_start = len(read_jsonl(context_trace_path))
                model_start = len(read_jsonl(model_trace_path))
                messages_before = current_agent_messages(runner, session_key)
                event = MessageEvent(
                    text=prompt,
                    message_type=MessageType.TEXT,
                    source=source,
                    message_id=f"agent-challenger-{RUN_ID}-{turn_id}",
                )
                started = datetime.now(timezone.utc)
                reply = await asyncio.wait_for(runner._handle_message(event), timeout=360)
                ended = datetime.now(timezone.utc)
                messages_after = current_agent_messages(runner, session_key)
                before_rows = conversation_rows(messages_before)
                after_rows = conversation_rows(messages_after)
                prefix = after_rows[: len(before_rows)] == before_rows
                new_messages = messages_after[len(messages_before):] if prefix else messages_after
                context_events = read_jsonl(context_trace_path)[context_start:]
                connection.send(
                    {
                        "event": "turn",
                        "turn_id": turn_id,
                        "actor": "leo",
                        "input_prompt": prompt,
                        "message": reply or "",
                        "started_at_utc": started.isoformat(),
                        "ended_at_utc": ended.isoformat(),
                        "process_identity_sha256": identity["process_identity_sha256"],
                        "handler_session_key_sha256": text_sha256(session_key),
                        "model_call_trace": read_jsonl(model_trace_path)[model_start:],
                        "model_response_trace": model_response_trace(new_messages),
                        "database_context_trace": context_events,
                        "database_tool_trace": database_tool_trace(context_events),
                        "tool_audit": tool_audit(new_messages),
                        "conversation_before": conversation_trace(messages_before),
                        "conversation_after": conversation_trace(messages_after),
                        "conversation_history_prefix_preserved": prefix,
                    }
                )

        asyncio.run(loop())
    except EOFError:
        return
    except Exception as exc:
        try:
            connection.send(
                {
                    "event": "error",
                    "error": type(exc).__name__,
                    "message": str(exc),
                    "traceback_tail": traceback.format_exc().splitlines()[-12:],
                    **identity,
                }
            )
        except Exception:
            pass
    finally:
        connection.close()


def recv_with_timeout(connection, timeout):
    if not connection.poll(timeout):
        raise TimeoutError(f"Leo worker did not reply within {timeout}s")
    value = connection.recv()
    if not isinstance(value, dict):
        raise RuntimeError("Leo worker returned a non-object")
    if value.get("event") == "error":
        raise RuntimeError(f"Leo worker error: {value.get('error')}: {value.get('message')}")
    return value


async def challenger_turn(prompt, session_id_sha256, process_row):
    if not OPENROUTER_KEY.is_file():
        raise RuntimeError("OpenRouter key file unavailable")
    key = OPENROUTER_KEY.read_text(encoding="utf-8").strip()
    started = datetime.now(timezone.utc)
    payload = {
        "model": CHALLENGER_MODEL,
        "messages": [
            {"role": "system", "content": protocol.CHALLENGER_SOUL},
            {"role": "user", "content": prompt},
        ],
        "max_tokens": CHALLENGER_MAX_TOKENS,
        "temperature": 0.0,
    }
    async with aiohttp.ClientSession() as session:
        async with session.post(
            OPENROUTER_URL,
            headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
            json=payload,
            timeout=aiohttp.ClientTimeout(total=180),
        ) as response:
            status = response.status
            data = await response.json()
    if status >= 400:
        raise RuntimeError(f"challenger provider returned HTTP {status}")
    message = str((data.get("choices") or [{}])[0].get("message", {}).get("content") or "").strip()
    if not message:
        raise RuntimeError("challenger provider returned an empty response")
    ended = datetime.now(timezone.utc)
    prompt_sha = text_sha256(prompt)
    reply_sha = text_sha256(message)
    provider_id_sha = text_sha256(str(data.get("id") or ""))
    usage = {
        key: value
        for key, value in (data.get("usage") or {}).items()
        if key in {"prompt_tokens", "completion_tokens", "total_tokens"} and isinstance(value, (int, float))
    }
    trace = [
        {
            "event": "turn_pre_llm_call",
            "session_id_sha256": session_id_sha256,
            "prompt_sha256": prompt_sha,
        },
        {
            "event": "post_api_request",
            "session_id_sha256": session_id_sha256,
            "prompt_sha256": prompt_sha,
            "provider": "openrouter",
            "model": CHALLENGER_MODEL,
            "response_model": data.get("model") or CHALLENGER_MODEL,
            "provider_response_id_sha256": provider_id_sha,
            "usage": usage,
        },
        {
            "event": "turn_post_llm_call",
            "session_id_sha256": session_id_sha256,
            "prompt_sha256": prompt_sha,
            "raw_assistant_response_sha256": reply_sha,
        },
    ]
    rows = [
        {
            "role": "user",
            "content_sha256": prompt_sha,
            "content_bytes": len(prompt.encode("utf-8")),
            "tool_calls_sha256": canonical_sha256([]),
            "tool_call_count": 0,
        },
        {
            "role": "assistant",
            "content_sha256": reply_sha,
            "content_bytes": len(message.encode("utf-8")),
            "tool_calls_sha256": canonical_sha256([]),
            "tool_call_count": 0,
        },
    ]
    return {
        "actor": "challenger",
        "input_prompt": prompt,
        "message": message,
        "started_at_utc": started.isoformat(),
        "ended_at_utc": ended.isoformat(),
        "process_identity_sha256": process_row["process_identity_sha256"],
        "handler_session_key_sha256": session_id_sha256,
        "model_call_trace": trace,
        "model_response_trace": [rows[-1]],
        "database_context_trace": [],
        "database_tool_trace": {
            "database_tool_call_count": 0,
            "database_tool_completed_count": 0,
            "database_tool_call_proven": False,
            "database_retrieval_receipt_proven": False,
            "database_tool_calls_read_only": True,
            "calls": [],
        },
        "tool_audit": {
            "tool_call_count": 0,
            "calls": [],
            "forbidden_mutation_detected": False,
            "unknown_tool_call_detected": False,
        },
        "conversation_before": {
            "message_count": 0,
            "messages_sha256": canonical_sha256([]),
            "last_message_role": None,
            "last_message_content_sha256": None,
        },
        "conversation_after": {
            "message_count": 2,
            "messages_sha256": canonical_sha256(rows),
            "last_message_role": "assistant",
            "last_message_content_sha256": reply_sha,
        },
        "conversation_history_prefix_preserved": True,
        "stateless_input_rebuilt": True,
        "usage": usage,
    }


def build_safety_gate(report):
    turns = report.get("turns") or []
    before = report.get("db_fingerprint_before") or {}
    after = report.get("db_fingerprint_after") or {}
    cleanup = report.get("cleanup") or {}
    service = report.get("service_before_after") or {}
    generation_audit = report.get("challenger_generation_audit") or {}
    checks = {
        "six_runtime_turns_completed": [str(item.get("turn_id") or "") for item in turns]
        == ["C1", "L1", "C2", "L2", "C3", "L3"]
        and [str(item.get("actor") or "") for item in turns]
        == ["challenger", "leo", "challenger", "leo", "challenger", "leo"]
        and all(str(item.get("message") or "").strip() for item in turns),
        "leo_worker_authorized": (report.get("handler") or {}).get("authorized") is True,
        "leo_model_tools_disabled": (report.get("isolation") or {}).get("roles", {}).get("leo", {}).get("tools_enabled") is False,
        "read_only_guard_blocked_mutation_preflight": (report.get("read_only_guard_preflight") or {}).get("blocked") is True,
        "no_turn_used_model_tools": all(protocol._safe_zero_tool_audit(item) for item in turns),
        "challenger_semantic_gate_passed_before_leo_advance": bool(
            generation_audit.get("accepted_attempt")
            and generation_audit.get("advanced_to_leo") is True
            and (turns[4].get("objection_gate") or {}).get("passes") is True
        )
        if len(turns) == 6
        else False,
        "no_telegram_post": report.get("posted_to_telegram") is False,
        "harness_declared_no_kb_mutation": report.get("mutates_kb_by_harness") is False,
        "database_fingerprint_unchanged": report.get("db_fingerprint_unchanged") is True
        and before.get("fingerprint_sha256") == after.get("fingerprint_sha256")
        and before.get("table_rows_sha256") == after.get("table_rows_sha256")
        and before.get("structure_sha256") == after.get("structure_sha256"),
        "live_behavior_unchanged": report.get("live_behavior_manifest_unchanged") is True,
        "gateway_service_unchanged": service.get("unchanged_from_preexisting_live_readback") is True,
        "workers_and_profiles_cleaned": cleanup.get("all_workers_stopped") is True
        and cleanup.get("temp_root_removed") is True
        and cleanup.get("orphan_worker_count") == 0,
        "no_runtime_error": not report.get("error"),
    }
    failed = [name for name, value in checks.items() if value is not True]
    return {"status": "pass" if not failed else "fail", "checks": checks, "failed_checks": failed}


def run_suite():
    protocol_module = types.ModuleType("leo_agent_challenger_protocol_embedded")
    protocol_module.__file__ = "<embedded leo_agent_challenger_protocol.py>"
    exec(compile(PROTOCOL_SOURCE, protocol_module.__file__, "exec"), protocol_module.__dict__)
    globals()["protocol"] = protocol_module

    behavior_module = types.ModuleType("leo_behavior_manifest_embedded")
    behavior_module.__file__ = "<embedded leo_behavior_manifest.py>"
    exec(compile(BEHAVIOR_MANIFEST_SOURCE, behavior_module.__file__, "exec"), behavior_module.__dict__)

    parent_identity = process_identity()
    before_service = service_state()
    before_fingerprint = safe_db_fingerprint()
    behavior_before = behavior_module.build_manifest(LIVE_PROFILE, AGENT_ROOT, DEPLOY_ROOT)
    tmp_root = Path(tempfile.mkdtemp(prefix="leo-agent-challenger-loop-"))
    profile = None
    worker = None
    parent_connection = None
    worker_ready = None
    stopped_event = None
    report = {
        "schema": protocol.REPORT_SCHEMA,
        "run_id": RUN_ID,
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "mode": "live_vps_isolated_agent_challenger_no_send_no_write",
        "objective": OBJECTIVE,
        "challenger_model": CHALLENGER_MODEL,
        "challenger_max_tokens_per_turn": CHALLENGER_MAX_TOKENS,
        "posted_to_telegram": False,
        "mutates_kb_by_harness": False,
        "db_fingerprint_before": before_fingerprint,
        "before_service": before_service,
        "live_behavior_manifest_before": behavior_before,
        "turns": [],
        "remote_report_path": str(REPORT_PATH),
    }
    write_report(report)
    try:
        profile, profile_contract = build_leo_profile(tmp_root)
        guard_env = os.environ.copy()
        guard_env["LEO_KB_GUARD_TRACE_PATH"] = str(profile / "state" / "guard-preflight.jsonl")
        guard = subprocess.run(
            [sys.executable, str(profile / "bin" / "kb_tool.py"), "--local", "propose-edge", "a", "supports", "b", "--rationale", "must block"],
            text=True,
            capture_output=True,
            env=guard_env,
            timeout=15,
        )
        report["read_only_guard_preflight"] = {
            "blocked": guard.returncode == 77,
            "returncode": guard.returncode,
            "stderr_sha256": text_sha256(guard.stderr),
        }
        if guard.returncode != 77:
            raise RuntimeError("read-only KB guard did not block mutation preflight")

        ctx = multiprocessing.get_context("fork")
        parent_connection, child_connection = ctx.Pipe()
        worker = ctx.Process(target=leo_worker, args=(profile, child_connection), name=f"leo-challenger-{RUN_ID}")
        worker.start()
        child_connection.close()
        worker_ready = recv_with_timeout(parent_connection, 90)
        if worker_ready.get("event") != "ready" or worker_ready.get("authorized") is not True:
            raise RuntimeError("Leo worker failed authorization preflight")

        challenger_session_id = f"challenger:{RUN_ID}:{uuid.uuid4().hex}"
        challenger_session_sha = text_sha256(challenger_session_id)
        challenger_role = {
            "role": "challenger",
            "runtime_kind": "stateless_openrouter_agent",
            **parent_identity,
            "session_key_sha256": challenger_session_sha,
            "profile_realpath": None,
            "profile_realpath_sha256": text_sha256("no-profile:stateless-openrouter-agent"),
            "profile_manifest": {
                "kind": "no_writable_profile",
                "system_prompt_sha256": text_sha256(protocol.CHALLENGER_SOUL),
                "manifest_sha256": canonical_sha256(
                    {"kind": "no_writable_profile", "system_prompt_sha256": text_sha256(protocol.CHALLENGER_SOUL)}
                ),
            },
            "memory_seed_empty": True,
            "session_seed_empty": True,
            "state_seed_empty": True,
            "tools_enabled": False,
            "kb_plugin_present": False,
            "soul_sha256": text_sha256(protocol.CHALLENGER_SOUL),
            "input_boundary": "fixed role contract plus user objective plus transcript-visible messages only",
        }
        challenger_role["contract_sha256"] = protocol.compute_runtime_role_contract_sha256(challenger_role)
        leo_role = {
            "role": "leo",
            "runtime_kind": "gatewayrunner_temp_profile_no_model_tools",
            **{key: worker_ready.get(key) for key in ("process_pid", "process_start_ticks", "process_identity_sha256")},
            "session_key_sha256": worker_ready.get("session_key_sha256"),
            **profile_contract,
            "input_boundary": "challenger message plus Leo-local prior conversation and read-only current DB context hook",
        }
        leo_role["contract_sha256"] = protocol.compute_runtime_role_contract_sha256(leo_role)
        report["handler"] = {
            "authorized": worker_ready.get("authorized"),
            "session_key_sha256": worker_ready.get("session_key_sha256"),
        }
        report["isolation"] = {
            "distinct_processes": challenger_role["process_identity_sha256"] != leo_role["process_identity_sha256"],
            "writable_roots_disjoint": True,
            "distinct_profile_roots": True,
            "transcript_only_bridge": True,
            "roles": {"challenger": challenger_role, "leo": leo_role},
        }
        write_report(report)

        async def exchange():
            for step in (1, 2, 3):
                base_prompt = protocol.build_challenger_prompt(step, OBJECTIVE, report["turns"])
                if step == 3:
                    candidate = protocol.extract_candidate_before(
                        str(report["turns"][-1].get("message") or "")
                    )
                    generation_audit = {
                        "turn_id": "C3",
                        "max_attempts": protocol.MAX_C3_GENERATION_ATTEMPTS,
                        "accepted_attempt": None,
                        "advanced_to_leo": False,
                        "attempts": [],
                    }
                    report["challenger_generation_audit"] = generation_audit
                    challenger = None
                    for attempt in range(1, protocol.MAX_C3_GENERATION_ATTEMPTS + 1):
                        prompt = (
                            base_prompt
                            if attempt == 1
                            else protocol.build_challenger_retry_prompt(base_prompt, attempt)
                        )
                        candidate_turn = await challenger_turn(
                            prompt, challenger_session_sha, parent_identity
                        )
                        candidate_turn["turn_id"] = "C3"
                        candidate_turn["visible_turn_ids"] = protocol.expected_visible_turn_ids(step)
                        candidate_turn["generation_attempt"] = attempt
                        candidate_turn["objection_gate"] = protocol.evaluate_challenger_objection(
                            candidate, candidate_turn["message"]
                        )
                        candidate_turn["turn_execution_sha256"] = (
                            protocol.compute_turn_execution_sha256(candidate_turn)
                        )
                        accepted = candidate_turn["objection_gate"]["passes"]
                        generation_audit["attempts"].append(
                            {
                                "attempt": attempt,
                                "accepted": accepted,
                                "turn": copy.deepcopy(candidate_turn),
                            }
                        )
                        write_report(report)
                        if accepted:
                            challenger = candidate_turn
                            generation_audit["accepted_attempt"] = attempt
                            generation_audit["advanced_to_leo"] = True
                            break
                    if challenger is None:
                        raise RuntimeError(
                            "C3 challenger exhausted semantic regeneration attempts before Leo advance"
                        )
                else:
                    challenger = await challenger_turn(
                        base_prompt, challenger_session_sha, parent_identity
                    )
                    challenger["turn_id"] = f"C{step}"
                    challenger["visible_turn_ids"] = protocol.expected_visible_turn_ids(step)
                    challenger["turn_execution_sha256"] = (
                        protocol.compute_turn_execution_sha256(challenger)
                    )

                report["turns"].append(challenger)
                write_report(report)

                parent_connection.send(
                    {
                        "action": "turn",
                        "turn_id": f"L{step}",
                        "prompt": challenger["message"],
                        "use_database_context": step < 3,
                    }
                )
                leo = await asyncio.to_thread(recv_with_timeout, parent_connection, 420)
                if leo.get("event") != "turn":
                    raise RuntimeError("unexpected Leo worker response")
                leo.pop("event", None)
                leo["visible_turn_ids"] = [f"C{step}"]
                leo["turn_execution_sha256"] = protocol.compute_turn_execution_sha256(leo)
                report["turns"].append(leo)
                write_report(report)

        asyncio.run(exchange())
        report["proposal_packet"] = protocol.build_proposal_packet(report)
    except Exception as exc:
        report["error"] = f"{type(exc).__name__}: {exc}"
        report["traceback_tail"] = traceback.format_exc().splitlines()[-16:]
    finally:
        if parent_connection is not None:
            try:
                parent_connection.send({"action": "stop"})
                stopped_event = recv_with_timeout(parent_connection, 30)
            except Exception:
                stopped_event = None
        if worker is not None:
            worker.join(timeout=30)
            if worker.is_alive():
                worker.terminate()
                worker.join(timeout=10)
        worker_exitcode = worker.exitcode if worker is not None else None
        worker_alive = worker.is_alive() if worker is not None else False

        after_fingerprint = safe_db_fingerprint()
        after_service = service_state()
        behavior_after = behavior_module.build_manifest(LIVE_PROFILE, AGENT_ROOT, DEPLOY_ROOT)
        report["db_fingerprint_after"] = after_fingerprint
        report["db_fingerprint_unchanged"] = bool(
            before_fingerprint.get("status") == "ok"
            and after_fingerprint.get("status") == "ok"
            and before_fingerprint.get("fingerprint_sha256") == after_fingerprint.get("fingerprint_sha256")
            and before_fingerprint.get("table_rows_sha256") == after_fingerprint.get("table_rows_sha256")
            and before_fingerprint.get("structure_sha256") == after_fingerprint.get("structure_sha256")
        )
        report["live_behavior_manifest_after"] = behavior_after
        report["live_behavior_manifest_unchanged"] = behavior_before.get("behavior_sha256") == behavior_after.get("behavior_sha256")
        report["service_before_after"] = {
            "before": before_service,
            "after": after_service,
            "unchanged_from_preexisting_live_readback": before_service == after_service,
        }
        leo_profile_removed = True
        if profile is not None and profile.exists():
            shutil.rmtree(profile, ignore_errors=True)
            leo_profile_removed = not profile.exists()
        if tmp_root.exists():
            shutil.rmtree(tmp_root, ignore_errors=True)
        report["cleanup"] = {
            "challenger_profile_removed": True,
            "leo_profile_removed": leo_profile_removed,
            "temp_root_removed": not tmp_root.exists(),
            "all_workers_stopped": not worker_alive and worker_exitcode == 0 and bool(stopped_event),
            "orphan_worker_count": 0 if not worker_alive else 1,
            "leo_worker_exitcode": worker_exitcode,
            "graceful_stop_receipt": bool(stopped_event and stopped_event.get("event") == "stopped"),
        }
        report["safety_gate"] = build_safety_gate(report)
        write_report(report)
    return report


print(json.dumps(run_suite(), sort_keys=True, default=str))
'''


def _patched_db_context_source() -> str:
    source = DB_CONTEXT_PLUGIN.read_text(encoding="utf-8")
    patched = source.replace('"--limit",\n        "0",', '"--limit",\n        "4",').replace(
        '"--context-limit",\n        "0",', '"--context-limit",\n        "6",'
    )
    patched = patched.replace(
        'def _pre_llm_call(**kwargs: Any) -> dict[str, str]:\n    user_message = str(kwargs.get("user_message") or "")\n',
        "def _pre_llm_call(**kwargs: Any) -> dict[str, str]:\n"
        '    user_message = str(kwargs.get("user_message") or "")\n'
        '    if os.getenv("LEO_CHALLENGER_SKIP_DB_CONTEXT") == "1":\n'
        '        query_sha256 = hashlib.sha256(user_message.encode("utf-8")).hexdigest()\n'
        '        _trace({"event": "pre_llm_call", "status": "skipped", "injected": False, '
        '"query_sha256": query_sha256, "reason": "isolated_challenger_final_revision"})\n'
        "        return {}\n",
    )
    patched = patched.replace(
        '    response = str(kwargs.get("assistant_response") or "")\n    base_record = {\n',
        '    response = str(kwargs.get("assistant_response") or "")\n'
        '    if os.getenv("LEO_CHALLENGER_CONTEXT_ONLY") == "1":\n'
        '        _trace({"event": "post_llm_call", "status": "context_only", "validated": False, '
        '"transformed": False, "fail_closed": False, "query_sha256": query_sha256, '
        '"response_sha256": hashlib.sha256(response.encode("utf-8")).hexdigest(), '
        '"delivered_response_sha256": hashlib.sha256(response.encode("utf-8")).hexdigest()})\n'
        "        return None\n"
        "    base_record = {\n",
    )
    if patched == source:
        raise RuntimeError("database context source no longer matches the bounded harness patch")
    for required in (
        "LEO_CHALLENGER_SKIP_DB_CONTEXT",
        "isolated_challenger_final_revision",
        "LEO_CHALLENGER_CONTEXT_ONLY",
        '"status": "context_only"',
    ):
        if required not in patched:
            raise RuntimeError(f"database context source patch missing {required}")
    return patched


def build_remote_script(
    run_id: str,
    *,
    objective: str = protocol.DEFAULT_OBJECTIVE,
    challenger_model: str = DEFAULT_MODEL,
    challenger_max_tokens: int = DEFAULT_MAX_TOKENS,
    report_prefix: str = DEFAULT_REPORT_PREFIX,
) -> str:
    if not re.fullmatch(r"[A-Za-z0-9._-]+", report_prefix):
        raise ValueError("report_prefix must contain only letters, numbers, dot, underscore, or hyphen")
    if not 128 <= challenger_max_tokens <= 2048:
        raise ValueError("challenger_max_tokens must be between 128 and 2048")
    wrapper = READ_ONLY_KB_WRAPPER.replace("__READ_ONLY_COMMANDS__", repr(set(READ_ONLY_COMMANDS)))
    return (
        REMOTE_SCRIPT.replace("__CHAT_ID__", handler_harness.CHAT_ID)
        .replace("__USER_ID__", handler_harness.USER_ID)
        .replace("__RUN_ID__", run_id)
        .replace("__REPORT_PREFIX_JSON__", json.dumps(report_prefix))
        .replace("__OBJECTIVE_JSON__", json.dumps(objective))
        .replace("__CHALLENGER_MODEL_JSON__", json.dumps(challenger_model))
        .replace("__CHALLENGER_MAX_TOKENS__", str(challenger_max_tokens))
        .replace(
            "__PROTOCOL_SOURCE_JSON__",
            json.dumps((ROOT / "scripts" / "leo_agent_challenger_protocol.py").read_text(encoding="utf-8")),
        )
        .replace("__DB_CONTEXT_PLUGIN_SOURCE_JSON__", json.dumps(_patched_db_context_source()))
        .replace("__KB_TOOL_SOURCE_JSON__", json.dumps(KB_TOOL.read_text(encoding="utf-8")))
        .replace("__READ_ONLY_KB_WRAPPER_SOURCE_JSON__", json.dumps(wrapper))
        .replace("__BEHAVIOR_MANIFEST_SOURCE_JSON__", json.dumps(BEHAVIOR_MANIFEST.read_text(encoding="utf-8")))
        .replace("__POSTGRES_MANIFEST_SQL_JSON__", json.dumps(POSTGRES_MANIFEST.read_text(encoding="utf-8")))
        .replace("__TURN_TRACE_PLUGIN_SOURCE_JSON__", json.dumps(TURN_TRACE_PLUGIN_SOURCE))
        .replace("__TURN_TRACE_PLUGIN_MANIFEST_JSON__", json.dumps(TURN_TRACE_PLUGIN_MANIFEST))
        .replace("__LEO_CHALLENGER_SOUL_OVERLAY_JSON__", json.dumps(LEO_CHALLENGER_SOUL_OVERLAY))
    )


def fetch_remote_report(run_id: str, *, report_prefix: str = DEFAULT_REPORT_PREFIX) -> dict[str, Any] | None:
    report_path = f"/tmp/{report_prefix}-{run_id}.json"
    for _attempt in range(1, REMOTE_REPORT_FETCH_ATTEMPTS + 1):
        try:
            proc = subprocess.run(
                [
                    "ssh",
                    "-i",
                    str(handler_harness.SSH_KEY),
                    "-o",
                    "BatchMode=yes",
                    "-o",
                    "ConnectTimeout=15",
                    "-o",
                    "ServerAliveInterval=10",
                    "-o",
                    "ServerAliveCountMax=2",
                    "-o",
                    "StrictHostKeyChecking=accept-new",
                    handler_harness.VPS,
                    "sudo -u teleo -H /home/teleo/.hermes/hermes-agent/venv/bin/python -",
                ],
                input=(
                    "from pathlib import Path\n"
                    f"p=Path({report_path!r})\n"
                    "if p.exists():\n"
                    " print(p.read_text(encoding='utf-8'))\n"
                ),
                text=True,
                capture_output=True,
                timeout=REMOTE_REPORT_FETCH_TIMEOUT_SECONDS,
            )
        except subprocess.TimeoutExpired:
            continue
        if proc.returncode != 0:
            continue
        if not proc.stdout.strip():
            return None
        try:
            return json.loads(handler_harness.redact(proc.stdout.strip()))
        except json.JSONDecodeError:
            continue
    return None


def unlink_remote_report(run_id: str, *, report_prefix: str = DEFAULT_REPORT_PREFIX) -> bool:
    report_path = f"/tmp/{report_prefix}-{run_id}.json"
    for _attempt in range(1, REMOTE_REPORT_FETCH_ATTEMPTS + 1):
        try:
            proc = subprocess.run(
                [
                    "ssh",
                    "-i",
                    str(handler_harness.SSH_KEY),
                    "-o",
                    "BatchMode=yes",
                    "-o",
                    "ConnectTimeout=15",
                    "-o",
                    "ServerAliveInterval=10",
                    "-o",
                    "ServerAliveCountMax=2",
                    "-o",
                    "StrictHostKeyChecking=accept-new",
                    handler_harness.VPS,
                    "sudo -u teleo -H /home/teleo/.hermes/hermes-agent/venv/bin/python -",
                ],
                input=(
                    "from pathlib import Path\n"
                    f"p=Path({report_path!r})\n"
                    "if p.exists():\n"
                    " p.unlink()\n"
                    "print('absent' if not p.exists() else 'present')\n"
                ),
                text=True,
                capture_output=True,
                timeout=REMOTE_REPORT_FETCH_TIMEOUT_SECONDS,
            )
        except subprocess.TimeoutExpired:
            continue
        if proc.returncode == 0 and proc.stdout.strip() == "absent":
            return True
    return False


def run_remote(
    *,
    objective: str = protocol.DEFAULT_OBJECTIVE,
    challenger_model: str = DEFAULT_MODEL,
    challenger_max_tokens: int = DEFAULT_MAX_TOKENS,
    report_prefix: str = DEFAULT_REPORT_PREFIX,
) -> dict[str, Any]:
    run_id = uuid.uuid4().hex[:12]
    remote_script = build_remote_script(
        run_id,
        objective=objective,
        challenger_model=challenger_model,
        challenger_max_tokens=challenger_max_tokens,
        report_prefix=report_prefix,
    )
    source_git_commit = subprocess.run(
        ["git", "rev-parse", "HEAD"],
        cwd=ROOT,
        text=True,
        capture_output=True,
        check=True,
    ).stdout.strip()
    source_worktree_clean = not subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=ROOT,
        text=True,
        capture_output=True,
        check=True,
    ).stdout.strip()
    proc = subprocess.run(
        [
            "ssh",
            "-i",
            str(handler_harness.SSH_KEY),
            "-o",
            "BatchMode=yes",
            "-o",
            "StrictHostKeyChecking=accept-new",
            handler_harness.VPS,
            "cd /home/teleo && sudo -u teleo -H /home/teleo/.hermes/hermes-agent/venv/bin/python -",
        ],
        input=remote_script,
        text=True,
        capture_output=True,
        timeout=3600,
    )
    result: dict[str, Any] = {
        "returncode": proc.returncode,
        "stdout": handler_harness.redact(proc.stdout),
        "stderr": handler_harness.redact(proc.stderr),
        "run_id": run_id,
        "execution_transport": {
            "schema": "livingip.leoAgentChallengerSshTransportReceipt.v1",
            "transport": "ssh_batch",
            "run_id": run_id,
            "ssh_returncode": proc.returncode,
            "ssh_endpoint_sha256": hashlib.sha256(handler_harness.VPS.encode("utf-8")).hexdigest(),
            "source_git_commit": source_git_commit,
            "source_worktree_clean_before_run": source_worktree_clean,
            "rendered_remote_script_sha256": hashlib.sha256(remote_script.encode("utf-8")).hexdigest(),
            "report_observed_on_stdout": False,
            "remote_report_fetched_and_unlinked": False,
        },
    }
    for line in reversed(proc.stdout.splitlines()):
        candidate = line.strip()
        if not candidate.startswith("{"):
            continue
        try:
            result["parsed"] = json.loads(handler_harness.redact(candidate))
            result["execution_transport"]["report_observed_on_stdout"] = True
            break
        except json.JSONDecodeError:
            continue
    fetched = fetch_remote_report(run_id, report_prefix=report_prefix)
    report_unlinked = unlink_remote_report(run_id, report_prefix=report_prefix) if fetched is not None else False
    result["execution_transport"]["remote_report_fetched_and_unlinked"] = bool(fetched is not None and report_unlinked)
    if "parsed" not in result and fetched is not None:
        result["parsed"] = fetched
        result["parsed_from_report_file"] = True
    return result


def _sha256_text(value: Any) -> str:
    return hashlib.sha256(str(value or "").encode("utf-8")).hexdigest()


def _manifest_content_projection(content: Any) -> dict[str, Any]:
    if not isinstance(content, dict):
        return {}
    return {
        "file_count": content.get("file_count"),
        "total_bytes": content.get("total_bytes"),
        "sha256": content.get("sha256"),
        "missing_count": len(content.get("missing") or []),
        "symlink_count": len(content.get("symlinks") or []),
    }


def compact_behavior_manifest_for_retention(manifest: Any) -> dict[str, Any]:
    """Keep behavior equality proof without retaining private per-file inventory."""

    if not isinstance(manifest, dict):
        return {}
    projection = manifest.get("retention_projection")
    if isinstance(projection, dict) and projection.get("schema") == BEHAVIOR_RETENTION_SCHEMA:
        return copy.deepcopy(manifest)
    components: dict[str, Any] = {}
    removed_file_entries = 0
    removed_missing_entries = 0
    removed_symlink_entries = 0
    raw_components = manifest.get("components") if isinstance(manifest.get("components"), dict) else {}
    for name, raw_component in sorted(raw_components.items()):
        component = raw_component if isinstance(raw_component, dict) else {}
        content = component.get("content") if isinstance(component.get("content"), dict) else {}
        removed_file_entries += len(content.get("files") or [])
        removed_missing_entries += len(content.get("missing") or [])
        removed_symlink_entries += len(content.get("symlinks") or [])
        components[str(name)] = {
            key: copy.deepcopy(component.get(key)) for key in ("role", "mutability", "replayability")
        }
        components[str(name)]["content"] = _manifest_content_projection(content)

    runtimes: dict[str, Any] = {}
    for name in ("hermes_runtime", "teleo_infrastructure_runtime"):
        runtime = manifest.get(name) if isinstance(manifest.get(name), dict) else {}
        source_tree = runtime.get("source_tree") if isinstance(runtime.get("source_tree"), dict) else {}
        removed_file_entries += len(source_tree.get("files") or [])
        removed_missing_entries += len(source_tree.get("missing") or [])
        runtimes[name] = {
            "git_head": runtime.get("git_head"),
            "git_state": copy.deepcopy(runtime.get("git_state")),
            "source_tree": _manifest_content_projection(source_tree),
        }

    removed_root_count = sum(1 for name in ("profile_root", "hermes_root", "deployment_root") if manifest.get(name))
    return {
        "schema": manifest.get("schema"),
        "generated_at_utc": manifest.get("generated_at_utc"),
        "behavior_sha256": manifest.get("behavior_sha256"),
        "model_runtime": copy.deepcopy(manifest.get("model_runtime")),
        "canonical_database": copy.deepcopy(manifest.get("canonical_database")),
        "components": components,
        **runtimes,
        "retention_projection": {
            "schema": BEHAVIOR_RETENTION_SCHEMA,
            "raw_manifest_sha256": protocol.canonical_sha256(manifest),
            "absolute_root_values_removed": True,
            "absolute_root_values_removed_count": removed_root_count,
            "per_file_inventories_removed": True,
            "removed_file_entries": removed_file_entries,
            "removed_missing_entries": removed_missing_entries,
            "removed_symlink_entries": removed_symlink_entries,
        },
    }


def sanitize_report_for_retention(report: dict[str, Any]) -> dict[str, Any]:
    """Project a live report into a review-safe, independently verifiable receipt."""

    retained = copy.deepcopy(report)
    projection = retained.get("retention_projection")
    if isinstance(projection, dict) and projection.get("schema") == RETENTION_PROJECTION_SCHEMA:
        assert_report_safe_for_retention(retained)
        return retained
    pre_projection_report_sha256 = protocol.canonical_sha256(retained)
    behavior_projection: dict[str, Any] = {}
    for field in ("live_behavior_manifest_before", "live_behavior_manifest_after"):
        raw_manifest = retained.get(field)
        compacted = compact_behavior_manifest_for_retention(raw_manifest)
        retained[field] = compacted
        behavior_projection[field] = copy.deepcopy(compacted.get("retention_projection"))

    redacted_fields: list[str] = []
    remote_report_path = retained.pop("remote_report_path", None)
    if remote_report_path:
        redacted_fields.append("remote_report_path")

    transport = retained.get("execution_transport")
    if isinstance(transport, dict) and isinstance(transport.get("recovery"), dict):
        recovery = transport["recovery"]
        remote_path = recovery.pop("remote_path", None)
        if remote_path:
            redacted_fields.append("execution_transport.recovery.remote_path")

    original_role_contracts: dict[str, str] = {}
    profile_realpath_hash_verified: dict[str, bool] = {}
    isolation = retained.get("isolation") if isinstance(retained.get("isolation"), dict) else {}
    roles = isolation.get("roles") if isinstance(isolation.get("roles"), dict) else {}
    for name, role in sorted(roles.items()):
        if not isinstance(role, dict):
            continue
        original_role_contracts[str(name)] = str(role.get("contract_sha256") or "")
        profile_realpath = role.get("profile_realpath")
        if profile_realpath:
            matches = _sha256_text(profile_realpath) == role.get("profile_realpath_sha256")
            profile_realpath_hash_verified[str(name)] = matches
            if not matches:
                raise ValueError(f"{name} profile realpath hash does not match before retention")
            redacted_fields.append(f"isolation.roles.{name}.profile_realpath")
            role["profile_realpath"] = None
        role["contract_sha256"] = protocol.compute_runtime_role_contract_sha256(role)

    recovery = transport.get("recovery") if isinstance(transport, dict) else {}
    retained["retention_projection"] = {
        "schema": RETENTION_PROJECTION_SCHEMA,
        "raw_remote_report_sha256": (recovery.get("recovered_report_sha256") if isinstance(recovery, dict) else None),
        "pre_projection_report_sha256": pre_projection_report_sha256,
        "behavior_manifests": behavior_projection,
        "redacted_fields": sorted(redacted_fields),
        "redacted_field_count": len(redacted_fields),
        "profile_realpath_hash_verified": profile_realpath_hash_verified,
        "original_runtime_role_contract_sha256": original_role_contracts,
        "absolute_runtime_paths_removed": True,
        "verifier_contract_recomputed_after_redaction": True,
    }
    assert_report_safe_for_retention(retained)
    return retained


def assert_report_safe_for_retention(report: dict[str, Any]) -> None:
    encoded = json.dumps(report, sort_keys=True, ensure_ascii=True)
    for label, pattern in RETENTION_FORBIDDEN_PATTERNS:
        if pattern.search(encoded):
            raise ValueError(f"retained report contains forbidden {label}")


def write_local_recovery_checkpoint(remote: dict[str, Any]) -> Path:
    """Persist the fetched raw report privately until durable projection succeeds."""

    LOCAL_RECOVERY_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)
    LOCAL_RECOVERY_DIR.chmod(0o700)
    run_id = str(remote.get("run_id") or "")
    safe_run_id = run_id if re.fullmatch(r"[0-9a-f]{12}", run_id) else _sha256_text(run_id)[:12]
    checkpoint = LOCAL_RECOVERY_DIR / f"{DEFAULT_REPORT_PREFIX}-raw-recovery-{safe_run_id}.json"
    descriptor = os.open(checkpoint, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump(remote, handle, indent=2, sort_keys=True)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        checkpoint.chmod(0o600)
    except BaseException:
        checkpoint.unlink(missing_ok=True)
        raise
    return checkpoint


def remove_local_recovery_checkpoint(checkpoint: Path) -> None:
    checkpoint.unlink()
    try:
        LOCAL_RECOVERY_DIR.rmdir()
    except OSError:
        # Preserve any checkpoint from a different run or concurrent process.
        pass


def write_outputs(remote: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    recovery_checkpoint = write_local_recovery_checkpoint(remote)
    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    report = copy.deepcopy(remote.get("parsed")) if isinstance(remote.get("parsed"), dict) else {}
    report["remote_returncode"] = remote.get("returncode")
    report["remote_run_id"] = remote.get("run_id")
    report["execution_transport"] = remote.get("execution_transport")
    if remote.get("stderr"):
        remote_stderr = str(remote["stderr"])
        report["remote_stderr_receipt"] = {
            "present": True,
            "sha256": _sha256_text(remote_stderr),
            "bytes": len(remote_stderr.encode("utf-8")),
            "line_count": len(remote_stderr.splitlines()),
        }
    report = sanitize_report_for_retention(report)
    OUTPUT_JSON.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    source_sha = hashlib.sha256(OUTPUT_JSON.read_bytes()).hexdigest()
    receipt = protocol.verify_report(report, source_report_sha256=source_sha)
    RECEIPT_JSON.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    negative = {
        "schema": "livingip.leoAgentChallengerNegativeControlReceipt.v1",
        "benchmark_id": protocol.BENCHMARK_ID,
        "source_report_sha256": source_sha,
        "status": (
            "pass"
            if receipt.get("negative_controls")
            and all(item.get("status") == "caught" for item in receipt["negative_controls"])
            else "fail"
        ),
        "controls": receipt.get("negative_controls") or [],
    }
    NEGATIVE_JSON.write_text(json.dumps(negative, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    ledger_event = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "benchmark_id": protocol.BENCHMARK_ID,
        "remote_run_id": remote.get("run_id"),
        "status": receipt.get("status"),
        "required_tier": receipt.get("required_tier"),
        "current_tier": receipt.get("current_tier"),
        "source_report_sha256": source_sha,
        "proposal_packet_sha256": receipt.get("proposal_packet_sha256"),
        "negative_controls": [item.get("status") for item in receipt.get("negative_controls") or []],
        "decision": "keep" if receipt.get("status") == "pass" else "repair",
    }
    with LEDGER_JSONL.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(ledger_event, sort_keys=True) + "\n")
    remove_local_recovery_checkpoint(recovery_checkpoint)
    return report, receipt


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--challenger-model", default=DEFAULT_MODEL)
    parser.add_argument("--challenger-max-tokens", type=int, default=DEFAULT_MAX_TOKENS)
    parser.add_argument("--objective", default=protocol.DEFAULT_OBJECTIVE)
    args = parser.parse_args()
    remote = run_remote(
        objective=args.objective,
        challenger_model=args.challenger_model,
        challenger_max_tokens=args.challenger_max_tokens,
    )
    report, receipt = write_outputs(remote)
    print(
        json.dumps(
            {
                "source": str(OUTPUT_JSON),
                "receipt": str(RECEIPT_JSON),
                "negative_controls": str(NEGATIVE_JSON),
                "runtime_safety": (report.get("safety_gate") or {}).get("status"),
                "verification": receipt.get("status"),
            },
            indent=2,
            sort_keys=True,
        )
    )
    return 0 if receipt.get("status") == "pass" else 1


if __name__ == "__main__":
    raise SystemExit(main())