"""Issue #23 + #24 — the create_session start path.

Four regressions live here, in the order of the issues:

1. a herdr call aimed at another server must not carry the caller pane's own
   workspace/pane/tab ids (#23 comments: reproduced on Linux and macOS);
2. blocking start-up modals are agent-adapter knowledge and are answered while
   waiting for the TUI (#24);
3. onboarding delivery is judged by the agent's own `started` event (#23);
4. a rollback verifies the kill and leaves no `running` row or dangling job (#23).

What is mocked and what is real is stated per test. The dialog tests fix OUR
contract - which pane string gets which keys - not that the keys drive a real
agy/grok/codex/opencode TUI; no such TUI can be started here.
"""

import json
import os
import subprocess
import textwrap

import pytest


# --------------------------------------------------------------------------
# helpers
# --------------------------------------------------------------------------

def _run_lib(tmp_path, body, env=None):
    """Source the sandbox lib.sh and run `body`. Unlike the shim-contract
    helper this keeps HERDR_* exactly as `env` sets it: these tests are about
    what the caller's pane leaks in."""
    lib_path = tmp_path / ".agents" / "skills" / "lib.sh"
    script = 'source "%s"\n_init_herdr_isolation\n%s\n' % (lib_path, body)
    run_env = dict(os.environ)
    for var in ("HERDR_SESSION", "HERDR_SESSION_NAME", "HERDR_WORKSPACE_ID",
                "HERDR_PANE_ID", "HERDR_TAB_ID"):
        run_env.pop(var, None)
    run_env.update(env or {})
    return subprocess.run(["bash", "-c", script], capture_output=True,
                          text=True, cwd=str(tmp_path), env=run_env)


def _mock_state(mock_herdr):
    with open(mock_herdr) as f:
        return json.load(f)


def _write_mock_state(mock_herdr, state):
    with open(mock_herdr, "w") as f:
        json.dump(state, f, indent=2)


def _seed_agent(mock_herdr, tmp_path, name, workspace_id, pane_id):
    state = _mock_state(mock_herdr)
    state.setdefault("workspaces", [])
    if not any(w.get("workspace_id") == workspace_id for w in state["workspaces"]):
        state["workspaces"].append(
            {"workspace_id": workspace_id, "label": workspace_id, "cwd": str(tmp_path)})
    state.setdefault("agents", {})[name] = {
        "name": name, "agent": "agy", "status": "running", "cwd": str(tmp_path),
        "workspace_id": workspace_id, "pane_id": pane_id, "pid": 9999,
        "command": "agy", "buffer": "Ready",
    }
    state.setdefault("panes", []).append(
        {"pane_id": pane_id, "workspace_id": workspace_id, "label": name, "buffer": "Ready"})
    _write_mock_state(mock_herdr, state)


def _install_env_recorder(tmp_path, monkeypatch):
    """A herdr earlier in PATH than the mock that dumps the env it was called
    with, then execs the mock. The shim resolves the first non-shim herdr on
    PATH, so this observes exactly what reached the real binary."""
    rec_dir = tmp_path / "recorder"
    rec_dir.mkdir(parents=True, exist_ok=True)
    env_log = tmp_path / "herdr_env.ndjson"
    real = tmp_path / "bin" / "herdr"
    script = rec_dir / "herdr"
    script.write_text(textwrap.dedent('''\
        #!/usr/bin/env python3
        import json, os, sys
        with open(%r, "a") as f:
            f.write(json.dumps({
                "argv": sys.argv[1:],
                "HERDR_WORKSPACE_ID": os.environ.get("HERDR_WORKSPACE_ID"),
                "HERDR_PANE_ID": os.environ.get("HERDR_PANE_ID"),
                "HERDR_TAB_ID": os.environ.get("HERDR_TAB_ID"),
                "HERDR_SESSION": os.environ.get("HERDR_SESSION"),
                "HERDR_SESSION_NAME": os.environ.get("HERDR_SESSION_NAME"),
            }) + "\\n")
        os.execv(%r, [%r] + sys.argv[1:])
        ''') % (str(env_log), str(real), str(real)))
    script.chmod(0o755)
    monkeypatch.setenv("PATH", "%s:%s" % (rec_dir, os.environ["PATH"]))
    return env_log


def _recorded(env_log):
    if not os.path.exists(env_log):
        return []
    return [json.loads(l) for l in open(env_log) if l.strip()]


# --------------------------------------------------------------------------
# item 1 — caller pane scope must not leak across servers
# --------------------------------------------------------------------------

CROSS = {"HERDR_SESSION": "default", "HERDR_SESSION_NAME": "mam-isolated",
         "HERDR_WORKSPACE_ID": "w5", "HERDR_PANE_ID": "w5:p2",
         "HERDR_TAB_ID": "w5:t1"}


def test_s1_cross_server_call_drops_the_caller_panes_scope_ids(
        mam_sandbox, mock_herdr, mock_agents, monkeypatch):
    """S1 (mock herdr): the caller sits in another server's pane. has-session,
    kill-session and agent prompt must find the agent, and the three ids must
    not reach herdr."""
    tmp_path = mam_sandbox
    env_log = _install_env_recorder(tmp_path, monkeypatch)
    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")

    res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01', env=CROSS)
    assert res.returncode == 0, res.stdout + res.stderr

    res = _run_lib(tmp_path, 'herdr agent prompt creator-agy-01 "hello"', env=CROSS)
    assert res.returncode == 0, res.stdout + res.stderr

    res = _run_lib(tmp_path, 'herdr kill-session -t creator-agy-01', env=CROSS)
    assert res.returncode == 0, res.stdout + res.stderr

    calls = _recorded(env_log)
    assert calls, "the recorder saw no herdr call"
    for call in calls:
        assert call["HERDR_WORKSPACE_ID"] is None, call
        assert call["HERDR_PANE_ID"] is None, call
        assert call["HERDR_TAB_ID"] is None, call

    # the kill really happened, i.e. this is not a silent no-op
    assert "creator-agy-01" not in _mock_state(mock_herdr).get("agents", {})


def test_s1b_the_delegate_job_wrappers_bare_herdr_is_the_lib_sh_function(
        mam_sandbox, mock_herdr, mock_agents, monkeypatch):
    """S1b (mock herdr): the wrapper sources lib.sh and then calls a bare
    `herdr` (wrapper :918 preflight). A shell function beats PATH, so that call
    is mam_herdr and is fixed by item 1 too - this is why #23's preflight
    failure is cured by the same change."""
    tmp_path = mam_sandbox
    _install_env_recorder(tmp_path, monkeypatch)
    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")

    res = _run_lib(tmp_path, 'echo "kind=$(type -t herdr)"\nherdr has-session -t creator-agy-01',
                   env=CROSS)
    assert "kind=function" in res.stdout, res.stdout + res.stderr
    assert res.returncode == 0, res.stdout + res.stderr


def test_s2_same_server_keeps_the_explicit_workspace_scope(
        mam_sandbox, mock_herdr, mock_agents, monkeypatch):
    """S2 (mock herdr): on the target's own server the exported scope still
    scopes - a same-named agent in another workspace is not addressed (H-22)."""
    tmp_path = mam_sandbox
    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
    same = {"HERDR_SESSION": "mam-isolated", "HERDR_SESSION_NAME": "mam-isolated",
            "HERDR_WORKSPACE_ID": "w2"}

    before = len(_mock_state(mock_herdr).get("calls", []))
    res = _run_lib(tmp_path, 'herdr agent prompt creator-agy-01 "hello from w2"', env=same)
    assert res.returncode != 0, res.stdout + res.stderr
    new = _mock_state(mock_herdr).get("calls", [])[before:]
    prompts = [c for c in new if len(c) >= 2 and c[0] == "agent" and c[1] == "prompt"]
    assert prompts == [], new


def test_s2b_a_caller_outside_any_pane_keeps_the_explicit_scope(
        mam_sandbox, mock_herdr, mock_agents):
    """S2b (mock herdr): no HERDR_SESSION means the caller is not in a pane at
    all, so an exported HERDR_WORKSPACE_ID is a deliberate scope, not a leak.
    This is the case H-18/H-21/H-23 fix."""
    tmp_path = mam_sandbox
    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")

    res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01',
                   env={"HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"})
    assert res.returncode == 1, "an explicit out-of-scope id must still hide the agent"

    res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01',
                   env={"HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w1"})
    assert res.returncode == 0, res.stdout + res.stderr


def test_s3_shim_contract_regressions_still_pass(mam_sandbox):
    """S3: H-18 / H-21 / H-23 are the tests that fix today's scoping meaning.
    Run them explicitly as this change's regression guard."""
    repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    res = subprocess.run(
        [os.path.join(repo, ".venv", "bin", "python"), "-m", "pytest", "-q",
         "-p", "no:cacheprovider",
         "tests/test_herdr_shim_contract.py", "-k",
         "h18 or h21 or h22 or h23"],
        capture_output=True, text=True, cwd=repo,
        env=dict(os.environ, PYTHONDONTWRITEBYTECODE="1"))
    assert res.returncode == 0, res.stdout[-4000:] + res.stderr[-2000:]


def test_s11_this_branch_does_not_change_the_generated_shim(mam_sandbox):
    """S11: the item-1 fix lives in mam_herdr, never in the heredoc that
    lib.sh materialises into $WORKSPACE_ROOT/.mam/shim/herdr. That file is
    shared by every checkout bound to one control plane, so keeping it
    byte-identical to base is what makes "the control plane is not modified"
    true at runtime."""
    repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    base = os.environ.get("MAM_BASE_SHA", "1e78dae")
    show = subprocess.run(["git", "show", "%s:.agents/skills/lib.sh" % base],
                          capture_output=True, text=True, cwd=repo)
    if show.returncode != 0:
        pytest.skip("base %s not available in this checkout" % base)
    with open(os.path.join(repo, ".agents", "skills", "lib.sh")) as f:
        head = f.read()
    assert _heredoc_body(head) == _heredoc_body(show.stdout), (
        "the shim heredoc changed: the generated .mam/shim/herdr of this branch "
        "would differ from the control root's, and the two would overwrite "
        "each other on every herdr call")


def _heredoc_body(text):
    lines = text.splitlines(True)
    start = None
    for i, line in enumerate(lines):
        if line.strip() == "cat <<'EOF' > \"$tmp_file\"":
            start = i + 1
            break
    assert start is not None, "shim heredoc not found in lib.sh"
    for j in range(start, len(lines)):
        if lines[j].rstrip("\n") == "EOF":
            return "".join(lines[start:j])
    raise AssertionError("unterminated shim heredoc")


@pytest.mark.parametrize("env,expected_rc", [(CROSS, 1), ({"HERDR_SESSION_NAME": "mam-isolated"}, 1)])
def test_s12_mam_herdr_never_aborts_its_caller_under_set_e(
        mam_sandbox, mock_herdr, mock_agents, env, expected_rc):
    """S12 (mock herdr): mam_herdr runs inside scripts using `set -euo
    pipefail`. Neither branch of the cross-server test may abort the shell,
    and the child's rc must be the function's rc."""
    tmp_path = mam_sandbox
    body = ('set -euo pipefail\n'
            'rc=0\n'
            'mam_herdr has-session -t nobody-here || rc=$?\n'
            'echo "missing_rc=$rc"\n'
            '_seen=0\n'
            'mam_herdr has-session -t creator-agy-01 || _seen=$?\n'
            'echo "present_rc=$_seen"\n'
            'echo REACHED_END\n')
    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
    lib_path = tmp_path / ".agents" / "skills" / "lib.sh"
    script = 'source "%s"\n%s' % (lib_path, body)
    run_env = dict(os.environ)
    for var in ("HERDR_SESSION", "HERDR_SESSION_NAME", "HERDR_WORKSPACE_ID",
                "HERDR_PANE_ID", "HERDR_TAB_ID"):
        run_env.pop(var, None)
    run_env.update(env)
    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
                         cwd=str(tmp_path), env=run_env)
    assert "REACHED_END" in res.stdout, res.stdout + res.stderr
    assert "missing_rc=%d" % expected_rc in res.stdout, res.stdout + res.stderr
    assert "present_rc=0" in res.stdout, res.stdout + res.stderr


def test_s13_the_loop_watcher_applies_the_same_rule(mam_sandbox):
    """S13 (pure unit): watcher.py builds its own herdr argv and never goes
    through mam_herdr, so it needs the same rule."""
    import sys
    repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    sys.path.insert(0, os.path.join(repo, ".agents", "skills"))
    from lib_py.loop import watcher

    caller_pane = {"HERDR_SESSION": "default", "HERDR_WORKSPACE_ID": "w5",
                   "HERDR_PANE_ID": "w5:p2", "HERDR_TAB_ID": "w5:t1", "PATH": "/usr/bin"}

    cross = watcher.scoped_env("mam-isolated", environ=caller_pane)
    assert "HERDR_WORKSPACE_ID" not in cross and "HERDR_PANE_ID" not in cross
    assert "HERDR_TAB_ID" not in cross and cross["PATH"] == "/usr/bin"

    same = watcher.scoped_env("default", environ=caller_pane)
    assert same["HERDR_WORKSPACE_ID"] == "w5"

    outside = watcher.scoped_env("mam-isolated", environ={"HERDR_WORKSPACE_ID": "w5"})
    assert outside["HERDR_WORKSPACE_ID"] == "w5"

    none_target = watcher.scoped_env(None, environ=caller_pane)
    assert none_target["HERDR_WORKSPACE_ID"] == "w5"

    seen = {}

    def _run(argv, **kw):
        seen["env"] = kw.get("env")

        class P:
            returncode, stdout, stderr = 0, "{}", ""
        return P()

    watcher.wait_once("agt", session="iso-x", run=_run)
    assert seen["env"] is not None and "HERDR_SESSION" not in ("",)