"""Tests for :class:`attractor.execution.local.LocalExec` (SPEC §6.8).

The :func:`exec_shell` fixture (see ``conftest.py``) picks whichever
shell the host actually has — ``pwsh`` on a properly-configured
Windows box, ``sh`` on Unix or Git-Bash-on-Windows. Tests parametrise
their scripts on :func:`is_pwsh` so they're portable across both."""

from __future__ import annotations

import asyncio
import os
import sys
import time
from collections.abc import Sequence
from datetime import timedelta
from pathlib import Path

import pytest

from attractor.execution import (
    DEFAULT_TIMEOUT,
    ELISION_HEAD,
    ELISION_TAIL,
    ExecError,
    ExecOutcome,
    LocalExec,
)


def _echo_script(text: str, *, is_pwsh: bool) -> str:
    """Return a shell script that prints ``text`` to stdout."""
    if is_pwsh:
        return f"Write-Output '{text}'"
    return f"echo '{text}'"


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_happy_path_echo(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """``echo hello`` exits 0 and "hello" appears in the capture."""
    del exec_shell
    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(_echo_script("hello", is_pwsh=is_pwsh))
    assert isinstance(outcome, ExecOutcome)
    assert outcome.exit_code == 0
    assert outcome.timed_out is False
    assert "hello" in outcome.captured_output


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_non_zero_exit_returns_outcome(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """A script that exits 1 returns ``ExecOutcome(exit_code=1)`` —
    *not* an :class:`ExecError`. The engine decides how to route
    non-zero exits (§6.3 FAILURE edges)."""
    del exec_shell, is_pwsh
    # ``exit 1`` works the same in sh and pwsh.
    script = "exit 1"
    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script)
    assert outcome.exit_code == 1
    assert outcome.timed_out is False


# [unit->REQ-EXEC-TOOL-CAPTURE]
@pytest.mark.asyncio
async def test_combined_stdout_and_stderr(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """Output written to both stdout and stderr appears in the same
    captured buffer, merged at OS level via ``subprocess.STDOUT``."""
    del exec_shell
    if is_pwsh:
        # PowerShell: native cmd `>&2` redirect inside a sub-call.
        script = "Write-Output 'to-out'; [Console]::Error.WriteLine('to-err')"
    else:
        script = "echo to-out; echo to-err 1>&2"
    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script)
    assert outcome.exit_code == 0
    assert "to-out" in outcome.captured_output
    assert "to-err" in outcome.captured_output


# [unit->REQ-EXEC-TOOL-CAPTURE]
@pytest.mark.asyncio
async def test_elision_when_output_exceeds_cap(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """A script that emits well over 512 KiB triggers the elision
    marker and the captured string stays bounded."""
    del exec_shell
    # 700 KiB of 'A' is comfortably over the 512 KiB threshold.
    payload_size = 700 * 1024
    if is_pwsh:
        # `-NoNewline` keeps Write-Host from inserting a newline per
        # call; we want a single contiguous run of bytes.
        script = (
            f"$s = 'A' * {payload_size}; "
            "[Console]::Out.Write($s)"
        )
    else:
        # printf is cheaper than `yes | head -c` for predictable byte
        # output and works in Git Bash's sh.
        script = (
            f"python -c \"import sys; sys.stdout.write('A' * {payload_size})\""
        )
    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script)
    assert outcome.exit_code == 0
    assert "[...elided" in outcome.captured_output
    # Capture is bounded: head + marker (< 64) + tail. Anything
    # under ~600 KiB chars is well within bounds.
    assert len(outcome.captured_output) < ELISION_HEAD + ELISION_TAIL + 1024


# [unit->REQ-EXEC-TOOL-CAPTURE]
@pytest.mark.asyncio
async def test_timeout_returns_timed_out_outcome(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """A script that outlives its timeout returns
    ``ExecOutcome(timed_out=True)`` with non-zero exit, and the wall
    clock stays within ``timeout + grace + slack``."""
    del exec_shell
    if is_pwsh:
        script = "Start-Sleep -Seconds 30"
    else:
        script = "sleep 30"
    exec_ = LocalExec(tmp_path)
    start = time.monotonic()
    outcome = await exec_.run(script, timeout=timedelta(milliseconds=300))
    elapsed = time.monotonic() - start

    assert outcome.timed_out is True
    assert outcome.exit_code != 0
    assert "[timeout:" in outcome.captured_output
    # 300ms timeout + 5s SIGINT grace + slack for shell startup +
    # taskkill / killpg. 15s is generous and lets a slow CI machine
    # still pass.
    assert elapsed < 15.0, f"kill ladder took {elapsed:.1f}s — too slow"


# [unit->REQ-EXEC-TOOL-CAPTURE]
@pytest.mark.asyncio
async def test_cancel_via_event(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """Setting the cancel event terminates the script the same way a
    timeout does (non-zero exit, bounded wall clock)."""
    del exec_shell
    if is_pwsh:
        script = "Start-Sleep -Seconds 30"
    else:
        script = "sleep 30"
    cancel = asyncio.Event()
    exec_ = LocalExec(tmp_path)

    async def _fire_cancel() -> None:
        await asyncio.sleep(0.2)
        cancel.set()

    start = time.monotonic()
    _, outcome = await asyncio.gather(
        _fire_cancel(),
        exec_.run(script, cancel=cancel),
    )
    elapsed = time.monotonic() - start

    assert isinstance(outcome, ExecOutcome)
    # Cancel is not a timeout; ``timed_out`` stays False.
    assert outcome.timed_out is False
    assert outcome.exit_code != 0
    assert "[cancelled]" in outcome.captured_output
    assert elapsed < 15.0, f"cancel took {elapsed:.1f}s — too slow"


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_env_scrub_anthropic_key(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """``ANTHROPIC_API_KEY`` is stripped from the child env even
    though it's set in the parent."""
    del exec_shell
    sentinel = "sk-not-a-real-key-12345"
    monkeypatch.setenv("ANTHROPIC_API_KEY", sentinel)

    if is_pwsh:
        script = "$env:ANTHROPIC_API_KEY"
    else:
        # `printf` so we don't depend on a default newline that might
        # leak the variable's emptiness vs. unset distinction.
        script = "printf '%s\\n' \"${ANTHROPIC_API_KEY:-MISSING}\""

    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script)
    assert outcome.exit_code == 0
    assert sentinel not in outcome.captured_output


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_env_scrub_multiple_providers(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Default-scrubbed provider keys (incl. Cerebras) are stripped."""
    del exec_shell
    monkeypatch.setenv("ANTHROPIC_API_KEY", "anth-sentinel")
    monkeypatch.setenv("OPENAI_API_KEY", "oai-sentinel")
    monkeypatch.setenv("GOOGLE_API_KEY", "ggl-sentinel")
    monkeypatch.setenv("CEREBRAS_API_KEY", "cbr-sentinel")

    if is_pwsh:
        script = (
            "$env:ANTHROPIC_API_KEY; $env:OPENAI_API_KEY; "
            "$env:GOOGLE_API_KEY; $env:CEREBRAS_API_KEY"
        )
    else:
        script = (
            "printf '%s\\n' \"${ANTHROPIC_API_KEY:-x}\" "
            "\"${OPENAI_API_KEY:-x}\" \"${GOOGLE_API_KEY:-x}\" "
            "\"${CEREBRAS_API_KEY:-x}\""
        )

    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script)
    assert outcome.exit_code == 0
    assert "anth-sentinel" not in outcome.captured_output
    assert "oai-sentinel" not in outcome.captured_output
    assert "ggl-sentinel" not in outcome.captured_output
    assert "cbr-sentinel" not in outcome.captured_output


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_scrub_env_vars_override(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Caller-supplied ``scrub_env_vars`` replaces the default — a
    var not in the override list reaches the child even though it's
    one of the defaults."""
    del exec_shell
    monkeypatch.setenv("ANTHROPIC_API_KEY", "keep-me")
    monkeypatch.setenv("CUSTOM_SECRET", "scrub-me")

    if is_pwsh:
        script = "Write-Output \"A=$env:ANTHROPIC_API_KEY C=$env:CUSTOM_SECRET\""
    else:
        script = "echo \"A=${ANTHROPIC_API_KEY:-} C=${CUSTOM_SECRET:-}\""

    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script, scrub_env_vars=("CUSTOM_SECRET",))
    assert outcome.exit_code == 0
    assert "keep-me" in outcome.captured_output
    assert "scrub-me" not in outcome.captured_output


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_path_passes_through_to_child(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """``PATH`` (or ``Path`` on Windows env-var conventions) is
    inherited by the child — SPEC §6.8 explicitly calls this out so
    tools like ``gh`` / ``cargo`` resolve without extra wiring."""
    del exec_shell
    if is_pwsh:
        script = "Write-Output $env:PATH"
    else:
        script = "printf '%s\\n' \"${PATH:-}\""
    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script)
    assert outcome.exit_code == 0
    # The captured PATH should overlap meaningfully with the parent's.
    parent_path = os.environ.get("PATH", "")
    segments = [p for p in parent_path.split(os.pathsep) if p]
    assert segments, "parent PATH is empty — can't run this test"
    # Git Bash's ``sh`` translates ``C:\Foo\Bar`` to ``/c/Foo/Bar``
    # when forwarding PATH, so we cannot expect verbatim equality.
    # Compare on basenames — every PATH segment's leaf name is the
    # same regardless of the path-style translation.
    parent_basenames = {
        os.path.basename(seg.rstrip("\\/")).lower()
        for seg in segments
        if os.path.basename(seg.rstrip("\\/"))
    }
    capture_lower = outcome.captured_output.lower()
    # Match the parent's unique basenames against the child's PATH.
    # On rich hosts (dev box, CI) we want at least three matches to
    # avoid false positives from common names ("bin", "system32"); on
    # minimal sandboxes the parent may only have two unique basenames
    # (e.g. /usr/local/bin:/usr/bin:/bin → {"bin"}, /sbin → {"sbin"}),
    # so we cap the requirement at the number of unique basenames
    # actually available. Either way, the test still proves propagation.
    hits = sum(1 for name in parent_basenames if name and name in capture_lower)
    required = min(len(parent_basenames), 3)
    assert hits >= required, (
        f"only {hits}/{required} PATH basenames reached the child "
        f"(parent had {len(parent_basenames)} unique basenames)"
    )


# [unit->REQ-EXEC-LOCAL-ENV]
def test_shell_not_found_message_on_windows_points_at_ps7_install(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Missing pwsh on Windows must surface the PS 7 install URL.

    SPEC §15 records that PS 5.1 fallback was rejected (workflow
    portability footgun). The error message is the user's path back
    to a working setup, so the install URL and the "PS 5.1 not
    supported" line are part of the contract.
    """
    from attractor.execution.local import (
        _shell_not_found_message,  # pyright: ignore[reportPrivateUsage]
    )
    monkeypatch.setattr(sys, "platform", "win32")
    msg = _shell_not_found_message("pwsh")
    assert "PowerShell 7" in msg
    assert "learn.microsoft.com" in msg
    assert "PowerShell 5.1" in msg
    assert "SPEC §15" in msg


# [unit->REQ-EXEC-LOCAL-ENV]
def test_shell_not_found_message_non_pwsh_stays_terse(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Non-pwsh / non-Windows misses keep the original short message."""
    from attractor.execution.local import (
        _shell_not_found_message,  # pyright: ignore[reportPrivateUsage]
    )
    monkeypatch.setattr(sys, "platform", "linux")
    assert _shell_not_found_message("sh") == "shell not found: 'sh'"
    # pwsh-name on a non-Windows host (artificial — would only happen
    # if a test monkeypatched the resolver). Still terse.
    assert "learn.microsoft.com" not in _shell_not_found_message("pwsh")


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_missing_shell_binary_wraps_with_actionable_error(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Spawn-time `FileNotFoundError` becomes an `ExecError` carrying
    the resolver's chosen shell name in the message — proves the
    error wrapper goes through `_shell_not_found_message`.
    """
    from attractor.execution import local
    monkeypatch.setattr(
        local, "_resolve_shell", lambda: ("definitely-not-a-shell", ("-c",))
    )
    exec_ = LocalExec(tmp_path)
    with pytest.raises(ExecError, match="definitely-not-a-shell"):
        await exec_.run("echo hi")


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_cwd_does_not_exist_raises(tmp_path: Path) -> None:
    """A missing cwd raises :class:`ExecError` — not a non-zero exit,
    not a silent fallback."""
    bogus = tmp_path / "does-not-exist"
    exec_ = LocalExec(bogus)
    with pytest.raises(ExecError):
        await exec_.run("echo hi")


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_cwd_is_a_file_raises(tmp_path: Path) -> None:
    """A cwd that points at a file (not a directory) raises
    :class:`ExecError`."""
    target = tmp_path / "regular.txt"
    target.write_text("not a directory")
    exec_ = LocalExec(target)
    with pytest.raises(ExecError):
        await exec_.run("echo hi")


# [unit->REQ-EXEC-LOCAL-ENV]
@pytest.mark.asyncio
async def test_runs_inside_specified_cwd(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """A file created by the script lands in the LocalExec cwd, not
    in the process's cwd."""
    del exec_shell
    if is_pwsh:
        script = "Set-Content -Path marker.txt -Value 'hi'"
    else:
        script = "echo hi > marker.txt"
    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script)
    assert outcome.exit_code == 0
    assert (tmp_path / "marker.txt").exists()


# [unit->REQ-EXEC-TOOL-CAPTURE]
@pytest.mark.asyncio
async def test_default_timeout_is_thirty_minutes() -> None:
    """The default timeout matches SPEC §6.8 verbatim."""
    assert DEFAULT_TIMEOUT == timedelta(minutes=30)


# [unit->REQ-EXEC-TOOL-CAPTURE]
@pytest.mark.asyncio
async def test_timeout_kill_does_not_leak_child(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """After a timeout the call returns and the child is reaped
    (no zombie / no hung pipe)."""
    del exec_shell
    if is_pwsh:
        script = "Start-Sleep -Seconds 60"
    else:
        script = "sleep 60"
    exec_ = LocalExec(tmp_path)
    outcome = await exec_.run(script, timeout=timedelta(milliseconds=200))
    assert outcome.timed_out is True


# [unit->REQ-EXEC-TOOL-CAPTURE]
@pytest.mark.asyncio
async def test_grandchild_killed_with_taskkill_on_windows(
    tmp_path: Path,
    exec_shell: tuple[str, Sequence[str]],
    is_pwsh: bool,
) -> None:
    """The Rust prototype found that ``Process.terminate`` on Windows
    only kills the immediate child — grandchildren orphan and hold
    pipes open. ``taskkill /F /T`` must tear down the whole tree.

    On non-Windows hosts this still exercises the process-group path
    via SIGKILL; the assertion is the same (kill returns promptly)."""
    del exec_shell
    if sys.platform != "win32":
        pytest.skip("Windows-specific kill-tree path; SIGKILL path covered elsewhere")
    if is_pwsh:
        # Parent spawns a long-lived powershell grandchild and waits.
        script = (
            "Start-Process -NoNewWindow powershell -ArgumentList "
            "'-NoProfile','-Command','Start-Sleep 60'; Start-Sleep 60"
        )
    else:
        # sh on Windows: subshell spawns a sleep, parent waits.
        script = "sleep 60 & sleep 60"
    exec_ = LocalExec(tmp_path)
    start = time.monotonic()
    outcome = await exec_.run(script, timeout=timedelta(milliseconds=200))
    elapsed = time.monotonic() - start
    assert outcome.timed_out is True
    assert elapsed < 15.0