#!/usr/bin/env python3
"""Issue #9 — herdr server bootstrap must not inherit the caller's agent-runtime
markers, and query-only skills must never bootstrap a server at all.

The bootstrap lives in the `.mam/shim/herdr` wrapper that lib.sh writes, so every
test drives that shim against a fake `herdr` binary placed first on PATH: it logs
each call, answers `session list --json` with "not running", and dumps its own
environment when asked to run as the `server`.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent
RECONCILE_SH = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
STATUS_SH = REPO_ROOT / ".agents" / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"

FAKE_HERDR = r'''#!/usr/bin/env python3
import json, os, sys
args = sys.argv[1:]
with open(os.environ["FAKE_HERDR_LOG"], "a") as f:
    f.write(json.dumps(args) + "\n")
if args[-1:] == ["server"]:
    with open(os.environ["FAKE_HERDR_ENV_DUMP"], "w") as f:
        json.dump(dict(os.environ), f)
    sys.exit(0)
if args[-3:] == ["session", "list", "--json"]:
    print(json.dumps({"sessions": []}))
    sys.exit(0)
print(json.dumps({"result": {"agents": []}}))
'''

# Per-session runtime markers of the calling agent CLI: none of these may reach
# the daemon (and through it, every pane it later spawns).
RUNTIME_MARKERS = {
    "CLAUDECODE": "1",
    "CLAUDE_CODE_CHILD_SESSION": "1",
    "CLAUDE_CODE_SESSION_ID": "orchestrator-uuid",
    "CLAUDE_CODE_MESSAGING_SOCKET": "/tmp/cc-socks/1.sock",
    "CLAUDE_PID": "12345",
    "ANTIGRAVITY_CONVERSATION_ID": "agy-uuid",
    "HERMES_SESSION_ID": "hermes-uuid",
    "GROK_SESSION_ID": "grok-uuid",
    "OPENCODE_SESSION_ID": "opencode-uuid",
}

# Ordinary configuration must keep flowing through unchanged.
KEPT_CONFIG = {
    "CLAUDE_CONFIG_DIR": "/keep/claude-config",
    "MAM_CUSTOM_SETTING": "kept",
}

# What herdr 0.9 writes to stderr (exit 1) for `agent list` against a stopped
# session server.
REAL_DOWN_ENVELOPE = {
    "id": "cli:agent:list",
    "error": {
        "code": "server_not_running",
        "message": "no herdr server is running at /tmp/sessions/down-srv/herdr.sock; "
                   "run `herdr session attach down-srv` to start or attach it",
    },
}

# A successful list in which the down-server code appears only as data: in an
# agent name (the sanitizer allows underscores) and in unrelated metadata.
LISTING_WITH_SUBSTRING = {"result": {"agents": [
    {"name": "server_not_running-codex", "agent": "codex"},
    {"name": "creator-claude-01", "agent": "claude", "title": "server_not_running"},
]}}


def _install_fake_herdr(sandbox: Path) -> dict:
    fakebin = sandbox / "fakebin"
    fakebin.mkdir(exist_ok=True)
    binary = fakebin / "herdr"
    binary.write_text(FAKE_HERDR)
    binary.chmod(0o755)
    env = dict(os.environ)
    env.update({
        "PATH": f"{fakebin}:{env['PATH']}",
        "FAKE_HERDR_LOG": str(sandbox / "fake_herdr_calls.log"),
        "FAKE_HERDR_ENV_DUMP": str(sandbox / "server_env.json"),
        "AGENT_PYTHON_BIN": sys.executable,
        "DELEGATE_JOB_PYTHON": sys.executable,
    })
    return env


def _fake_agent_list(sandbox: Path, body: str) -> None:
    """Replace the fake's fallback answer (what `agent list` gets) with `body`."""
    fake = sandbox / "fakebin" / "herdr"
    fake.write_text(FAKE_HERDR.replace('print(json.dumps({"result": {"agents": []}}))', body))


def _calls(env: dict) -> list:
    log = Path(env["FAKE_HERDR_LOG"])
    if not log.exists():
        return []
    return [json.loads(line) for line in log.read_text().splitlines() if line.strip()]


def _run_through_shim(sandbox: Path, env: dict, session: str) -> subprocess.CompletedProcess:
    """`herdr agent list` through lib.sh's shim, scoped to an isolated session
    whose server the fake reports as not running."""
    lib = sandbox / ".agents" / "skills" / "lib.sh"
    env = dict(env, HERDR_SESSION_NAME=session)
    return subprocess.run(
        ["bash", "-c", f'source "{lib}" >/dev/null 2>&1; herdr agent list'],
        capture_output=True, text=True, cwd=str(sandbox), env=env, timeout=60,
    )


def _shim_ls(sandbox: Path, env: dict, server: str) -> subprocess.CompletedProcess:
    """The exact `ls` query reconcile.sh sends for each registered server."""
    lib = sandbox / ".agents" / "skills" / "lib.sh"
    return subprocess.run(
        ["bash", "-c", f'source "{lib}" >/dev/null 2>&1; herdr -L {server} ls -F "#{{session_name}}|#{{session_created}}"'],
        capture_output=True, text=True, cwd=str(sandbox), env=env, timeout=60,
    )


def test_bootstrap_strips_agent_runtime_markers(mam_sandbox):
    env = _install_fake_herdr(mam_sandbox)
    env.update(RUNTIME_MARKERS)
    env.update(KEPT_CONFIG)

    res = _run_through_shim(mam_sandbox, env, "issue9-probe")
    assert res.returncode == 0, res.stderr

    server_calls = [c for c in _calls(env) if c[-1:] == ["server"]]
    assert server_calls == [["--session", "issue9-probe", "server"]], _calls(env)

    dump = Path(env["FAKE_HERDR_ENV_DUMP"])
    assert dump.exists(), "server was bootstrapped but never received an environment"
    server_env = json.loads(dump.read_text())

    leaked = sorted(k for k in RUNTIME_MARKERS if k in server_env)
    assert not leaked, f"runtime markers leaked into the herdr server: {leaked}"
    for key, value in KEPT_CONFIG.items():
        assert server_env.get(key) == value, f"{key} must be inherited unchanged"
    # MAM's own configuration is configuration, not a marker.
    assert server_env.get("CLAUDE_PROJECT_DIR") == os.environ["CLAUDE_PROJECT_DIR"]
    for key in ("HOME", "PATH", "WORKSPACE_ROOT"):
        assert key in server_env


def test_no_autostart_skips_bootstrap_but_still_routes_the_call(mam_sandbox):
    env = _install_fake_herdr(mam_sandbox)
    env["MAM_NO_AUTOSTART"] = "1"

    res = _run_through_shim(mam_sandbox, env, "issue9-quiet")
    assert res.returncode == 0, res.stderr

    calls = _calls(env)
    assert not any(c[-1:] == ["server"] for c in calls), f"server must not be started: {calls}"
    assert not Path(env["FAKE_HERDR_ENV_DUMP"]).exists()
    # The command itself still goes to the named session, whose server answers.
    assert ["--session", "issue9-quiet", "agent", "list"] in calls, calls


def test_status_query_never_starts_a_server(mam_sandbox):
    """status.sh sweeps every herdr_session named in the YAML; a stopped
    isolated server must be reported dead, not bootstrapped to answer."""
    env = _install_fake_herdr(mam_sandbox)
    (mam_sandbox / ".mam" / "agent-sessions.yaml").write_text(
        "herdr_sessions:\n"
        "  - name: ghost-creator-claude\n"
        "    status: stopped\n"
        "    herdr_session: ghost-srv\n"
        "    pane:\n"
        "      cwd: /tmp/nonexistent-workspace\n"
    )
    status_sh = mam_sandbox / "skills" / "multi-agent-mux-status" / "scripts" / "status.sh"

    res = subprocess.run(["bash", str(status_sh), "--json"], capture_output=True, text=True,
                         cwd=str(mam_sandbox), env=env, timeout=120)
    assert res.returncode == 0, res.stderr

    calls = _calls(env)
    assert any("--session" in c and "ghost-srv" in c for c in calls), \
        f"the ghost-srv session was never queried: {calls}"
    assert not any(c[-1:] == ["server"] for c in calls), f"status.sh started a server: {calls}"
    assert not Path(env["FAKE_HERDR_ENV_DUMP"]).exists()


def test_reconcile_and_status_opt_out_of_autostart_statically():
    for script in (RECONCILE_SH, STATUS_SH):
        assert "export MAM_NO_AUTOSTART=1" in script.read_text(), \
            f"{script.name} must opt out of the shim's server bootstrap"


def _fail_with(text: str, stream: str) -> str:
    """Fake `agent list` body: write `text` to `stream` and exit 1."""
    return f'sys.{stream}.write({json.dumps(text)} + "\\n"); sys.exit(1)'


# reconcile.sh reads a nonzero `ls` whose stderr contains any of these as a
# confirmed-EMPTY server (`is_empty`); anything else as unknown.
RECONCILE_EMPTY_PHRASES = ("no server running", "no sessions", "failed to connect")

# An unrelated structured failure whose message happens to mention the code.
TRANSPORT_ERROR = json.dumps({"error": {"code": "transport_error",
                                        "message": "socket reset while probing server_not_running"}})


@pytest.mark.parametrize("envelope,stream", [
    ({"error": {"code": "server_not_running"}}, "stderr"),
    (REAL_DOWN_ENVELOPE, "stderr"),
    ({"error": {"code": "server_not_running"}}, "stdout"),
], ids=["minimal", "herdr-0.9", "minimal-on-stdout"])
def test_ls_reports_down_server_as_no_server_running(mam_sandbox, envelope, stream):
    """reconcile.sh treats a nonzero `ls` as 'empty' only when stderr says
    'no server running'; anything else flips herdr_confirmed=False for every
    server. The shim must therefore translate server_not_running explicitly."""
    env = _install_fake_herdr(mam_sandbox)
    env["MAM_NO_AUTOSTART"] = "1"
    _fake_agent_list(mam_sandbox, _fail_with(json.dumps(envelope), stream))

    res = _shim_ls(mam_sandbox, env, "down-srv")
    assert res.returncode != 0
    assert "no server running" in res.stderr, res.stderr
    assert res.stdout.strip() == ""


def test_ls_lists_agents_whose_names_contain_server_not_running(mam_sandbox):
    """A successful list is never a down server, whatever text it carries.
    Matching the bare substring let one agent named server_not_running-codex
    hide every agent its server returned."""
    env = _install_fake_herdr(mam_sandbox)
    env["MAM_NO_AUTOSTART"] = "1"
    _fake_agent_list(mam_sandbox, f"print({json.dumps(json.dumps(LISTING_WITH_SUBSTRING))})")

    res = _shim_ls(mam_sandbox, env, "live-srv")
    assert res.returncode == 0, res.stderr
    assert "no server running" not in res.stderr, res.stderr
    names = [line.split("|", 1)[0] for line in res.stdout.splitlines() if line.strip()]
    assert names == ["server_not_running-codex", "creator-claude-01"], res.stdout


@pytest.mark.parametrize("text,stream", [
    (TRANSPORT_ERROR, "stderr"),
    (TRANSPORT_ERROR, "stdout"),
    ("warning: server_not_running cache entry is stale", "stderr"),
    ("", "stderr"),
], ids=["other-code", "other-code-on-stdout", "plain-text", "empty"])
def test_ls_leaves_other_failures_unknown(mam_sandbox, text, stream):
    """Only the structured code means "down". Any other failure must stay
    unknown - nonzero exit, no empty-server phrase - wherever its error
    document lands. On stdout it used to reach the list parser, which read a
    JSON error without result.agents as a successful EMPTY list (rc 0)."""
    env = _install_fake_herdr(mam_sandbox)
    env["MAM_NO_AUTOSTART"] = "1"
    _fake_agent_list(mam_sandbox, _fail_with(text, stream))

    res = _shim_ls(mam_sandbox, env, "flaky-srv")
    assert res.returncode != 0, f"a failed list must not exit 0: {res.stdout!r} {res.stderr!r}"
    for phrase in RECONCILE_EMPTY_PHRASES:
        assert phrase not in res.stderr.lower(), res.stderr
    assert res.stdout.strip() == ""


def _write_running_rows(sandbox: Path, names, server: str = "live-srv") -> None:
    rows = "".join(
        f"  - name: {name}\n"
        "    status: running\n"
        f"    herdr_session: {server}\n"
        "    pane:\n"
        "      cwd: /tmp/nonexistent-workspace\n"
        for name in names
    )
    (sandbox / ".mam" / "agent-sessions.yaml").write_text("herdr_sessions:\n" + rows)


def _reconcile_dry_run(sandbox: Path, env: dict) -> dict:
    reconcile = sandbox / "skills" / "multi-agent-mux-monitor" / "scripts" / "reconcile.sh"
    res = subprocess.run(["bash", str(reconcile), "--once", "--emit-diff", "--dry-run"],
                         capture_output=True, text=True, cwd=str(sandbox), env=env, timeout=120)
    assert res.returncode == 0, res.stderr
    return json.loads(res.stdout)


def test_reconcile_keeps_live_rows_when_a_name_contains_server_not_running(mam_sandbox):
    """End to end through reconcile.sh: the server answers a successful list in
    which the down-server code appears as data. Both registered rows must stay
    alive with herdr_confirmed intact, so drift A cannot mark them terminated."""
    env = _install_fake_herdr(mam_sandbox)
    _fake_agent_list(mam_sandbox, f"print({json.dumps(json.dumps(LISTING_WITH_SUBSTRING))})")
    _write_running_rows(mam_sandbox, ("server_not_running-codex", "creator-claude-01"))

    report = _reconcile_dry_run(mam_sandbox, env)
    assert report["herdr_confirmed"] is True, report
    alive = set(report["herdr_sessions_alive"])
    assert {"server_not_running-codex|live-srv", "creator-claude-01|live-srv"} <= alive, alive
    drift_a = [d for d in report["drifts"] if d.get("class") == "A"]
    assert not drift_a, drift_a


def test_reconcile_treats_other_list_failures_as_unknown(mam_sandbox):
    """End to end through reconcile.sh: a server whose `agent list` fails with
    an unrelated structured error on stdout must drop herdr_confirmed, so
    drift A never marks its live row terminated."""
    env = _install_fake_herdr(mam_sandbox)
    _fake_agent_list(mam_sandbox, _fail_with(TRANSPORT_ERROR, "stdout"))
    _write_running_rows(mam_sandbox, ("creator-claude-01",))

    report = _reconcile_dry_run(mam_sandbox, env)
    assert report["herdr_confirmed"] is False, report
    drift_a = [d for d in report["drifts"] if d.get("class") == "A"]
    assert not drift_a, drift_a
    terminated = [a for a in report.get("actions", []) if str(a).startswith("terminated")]
    assert not terminated, terminated