#!/usr/bin/env python3
"""Run the m3taversal-style direct-claim suite through live VPS GatewayRunner.

The harness does not post to Telegram and does not write to the production KB.
It copies only static behavior surfaces from the live leoclean profile to a
temporary profile on the VPS, invokes GatewayRunner._handle_message with
Telegram-shaped MessageEvents, and removes the temporary profile after the
run. Prior sessions, memory, state databases, and generated prompt caches are
excluded so the result can be attributed to the declared runtime inputs.
"""

from __future__ import annotations

import json
import re
import subprocess
import uuid
from pathlib import Path
from typing import Any

import working_leo_open_ended_benchmark as benchmark

try:
    from leo_turn_execution_manifest import attach_execution_manifests
except ImportError:  # pragma: no cover - imported as scripts.* in tests
    from scripts.leo_turn_execution_manifest import attach_execution_manifests

ROOT = Path(__file__).resolve().parents[1]
REPORT_DIR = ROOT / "docs" / "reports" / "leo-working-state-20260709"
OUTPUT_JSON = REPORT_DIR / "telegram-handler-direct-claim-suite-current.json"
SSH_KEY = Path.home() / ".ssh/livingip_hetzner_20260604_ed25519"
VPS = "root@77.42.65.182"

DEFAULT_SUITE_MODE = "live_vps_gatewayrunner_temp_profile_direct_claim_suite"
DEFAULT_REPORT_PREFIX = "leo-direct-claim-handler-report"
DEFAULT_PROMPT_NOTE = "Prompts are the exact DC-01..DC-06 direct-claim benchmark messages."

CHAT_ID = "-5146042086"
USER_ID = "9070919"
USER_NAME = "codex handler direct claim"

SECRET_PATTERNS = [
    re.compile(r"\b\d{6,}:[A-Za-z0-9_-]{20,}\b"),
    re.compile(r"(Authorization: Bearer )([A-Za-z0-9._-]+)", re.IGNORECASE),
    re.compile(r"((?:api[_-]?key|token|secret|password)[=:]\s*)([A-Za-z0-9._-]+)", re.IGNORECASE),
]

DB_CONTEXT_PLUGIN = (
    ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
)


REMOTE_SCRIPT = r'''
import asyncio
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import traceback
from datetime import datetime, timezone
from pathlib import Path

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")
CHAT_ID = "__CHAT_ID__"
USER_ID = "__USER_ID__"
USER_NAME = "__USER_NAME__"
RUN_ID = "__RUN_ID__"
PROMPTS = __PROMPTS_JSON__
REPORT_PREFIX = __REPORT_PREFIX_JSON__
REPORT_PATH = Path(f"/tmp/{REPORT_PREFIX}-{RUN_ID}.json")
DB_CONTEXT_PLUGIN_SOURCE = __DB_CONTEXT_PLUGIN_SOURCE_JSON__
TURN_TRACE_PLUGIN_SOURCE = """\
import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path


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


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


def _post_api_request(**kwargs):
    raw_path = os.getenv("LEO_TURN_API_TRACE_PATH", "").strip()
    if not raw_path:
        return None
    path = Path(raw_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    record = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "event": "post_api_request",
        "task_id_sha256": _sha(kwargs.get("task_id")),
        "session_id_sha256": _sha(kwargs.get("session_id")),
        "model": kwargs.get("model"),
        "provider": kwargs.get("provider"),
        "base_url_sha256": _sha(kwargs.get("base_url")),
        "api_mode": kwargs.get("api_mode"),
        "api_call_count": kwargs.get("api_call_count"),
        "finish_reason": kwargs.get("finish_reason"),
        "message_count": kwargs.get("message_count"),
        "response_model": kwargs.get("response_model"),
        "usage": _safe_usage(kwargs.get("usage")),
        "assistant_content_chars": kwargs.get("assistant_content_chars"),
        "assistant_tool_call_count": kwargs.get("assistant_tool_call_count"),
        "provider_response_id": None,
        "provider_response_id_status": "not_exposed_by_hermes_post_api_request_hook",
    }
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(record, sort_keys=True) + "\\n")
    try:
        path.chmod(0o600)
    except OSError:
        pass
    return None


def register(ctx):
    ctx.register_hook("post_api_request", _post_api_request)
"""
TURN_TRACE_PLUGIN_MANIFEST = """\
name: leo-turn-execution-trace
version: "1.0.0"
description: Secret-safe model-call trace for no-send Leo harnesses.
provides_hooks:
  - post_api_request
"""


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


def db_counts():
    sql = """
select jsonb_build_object(
  'public.claims', (select count(*) from public.claims),
  'public.sources', (select count(*) from public.sources),
  'public.claim_evidence', (select count(*) from public.claim_evidence),
  'public.claim_edges', (select count(*) from public.claim_edges),
  'kb_stage.kb_proposals', (select count(*) from kb_stage.kb_proposals)
)::text;
"""
    try:
        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())
    except Exception as exc:
        return {"error": str(exc)}


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():
    sql_path = DEPLOY_ROOT / "ops" / "postgres_parity_manifest.sql"
    before = db_read_marker()
    raw = subprocess.check_output(
        ["docker", "exec", "-i", "teleo-pg", "psql", "-U", "postgres", "-d", "teleo", "-At", "-q"],
        input=sql_path.read_text(encoding="utf-8"),
        text=True,
        stderr=subprocess.STDOUT,
        timeout=180,
    )
    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",
        }
    }
    identity_rows = [row for row in rows if row.get("kind") == "identity"]
    if len(identity_rows) != 1 or not tables:
        raise RuntimeError("canonical database fingerprint manifest is incomplete")
    identity = identity_rows[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: hashlib.sha256(
                json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
            ).hexdigest()
            for key, value in sorted(singleton.items())
        },
        "tables": tables,
    }
    return {
        "schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
        "status": "ok",
        "fingerprint_sha256": hashlib.sha256(
            json.dumps(stable, sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest(),
        "table_count": len(tables),
        "total_rows": sum(item["row_count"] for item in tables.values()),
        "table_rows_sha256": hashlib.sha256(
            json.dumps(tables, sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest(),
        "structure_sha256": hashlib.sha256(
            json.dumps(stable["singleton_sha256"], sort_keys=True, separators=(",", ":")).encode("utf-8")
        ).hexdigest(),
        "read_consistency": {
            "status": "stable_wal_marker" if before == after else "wal_changed_during_fingerprint",
            "before": before,
            "after": 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 copy_profile() -> Path:
    tmp_root = Path(tempfile.mkdtemp(prefix="leo-direct-claim-handler-"))
    target = tmp_root / "profile"

    def ignore(_dir, names):
        ignored = {
            "logs",
            "cache",
            "image_cache",
            "memory_backups",
            "memories",
            "sessions",
            "state",
            ".pytest_cache",
            "__pycache__",
            "gateway.pid",
            "auth.lock",
            ".skills_prompt_snapshot.json",
            "state.db",
            "state.db-shm",
            "state.db-wal",
        }
        return {
            name
            for name in names
            if name in ignored
            or name.endswith(".lock")
            or ".bak-" in name
            or name.endswith(".bak")
        }

    try:
        shutil.copytree(LIVE_PROFILE, target, symlinks=True, ignore=ignore)
        for directory in ("memories", "sessions", "state"):
            (target / directory).mkdir(mode=0o700, exist_ok=True)
    except Exception:
        shutil.rmtree(tmp_root, ignore_errors=True)
        raise
    return target


def install_harness_plugins(profile):
    db_context = profile / "plugins" / "leo-db-context" / "__init__.py"
    db_context.parent.mkdir(parents=True, exist_ok=True)
    db_context.write_text(DB_CONTEXT_PLUGIN_SOURCE, encoding="utf-8")

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


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


def remove_tree(path):
    if not path or not path.exists():
        return True
    try:
        shutil.rmtree(path)
    except Exception:
        for root, dirs, files in os.walk(path, topdown=False):
            for name in files:
                p = Path(root) / name
                try:
                    p.chmod(0o600)
                    p.unlink()
                except Exception:
                    pass
            for name in dirs:
                p = Path(root) / name
                try:
                    p.chmod(0o700)
                    p.rmdir()
                except Exception:
                    pass
        try:
            path.chmod(0o700)
            path.rmdir()
        except Exception:
            pass
    return not path.exists()


def write_report_snapshot(report):
    try:
        REPORT_PATH.write_text(json.dumps(report, sort_keys=True), encoding="utf-8")
    except Exception:
        pass


def read_database_context_trace(path):
    return read_jsonl(path)


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 []
    agent = cached[0]
    messages = getattr(agent, "_session_messages", None) or []
    return [dict(message) for message in messages if isinstance(message, dict)]


async def run_suite():
    sys.path.insert(0, str(DEPLOY_ROOT / "scripts"))
    import leo_behavior_manifest as behavior_manifest_module
    from leo_tool_trace import extract_kb_tool_trace

    def build_behavior_manifest(profile):
        try:
            return behavior_manifest_module.build_manifest(profile, AGENT_ROOT, DEPLOY_ROOT)
        except TypeError:
            manifest = behavior_manifest_module.build_manifest(profile, AGENT_ROOT)
            manifest["teleo_infrastructure_runtime"] = {
                "git_head": behavior_manifest_module.git_head(DEPLOY_ROOT),
                "source_tree": behavior_manifest_module.path_manifest(
                    profile,
                    (
                        DEPLOY_ROOT / "scripts" / "leo_behavior_manifest.py",
                        DEPLOY_ROOT / "scripts" / "leo_tool_trace.py",
                    ),
                ),
            }
            stable = {
                key: value
                for key, value in manifest.items()
                if key
                not in {
                    "generated_at_utc",
                    "profile_root",
                    "hermes_root",
                    "deployment_root",
                    "behavior_sha256",
                }
            }
            manifest["deployment_root"] = str(DEPLOY_ROOT)
            manifest["behavior_sha256"] = behavior_manifest_module.canonical_sha256(stable)
            return manifest

    before_service = service_state()
    before_counts = db_counts()
    before_fingerprint = safe_db_fingerprint()
    behavior_before = build_behavior_manifest(LIVE_PROFILE)
    temp_profile = None
    context_trace_path = None
    model_trace_path = None
    temp_removed = False
    report = {
        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
        "mode": __SUITE_MODE_JSON__,
        "posted_to_telegram": False,
        "mutates_kb_by_harness": False,
        "live_behavior_manifest_before": behavior_before,
        "temp_profile_seed": {
            "mode": "static_runtime_surfaces_only",
            "excluded": [
                "memories",
                "sessions",
                "state",
                "state.db",
                "state.db-shm",
                "state.db-wal",
                ".skills_prompt_snapshot.json",
            ],
            "same_session_continuity_starts_fresh": True,
        },
        "source": {
            "platform": "telegram",
            "chat_id": CHAT_ID,
            "chat_name": "Leo",
            "chat_type": "group",
            "user_id": USER_ID,
            "user_name": USER_NAME,
        },
        "harness_notes": [
            __PROMPT_NOTE_JSON__,
            "The harness uses a temporary copy of the leoclean profile and does not post to Telegram.",
            "This proves handler-level live VPS GatewayRunner behavior, not human-visible Telegram delivery.",
        ],
        "db_counts_before": before_counts,
        "db_fingerprint_before": before_fingerprint,
        "before_service": before_service,
        "remote_report_path": str(REPORT_PATH),
    }
    write_report_snapshot(report)
    try:
        temp_profile = copy_profile()
        install_harness_plugins(temp_profile)
        os.environ["HERMES_HOME"] = str(temp_profile)
        os.environ["HOME"] = "/home/teleo"
        context_trace_path = temp_profile / "state" / "leo-db-context-trace.jsonl"
        model_trace_path = temp_profile / "state" / "leo-turn-api-trace.jsonl"
        os.environ["LEO_DB_CONTEXT_TRACE_PATH"] = str(context_trace_path)
        os.environ["LEO_TURN_API_TRACE_PATH"] = str(model_trace_path)
        report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)
        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",
            chat_type="group",
            user_id=USER_ID,
            user_name=USER_NAME,
        )
        runner = GatewayRunner()
        session_key = runner._session_key_for_source(source)
        authorized = runner._is_user_authorized(source)
        report["handler"] = {
            "temp_profile": str(temp_profile),
            "session_key": session_key,
            "authorized": authorized,
        }
        write_report_snapshot(report)
        if not authorized:
            report["results"] = []
            report["pass_runtime"] = False
            report["failure"] = "telegram group source was not authorized by live config"
            write_report_snapshot(report)
            return report

        results = []
        for index, prompt in enumerate(PROMPTS, start=1):
            trace_start = len(read_database_context_trace(context_trace_path))
            model_trace_start = len(read_jsonl(model_trace_path))
            event = MessageEvent(
                text=prompt["message"],
                message_type=MessageType.TEXT,
                source=source,
                message_id=f"handler-suite-{RUN_ID}-{prompt['id']}",
            )
            started = datetime.now(timezone.utc)
            reply = await asyncio.wait_for(runner._handle_message(event), timeout=300)
            ended = datetime.now(timezone.utc)
            database_tool_trace = extract_kb_tool_trace(current_agent_messages(runner, session_key))
            results.append(
                {
                    "turn": index,
                    "prompt_id": prompt["id"],
                    "dimension": prompt["dimension"],
                    "prompt": prompt["message"],
                    "reply": reply or "",
                    "started_at_utc": started.isoformat(),
                    "ended_at_utc": ended.isoformat(),
                    "ok": bool((reply or "").strip()),
                    "mutates_kb": False,
                    "evidence_tier": "live_vps_gatewayrunner_temp_profile",
                    "claim_ceiling": "Live VPS GatewayRunner reply from temp leoclean profile; no Telegram post; no production DB apply authorized.",
                    "database_context_trace": read_database_context_trace(context_trace_path)[trace_start:],
                    "database_tool_trace": database_tool_trace,
                    "model_call_trace": read_jsonl(model_trace_path)[model_trace_start:],
                }
            )
            report["results"] = results
            write_report_snapshot(report)
        report["results"] = results
        report["pass_runtime"] = all(row["ok"] for row in results)
        return report
    except Exception as exc:
        report["pass_runtime"] = False
        report["error"] = str(exc)
        report["traceback_tail"] = traceback.format_exc().splitlines()[-12:]
        write_report_snapshot(report)
        return report
    finally:
        after_counts = db_counts()
        after_fingerprint = safe_db_fingerprint()
        after_service = service_state()
        behavior_after = build_behavior_manifest(LIVE_PROFILE)
        report["db_counts_after"] = after_counts
        report["db_counts_changed"] = after_counts != before_counts
        report["db_fingerprint_after"] = after_fingerprint
        report["db_fingerprint_unchanged"] = (
            before_fingerprint.get("fingerprint_sha256") == after_fingerprint.get("fingerprint_sha256")
        )
        report["live_behavior_manifest_after"] = behavior_after
        report["changed_live_profile"] = behavior_before["behavior_sha256"] != behavior_after["behavior_sha256"]
        report["live_behavior_manifest_unchanged"] = not report["changed_live_profile"]
        report["service_before_after"] = {
            "before": before_service,
            "after": after_service,
            "unchanged_from_preexisting_live_readback": before_service == after_service,
        }
        database_context_trace = read_database_context_trace(context_trace_path)
        model_call_trace = read_jsonl(model_trace_path)
        if temp_profile is not None:
            tmp_root = temp_profile.parent
            temp_removed = remove_tree(tmp_root)
        report["temp_profile_removed"] = temp_removed
        report["database_context_trace"] = database_context_trace
        report["model_call_trace_count"] = len(model_call_trace)
        report["model_call_trace_all_bound"] = bool(model_call_trace) and all(
            item.get("event") == "post_api_request"
            and item.get("model")
            and item.get("provider")
            and item.get("response_model")
            for item in model_call_trace
        )
        context_injections = [
            item for item in database_context_trace if item.get("event", "pre_llm_call") == "pre_llm_call"
        ]
        response_validations = [
            item for item in database_context_trace if item.get("event") == "post_llm_call"
        ]
        report["database_context_injection_count"] = len(context_injections)
        report["database_context_all_ok"] = bool(context_injections) and all(
            item.get("status") == "ok" and item.get("injected") is True
            for item in context_injections
        )
        report["database_response_validation_count"] = len(response_validations)
        report["database_response_validation_all_ok"] = bool(response_validations) and all(
            item.get("status") == "ok" and item.get("validated") is True
            for item in response_validations
        )
        report["database_response_transform_count"] = sum(
            item.get("transformed") is True for item in response_validations
        )
        tool_traces = [
            result.get("database_tool_trace") or {}
            for result in report.get("results") or []
        ]
        report["database_tool_call_proven_count"] = sum(
            trace.get("database_tool_call_proven") is True for trace in tool_traces
        )
        report["database_tool_call_proven"] = any(
            trace.get("database_tool_call_proven") is True for trace in tool_traces
        )
        report["database_retrieval_receipt_proven"] = any(
            trace.get("database_retrieval_receipt_proven") is True for trace in tool_traces
        )
        tool_traces_with_calls = [
            trace for trace in tool_traces if trace.get("database_tool_call_count")
        ]
        report["database_tool_calls_read_only"] = bool(tool_traces_with_calls) and all(
            trace.get("database_tool_calls_read_only") is True
            for trace in tool_traces_with_calls
        )
        write_report_snapshot(report)


print(json.dumps(asyncio.run(run_suite()), sort_keys=True))
'''


def direct_claim_prompts() -> list[dict[str, Any]]:
    return [
        {
            "id": prompt["id"],
            "dimension": prompt["dimension"],
            "message": prompt["message"],
        }
        for prompt in benchmark.M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS
    ]


def build_remote_script(
    run_id: str,
    *,
    prompts: list[dict[str, Any]] | None = None,
    suite_mode: str = DEFAULT_SUITE_MODE,
    report_prefix: str = DEFAULT_REPORT_PREFIX,
    prompt_note: str = DEFAULT_PROMPT_NOTE,
) -> 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")
    selected_prompts = direct_claim_prompts() if prompts is None else prompts
    return (
        REMOTE_SCRIPT.replace("__CHAT_ID__", CHAT_ID)
        .replace("__USER_ID__", USER_ID)
        .replace("__USER_NAME__", USER_NAME)
        .replace("__RUN_ID__", run_id)
        .replace("__PROMPTS_JSON__", json.dumps(selected_prompts))
        .replace("__SUITE_MODE_JSON__", json.dumps(suite_mode))
        .replace("__REPORT_PREFIX_JSON__", json.dumps(report_prefix))
        .replace("__PROMPT_NOTE_JSON__", json.dumps(prompt_note))
        .replace("__DB_CONTEXT_PLUGIN_SOURCE_JSON__", json.dumps(DB_CONTEXT_PLUGIN.read_text(encoding="utf-8")))
    )


def redact(text: str) -> str:
    result = text
    for pattern in SECRET_PATTERNS:
        if pattern.groups >= 2:
            result = pattern.sub(r"\1[REDACTED]", result)
        else:
            result = pattern.sub("[REDACTED_TOKEN]", result)
    return result


def fetch_remote_report(
    run_id: str,
    *,
    report_prefix: str = DEFAULT_REPORT_PREFIX,
    unlink: bool = True,
) -> dict[str, Any] | None:
    report_path = f"/tmp/{report_prefix}-{run_id}.json"
    unlink_line = "    p.unlink()\n" if unlink else ""
    proc = subprocess.run(
        [
            "ssh",
            "-i",
            str(SSH_KEY),
            "-o",
            "BatchMode=yes",
            "-o",
            "StrictHostKeyChecking=accept-new",
            VPS,
            "sudo -u teleo -H /home/teleo/.hermes/hermes-agent/venv/bin/python -",
        ],
        input=(
            "import json\n"
            "from pathlib import Path\n"
            f"p = Path({report_path!r})\n"
            "if p.exists():\n"
            "    print(p.read_text(encoding='utf-8'))\n"
            f"{unlink_line}"
        ),
        text=True,
        capture_output=True,
        timeout=60,
    )
    if proc.returncode != 0 or not proc.stdout.strip():
        return None
    return json.loads(redact(proc.stdout.strip()))


def run_remote(
    *,
    prompts: list[dict[str, Any]] | None = None,
    suite_mode: str = DEFAULT_SUITE_MODE,
    report_prefix: str = DEFAULT_REPORT_PREFIX,
    prompt_note: str = DEFAULT_PROMPT_NOTE,
) -> dict[str, Any]:
    run_id = uuid.uuid4().hex[:12]
    proc = subprocess.run(
        [
            "ssh",
            "-i",
            str(SSH_KEY),
            "-o",
            "BatchMode=yes",
            "-o",
            "StrictHostKeyChecking=accept-new",
            VPS,
            "cd /home/teleo && sudo -u teleo -H /home/teleo/.hermes/hermes-agent/venv/bin/python -",
        ],
        input=build_remote_script(
            run_id,
            prompts=prompts,
            suite_mode=suite_mode,
            report_prefix=report_prefix,
            prompt_note=prompt_note,
        ),
        text=True,
        capture_output=True,
        timeout=2100,
    )
    result: dict[str, Any] = {
        "returncode": proc.returncode,
        "stdout": redact(proc.stdout),
        "stderr": redact(proc.stderr),
        "run_id": run_id,
    }
    if proc.returncode == 0 and proc.stdout.strip():
        for line in reversed(proc.stdout.splitlines()):
            candidate = line.strip()
            if not candidate.startswith("{"):
                continue
            try:
                result["parsed"] = json.loads(candidate)
                break
            except json.JSONDecodeError:
                continue
        if "parsed" not in result:
            result["parse_error"] = "remote stdout contained no parseable JSON object line"
    if "parsed" not in result:
        fetched = fetch_remote_report(run_id, report_prefix=report_prefix)
        if fetched is not None:
            result["parsed"] = fetched
            result["parsed_from_report_file"] = True
    else:
        fetch_remote_report(run_id, report_prefix=report_prefix)
    return result


def write_output(remote: dict[str, Any], *, output_json: Path = OUTPUT_JSON) -> dict[str, Any]:
    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    parsed = remote.get("parsed") or {}
    report = parsed if isinstance(parsed, dict) else {}
    report["source_report_path"] = str(output_json)
    report["remote_returncode"] = remote.get("returncode")
    report["remote_run_id"] = remote.get("run_id")
    if remote.get("stderr"):
        report["remote_stderr"] = remote["stderr"]
    attach_execution_manifests(report, repo_root=ROOT)
    output_json.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return report


def main() -> int:
    remote = run_remote()
    report = write_output(remote)
    print(json.dumps({"json": str(OUTPUT_JSON), "pass_runtime": report.get("pass_runtime")}, indent=2))
    return 0 if remote.get("returncode") == 0 and report.get("pass_runtime") else 1


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