"""Orchestrate the interceptor and agent Docker containers for a spens session."""

from __future__ import annotations

import fnmatch
import os
import re
import subprocess
import sys
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from spens import events, pricing, ui
from spens.builder import (
    INTERCEPTOR_PROXY_ALIAS,
    agent_user,
    build_agent_image,
    build_interceptor_image,
    expand_agent_path,
    generate_interceptor_config,
)
from spens.config import CONFIG_FILENAME, load_config
from spens.sessions import SPENS_DIR_NAME
from spens.sinks import FileSink, get_sink
from spens.summarizer import write_summary
from spens.templates import LOCAL_TEMPLATES_DIRNAME, load_agent, load_environment


class SpensUsageError(ValueError):
    """Invalid CLI usage / arguments (exit code 1, no session state created)."""


# A --session-id is interpolated into docker CLI arguments and object names
# (containers, networks, volume), so it must be a short, lowercase,
# dash-safe slug.
SESSION_ID_RE = r"^[a-z0-9][a-z0-9-]{0,31}$"

# How long to wait for the interceptor's readiness markers before giving up.
INTERCEPTOR_READY_TIMEOUT_SECS = 60
# ``SPENS_DIR_NAME`` (imported above) is the directory spens uses on the HOST
# to store session state: interceptor traces, nono audit/rollback data and the
# per-session interceptor config.  The whole workspace is bind-mounted into the
# agent container, which would otherwise expose this directory -- letting the
# untrusted agent read the logs/traces of its own (and previous) sessions and
# potentially exfiltrate that data.  We overlay it with an empty tmpfs inside
# the agent container so the host's ``.spens`` is never visible to the agent,
# while the host copy is left untouched.
#
# The workspace is bind-mounted read-write into the agent at this path.
AGENT_WORKSPACE_MOUNTPOINT = "/workspace"
AGENT_SPENS_MOUNTPOINT = f"{AGENT_WORKSPACE_MOUNTPOINT}/{SPENS_DIR_NAME}"
# Path of the readiness markers inside the interceptor container. The addon's
# `running()` hook writes its marker once it is configured and the proxy is
# listening; the DNS forwarder writes its marker once it has bound port 53.
# Both live in the shared cert volume so the agent sees them at /certs/.
ADDON_READY_MARKER = "/root/.mitmproxy/spens_addon_ready"
DNS_READY_MARKER = "/root/.mitmproxy/spens_dns_ready"


def _docker_path(p: Path) -> str:
    """Convert a Path to a Docker-compatible volume mount string."""
    return str(p).replace("\\", "/")


# Path tokens accepted in agent-template ``configuration`` ``source``
# values. ``{workspace}`` expands to the resolved ``--workspace`` directory;
# ``{spens_dir}`` to the session's spens directory (``--spens-dir`` when
# given, else ``<workspace>/.spens``). Like the ``{prompt}`` placeholder in
# ``yolo_command``, braces keep the value JSON-friendly and hard to typo
# silently.
WORKSPACE_SOURCE_TOKEN = "{workspace}"
SPENS_DIR_SOURCE_TOKEN = "{spens_dir}"


def _expand_source_path(
    source: str,
    workspace_path: Path | None,
    spens_dir: Path | None,
) -> Path:
    """Expand a ``configuration`` ``source`` into a host ``Path``.

    ``{workspace}`` and ``{spens_dir}`` tokens are replaced with the
    corresponding directories; using a token without providing its directory
    is an error rather than a silent skip. Relative sources (no token, not
    absolute, not ``~``-relative) resolve against ``workspace_path`` when
    given, else against the process CWD (legacy behavior).
    """
    for token, path in (
        (WORKSPACE_SOURCE_TOKEN, workspace_path),
        (SPENS_DIR_SOURCE_TOKEN, spens_dir),
    ):
        if token not in source:
            continue
        if path is None:
            raise ValueError(
                f"configuration source '{source}' uses {token}, but no "
                "corresponding directory was provided"
            )
        source = source.replace(token, str(path))
    expanded = Path(source).expanduser()
    if not expanded.is_absolute() and workspace_path is not None:
        expanded = workspace_path / expanded
    return expanded


def _configuration_mounts(
    configuration: list[dict[str, Any]],
    agent_home: str = "/home/spens",
    workspace_path: Path | None = None,
    spens_dir: Path | None = None,
) -> list[tuple[str, str]]:
    """Resolve runtime configuration mounts from an agent template.

    ``configuration`` is a list of entries, each mapping a host ``source``
    directory to an in-container ``destination``. ``source`` supports the
    ``{workspace}`` and ``{spens_dir}`` tokens (see ``_expand_source_path``);
    plain relative sources resolve against ``workspace_path`` when given.
    Files matching ``include`` patterns (or every file in the directory when
    ``include`` is omitted) are bind-mounted individually, minus anything
    matching ``exclude`` patterns. An entry whose ``source`` does not exist
    on the host is skipped, and only files that actually exist are mounted.

    When several entries resolve files to the same container ``destination``
    path (e.g. a global ``~/.pi/agent`` entry and a local ``{spens_dir}/.pi/agent``
    override of the same file), the entry listed FIRST in the configuration
    wins and later duplicates are skipped -- a container path is only ever
    mounted once, from one host file.

    ``destination`` paths beginning with ``~`` are expanded to the agent
    user's home directory (legacy ``/root`` paths likewise).

    Returns a list of ``(host_path, container_path)`` tuples ready to turn into
    read-only ``docker -v`` arguments.
    """
    mounts: list[tuple[str, str]] = []
    claimed_destinations: set[str] = set()
    for entry in configuration:
        source = _expand_source_path(entry["source"], workspace_path, spens_dir)
        if not source.is_dir():
            continue
        destination = expand_agent_path(entry["destination"], agent_home).rstrip("/")
        include = entry.get("include") or []
        exclude = entry.get("exclude") or []

        if include:
            candidates = {
                p.relative_to(source) for pat in include for p in source.glob(pat)
            }
        else:
            candidates = {
                p.relative_to(source) for p in source.iterdir() if p.is_file()
            }

        for rel in sorted(candidates, key=lambda r: r.as_posix()):
            if not rel.parts:
                continue
            rel_posix = rel.as_posix()
            if any(fnmatch.fnmatch(rel_posix, pat) for pat in exclude):
                continue
            host_file = (source / rel).resolve()
            if host_file.is_file():
                container_path = f"{destination}/{rel_posix}"
                if container_path in claimed_destinations:
                    # An earlier configuration entry already mounted this
                    # container path; configuration order is authoritative
                    # (first entry wins), so skip the duplicate rather than
                    # bind-mounting the same container path twice.
                    continue
                claimed_destinations.add(container_path)
                mounts.append((_docker_path(host_file), container_path))
    return mounts


def _run(cmd: list[str], **kw: Any) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, check=False, **kw)


def _run_checked(cmd: list[str], error: str) -> subprocess.CompletedProcess:
    """Run a docker command, raising ``RuntimeError`` with stderr on failure.

    Every setup step aborts the session on failure, so the check belongs with
    the call rather than repeated at each site.
    """
    result = _run(
        cmd, capture_output=True, text=True, encoding="utf-8", errors="replace"
    )
    if result.returncode != 0:
        raise RuntimeError(f"{error}: {result.stderr}")
    return result


def _wait_for_interceptor_ready(
    container: str, timeout: int = INTERCEPTOR_READY_TIMEOUT_SECS
) -> bool:
    """Wait until the mitmproxy addon AND the DNS forwarder are ready.

    Both write readiness markers into the cert volume: the addon once it is
    loaded, configured and listening (waiting on the CA cert file alone is
    not enough -- the cert can exist before the addon is active), and the DNS
    forwarder once it has bound port 53.  Gating the agent on both markers
    means no agent traffic (HTTP or DNS) can leave before interception works.
    """
    deadline = time.time() + timeout
    while time.time() < deadline:
        result = _run(
            [
                "docker", "exec", container,
                "test", "-f", ADDON_READY_MARKER,
                "-a", "-f", DNS_READY_MARKER,
            ],
            capture_output=True,
        )
        if result.returncode == 0:
            return True
        time.sleep(1)
    return False


def _network_ip(container: str, network: str) -> str | None:
    """Return ``container``'s IPv4 address on ``network`` (None if unknown).

    Used to point the agent's resolv.conf (``docker run --dns``) at the
    interceptor's DNS forwarder.
    """
    fmt = f'{{{{(index .NetworkSettings.Networks "{network}").IPAddress}}}}'
    result = _run(
        ["docker", "inspect", "-f", fmt, container],
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    if result.returncode == 0:
        ip = (result.stdout or "").strip()
        if ip:
            return ip
    return None


def _interceptor_network_args(internal_net: str) -> list[str]:
    """docker-run flags placing the interceptor on the internal network.

    The alias lets the agent reach the proxy at a session-independent name
    (``http://spens-interceptor:9090``) via Docker's embedded DNS.
    """
    return ["--network", internal_net, "--network-alias", INTERCEPTOR_PROXY_ALIAS]


def _agent_spens_mask_args(
    workspace_path: Path,
    spens_dir: Path,
) -> list[str]:
    """docker-run flags masking the spens session directory from the agent.

    The workspace is bind-mounted read-write into the agent, so without this
    the agent could read (and exfiltrate) the host's session traces, nono
    audit data and interceptor config stored in the spens directory.  This
    matters both for the default ``<workspace>/.spens`` and for a custom
    ``--spens-dir`` that happens to live inside the workspace.  Overlaying an
    empty tmpfs at the in-container path hides the host directory from the
    agent while leaving the host copy intact; anything the agent writes there
    is ephemeral and discarded when the container exits.  The tmpfs is
    hardened (``noexec,nosuid,nodev``) and made world-writable (mode ``1777``)
    so the unprivileged agent user can still create files inside it.

    ``/workspace/.spens`` is always masked (a stale default directory may
    still exist there); a custom ``--spens-dir`` inside the workspace is
    masked at its own mountpoint too.  A spens directory outside the
    workspace is not reachable through the agent's mounts at all.
    """
    mountpoints = {AGENT_SPENS_MOUNTPOINT}
    try:
        rel = spens_dir.resolve().relative_to(workspace_path)
    except ValueError:
        # spens dir lives outside the workspace; not visible to the agent.
        pass
    else:
        if rel.parts:
            mountpoints.add(f"{AGENT_WORKSPACE_MOUNTPOINT}/{rel.as_posix()}")
    return [
        arg
        for mountpoint in sorted(mountpoints)
        for arg in ("--tmpfs", f"{mountpoint}:rw,noexec,nosuid,nodev,mode=1777")
    ]


def _agent_config_mask_args(
    workspace_path: Path,
    config: dict[str, Any] | None,
) -> list[str]:
    """docker-run flags masking spens' own config inputs from the agent.

    The workspace is bind-mounted read-write into the agent, and it is also
    where spens reads its trust-sensitive inputs:

      * ``.spens.config.json`` -- egress mode, domain rules,
        ``pre_sandbox_commands``, secret-injection rules;
      * the ``templates/`` override directory -- agent/environment templates
        that control the container user (potentially uid 0!), host
        configuration mounts (``~/.ssh``, ``~/.aws``, ...) and build commands;
      * the ``nono_override`` profile named by the config -- the raw nono
        sandbox policy.

    spens has already consumed all of these (baking them into the built image,
    the generated nono profile and the interceptor config) by the time the
    agent starts, so the agent never needs to see them.  Left visible and
    writable, an untrusted agent -- or a cloned malicious repo -- could rewrite
    the very policy the *next* session honors: downgrade egress to ``legacy``,
    clear the domain rules, add ``pre_sandbox_commands``, define an
    environment with ``agent_user`` uid 0, or an agent whose ``configuration``
    block mounts ``~/.ssh`` / ``~/.aws`` from the host.

    So, exactly like the ``.spens`` masking, we overlay each input with a
    read-only empty mask inside the agent container -- ``/dev/null`` for
    files, an empty hardened tmpfs for directories.  The host copies are left
    untouched; the agent simply cannot read them, and anything it writes to
    the masked path is ephemeral and discarded when the container exits, so it
    cannot persist a poisoned policy for a later run.
    """
    args: list[str] = []

    def _container_path(target: Path) -> str | None:
        """Return the in-agent path for ``target`` if it lives in the workspace."""
        try:
            rel = target.resolve().relative_to(workspace_path)
        except ValueError:
            # Input resolved outside the workspace (e.g. an absolute or
            # ``..`` nono_override path).  It is not reachable through the
            # agent's ``/workspace`` mount, so there is nothing to mask.
            return None
        if not rel.parts:
            return None
        return f"{AGENT_WORKSPACE_MOUNTPOINT}/{rel.as_posix()}"

    # File inputs: bind-mount /dev/null read-only so the agent sees an empty,
    # unwritable file in place of the real policy.
    file_inputs = [workspace_path / CONFIG_FILENAME]
    if config and config.get("nono_override"):
        file_inputs.append(workspace_path / config["nono_override"])
    for target in file_inputs:
        if not target.is_file():
            continue
        container_path = _container_path(target)
        if container_path:
            args.extend(["-v", f"/dev/null:{container_path}:ro"])

    # Directory inputs: overlay an empty hardened tmpfs, like ``.spens``.
    dir_inputs = [workspace_path / LOCAL_TEMPLATES_DIRNAME]
    for target in dir_inputs:
        if not target.is_dir():
            continue
        container_path = _container_path(target)
        if container_path:
            args.extend([
                "--tmpfs",
                f"{container_path}:rw,noexec,nosuid,nodev,mode=1777",
            ])

    return args


def _agent_network_args(internal_net: str, dns_ip: str | None) -> list[str]:
    """docker-run flags isolating the agent on the internal network.

    The network is ``--internal``, so the agent has no routed egress: its
    only reachable neighbor is the interceptor, and every byte it sends
    flows through mitmproxy (domain rules, secret substitution, capture).
    NET_RAW is dropped to block raw sockets / ARP spoofing toward the
    interceptor (cost: ``ping`` no longer works).  ``--dns`` points the
    agent's resolv.conf at the interceptor's logging DNS forwarder so name
    resolution is captured as well.
    """
    args = ["--network", internal_net, "--cap-drop", "NET_RAW"]
    if dns_ip:
        args.extend(["--dns", dns_ip])
    return args


def _stop_and_remove(container: str) -> None:
    _run(["docker", "stop", container], capture_output=True, timeout=15)
    _run(["docker", "rm", "-f", container], capture_output=True, timeout=15)


def _inject_env_mapping(
    config: dict[str, Any] | None,
) -> tuple[list[tuple[str, str]], list[tuple[str, str]]]:
    """Compute ``(agent_envs, interceptor_envs)`` for ``inject_headers`` rules.

    Each rule has a ``placeholder`` (the literal string the agent is given as
    its "key") and an ``env_var`` (the host environment variable holding the
    real secret).  The agent container receives the placeholder as its env var
    value so the real secret never enters the agent.  The interceptor receives
    the real value and substitutes the placeholder at the proxy layer.
    """
    agent_envs: list[tuple[str, str]] = []
    interceptor_envs: list[tuple[str, str]] = []
    if config:
        for rule in config.get("inject_headers", []):
            placeholder = rule.get("placeholder", "")
            env_var = rule.get("env_var", "")
            if env_var:
                interceptor_envs.append((env_var, os.environ.get(env_var, "")))
            if placeholder:
                agent_envs.append((placeholder, placeholder))
    return agent_envs, interceptor_envs


# ---------------------------------------------------------------------------
# Session scaffolding
# ---------------------------------------------------------------------------


@dataclass
class SessionNames:
    """Names of the Docker objects backing one session.

    Every object is suffixed with the session id so concurrent runs cannot
    collide, and so cleanup can remove exactly this session's objects.
    """

    session_id: str
    internal_net: str
    egress_net: str
    cert_volume: str
    interceptor: str
    agent: str

    @classmethod
    def for_session(cls, session_id: str) -> SessionNames:
        return cls(
            session_id=session_id,
            internal_net=f"spens-internal-{session_id}",
            egress_net=f"spens-egress-{session_id}",
            cert_volume=f"spens-certs-{session_id}",
            interceptor=f"spens-interceptor-{session_id}",
            agent=f"spens-agent-{session_id}",
        )


@dataclass
class SessionPaths:
    """Host-side directories and files holding one session's state."""

    session_dir: Path
    traces_dir: Path
    audit_dir: Path
    interceptor_config: Path


def _prepare_session_paths(
    spens_root: Path, session_id: str, config: dict[str, Any] | None
) -> SessionPaths:
    """Create the session's host directories and write the interceptor config."""
    session_dir = spens_root / "sessions" / session_id
    paths = SessionPaths(
        session_dir=session_dir,
        traces_dir=session_dir / "traces",
        audit_dir=session_dir / "nono-audit",
        interceptor_config=session_dir / "interceptor_config.json",
    )
    paths.traces_dir.mkdir(parents=True, exist_ok=True)
    paths.audit_dir.mkdir(parents=True, exist_ok=True)
    # The agent (an unprivileged user in the container) must be able to write
    # its nono audit/rollback state into this bind-mounted directory.
    try:
        paths.audit_dir.chmod(0o777)
    except OSError:
        events.emit(
            "warning",
            message=(
                f"[spens] Warning: could not make {paths.audit_dir} world-writable; "
                "the agent may fail to write its nono audit state."
            ),
        )
    paths.interceptor_config.write_text(
        generate_interceptor_config(config), encoding="utf-8"
    )
    return paths


def _warn_about_egress_mode(legacy_egress: bool) -> None:
    """Warn loudly when egress enforcement has been turned off in config."""
    if not legacy_egress:
        return
    events.emit(
        "warning",
        message=(
            "[spens] Warning: egress mode 'legacy' -- the agent shares the "
            "interceptor's network namespace with UNRESTRICTED internet access. "
            "Proxy routing is advisory only: unsetting HTTPS_PROXY, raw sockets, "
            "git-over-SSH or DNS traffic bypass capture and policy entirely. "
            "Remove \"egress\": \"legacy\" from .spens.config.json to restore "
            "enforcement."
        ),
    )


def _warn_about_inject_headers(
    config: dict[str, Any] | None, interceptor_envs: list[tuple[str, str]]
) -> None:
    """Warn early about ``inject_headers`` rules that cannot substitute.

    Both cases surface downstream as an opaque auth failure from the API
    provider, so they are called out before the agent starts rather than left
    to be diagnosed from a 401.
    """
    # A missing host env var means the interceptor has no real secret to
    # substitute the placeholder with.
    for env_var, value in interceptor_envs:
        if not value:
            events.emit(
                "warning",
                message=(
                    f"[spens] Warning: inject_headers env var '{env_var}' is not set on the "
                    "host. The placeholder will NOT be replaced by the interceptor, and API "
                    "requests using it will fail authentication. Export the variable in the "
                    "shell running spens, or fix 'env_var' in .spens.config.json."
                ),
            )

    # Secrets are only substituted on URLs a rule's for_domains authorizes, so
    # a rule without any authorizes nothing.
    for rule in (config or {}).get("inject_headers", []):
        placeholder = rule.get("placeholder", "")
        if placeholder and not rule.get("for_domains"):
            events.emit(
                "warning",
                message=(
                    f"[spens] Warning: inject_headers rule (placeholder '{placeholder}') "
                    "has no 'for_domains' patterns. Secrets are only substituted on "
                    "authorized URLs, so this placeholder will NEVER be substituted. "
                    "Add a for_domains list (e.g. [\"*api.anthropic.com*\"]) in "
                    ".spens.config.json."
                ),
            )


def _create_session_networks(names: SessionNames) -> None:
    """Create the session's two networks (enforced egress).

    ``internal`` holds the agent + interceptor only; Docker drops all routed
    egress from ``--internal`` networks, so the agent's only reachable
    neighbor is the interceptor.  ``egress`` is the interceptor's (and only
    the interceptor's) path to the internet.
    """
    for name, extra in ((names.internal_net, ["--internal"]), (names.egress_net, [])):
        _run_checked(
            ["docker", "network", "create", *extra, name],
            f"Failed to create network {name}",
        )


def _start_interceptor(
    names: SessionNames,
    paths: SessionPaths,
    tag: str,
    interceptor_envs: list[tuple[str, str]],
    legacy_egress: bool,
) -> str | None:
    """Start the interceptor container and return its internal-network IP.

    The returned IP is what the agent's resolv.conf is pointed at, so its DNS
    is captured too; None means DNS falls back to Docker's resolver.
    """
    # The "[spens] Starting interceptor ..." line is emitted by the caller
    # as the ``interceptor_starting`` event (it doubles as the state
    # transition marker), so it is not printed here.
    cmd = [
        "docker", "run", "-d",
        "-i", "-t",
        "--name", names.interceptor,
        "-v", f"{names.cert_volume}:/root/.mitmproxy",
        "-v", f"{_docker_path(paths.traces_dir)}:/app/traces",
        "-v", f"{_docker_path(paths.interceptor_config)}:/app/spens_interceptor_config.json:ro",
    ]
    if not legacy_egress:
        cmd.extend(_interceptor_network_args(names.internal_net))

    # Forward the env vars backing inject_headers to the interceptor.  The
    # agent receives the *placeholder* string as its env var value (e.g.
    # ANTHROPIC_API_KEY=ANTHROPIC_API_KEY) so the real secret never enters the
    # agent container; the interceptor holds the real value and substitutes it
    # at the proxy layer.
    for env_var, value in interceptor_envs:
        cmd.extend(["-e", f"{env_var}={value}"])

    cmd.append(tag)
    _run_checked(cmd, "Failed to start interceptor")

    if legacy_egress:
        return None

    # Dual-home the interceptor: created on the internal (agent-facing)
    # network above, now attached to the egress network for upstream internet
    # access.  Done before the readiness wait so no agent traffic can arrive
    # before upstream connectivity exists.
    _run_checked(
        ["docker", "network", "connect", names.egress_net, names.interceptor],
        "Failed to attach interceptor to egress network",
    )

    dns_ip = _network_ip(names.interceptor, names.internal_net)
    if dns_ip:
        events.emit(
            "info",
            message=(
                f"[spens] Interceptor reachable as {INTERCEPTOR_PROXY_ALIAS} "
                f"({dns_ip}) on the internal network; agent DNS will be captured."
            ),
        )
    else:
        events.emit(
            "warning",
            message=(
                "[spens] Warning: could not determine the interceptor's internal "
                "IP; agent DNS falls back to Docker's embedded resolver "
                "(uncaptured). HTTP egress is still enforced."
            ),
        )
    return dns_ip


def _build_agent_command(
    names: SessionNames,
    paths: SessionPaths,
    *,
    tag: str,
    workspace_path: Path,
    spens_root: Path,
    agent: dict[str, Any],
    agent_home: str,
    config: dict[str, Any] | None,
    inject_agent_envs: list[tuple[str, str]],
    dns_ip: str | None,
    legacy_egress: bool,
    prompt: str | None,
    tty: bool = True,
) -> list[str]:
    """Assemble the ``docker run`` command line for the agent container.

    ``tty=True`` (the interactive case the spec exempts) runs ``docker run
    -it`` with the agent's terminal passed straight through.  Non-interactive
    output modes (jsonl / background) drop ``-it`` so the container's
    stdout/stderr can be collected by the spens process and re-emitted as
    ``agent_output`` events.
    """
    cmd: list[str] = [
        "docker", "run",
        *(("-it",) if tty else ()),
        "--init",
        "--name", names.agent,
        "--security-opt", "seccomp:unconfined",
        "-v", f"{names.cert_volume}:/certs:ro",
        "-v", f"{_docker_path(workspace_path)}:{AGENT_WORKSPACE_MOUNTPOINT}",
        # Hide the host's spens session directory (traces/audit/config) from
        # the untrusted agent by overlaying it with an empty tmpfs.
        *_agent_spens_mask_args(workspace_path, spens_root),
        # Hide spens' own trust-sensitive config inputs (.spens.config.json,
        # templates/, the nono_override profile) so the agent can neither read
        # them nor persistently tamper with the policy the next session honors.
        *_agent_config_mask_args(workspace_path, config),
        "-v", f"{_docker_path(paths.audit_dir)}:{agent_home}/.local/state/nono",
    ]

    if legacy_egress:
        # advisory-only: shared netns, unrestricted egress (see warning)
        cmd.extend(["--network", f"container:{names.interceptor}"])
    else:
        # enforced: isolated internal network, no routed egress, captured DNS
        cmd.extend(_agent_network_args(names.internal_net, dns_ip))

    # Mount agent configuration files from the host, read-only: the agent runs
    # untrusted code, so it must not be able to tamper with (or exfiltrate by
    # rewriting) the host's configuration files.
    for host, dest in _configuration_mounts(
        agent.get("configuration") or [],
        agent_home,
        workspace_path=workspace_path,
        spens_dir=spens_root,
    ):
        cmd.extend(["-v", f"{host}:{dest}:ro"])

    # Forward environment variables listed in .spens.config.json
    for var_name in (config or {}).get("env") or []:
        cmd.extend(["-e", f"{var_name}={os.environ.get(var_name, '')}"])

    # Placeholder env vars for secret substitution -- the agent gets the
    # placeholder string, the interceptor holds the real value.
    for var_name, placeholder_val in inject_agent_envs:
        cmd.extend(["-e", f"{var_name}={placeholder_val}"])

    # Yolo mode: the entrypoint runs the agent non-interactively via the
    # configured yolo_command when this is set.
    if prompt:
        cmd.extend(["-e", f"SPENS_YOLO_PROMPT={prompt}"])

    cmd.append(tag)
    return cmd


def _cleanup_session(names: SessionNames, legacy_egress: bool) -> None:
    """Remove the session's containers, networks and volume (best effort)."""
    _stop_and_remove(names.interceptor)
    _stop_and_remove(names.agent)
    if not legacy_egress:
        # Both containers are gone, so the networks have no endpoints left;
        # tolerate failure (stale per-session networks are harmless).
        _run(
            ["docker", "network", "rm", names.internal_net, names.egress_net],
            capture_output=True,
            timeout=15,
        )
    _run(["docker", "volume", "rm", names.cert_volume], capture_output=True, timeout=15)


def _run_agent_streaming(cmd: list[str]) -> int:
    """Run the agent container attached, re-emitting its output as events.

    Used in the non-interactive output modes (jsonl / background): the
    container runs without ``-it``, its combined stdout/stderr is read
    line-by-line and each line becomes an ``agent_output`` event -- so the
    output lands in ``events.jsonl`` regardless of CLI renderer, and in
    jsonl mode also streams to the caller's stdout.
    """
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        encoding="utf-8",
        errors="replace",
        bufsize=1,
    )
    assert proc.stdout is not None
    for line in proc.stdout:
        events.emit("agent_output", chunk=line.rstrip("\n"))
    return proc.wait()


def _report_session_end(session_id: str, paths: SessionPaths, exit_code: int) -> None:
    """Write the session summary and emit the finished event.

    The summary travels in the event as a JSON dict (``data["summary"]``),
    not as pre-formatted text: machine consumers (jsonl output, attach,
    orchestrators) read the structured numbers, and the tty sink renders the
    recap panel from the same dict.  ``message`` stays a single line -- the
    multi-line recap that used to live there made ``events.jsonl`` noisy and
    forced every consumer to parse prose.
    """
    location = f"Traces: {paths.traces_dir}  Audit: {paths.audit_dir}"
    try:
        summary_path, summary = write_summary(paths.session_dir)
    except Exception as exc:  # noqa: BLE001 -- summary failure is not fatal
        events.emit(
            "finished",
            exit_code=exit_code,
            message=f"[spens] Session {session_id} finished. {location}",
        )
        events.emit(
            "warning",
            message=f"[spens] Warning: could not generate session summary: {exc}",
        )
        return
    events.emit(
        "finished",
        exit_code=exit_code,
        summary_path=str(summary_path),
        summary=summary,
        message=(
            f"[spens] Session {session_id} finished. {location}  "
            f"Summary: {summary_path}"
        ),
    )


# ---------------------------------------------------------------------------
# Session entry point
# ---------------------------------------------------------------------------


def validate_session_id(session_id: str) -> None:
    """Validate an explicit --session-id (interpolated into docker arguments)."""
    if not re.fullmatch(SESSION_ID_RE, session_id):
        raise SpensUsageError(
            f"Invalid --session-id '{session_id}': must match {SESSION_ID_RE} "
            "(lowercase letters, digits and dashes, at most 32 characters)."
        )


def _validate_yolo_mode(prompt: str | None, agent: dict[str, Any]) -> None:
    if prompt and not agent.get("yolo_command"):
        raise SpensUsageError(
            f"Agent '{agent['name']}' has no 'yolo_command' configured and "
            "cannot run in yolo mode. Add a 'yolo_command' (with a {prompt} "
            "placeholder) to the agent template."
        )


def validate_non_interactive(
    output_mode: str,
    prompt: str | None,
    agent: dict[str, Any],
    accept_changes: bool,
    reject_changes: bool,
) -> None:
    """Enforce the non-interactive-mode prerequisites BEFORE any session state.

    In yolo mode with neither change flag, the nono save/rollback prompt
    baked into the agent entrypoint would block forever in a
    non-interactive container -- so an explicit decision is mandatory.
    """
    if output_mode not in ("jsonl", "background"):
        return
    if not prompt:
        raise SpensUsageError(
            f"--output {output_mode} requires a prompt argument (yolo mode): "
            "there is no interactive terminal to drive the agent with."
        )
    _validate_yolo_mode(prompt, agent)
    if accept_changes == reject_changes:
        raise SpensUsageError(
            f"--output {output_mode} requires exactly one of "
            "--accept-changes / --reject-changes: without an explicit "
            "decision the nono save/rollback prompt would block the "
            "non-interactive session forever."
        )


@dataclass
class _SessionBoot:
    """Everything the session loop needs after the session is booted."""

    session_id: str
    names: SessionNames
    paths: SessionPaths
    env: dict[str, Any]
    agent: dict[str, Any]
    config: dict[str, Any] | None
    workspace_path: Path
    spens_root: Path
    legacy_egress: bool
    started: bool


def _boot_session(
    env_name: str,
    agent_name: str,
    workspace: str | Path,
    *,
    prompt: str | None,
    accept_changes: bool,
    reject_changes: bool,
    spens_dir: str | Path | None,
    session_id: str | None,
    output_mode: str,
    reuse_session_id: bool = False,
) -> _SessionBoot:
    """Load inputs, validate, create session state and emit ``started``.

    Shared by the foreground session loop (:func:`run_session`) and the
    background-mode parent (:mod:`spens.background`), which boots the
    session (creating ``state.json`` / ``events.jsonl`` and the ``started``
    event) and then spawns the detached child that reuses it.

    Raises before creating any session state on invalid usage, so the CLI
    can report a bare ``error`` event with no session files left behind.
    """
    workspace_path = Path(workspace).resolve()
    if not workspace_path.is_dir():
        raise FileNotFoundError(f"Workspace not found: {workspace_path}")

    env = load_environment(env_name, workspace_path)
    agent = load_agent(agent_name, workspace_path)
    config = load_config(workspace_path)

    _validate_yolo_mode(prompt, agent)
    validate_non_interactive(
        output_mode, prompt, agent, accept_changes, reject_changes
    )

    if session_id:
        validate_session_id(session_id)
    sid = session_id or uuid.uuid4().hex[:16]
    names = SessionNames.for_session(sid)

    # Egress enforcement mode (specs/enforced_egress.md).  Default (enforced):
    # the agent runs on an isolated ``--internal`` Docker network whose only
    # other member is the interceptor, so ALL egress is forced through
    # mitmproxy and non-proxied protocols fail closed.  "legacy": the agent
    # shares the interceptor's network namespace with unrestricted internet
    # access; HTTP_PROXY env vars are then advisory only.
    legacy_egress = bool(config) and config.get("egress") == "legacy"

    # Costing the session normally refreshes the Portkey pricing data from
    # the network (on the HOST, after the agent has exited).  A workspace can
    # turn that off with ``"pricing": {"live_fetch": false}`` and be costed
    # from the vendored snapshot instead.
    pricing_config = config.get("pricing") if isinstance(config, dict) else None
    if isinstance(pricing_config, dict) and pricing_config.get("live_fetch") is False:
        pricing.set_live_fetch(False)

    # Session state lives under <spens_dir>/sessions/<id>, defaulting to
    # <workspace>/.spens.  A custom --spens-dir keeps it out of the workspace.
    spens_root = Path(spens_dir).resolve() if spens_dir else workspace_path / SPENS_DIR_NAME

    if session_id and not reuse_session_id:
        existing = spens_root / "sessions" / session_id
        if existing.exists():
            raise SpensUsageError(
                f"Session id '{session_id}' already exists ({existing}); "
                "session ids are never overwritten. Choose a new --session-id."
            )

    # Wire the sinks: FileSink is ALWAYS installed (events.jsonl is the
    # single source of truth in every mode); --output only selects the
    # additional CLI renderer (none in background mode).
    session_dir = spens_root / "sessions" / sid
    base = {
        "state": "building",
        "session_id": sid,
        "exit_code": None,
        "agent_container": names.agent,
        "interceptor_container": names.interceptor,
        "started_at": events.utc_now_iso(),
        "updated_at": events.utc_now_iso(),
        "prompt": prompt or "",
        "env": env_name,
        "agent": agent_name,
    }
    sinks: list[events.Sink] = [FileSink(session_dir, base)]
    renderer = get_sink(output_mode)
    if renderer is not None:
        sinks.append(renderer)
    events.configure(events.Emitter(sid, sinks))

    paths = _prepare_session_paths(spens_root, sid, config)

    started = not reuse_session_id
    if started:
        events.emit(
            "started",
            session_dir=str(paths.session_dir),
            traces_dir=str(paths.traces_dir),
            audit_dir=str(paths.audit_dir),
        )

    # With the emitter wired, surface the config warnings (these used to
    # print before any session state existed).
    _warn_about_egress_mode(legacy_egress)
    inject_agent_envs, inject_interceptor_envs = _inject_env_mapping(config)
    _warn_about_inject_headers(config, inject_interceptor_envs)

    return _SessionBoot(
        session_id=sid,
        names=names,
        paths=paths,
        env=env,
        agent=agent,
        config=config,
        workspace_path=workspace_path,
        spens_root=spens_root,
        legacy_egress=legacy_egress,
        started=started,
    )


def run_session(
    env_name: str,
    agent_name: str,
    workspace: str | Path,
    no_cache: bool = False,
    prompt: str | None = None,
    accept_changes: bool = False,
    spens_dir: str | Path | None = None,
    reject_changes: bool = False,
    output_mode: str = "tty",
    session_id: str | None = None,
    reuse_session_id: bool = False,
) -> int:
    """Build the images, run the agent against ``workspace``, and clean up.

    Returns the agent's exit code (1 if the session could not be set up,
    130 on SIGINT).  ``reuse_session_id`` marks a background-mode child that
    reuses an already-booted session (its ``started`` event and state were
    written by the parent).
    """
    boot = _boot_session(
        env_name, agent_name, workspace,
        prompt=prompt,
        accept_changes=accept_changes,
        reject_changes=reject_changes,
        spens_dir=spens_dir,
        session_id=session_id,
        output_mode=output_mode,
        reuse_session_id=reuse_session_id,
    )
    session_id = boot.session_id
    names = boot.names
    paths = boot.paths
    legacy_egress = boot.legacy_egress
    agent_home = agent_user(boot.env)["home"]

    agent_tag = f"spens-agent:{boot.env['name']}-{boot.agent['name']}"
    interceptor_tag = "spens-interceptor:latest"

    inject_agent_envs, inject_interceptor_envs = _inject_env_mapping(boot.config)

    exit_code = 1
    try:
        _run_checked(
            ["docker", "volume", "create", names.cert_volume],
            f"Failed to create volume {names.cert_volume}",
        )

        # In tty mode the docker build output is streamed into a
        # constrained live panel (spens.ui) instead of flooding the
        # terminal; jsonl/background keep redirecting build stdout to
        # stderr so the stdout event stream stays clean.
        agent_progress = (
            ui.BuildProgress(title=f"agent image {agent_tag}")
            if output_mode == "tty" else None
        )
        build_agent_image(
            agent_tag, boot.env, boot.agent, boot.config, boot.workspace_path,
            no_cache=no_cache,
            accept_changes=accept_changes,
            reject_changes=reject_changes,
            stdout=None if output_mode == "tty" else sys.stderr,
            progress=agent_progress,
        )
        interceptor_progress = (
            ui.BuildProgress(title=f"interceptor image {interceptor_tag}")
            if output_mode == "tty" else None
        )
        build_interceptor_image(
            interceptor_tag, no_cache=no_cache,
            stdout=None if output_mode == "tty" else sys.stderr,
            progress=interceptor_progress,
        )

        if not legacy_egress:
            _create_session_networks(names)

        events.emit(
            "interceptor_starting",
            message=(
                f"[spens] Starting interceptor for session {session_id} ..."
            ),
        )
        dns_ip = _start_interceptor(
            names, paths, interceptor_tag, inject_interceptor_envs, legacy_egress
        )

        # Gate on the readiness markers (written once the addon is configured
        # and the proxy is listening, and once the DNS forwarder has bound
        # port 53), NOT the CA cert, so the agent cannot start before
        # interception actually works.
        events.emit(
            "info",
            message="[spens] Waiting for interceptor addon and DNS forwarder ...",
        )
        if not _wait_for_interceptor_ready(names.interceptor):
            raise TimeoutError(
                "Timed out waiting for the interceptor (addon + DNS forwarder) "
                "to become ready"
            )
        events.emit(
            "interceptor_ready",
            message=(
                "[spens] Interceptor ready (addon + DNS forwarder up, "
                "proxy listening)."
            ),
        )

        agent_cmd = _build_agent_command(
            names, paths,
            tag=agent_tag,
            workspace_path=boot.workspace_path,
            spens_root=boot.spens_root,
            agent=boot.agent,
            agent_home=agent_home,
            config=boot.config,
            inject_agent_envs=inject_agent_envs,
            dns_ip=dns_ip,
            legacy_egress=legacy_egress,
            prompt=prompt,
            tty=output_mode == "tty",
        )

        events.emit(
            "agent_started",
            container=names.agent,
            message=(
                f"[spens] Starting agent {agent_name} in {env_name} ..."
            ),
        )
        if prompt:
            events.emit("info", message=f"[spens] Yolo mode: {prompt}")

        try:
            # Interactive exemption: in tty mode the agent's docker -it
            # stream (a TUI) is passed through to the terminal directly.
            exit_code = (
                _run(agent_cmd).returncode
                if output_mode == "tty"
                else _run_agent_streaming(agent_cmd)
            )
        except KeyboardInterrupt:
            # Terminal ``canceled`` event before exiting, so status readers
            # see a coherent terminal state rather than a stalled one.  The
            # recap still renders afterwards (via the finished event), but
            # the first terminal *state* wins: state.json keeps ``canceled``.
            events.emit(
                "canceled",
                reason="sigint",
                message="\n[spens] Interrupted, cleaning up ...",
            )
            exit_code = 130

        events.emit("agent_exited", exit_code=exit_code)

    except Exception as exc:  # noqa: BLE001
        events.emit("error", message=f"[spens] Error: {exc}")
        exit_code = 1

    finally:
        _cleanup_session(names, legacy_egress)

    _report_session_end(session_id, paths, exit_code)
    return exit_code
