"""CagedCodexRunner: fuehrt Codex in einem systemd --user --scope aus, der iso_exec.py
mit einem pro-Lauf erzeugten Landlock/seccomp-Profil startet.

Ablauf je Lauf:
  1. ephemere Verzeichnisse unter workspace_root anlegen (ws/home/codex-home/authdir/tmp).
  2. das EINE Credential als read-only KOPIE ins authdir legen; im codex-home einen Verweis
     (Symlink) darauf (M1). Das echte ~/.codex ist nie in der Allowlist.
  3. Profil-Konfiguration (text|tools) bauen und als cfg.json schreiben. tools: Home nur lesend,
     Schreiben nur in den Lauf-Verzeichnissen und in write_allowlist (E44 Option 2; ungueltiger
     Eintrag beim Profilbau -> WRITE_ALLOWLIST_INVALID, nichts startet).
  4. Platzhalter {unit}/{cfg}/{workdir} in der (geprueften) Argumentliste ersetzen.
  5. Argumentliste pruefen: tools-Profil erlaubt die Bypass-Tokens NUR, wenn ein echter Kaefig
     (systemd-run ... iso_exec ... --) davor steht; sonst/alle anderen Profile: hart ablehnen.
  6. Popen mit EXPLIZITER Umgebung (nur XDG_RUNTIME_DIR, DBUS_SESSION_BUS_ADDRESS, PATH),
     keine Shell, eigene Prozessgruppe. Supervision wie SubprocessRunner (Limit/Timeout/Kill).
  7. bei Timeout/Ende: Prozessgruppe toeten UND `systemctl --user stop <unit>` (cgroup-Kill der
     Enkel). last.txt aus dem Workspace lesen (M3-Quercheck), dann aufraeumen + reset-failed.

Fehlt der User-Bus, liefert run() den strukturierten Fehler NO_USER_BUS und startet NICHTS
ohne Scope (M3). Jeder andere Setup-Fehler ist ebenfalls fail-closed.
"""
from __future__ import annotations

import dataclasses
import json
import os
import shutil
import signal
import stat
import subprocess
import tempfile
import time
import uuid
from pathlib import Path
from typing import Mapping, Optional

from . import ISO_EXEC_PATH, PKG_REPO_ROOT
from . import profiles as P
from ..runner import RunResult, RunSpec, SubprocessRunner, UnsafeCommandError, assert_safe_argv

CODEX_ARG_SEP = "--"
LAST_TXT = "last.txt"

_MISSING_HANDLER = object()


class _BridgeTerminated(BaseException):
    """Wird vom SIGTERM-Handler geworfen, damit der finally-Block den Scope stoppt (Review #9)."""


def _raise_terminated(signum, frame):
    raise _BridgeTerminated()

# Das laufende V3-Repo (python/aria_router/isolation/caged_runner.py -> <repo>). Wird im tools-Profil
# gesperrt (Selbstmodifikation, Review #6). Gleiche Ableitung wie isolation.PKG_REPO_ROOT (dort auch fuer
# die Allowlist-Pruefung der Konfiguration genutzt).
_PKG_REPO_ROOT = PKG_REPO_ROOT


def default_bus_env(uid: Optional[int] = None) -> dict:
    uid = os.getuid() if uid is None else uid
    return {
        "XDG_RUNTIME_DIR": f"/run/user/{uid}",
        "DBUS_SESSION_BUS_ADDRESS": f"unix:path=/run/user/{uid}/bus",
        "PATH": "/usr/bin:/bin",
    }


def _res_to_rlimits(res) -> dict:
    """ResourceLimits -> iso_exec rlimits. M3: AS/CPU/FSIZE/NOFILE als harter Boden im Wrapper."""
    lim = {"CORE": 0}
    if res is None:
        return lim
    if res.address_space_bytes is not None:
        lim["AS"] = res.address_space_bytes
    if res.cpu_seconds is not None:
        lim["CPU"] = res.cpu_seconds
    if res.max_file_bytes is not None:
        lim["FSIZE"] = res.max_file_bytes
    if res.max_open_files is not None:
        lim["NOFILE"] = res.max_open_files
    return lim


SYSTEMD_RUN = "/usr/bin/systemd-run"
# Feste Kette unmittelbar vor <iso_exec> <cfg>: env -i python3 -I. Danach iso_exec.py und cfg,
# dann der Separator '--' und das Codex-Binary.
CAGE_FIXED_TAIL = ("/usr/bin/env", "-i", "/usr/bin/python3", "-I")
# In der Mitte (zwischen '--scope' und der festen Kette) erlaubte systemd-Optionen.
_CAGE_MIDDLE_FLAGS = frozenset({"--quiet"})          # ohne Wert
_CAGE_MIDDLE_PAIRS = frozenset({"--unit", "-p"})     # mit genau einem Wert danach


def wrapper_is_tools_cage(argv) -> bool:
    """Prueft STRUKTURELL (nicht per Namenssuche), dass vor dem Codex-Aufruf ein echter Kaefig steht:

        /usr/bin/systemd-run --user --scope [ --quiet | --unit X | -p KEY=VAL ]* \\
            /usr/bin/env -i /usr/bin/python3 -I <abs .../iso_exec.py> <cfg> -- <codex> ...

    Nur exakt diese Form (richtige Reihenfolge, feste Kette direkt vor dem Separator, keine fremden
    Tokens in der Mitte) gilt als Kaefig. python3 -c / sh -c / Decoy-Argumente werden abgelehnt (Review #1)."""
    if not isinstance(argv, (list, tuple)) or CODEX_ARG_SEP not in argv:
        return False
    sep = argv.index(CODEX_ARG_SEP)
    w = list(argv[:sep])
    # Kopf (3) + feste Kette (4) + iso_exec (1) + cfg (1) = Minimum 9 Tokens.
    if len(w) < 9:
        return False
    if w[0] != SYSTEMD_RUN or w[1] != "--user" or w[2] != "--scope":
        return False
    if tuple(w[-6:-2]) != CAGE_FIXED_TAIL:
        return False
    iso = w[-2]
    if not (os.path.isabs(iso) and os.path.basename(iso) == "iso_exec.py"):
        return False
    cfg = w[-1]
    if not (cfg == "{cfg}" or os.path.isabs(cfg)):
        return False
    middle = w[3:-6]
    i = 0
    while i < len(middle):
        tok = middle[i]
        if tok in _CAGE_MIDDLE_FLAGS:
            i += 1
        elif tok in _CAGE_MIDDLE_PAIRS:
            if i + 1 >= len(middle):
                return False
            i += 2                            # Option + genau ein Wert
        else:
            return False                      # fremdes Token (z. B. -c, sh, python3) -> kein Kaefig
    return True


class CagedCodexRunner(SubprocessRunner):
    def __init__(self, workspace_root, *, profile: str = "text", resource_limits=None,
                 auth_file: str = "/home/aria/.codex/auth.json", codex_auth_writable: bool = False,
                 home_root: str = "/home/aria", bus_env: Optional[Mapping[str, str]] = None,
                 systemctl_path: str = "/usr/bin/systemctl", iso_min_abi: int = 4,
                 secret_files: Optional[set] = None, exit_drain_seconds: float = 0.2,
                 bus_check_timeout: float = 5.0, scope_stop_timeout: float = 5.0,
                 state_dir: Optional[str] = None, repo_root: Optional[str] = None,
                 protect_startup: bool = True, write_allowlist=()):
        # resource_limits werden im Kaefig als iso_exec-rlimits durchgesetzt (nicht als prlimit-Praefix);
        # wir halten sie dennoch auf self.resource_limits fuer Introspektion/Kompatibilitaet.
        super().__init__(workspace_root, resource_limits=resource_limits, exit_drain_seconds=exit_drain_seconds)
        if profile not in ("text", "tools"):
            raise ValueError("profile must be text or tools")
        if profile != "tools" and write_allowlist:
            # text hat keinen Home-Zugriff; eine Allowlist dort waere ein stiller Schein-Schalter.
            raise ValueError("write_allowlist is only valid for profile tools")
        if profile == "tools" and getattr(resource_limits, "address_space_bytes", None) is not None:
            # siehe config.py: RLIMIT_AS bricht den V8-Host der Codex-Werkzeuge (SIGTRAP)
            raise ValueError("address_space_bytes is incompatible with profile tools")
        self.profile = profile
        self.auth_file = str(auth_file)
        self.codex_auth_writable = bool(codex_auth_writable)
        self.home_root = os.path.abspath(home_root)
        self.bus_env = dict(bus_env) if bus_env is not None else default_bus_env()
        self.systemctl_path = systemctl_path
        self.iso_min_abi = iso_min_abi
        self._secret_files = secret_files
        self.bus_check_timeout = bus_check_timeout
        self.scope_stop_timeout = scope_stop_timeout
        self.state_dir = os.path.abspath(state_dir) if state_dir else None
        # Default: das laufende V3-Repo sperren (Selbstmodifikation). Explizit None ueberschreibbar.
        self.repo_root = os.path.abspath(repo_root) if repo_root else _PKG_REPO_ROOT
        self.protect_startup = bool(protect_startup)
        # E44 Option 2: die einzigen Home-Ordner mit Schreibrecht (Default leer). Schon hier pruefen
        # (fail-closed beim Bau, WriteAllowlistError ist ein ValueError); je Lauf prueft tools_profile
        # erneut gegen die erst dann bekannten Sperren (Secrets, Cron-Ziele, Hooks, Lauf-Root).
        self.write_allowlist = P.validate_write_allowlist(
            list(write_allowlist), home_root=self.home_root, state_dir=self.state_dir,
            repo_root=self.repo_root, runs_root=os.path.abspath(str(self.workspace_root)))

    # -- bus ----------------------------------------------------------------
    def bus_reachable(self) -> bool:
        try:
            proc = subprocess.run([self.systemctl_path, "--user", "show", "-p", "Version"],
                                  env=dict(self.bus_env), capture_output=True,
                                  timeout=self.bus_check_timeout, check=False)
        except (OSError, subprocess.TimeoutExpired):
            return False
        return proc.returncode == 0 and b"Version=" in proc.stdout

    # -- run dirs / auth / cfg ---------------------------------------------
    def _make_run_dirs(self) -> P.RunDirs:
        root = tempfile.mkdtemp(prefix="codexrun-", dir=str(self.workspace_root))
        sub = {}
        for name in ("ws", "home", "codex-home", "authdir", "tmp"):
            d = os.path.join(root, name)
            os.mkdir(d, 0o700)
            sub[name] = d
        return P.RunDirs(root, sub["ws"], sub["home"], sub["codex-home"], sub["authdir"], sub["tmp"])

    def _stage_auth(self, dirs: P.RunDirs) -> None:
        # read-only KOPIE (nie Symlink auf das echte ~/.codex -> Review M1)
        shutil.copyfile(self.auth_file, dirs.auth_copy)
        os.chmod(dirs.auth_copy, 0o600 if self.codex_auth_writable else 0o400)
        # Verweis im ephemeren CODEX_HOME auf die Kopie (nicht in der Allowlist gelistet -> N4 ok)
        os.symlink(dirs.auth_copy, dirs.codex_home_auth_link)

    def _build_cfg(self, dirs: P.RunDirs, codex_bin: str) -> dict:
        rlimits = _res_to_rlimits(self.resource_limits)
        if self.profile == "tools":
            return P.tools_profile(dirs, codex_bin=codex_bin, rlimits=rlimits, home_root=self.home_root,
                                   writable_auth=self.codex_auth_writable, min_abi=self.iso_min_abi,
                                   secret_files=self._secret_files, state_dir=self.state_dir,
                                   repo_root=self.repo_root, protect_startup=self.protect_startup,
                                   write_allowlist=self.write_allowlist,
                                   runs_root=os.path.abspath(str(self.workspace_root)))
        return P.text_profile(dirs, codex_bin=codex_bin, rlimits=rlimits,
                              writable_auth=self.codex_auth_writable, min_abi=self.iso_min_abi)

    def _compose_argv(self, template_argv, dirs: P.RunDirs, cfg_path: str, unit: str):
        if CODEX_ARG_SEP not in template_argv:
            raise UnsafeCommandError("INVALID_CAGE_ARGV")
        subs = {"{unit}": unit, "{cfg}": cfg_path, "{workdir}": dirs.workspace}
        argv = []
        for a in template_argv:
            for k, v in subs.items():
                a = a.replace(k, v)
            argv.append(a)
        argv = tuple(argv)
        # Kaefig-Pflicht fuer tools; Bypass-Tokens nur dann erlaubt.
        allow_bypass = self.profile == "tools"
        if allow_bypass and not wrapper_is_tools_cage(argv):
            raise UnsafeCommandError("TOOLS_PROFILE_REQUIRES_CAGE")
        assert_safe_argv(argv, allow_bypass_sandbox=allow_bypass)
        return argv

    def _codex_bin_of(self, template_argv) -> str:
        bin_ = template_argv[template_argv.index(CODEX_ARG_SEP) + 1]
        if os.path.basename(bin_).lower() == "claude":
            raise UnsafeCommandError("CODEX_BIN_IS_CLAUDE")
        return os.path.realpath(bin_)

    # -- scope teardown -----------------------------------------------------
    def _systemctl(self, *args) -> None:
        try:
            subprocess.run([self.systemctl_path, "--user", *args], env=dict(self.bus_env),
                           capture_output=True, timeout=self.scope_stop_timeout, check=False)
        except (OSError, subprocess.TimeoutExpired):
            pass

    def stop_scope(self, unit: str) -> None:
        self._systemctl("stop", unit)

    def reset_scope(self, unit: str) -> None:
        self._systemctl("reset-failed", unit)

    # -- run ----------------------------------------------------------------
    def run(self, spec: RunSpec) -> RunResult:
        started = time.monotonic()
        try:
            assert_safe_argv(spec.argv, allow_bypass_sandbox=(self.profile == "tools"))
        except UnsafeCommandError as exc:
            return RunResult(False, exc.code)
        if CODEX_ARG_SEP not in spec.argv:
            return RunResult(False, "INVALID_CAGE_ARGV")
        if not isinstance(spec.stdin, bytes):
            return RunResult(False, "INVALID_STDIN")
        # Review #15: der Kaefig-Runner stellt RunSpec.files NICHT bereit (der Prompt nennt sie aber
        # als attachments/...). Fail-closed statt still ignorieren: keine Anhaenge im tools/text-Kaefig.
        if spec.files:
            return RunResult(False, "ATTACHMENTS_UNSUPPORTED")
        if not self.workspace_root.is_dir():
            return RunResult(False, "WORKSPACE_UNAVAILABLE")
        if not os.path.isfile(self.auth_file):
            return RunResult(False, "AUTH_UNAVAILABLE")
        if not self.bus_reachable():
            # M3: ohne Scope NICHT ungeschuetzt ausfuehren.
            return RunResult(False, "NO_USER_BUS")
        try:
            codex_bin = self._codex_bin_of(spec.argv)
        except UnsafeCommandError as exc:
            return RunResult(False, exc.code)
        try:
            dirs = self._make_run_dirs()
        except OSError:
            return RunResult(False, "WORKSPACE_UNAVAILABLE")
        # explizites .scope-Suffix: ohne es haengt systemctl ".service" an und der Stop
        # trifft den falschen Unit-Typ -> der cgroup-Kill (inkl. Enkel) bliebe aus.
        unit = f"aria-codex-{uuid.uuid4().hex}.scope"
        # Review #9: Toetet das Plugin die Bridge per SIGTERM, muss der separat gestartete Codex-Scope
        # mitgenommen werden. Ein SIGTERM-Handler wirft -> der finally-Block unten stoppt den Scope
        # (und _supervise killt die Prozessgruppe). Ohne Handler liefe der Scope ungekaefigt weiter.
        prev_handler = _MISSING_HANDLER
        try:
            prev_handler = signal.signal(signal.SIGTERM, _raise_terminated)
        except (ValueError, OSError):
            prev_handler = _MISSING_HANDLER   # nicht im Haupt-Thread (z. B. Tests) -> ohne Handler
        try:
            try:
                self._stage_auth(dirs)
            except OSError:
                return RunResult(False, "AUTH_STAGE_FAILED")
            try:
                cfg = self._build_cfg(dirs, codex_bin)
            except P.WriteAllowlistError:
                # E44: ein Allowlist-Eintrag verletzt beim Profilbau eine Sperre (z. B. liegt in einem
                # Cron-Ziel oder wurde inzwischen gegen einen Symlink getauscht) -> nichts starten.
                return RunResult(False, "WRITE_ALLOWLIST_INVALID")
            except Exception:
                return RunResult(False, "PROFILE_BUILD_FAILED")
            cfg_path = os.path.join(dirs.root, "cfg.json")
            with open(cfg_path, "w", encoding="utf-8") as fh:
                json.dump(cfg, fh)
            try:
                argv = self._compose_argv(spec.argv, dirs, cfg_path, unit)
            except UnsafeCommandError as exc:
                return RunResult(False, exc.code)
            try:
                proc = subprocess.Popen(
                    argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                    cwd=dirs.root, env=dict(self.bus_env), shell=False, close_fds=True,
                    start_new_session=True,
                )
            except OSError:
                return RunResult(False, "START_FAILED", duration_seconds=time.monotonic() - started)
            result = self._supervise(proc, spec, started)
            # Review #7: ZUERST den Scope (inkl. aller Enkel) stoppen, DANN last.txt lesen -
            # sonst koennte ein noch laufender setsid-Enkel die Datei gegen einen Symlink tauschen.
            self.stop_scope(unit)
            artifact = self._read_last(dirs, spec.max_output_bytes)
            return dataclasses.replace(result, artifact=artifact)
        except _BridgeTerminated:
            # SIGTERM vom Plugin: der finally-Block stoppt den Scope; strukturierter Fehler.
            return RunResult(False, "BRIDGE_TERMINATED", duration_seconds=time.monotonic() - started)
        finally:
            if prev_handler is not _MISSING_HANDLER:
                try:
                    signal.signal(signal.SIGTERM, prev_handler)
                except (ValueError, OSError):
                    pass
            self.stop_scope(unit)
            self.reset_scope(unit)
            shutil.rmtree(dirs.root, ignore_errors=True)

    def _read_last(self, dirs: P.RunDirs, limit: int) -> bytes:
        """Liest <workspace>/last.txt rennfrei: EIN open mit O_NOFOLLOW|O_NONBLOCK (kein Symlink-
        Folgen), fstat auf den geoeffneten fd (S_ISREG), und der aufgeloeste Pfad muss direkt
        unter dem realpath-Workspace liegen. Wird erst nach stop_scope aufgerufen (Review #7)."""
        ws = os.path.realpath(dirs.workspace)
        path = os.path.join(ws, LAST_TXT)
        try:
            fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC)
        except OSError:
            return b""                                   # u. a. ELOOP, wenn last.txt ein Symlink ist
        try:
            st = os.fstat(fd)
            if not stat.S_ISREG(st.st_mode):
                return b""                               # kein regulaeres File (FIFO/Device/Dir)
            real = os.path.realpath(path)
            if real != path or os.path.dirname(real) != ws:
                return b""                               # Pfad nicht direkt unter dem Workspace
            data = bytearray()
            want = max(1, limit) + 1
            while len(data) < want:
                try:
                    chunk = os.read(fd, want - len(data))
                except BlockingIOError:
                    break
                if not chunk:
                    break
                data.extend(chunk)
            return bytes(data[: max(1, limit)])
        finally:
            os.close(fd)


__all__ = ["CagedCodexRunner", "default_bus_env", "wrapper_is_tools_cage"]