#!/usr/bin/env python3
"""Issue #20 - test state stays out of every real control plane.

conftest records the control-plane binding an agent pane carries and drops it
before collection (pytest_configure), gives every test a registry and an
audit-log root under its own tmp_path (_isolate_mam_state_dirs), and a
session guard reports new entries in a real `.mam/jobs` or
`.mam/delegate_job_logs` (mam_state_leak_guard, helpers in _mam_state_guard).
"""
from __future__ import annotations

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

import pytest

import _mam_state_guard as guard

REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"


def _fake_control_plane(root: Path, sessions=()) -> Path:
    """A checkout-shaped tree with a `.mam` registry, audit-log root and session yaml."""
    mam = root / ".mam"
    (mam / "jobs").mkdir(parents=True)
    (mam / "delegate_job_logs").mkdir()
    rows = "".join(f"- name: {s}\n  status: running\n  mcp_attachments:\n  - name: stitch\n" for s in sessions)
    (mam / "agent-sessions.yaml").write_text("herdr_sessions:\n" + rows if rows else "herdr_sessions: []\n")
    return root


def _record(mam: Path, jid: str, session) -> None:
    rec = {"job_id": jid, "prompt": f"prompt of {jid}"}
    if session:
        rec["agent_session"] = session
    (mam / "jobs" / f"{jid}.json").write_text(json.dumps(rec))


def _log(mam: Path, jid: str, session=None) -> None:
    d = mam / "delegate_job_logs" / jid
    d.mkdir()
    if session:
        (d / "meta.json").write_text(json.dumps({"job_id": jid, "agent_session": session}))


# ---------------------------------------------------------------------------
# the session guard (S4, S5)
# ---------------------------------------------------------------------------
def test_s4_guard_flags_only_entries_no_live_delegation_explains(tmp_path):
    main = _fake_control_plane(tmp_path / "main", sessions=("creator-real",))
    mam = main / ".mam"
    _record(mam, "old00001", "herdr:nobody")        # before the snapshot: not judged
    _log(mam, "old00001")
    recorded = {"WORKSPACE_ROOT": str(main)}
    roots = guard.watch_roots(tmp_path / "repo", recorded)
    before = guard.snapshot(roots, recorded)

    _record(mam, "live0001", "herdr:creator-real")  # a live delegation: record first, registered session
    _log(mam, "live0001", "herdr:creator-real")
    _record(mam, "live0002", "remote:creator-real")
    _log(mam, "orphan01", "herdr:creator-real")     # (a) audit log with no record, even for a known session
    _record(mam, "stray001", "herdr:nobody")         # (b) unknown session
    _record(mam, "stitch01", "herdr:stitch")         # (b) `stitch` is an mcp attachment, not a session
    _record(mam, "anon0001", None)                   # (b) no session at all

    found = {Path(f["path"]).name: f for f in guard.leaks(roots, before, recorded)}
    assert sorted(found) == ["anon0001.json", "orphan01", "stitch01.json", "stray001.json"], found
    assert found["orphan01"]["reason"] == "audit log without a registry record"
    assert found["anon0001.json"]["reason"] == "no agent_session"
    assert found["stray001.json"]["reason"] == "agent_session names no registered session"
    assert found["stray001.json"]["prompt"] == "prompt of stray001"
    assert "orphan01" in guard.describe(list(found.values()))


def test_s5_watch_roots_follow_the_recorded_binding(tmp_path):
    repo = _fake_control_plane(tmp_path / "repo")
    main = _fake_control_plane(tmp_path / "main")
    (tmp_path / "main-link").symlink_to(main)
    own, other = str((repo / ".mam").resolve()), str((main / ".mam").resolve())

    assert guard.watch_roots(repo, {}) == [own]
    # WORKSPACE_ROOT naming this checkout is counted once
    assert guard.watch_roots(repo, {"WORKSPACE_ROOT": str(repo)}) == [own]
    # a worktree creator's pane: all four at another checkout, reached through a symlink too
    pane = {"WORKSPACE_ROOT": str(tmp_path / "main-link"),
            "AGENT_SESSIONS_YAML": str(main / ".mam" / "agent-sessions.yaml"),
            "MAM_JOBS_DIR": str(main / ".mam" / "jobs"),
            "DELEGATE_JOB_LOGS_DIR": str(main / ".mam" / "delegate_job_logs")}
    assert guard.watch_roots(repo, pane) == [own, other]
    # a recorded root that does not exist is skipped; the checkout's own .mam never is
    bare = tmp_path / "bare"
    assert guard.watch_roots(bare, {"WORKSPACE_ROOT": str(tmp_path / "gone")}) == [str((bare / ".mam").resolve())]
    # a registry outside any <root>/jobs is watched directly
    odd = tmp_path / "elsewhere" / "registry"
    odd.mkdir(parents=True)
    rec = {"MAM_JOBS_DIR": str(odd)}
    jobs, logs = guard.state_dirs(guard.watch_roots(repo, rec), rec)
    assert str(odd.resolve()) in jobs and str(odd.resolve()) not in logs


# ---------------------------------------------------------------------------
# per-test defaults and the dropped binding (S1-S3, S6, S7)
# ---------------------------------------------------------------------------
# Copied at import, like test_orc_onboard.SCRUBBED_ENV: the binding has to be
# gone before collection, not merely before the first test (S6).
_IMPORT_ENV = dict(os.environ)
PROBE_ENV = "MAM_ISSUE20_PROBE"
BINDING = ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "MAM_JOBS_DIR", "DELEGATE_JOB_LOGS_DIR")


def _tree(root: Path) -> dict:
    return {str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() if p.is_file() else None
            for p in sorted(root.rglob("*"))}


def _binding(root: Path) -> dict:
    """The four variables a worktree creator's pane carries, aimed at `root`."""
    mam = root / ".mam"
    return {"WORKSPACE_ROOT": str(root), "AGENT_SESSIONS_YAML": str(mam / "agent-sessions.yaml"),
            "MAM_JOBS_DIR": str(mam / "jobs"), "DELEGATE_JOB_LOGS_DIR": str(mam / "delegate_job_logs")}


def test_s1_every_test_defaults_to_state_dirs_under_tmp_path(tmp_path, monkeypatch):
    from lib_py.delegation import mqtt_common

    state = tmp_path / ".mam-test-state"
    assert os.environ["MAM_JOBS_DIR"] == str(state / "jobs")
    assert os.environ["DELEGATE_JOB_LOGS_DIR"] == str(state / "delegate_job_logs")
    # a WORKSPACE_ROOT on a checkout with a .mam no longer decides where state goes
    monkeypatch.setenv("WORKSPACE_ROOT", str(_fake_control_plane(tmp_path / "repo")))
    assert mqtt_common.default_registry_dir() == str(state / "jobs")
    assert mqtt_common.get_logs_dir() == str(state / "delegate_job_logs")


def test_s2_mam_sandbox_overrides_the_default(mam_sandbox):
    assert os.environ["MAM_JOBS_DIR"] == str(mam_sandbox / ".mam" / "jobs")
    assert os.environ["DELEGATE_JOB_LOGS_DIR"] == str(mam_sandbox / ".mam" / "delegate_job_logs")


def test_s3_submit_with_workspace_root_on_a_checkout_leaves_its_mam_alone(tmp_path):
    """The issue's acceptance check in small: WORKSPACE_ROOT names a checkout
    with a real-looking .mam and the test sets nothing else."""
    repo = _fake_control_plane(tmp_path / "repo", sessions=("creator-real",))
    before = _tree(repo)
    env = dict(os.environ, WORKSPACE_ROOT=str(repo), MAM_ENV_FILE=str(tmp_path / ".mam.env.absent"),
               PYTHONPATH=f"{SKILLS}:{os.environ.get('PYTHONPATH', '')}")
    res = subprocess.run(
        [sys.executable, "-m", "lib_py.delegation.cli", "submit", "--prompt", "issue 20 probe",
         "--agent", "claude-code", "--agent-session", "herdr:issue20-probe"],
        capture_output=True, text=True, env=env, cwd=str(REPO_ROOT), timeout=60,
    )
    assert res.returncode == 0, res.stderr
    state = tmp_path / ".mam-test-state"
    records = list((state / "jobs").glob("*.json"))
    assert len(records) == 1, records
    assert (state / "delegate_job_logs" / records[0].stem).is_dir()
    assert _tree(repo) == before


def test_s6_a_pane_binding_is_dropped_before_collection(tmp_path):
    """A pytest started with a worktree creator pane's binding - all four at
    another checkout F - sees none of it, not even in a module-level copy of
    os.environ, and leaves F untouched."""
    fake_main = _fake_control_plane(tmp_path / "main", sessions=("creator-real",))
    before = _tree(fake_main)
    env = dict(os.environ, **_binding(fake_main), **{PROBE_ENV: "1"})
    # An explicit basetemp: the inner run must not clean up the outer one's.
    res = subprocess.run(
        [sys.executable, "-m", "pytest", f"{__file__}::test_s6_probe", "-q", "-p", "no:cacheprovider",
         f"--basetemp={tmp_path / 'inner'}"],
        capture_output=True, text=True, env=env, cwd=str(REPO_ROOT), timeout=300,
    )
    assert res.returncode == 0, res.stdout[-4000:] + res.stderr[-2000:]
    assert "1 passed" in res.stdout, res.stdout[-2000:]
    assert _tree(fake_main) == before


@pytest.mark.skipif(not os.environ.get(PROBE_ENV), reason="the inner half of S6 (nested pytest) only")
def test_s6_probe(tmp_path):
    assert not [k for k in BINDING if k in _IMPORT_ENV], {k: _IMPORT_ENV[k] for k in BINDING if k in _IMPORT_ENV}
    assert "AGENT_SESSIONS_YAML" not in os.environ
    assert "WORKSPACE_ROOT" not in os.environ  # it named another checkout
    state = tmp_path / ".mam-test-state"
    assert os.environ["MAM_JOBS_DIR"] == str(state / "jobs")
    assert os.environ["DELEGATE_JOB_LOGS_DIR"] == str(state / "delegate_job_logs")


def test_s7_lib_sh_takes_the_session_registry_from_a_tests_workspace_root(tmp_path):
    """An ambient AGENT_SESSIONS_YAML (a worktree pane's) would win over the
    test's WORKSPACE_ROOT in lib.sh; conftest dropped it before collection."""
    assert "AGENT_SESSIONS_YAML" not in os.environ
    repo = tmp_path / "repo"
    repo.mkdir()
    script = f'source "{SKILLS / "lib.sh"}" >/dev/null 2>&1; printf %s "$AGENT_SESSIONS_YAML"'
    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
                         env=dict(os.environ, WORKSPACE_ROOT=str(repo)), timeout=60)
    assert res.stdout == str(repo / ".mam" / "agent-sessions.yaml"), (res.stdout, res.stderr)