"""Spawn a CLI engine and stream its output as BridgeEvents."""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import tempfile
from collections.abc import AsyncGenerator
from pathlib import Path
from typing import Optional

from atashi_bridge.config import BridgeConfig
from atashi_bridge.engines import build_command
from atashi_bridge.events import BridgeEvent
from atashi_bridge.isolation import ensure_claude_config_home, write_agent_identity
from atashi_bridge.mcp import (
    cleanup_paths,
    ensure_codex_home,
    provision_claude_mcp,
    provision_codex_mcp,
)
from atashi_bridge.transcript import TranscriptWriter

logger = logging.getLogger(__name__)

_ENV_KEY = {
    "claude_code": "ANTHROPIC_API_KEY",
    "codex": "OPENAI_API_KEY",
}

# Directive env (connector REST-fallback vars like CLICKUP_API_KEY) is
# backend-supplied input crossing onto the user's machine. Only plain
# uppercase variables may pass, and never loader/interpreter hooks or
# bridge-owned vars — a compromised backend must not be able to LD_PRELOAD
# the host or redirect an engine's config home.
_EXTRA_ENV_KEY_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$")
_EXTRA_ENV_DENYLIST = {
    "PATH", "HOME", "SHELL", "IFS", "ENV", "BASH_ENV",
    "PYTHONPATH", "PYTHONSTARTUP", "NODE_OPTIONS",
    "CLAUDE_CONFIG_DIR", "CODEX_HOME",
} | set(_ENV_KEY.values())


# ── R2: what the spawned engine inherits from THIS machine ──────────────────
#
# The denylist above filters the SERVER's slice of the child environment. Until
# 2026-08-15 the LOCAL slice was not filtered at all: the spawn site did
# `env = {**os.environ}`, so a headless Claude Code / Codex / agy process
# started on the user's own machine received every variable the daemon had —
# AWS keys, SSH agent sockets, database URLs, unrelated vendor tokens — when
# all it needs is enough environment to find its binary, read its own
# credentials, and reach the network.
#
# No permission tier addresses this. `--sandbox read-only` and
# `--permission-mode plan` constrain what the process may DO; neither changes
# what it was handed at exec time. So this holds at every tier, which is why it
# ships independently of R1.
#
# ALLOWLIST, NOT DENYLIST — the same reasoning that fixed the target guard the
# night before: a denylist must successfully predict every secret name a user
# might export and fails open on the first one it has not met. This fails
# closed, and an unknown-but-needed variable is one documented export away
# rather than a silent leak.
_ENV_ALLOW_EXACT = {
    # Locating and running the binary at all.
    "PATH", "HOME", "SHELL", "USER", "LOGNAME", "TMPDIR", "TERM", "PWD",
    # Text handling. Dropping these turns non-ASCII filenames into mojibake or
    # UnicodeDecodeErrors deep inside the engine.
    "LANG", "LANGUAGE", "LC_ALL", "LC_CTYPE",
    # Corporate networks. Without the proxy and CA variables the engine simply
    # cannot reach its own API, which presents as an unexplained hang.
    "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY",
    "http_proxy", "https_proxy", "all_proxy", "no_proxy",
    "SSL_CERT_FILE", "SSL_CERT_DIR", "REQUESTS_CA_BUNDLE", "NODE_EXTRA_CA_CERTS",
    # XDG config discovery, used by the CLIs' own config resolution.
    "XDG_CONFIG_HOME", "XDG_CACHE_HOME", "XDG_DATA_HOME", "XDG_RUNTIME_DIR",
    # Windows essentials — the bridge is Linux-first today, but omitting these
    # would make any future Windows run fail in a way that looks like a bug in
    # the engine rather than in this list.
    "SYSTEMROOT", "SystemRoot", "APPDATA", "LOCALAPPDATA", "USERPROFILE",
    "COMSPEC", "PATHEXT", "TEMP", "TMP",
}

# Interpreter and loader hooks. These are refused even when an operator names
# them in the passthrough: each one converts "let one more variable through"
# into arbitrary code execution inside the engine process, which is a strictly
# larger grant than the passthrough is meant to give.
_ENV_NEVER_PASSTHROUGH = {
    "LD_PRELOAD", "LD_LIBRARY_PATH", "LD_AUDIT",
    "DYLD_INSERT_LIBRARIES", "DYLD_LIBRARY_PATH",
    "PYTHONPATH", "PYTHONSTARTUP", "NODE_OPTIONS", "BASH_ENV", "ENV", "IFS",
    "CLAUDE_CONFIG_DIR", "CODEX_HOME",
} | set(_ENV_KEY.values())


def _env_allowlist_enabled() -> bool:
    """Kill switch. Fleet flag convention: literal parse, never == "true".

    Defaults ON — the leak is real and present, so the secure state is the
    unattended one. If this stripping ever breaks a live run, recovery is a
    single export rather than a redeploy.
    """
    raw = (os.environ.get("ATASHI_BRIDGE_ENV_ALLOWLIST") or "").strip().lower()
    if raw in ("0", "false", "no", "off"):
        return False
    return True


def build_child_env(
    engine: str,
    api_key: Optional[str],
    extra_env: Optional[dict],
) -> dict:
    """Assemble the environment handed to the spawned engine process.

    Order is deliberate and matches the pre-R2 behaviour exactly: base
    environment first, then the engine's own API key, then the sanitised
    server-supplied extras — so bridge-owned vars keep winning and the extra-env
    denylist keeps applying.
    """
    if _env_allowlist_enabled():
        passthrough = {
            name.strip()
            for name in (os.environ.get("ATASHI_BRIDGE_ENV_PASSTHROUGH") or "").split(",")
            if name.strip()
        } - _ENV_NEVER_PASSTHROUGH
        allowed = _ENV_ALLOW_EXACT | passthrough
        env = {k: v for k, v in os.environ.items() if k in allowed}
    else:
        env = {**os.environ}

    env_key = _ENV_KEY.get(engine)
    if api_key and env_key:
        env[env_key] = api_key
    # Merged before the engine-specific config so bridge-owned vars
    # (CLAUDE_CONFIG_DIR, CODEX_HOME) always win — and they are denylisted
    # in the sanitiser anyway.
    env.update(sanitize_extra_env(extra_env))
    return env


def sanitize_extra_env(extra_env: Optional[dict]) -> dict:
    out: dict[str, str] = {}
    for key, value in (extra_env or {}).items():
        if not isinstance(key, str) or not isinstance(value, str):
            continue
        if not _EXTRA_ENV_KEY_RE.match(key):
            continue
        if key in _EXTRA_ENV_DENYLIST or key.startswith(("LD_", "DYLD_")):
            continue
        out[key] = value
    if out:
        # Key names only — the values are secrets.
        logger.info("directive env injected: %s", ", ".join(sorted(out)))
    return out


# The one directory an unconfigured bridge may work inside. A user-visible
# folder on purpose: the confinement should be somewhere they can see, open and
# put projects in, not a hidden config path they would never think to look at.
ATASHI_HOME_DIRNAME = "Atashi"


def _allowed_roots() -> list[str]:
    """Directories the bridge will run an engine inside.

    Defaults to the user's home. Override with ATASHI_BRIDGE_ALLOWED_ROOTS
    (comma-separated). This is a local jail on the user's own machine: it stops a
    compromised/buggy server from directing the bridge to run in '/', '/etc', etc.
    """
    raw = os.environ.get("ATASHI_BRIDGE_ALLOWED_ROOTS", "")
    if raw.strip():
        # An EXPLICIT grant is a deliberate human act — honoured exactly as given
        # and never narrowed by the default below.
        return [os.path.realpath(os.path.expanduser(r.strip())) for r in raw.split(",") if r.strip()]

    # UNCONFIGURED DEFAULT: the Atashi folder only. Sanjay 2026-08-14 05:54,
    # after we RAN the packaged AppImage and read the daemon's own warning:
    # "let the bridge be confined to the atashi folder itself, no other folder,
    # only inside atashi folder they can work ... and then can be expanded."
    #
    # This used to be $HOME plus the temp dir, which A7.1 measured as permitting
    # ~/.ssh's parent, every agent config directory and $HOME itself. The two
    # extra refusals in validate_workspace narrowed the symptoms; they did not
    # narrow the BOUND. On a fresh install with no grant and no bound folder,
    # every ordinary directory under home was a legal workspace — confirmed live
    # in the packaged app, not inferred from the code.
    #
    # Start closed and widen deliberately: the desktop app adds each folder the
    # user actually binds (ProjectFolderStore -> resolveBridgeAllowedRoots), and
    # a real workspace grant widens it further. Nothing here can widen itself.
    home = Path.home()
    return [
        os.path.realpath(str(home / ATASHI_HOME_DIRNAME)),
        os.path.realpath(tempfile.gettempdir()),
    ]


def validate_workspace(workspace_dir: str) -> str:
    """Resolve + jail the requested workspace. Raises ValueError if outside roots.

    A7.2, 2026-08-01: when no roots are CONFIGURED the default bound is the whole
    home directory, and A7.1 measured what that permits -- ``~/.ssh``, the fleet
    token directories, every agent config directory, and ``$HOME`` itself were all
    valid workspaces. ``/etc`` and ``/`` refused correctly the whole time, which is
    why it read as a working jail.

    So the default bound is now hardened by two extra refusals. They apply ONLY
    when nothing was configured: an EXPLICIT root is a deliberate human grant and
    is honoured exactly as given (granting ``~/.config/thing`` is the operator's
    call). The hardening is a floor for the unconfigured case, never a second
    opinion on a real grant.
    """
    resolved = os.path.realpath(os.path.expanduser(workspace_dir or ""))
    if not os.path.isdir(resolved):
        raise ValueError(f"workspace is not a directory: {workspace_dir!r}")
    roots = _allowed_roots()
    if not any(resolved == r or resolved.startswith(r + os.sep) for r in roots):
        raise ValueError("workspace is outside the bridge's allowed roots")

    if not os.environ.get("ATASHI_BRIDGE_ALLOWED_ROOTS", "").strip():
        # (1) A bare root is not a workspace. Running an engine AT $HOME exposes
        #     every sibling project and every dotfile in one grant.
        if resolved in roots:
            raise ValueError(
                "workspace is the unconfigured default root itself; a workspace must "
                "be a specific project directory inside it. Set "
                "ATASHI_BRIDGE_ALLOWED_ROOTS to scope the jail."
            )
        # (2) No hidden directories. Config and state live in dot-directories --
        #     credentials, agent memory, tokens -- and none of them is a project.
        for root in roots:
            if resolved == root or resolved.startswith(root + os.sep):
                tail = resolved[len(root) :].strip(os.sep)
                if any(part.startswith(".") for part in tail.split(os.sep) if part):
                    raise ValueError(
                        "workspace is inside a hidden directory, which the "
                        "unconfigured default bound refuses (config/state/credential "
                        "directories are not project workspaces). Set "
                        "ATASHI_BRIDGE_ALLOWED_ROOTS to grant it deliberately."
                    )
                break
    return resolved


def _parse_claude_line(line: str) -> Optional[BridgeEvent]:
    try:
        obj = json.loads(line)
    except json.JSONDecodeError:
        return None
    if obj.get("type") == "system" and obj.get("subtype") == "init":
        session_id = obj.get("session_id")
        if session_id:
            return BridgeEvent.session(str(session_id))
        return None
    if obj.get("type") == "assistant":
        parts = obj.get("message", {}).get("content", [])
        text = "".join(p.get("text", "") for p in parts if p.get("type") == "text")
        if text:
            return BridgeEvent.text(text)
        for p in parts:
            if p.get("type") == "tool_use":
                # L11: `input` is the tool call's real arguments, available
                # synchronously on this same block — unlike the result, which
                # arrives on a later, separately-correlated line.
                return BridgeEvent.tool(p.get("name", "tool"), args=p.get("input"))
    if obj.get("type") == "result":
        usage = obj.get("usage") or {}
        it = usage.get("input_tokens")
        if isinstance(it, int):
            return BridgeEvent.usage(it)
        return None
    return None


def _parse_codex_line(line: str) -> Optional[BridgeEvent]:
    """codex exec --json JSONL (v0.137). Non-JSON lines (warnings, banners)
    pass through as text. Unknown JSON event types are protocol noise and are
    dropped; transient `error` events (e.g. reconnect chatter) surface as
    status lines, with only turn.failed treated as terminal."""
    try:
        obj = json.loads(line)
    except json.JSONDecodeError:
        return BridgeEvent.text(line)
    etype = obj.get("type", "")
    if etype == "thread.started":
        thread_id = obj.get("thread_id")
        return BridgeEvent.session(str(thread_id)) if thread_id else None
    if etype in ("item.completed", "item.started", "item.updated"):
        item = obj.get("item") or {}
        itype = item.get("type", "")
        if etype == "item.completed" and itype == "agent_message":
            text = item.get("text") or ""
            return BridgeEvent.text(text) if text else None
        if etype == "item.started" and itype in ("command_execution", "file_change", "mcp_tool_call", "tool_call"):
            name = item.get("command") or item.get("tool") or itype
            return BridgeEvent.tool(str(name))
        return None
    if etype == "turn.failed":
        message = (obj.get("error") or {}).get("message") or "turn failed"
        return BridgeEvent.error(message)
    if etype == "error":
        message = obj.get("message") or "engine error"
        return BridgeEvent.status(message)
    if etype == "turn.completed":
        usage = obj.get("usage") or {}
        it = usage.get("input_tokens")
        if isinstance(it, int):
            return BridgeEvent.usage(it)
        return None
    return None


async def _run_once(
    engine: str,
    prompt: str,
    model: Optional[str],
    safe_workspace: str,
    api_key: Optional[str],
    kind: str,
    resume_session_id: Optional[str],
    mcp_servers: Optional[list],
    extra_env: Optional[dict] = None,
    agent_identity: Optional[dict] = None,
    tier: Optional[str] = None,
) -> AsyncGenerator[BridgeEvent, None]:
    # L11: the local transcript is keyed on L7's stable agent_uuid — never on
    # a name (renamable) or a catalog id (shared across up to five agents,
    # the ATA-GEN-0001 collision class). No id means no transcript, not a
    # write to a folder named after a fabricated identity.
    agent_uuid = (agent_identity or {}).get("id")
    transcript = TranscriptWriter(
        enabled=bool(agent_uuid) and BridgeConfig.load().transcripts_enabled,
        engine=engine,
        agent_id=agent_uuid or "unknown",
    )

    # R2 (2026-08-15): was `env = {**os.environ}` — the whole local environment.
    # See build_child_env for what changed and why an allowlist rather than a
    # denylist. Same ordering guarantees as before.
    env = build_child_env(engine=engine, api_key=api_key, extra_env=extra_env)

    # Ephemeral secret-bearing config provisioned for this run; removed in
    # finally so connector tokens never persist on disk between runs.
    ephemeral: list[str] = []
    mcp_config_path: Optional[str] = None

    if engine == "claude_code":
        # OBS-10: run inside the workspace's Atashi-owned config home so the
        # engine reads the seeded Atashi setup memory, not the host user's
        # personal ~/.claude context. Provisioning failures (e.g. a read-only
        # workspace) fall back to the host config rather than killing the
        # run — the isolation is a quality fix, not a safety gate.
        config_home: Optional[str] = None
        try:
            config_home = ensure_claude_config_home(safe_workspace)
            env["CLAUDE_CONFIG_DIR"] = config_home
            # L7: name the agent in claude_code's own native file.
            write_agent_identity(config_home, agent_identity)
        except OSError as exc:
            logger.warning(
                "could not provision .atashi config home, using host config: %s", exc
            )
        if mcp_servers and config_home:
            try:
                mcp_config_path = provision_claude_mcp(config_home, mcp_servers)
                ephemeral.append(mcp_config_path)
            except OSError as exc:
                logger.warning("could not provision MCP config, run proceeds without: %s", exc)
    elif engine == "codex":
        # Codex twin of OBS-10: relocate CODEX_HOME into the workspace's
        # .atashi folder (host auth.json symlinked in). MCP servers ride
        # config.toml inside it.
        try:
            codex_home = ensure_codex_home(safe_workspace)
            env["CODEX_HOME"] = codex_home
            # L7 part 3: name the agent in codex's own native file. Measured
            # 2026-08-13 with a codeword probe and a negative control — codex
            # reads AGENTS.md out of CODEX_HOME, and the home above is already
            # relocated into the workspace, so this lands per-project for free.
            write_agent_identity(codex_home, agent_identity, filename="AGENTS.md")
            if mcp_servers:
                ephemeral.append(provision_codex_mcp(codex_home, mcp_servers))
        except OSError as exc:
            logger.warning(
                "could not provision .atashi codex home, using host config: %s", exc
            )

    try:
        cmd, stdin_payload = build_command(
            engine,
            prompt=prompt,
            model=model,
            kind=kind,
            resume_session_id=resume_session_id,
            mcp_config_path=mcp_config_path,
            tier=tier,
        )
    except ValueError as exc:
        cleanup_paths(ephemeral)
        yield BridgeEvent.error(str(exc))
        return

    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=safe_workspace,
            env=env,
            # BUG-21: stream-json engines emit the WHOLE result as one line —
            # a generated document blows past asyncio's 64 KB default and
            # readline raises "chunk is longer than limit", killing the run
            # AFTER the engine did the work. 16 MB is allocated lazily.
            limit=16 * 1024 * 1024,
        )
    except FileNotFoundError:
        cleanup_paths(ephemeral)
        yield BridgeEvent.error(f"{cmd[0]} not found on PATH")
        return

    assert proc.stdin is not None
    assert proc.stdout is not None
    assert proc.stderr is not None

    async def _feed_stdin():
        try:
            proc.stdin.write((stdin_payload or "").encode("utf-8"))
            await proc.stdin.drain()
        except (ConnectionResetError, BrokenPipeError):
            pass  # engine exited before reading the whole prompt
        finally:
            proc.stdin.close()

    stdin_task = asyncio.create_task(_feed_stdin())

    async def _drain(stream):
        return await stream.read()

    stderr_task = asyncio.create_task(_drain(proc.stderr))

    try:
        async for raw in proc.stdout:
            line = raw.decode("utf-8", errors="replace").rstrip()
            if not line:
                continue
            if engine == "claude_code":
                event = _parse_claude_line(line)
            elif engine == "codex":
                event = _parse_codex_line(line)
            else:
                event = BridgeEvent.text(line)
            if event:
                # Written from the event stream itself, never a self-report —
                # a no-op when disabled or the writer never got a real id.
                transcript.record(event)
                yield event

        stderr_bytes = await stderr_task
        await proc.wait()
        if proc.returncode != 0:
            msg = stderr_bytes.decode("utf-8", errors="replace").strip() or f"exit {proc.returncode}"
            yield BridgeEvent.error(msg)
            return
        yield BridgeEvent.done()
    finally:
        # Cancel path (generator closed mid-run): kill the engine rather than
        # orphan it. On normal completion returncode is already set and this
        # is a no-op. The reaper task collects the exit status so the dead
        # process never lingers as a zombie.
        if proc.returncode is None:
            proc.kill()
            asyncio.ensure_future(proc.wait())
        for task in (stdin_task, stderr_task):
            if not task.done():
                task.cancel()
        cleanup_paths(ephemeral)


# L3 (2026-08-13 spec, D4): "previous session unavailable" named no cause.
# Keyed on the same reason strings services/work_sessions.py hands back — see
# that module's COLD_START_* constants, not duplicated here as an import
# because the bridge package ships and updates independently of the backend.
#
# Deliberately NOT keyed on "no_session": that cause fires on every genuinely
# first-ever turn (a brand new conversation, or an engine that never ran here)
# — today's silent behaviour is already correct for it, and naming it would
# turn ordinary first turns into a status line that implies something was
# lost when nothing was. Only the two SURPRISING causes — the user's context
# visibly changed out from under them — earn a message.
_COLD_START_MESSAGES = {
    "workspace_changed": "previous session was for a different folder — starting fresh here",
    "engine_changed": "previous session ran on a different engine — starting fresh",
}


async def run_engine(
    engine: str,
    prompt: str,
    model: Optional[str],
    workspace_dir: str,
    api_key: Optional[str],
    kind: str = "run",
    resume_session_id: Optional[str] = None,
    mcp_servers: Optional[list] = None,
    extra_env: Optional[dict] = None,
    cold_start_reason: Optional[str] = None,
    agent_identity: Optional[dict] = None,
    tier: Optional[str] = None,
) -> AsyncGenerator[BridgeEvent, None]:
    try:
        safe_workspace = validate_workspace(workspace_dir)
    except ValueError as exc:
        yield BridgeEvent.error(str(exc))
        return

    # Resume fallback: a stale or wrong-machine session id must degrade to a
    # cold start, not a dead run. A resume attempt that errors BEFORE
    # producing any output retries once without the id; an error after real
    # output is a genuine run failure and surfaces as-is.
    attempts = [resume_session_id, None] if resume_session_id else [None]
    if not resume_session_id and cold_start_reason in _COLD_START_MESSAGES:
        # No resume was ever offered, and the backend's reason is one of the
        # two SURPRISING causes -- no attempt to fail here, so name the cause
        # up front rather than starting silent (see _COLD_START_MESSAGES for
        # why "no_session" / an unrecognised reason stays silent instead).
        yield BridgeEvent.status(_COLD_START_MESSAGES[cold_start_reason])
    for attempt_index, attempt_resume in enumerate(attempts):
        produced_output = False
        resume_failed = False
        gen = _run_once(
            engine,
            prompt=prompt,
            model=model,
            safe_workspace=safe_workspace,
            api_key=api_key,
            kind=kind,
            resume_session_id=attempt_resume,
            mcp_servers=mcp_servers,
            extra_env=extra_env,
            agent_identity=agent_identity,
            tier=tier,
        )
        async for event in gen:
            if event.type in ("text", "tool"):
                produced_output = True
            if (
                event.type == "error"
                and attempt_resume is not None
                and not produced_output
                and attempt_index + 1 < len(attempts)
            ):
                resume_failed = True
                logger.warning("resume of session %s failed (%s); cold starting",
                               attempt_resume, event.payload.get("message", ""))
                break  # closes gen → its finally kills the proc + cleans up
            yield event
        if not resume_failed:
            return
        # A fourth, genuinely local cause: the backend DID offer a resume id
        # (get_resume_session_id found a matching row), but the engine itself
        # rejected it on THIS machine -- invisible to the backend by
        # construction, so it gets its own distinct wording rather than
        # reusing one of the three backend-named causes above.
        yield BridgeEvent.status("previous session could not be resumed — starting fresh")
