"""Bounded session launcher: runs `claude --bare -p` on an executor node over SSH, streams stream-json to a local raw file,
normalises events into the state store, enforces a wall-clock limit by killing the remote process (matched by session id),
and supports resumption of an existing session. Never logs the API key.

Nested-invocation guard: the executor is given no Agent tool and the launcher refuses to start if the environment variable
PGAR_DEPTH is already set (the controller runs at depth 0; executors run on another host at depth 1; nothing goes deeper).
"""
from __future__ import annotations
import json, os, shlex, subprocess, threading, time
from .contracts import utcnow, new_session_id
from .events import normalise, parse_result_line

DEFAULT_TOOLS = ["Bash", "Read", "Write", "Edit", "Glob", "Grep"]
EXECUTOR_ENTRYPOINT = "claude"   # allow-listed executor entry point

class LaunchError(RuntimeError):
    pass

def build_command(workspace: str, ctl_dir: str, prompt_file: str, system_append_file: str, model: str, session_id: str,
                  max_budget_usd: float, max_turns: int, resume: bool = False, tools=None, python_env="~/wp29/.env", effort=None) -> list[str]:
    """Return the remote argv (list) for the executor process. The API key is sourced from the node-side env file."""
    tools = tools or DEFAULT_TOOLS
    argv = [EXECUTOR_ENTRYPOINT, "--bare", "-p", f"$(cat {shlex.quote(prompt_file)})",
            "--output-format", "stream-json", "--verbose",
            "--model", model,
            "--tools", ",".join(tools), "--allowedTools", *tools,
            "--permission-prompts", "none",
            "--max-budget-usd", f"{max_budget_usd:.2f}", "--max-turns", str(max_turns),
            "--append-system-prompt-file", system_append_file,
            "--no-chrome", "--disable-slash-commands"]
    if effort:
        argv += ["--effort", effort]
    if resume:
        argv += ["--resume", session_id]
    else:
        argv += ["--session-id", session_id]
    return argv

def remote_shell_line(workspace: str, argv: list[str], env_file="~/wp29/.env") -> str:
    """Compose the remote shell line. `$(cat prompt)` must stay unquoted so the shell expands it; everything else is quoted."""
    parts = []
    for a in argv:
        parts.append(a if a.startswith("$(cat ") else shlex.quote(a))
    cmd = " ".join(parts)
    return (f"cd {shlex.quote(workspace)} && export PATH=/opt/homebrew/bin:/usr/local/bin:$PATH PGAR_DEPTH=1 "
            f"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 DISABLE_AUTOUPDATER=1 && set -a && . {env_file} && set +a && exec {cmd}")

def run_session(node: str, workspace: str, argv: list[str], raw_path: str, run_id: str, session_id: str, store, wall_seconds: int,
                seq_start: int = 0, on_event=None, env_file="~/wp29/.env") -> dict:
    """Execute one session; return {'exit_code','result','timed_out','n_events','seq_end','started_utc','ended_utc'}."""
    if os.environ.get("PGAR_DEPTH"):
        raise LaunchError("refusing to launch: PGAR_DEPTH is set (nested controller)")
    line = remote_shell_line(workspace, argv, env_file)
    ssh = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", "-o", "ServerAliveInterval=30", node, line]
    started = utcnow(); t0 = time.monotonic()
    os.makedirs(os.path.dirname(raw_path), exist_ok=True)
    proc = subprocess.Popen(ssh, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)
    result = {}; seq = seq_start; timed_out = False; stderr_buf = []
    def _drain_err():
        for l in proc.stderr:
            stderr_buf.append(l)
    threading.Thread(target=_drain_err, daemon=True).start()
    with open(raw_path, "a", encoding="utf-8") as raw:
        while True:
            if time.monotonic() - t0 > wall_seconds:
                timed_out = True
                kill_remote(node, session_id)
                try: proc.wait(timeout=15)
                except subprocess.TimeoutExpired: proc.kill()
                break
            l = proc.stdout.readline()
            if not l:
                if proc.poll() is not None:
                    break
                time.sleep(0.05); continue
            raw.write(l); raw.flush()
            seq += 1
            ev = normalise(l, run_id, seq, time.monotonic() - t0)
            if ev:
                ev["session_id"] = ev.get("session_id") or session_id
                store.add_event(ev)
                if on_event:
                    try: on_event(ev)
                    except Exception: pass  # noqa: BLE001
            r = parse_result_line(l)
            if r: result = r
        # drain remainder after process end
        if not timed_out:
            for l in proc.stdout:
                raw.write(l); seq += 1
                ev = normalise(l, run_id, seq, time.monotonic() - t0)
                if ev:
                    ev["session_id"] = ev.get("session_id") or session_id; store.add_event(ev)
                r = parse_result_line(l)
                if r: result = r
    code = proc.poll()
    ended = utcnow()
    err = "".join(stderr_buf)[-2000:]
    return {"exit_code": code, "result": result, "timed_out": timed_out, "seq_end": seq, "started_utc": started, "ended_utc": ended,
            "wall_s": round(time.monotonic() - t0, 1), "stderr_tail": err}

def kill_remote(node: str, session_id: str) -> int:
    """Kill the executor process on the node by matching its unique session id on the command line (and children)."""
    cmd = f"pkill -f {shlex.quote(session_id)}; sleep 1; pkill -9 -f {shlex.quote(session_id)}; true"
    return subprocess.run(["ssh", "-o", "BatchMode=yes", node, cmd], capture_output=True, timeout=60).returncode

def ssh(node: str, cmd: str, timeout=600, check=False) -> subprocess.CompletedProcess:
    r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", node, cmd], capture_output=True, text=True, timeout=timeout)
    if check and r.returncode != 0:
        raise LaunchError(f"ssh {node} failed ({r.returncode}): {cmd[:200]}\n{r.stderr[-800:]}")
    return r

def rsync_to(node: str, src_dir: str, dst_dir: str, excludes=()) -> None:
    cmd = ["rsync", "-a", "--delete", "-e", "ssh -o BatchMode=yes"]
    for e in excludes: cmd += ["--exclude", e]
    cmd += [src_dir.rstrip("/") + "/", f"{node}:{dst_dir.rstrip('/')}/"]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    if r.returncode != 0:
        raise LaunchError(f"rsync to {node} failed: {r.stderr[-500:]}")

def rsync_from(node: str, src_dir: str, dst_dir: str, excludes=()) -> bool:
    os.makedirs(dst_dir, exist_ok=True)
    cmd = ["rsync", "-a", "-e", "ssh -o BatchMode=yes"]
    for e in excludes: cmd += ["--exclude", e]
    cmd += [f"{node}:{src_dir.rstrip('/')}/", dst_dir.rstrip("/") + "/"]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    return r.returncode == 0