"""Tests for async shell tool with timeout-to-handle support."""

from __future__ import annotations

import asyncio
import contextlib
import os
import signal
import sys
import time
from typing import TYPE_CHECKING
from unittest.mock import patch

import pytest

from mindroom.constants import RuntimePaths, resolve_runtime_paths, workspace_home_identity_env
from mindroom.shell_execution import _MAX_OUTPUT_BYTES
from mindroom.tool_system.metadata import get_tool_by_name
from mindroom.tools.shell import (
    _process_registry,
    _shell_subprocess_env,
    _workspace_home_contract_env_from_process_env,
    shell_tools,
)

if TYPE_CHECKING:
    from pathlib import Path

    from agno.tools.function import Function
    from agno.tools.toolkit import Toolkit

    from mindroom.constants import RuntimePaths


def _make_runtime_paths(tmp_path: Path, *, process_env: dict[str, str] | None = None) -> RuntimePaths:
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (tmp_path / ".env").write_text("", encoding="utf-8")
    return resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={} if process_env is None else process_env,
    )


def _get_toolkit(tmp_path: Path) -> Toolkit:
    runtime_paths = _make_runtime_paths(tmp_path)
    return get_tool_by_name("shell", runtime_paths, disable_sandbox_proxy=True, worker_target=None)


def _fork_child_holding_stdio_script(*, parent_stderr: str = "", parent_exit_code: int = 0) -> str:
    return (
        "import os\n"
        "import pathlib\n"
        "import signal\n"
        "import sys\n"
        "import time\n"
        "stop_file = pathlib.Path(sys.argv[2])\n"
        "def stop_child(signum, frame):\n"
        "    stop_file.write_text('stopped', encoding='utf-8')\n"
        "    os._exit(0)\n"
        f"sys.stderr.write({parent_stderr!r})\n"
        "sys.stderr.flush()\n"
        "pid = os.fork()\n"
        "if pid:\n"
        "    pathlib.Path(sys.argv[1]).write_text(str(pid), encoding='utf-8')\n"
        f"    sys.exit({parent_exit_code})\n"
        "signal.signal(signal.SIGTERM, stop_child)\n"
        "time.sleep(30)\n"
    )


async def _wait_for_pid_file(pid_file: Path) -> int:
    last_text: str | None = None
    for _ in range(50):
        if pid_file.exists():
            last_text = pid_file.read_text(encoding="utf-8").strip()
            if last_text:
                return int(last_text)
        await asyncio.sleep(0.05)
    message = f"PID file was not written: {pid_file}" if last_text is None else f"PID file was empty: {pid_file}"
    raise AssertionError(message)


async def _stop_recorded_child(pid: int, stop_file: Path) -> None:
    with contextlib.suppress(ProcessLookupError, PermissionError):
        os.kill(pid, signal.SIGTERM)

    for _ in range(20):
        if stop_file.exists():
            return
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            return
        await asyncio.sleep(0.05)

    with contextlib.suppress(ProcessLookupError, PermissionError):
        os.kill(pid, signal.SIGKILL)

    for _ in range(20):
        try:
            os.kill(pid, 0)
        except ProcessLookupError:
            return
        await asyncio.sleep(0.05)

    message = f"Child process {pid} did not stop"
    raise AssertionError(message)


def _get_run_shell_command_function(tool: Toolkit) -> Function:
    return tool.async_functions["run_shell_command"]


# ---------------------------------------------------------------------------
# run_shell_command
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_shell_command_returns_output(tmp_path: Path) -> None:
    """Fast command should return stdout directly."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["echo", "hello world"])
    assert result == "hello world"


@pytest.mark.asyncio
async def test_run_shell_command_bash_echo_returns_output(tmp_path: Path) -> None:
    """Normal shell output should still be returned."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["bash", "-lc", "echo hello"])
    assert result == "hello"


@pytest.mark.asyncio
async def test_run_shell_command_accepts_shell_command_string(tmp_path: Path) -> None:
    """Natural shell command strings should execute through bash."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint("echo $HOME")
    assert result
    assert not result.startswith("Error:")


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("command", "expected"),
    [
        ("[ 1 -eq 1 ] && echo ok", "ok"),
        ("{ echo ok; }", "ok"),
    ],
)
async def test_run_shell_command_accepts_bracketed_shell_command_strings(
    tmp_path: Path,
    command: str,
    expected: str,
) -> None:
    """Shell grammar that starts with JSON-like characters should still execute through bash."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(command)
    assert result == expected


@pytest.mark.asyncio
async def test_run_shell_command_accepts_single_item_shell_command_line(tmp_path: Path) -> None:
    """LLMs often send one shell command line inside the argv array."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["printf '%s' ok"])
    assert result == "ok"


def test_run_shell_command_schema_accepts_string_or_string_array(tmp_path: Path) -> None:
    """The tool schema should expose both natural command strings and argv lists."""
    tool = _get_toolkit(tmp_path)
    args_schema = _get_run_shell_command_function(tool).parameters["properties"]["args"]

    assert args_schema["anyOf"] == [
        {"type": "array", "items": {"type": "string"}},
        {"type": "string"},
    ]


@pytest.mark.asyncio
async def test_run_shell_command_parses_json_string_args(tmp_path: Path) -> None:
    """JSON-stringified args should be parsed into a shell argv list."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint('["echo", "hello world"]')
    assert result == "hello world"


@pytest.mark.asyncio
async def test_run_shell_command_parses_single_item_json_args(tmp_path: Path) -> None:
    """A single command item in JSON string form should execute normally."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint('["ls"]')
    assert result
    assert not result.startswith("Error:")


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "args",
    [
        '["echo",',
        '{"cmd": "ls"}',
        "",
        '["echo", 42]',
        '["bash", ["-c", "ls"]]',
        '["echo", {"x": 1}]',
    ],
)
async def test_run_shell_command_rejects_invalid_stringified_args(tmp_path: Path, args: str) -> None:
    """Malformed or non-flat stringified args should fail validation."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(args)
    assert result.startswith("Error:")
    assert "'args' must be a shell command string or a flat list of strings" in result


@pytest.mark.asyncio
async def test_run_shell_command_empty_json_list_uses_existing_empty_behavior(tmp_path: Path) -> None:
    """An empty JSON list should fall through to the existing subprocess error path."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint("[]")
    assert result.startswith("Error:")


@pytest.mark.asyncio
@pytest.mark.parametrize("args", ["", "   "])
async def test_run_shell_command_entrypoint_returns_error_for_empty_string_args(tmp_path: Path, args: str) -> None:
    """Direct entrypoint calls should return validation errors instead of raising."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(args)
    assert result.startswith("Error:")
    assert "'args' must be a shell command string or a flat list of strings" in result


@pytest.mark.asyncio
async def test_run_shell_command_returns_error_on_nonzero_exit(tmp_path: Path) -> None:
    """Non-zero exit should return stderr."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["bash", "-c", "echo oops >&2; exit 1"])
    assert result.startswith("Error:")
    assert "oops" in result


@pytest.mark.asyncio
@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
async def test_run_shell_command_returns_after_foreground_exit_with_inherited_pipe_child(tmp_path: Path) -> None:
    """Regression: tmux/auth helpers/watchers/daemons can inherit pipes after foreground exit."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    pid_file = tmp_path / "child.pid"
    stop_file = tmp_path / "child.stopped"
    command = [sys.executable, "-c", _fork_child_holding_stdio_script(), str(pid_file), str(stop_file)]
    started = time.perf_counter()
    task = asyncio.create_task(entrypoint(command, timeout=10))
    child_pid: int | None = None

    try:
        child_pid = await _wait_for_pid_file(pid_file)
        result = await asyncio.wait_for(task, timeout=2)
    finally:
        if child_pid is not None:
            await _stop_recorded_child(child_pid, stop_file)
        if not task.done():
            task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await task

    elapsed = time.perf_counter() - started
    assert elapsed < 2
    assert result == ""


@pytest.mark.asyncio
@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork")
async def test_run_shell_command_nonzero_foreground_exit_with_inherited_pipe_child(tmp_path: Path) -> None:
    """Nonzero foreground exit should return buffered stderr without waiting for descendant EOF."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    pid_file = tmp_path / "child-error.pid"
    stop_file = tmp_path / "child-error.stopped"
    command = [
        sys.executable,
        "-c",
        _fork_child_holding_stdio_script(parent_stderr="foreground failed\n", parent_exit_code=7),
        str(pid_file),
        str(stop_file),
    ]
    started = time.perf_counter()
    task = asyncio.create_task(entrypoint(command, timeout=10))
    child_pid: int | None = None

    try:
        child_pid = await _wait_for_pid_file(pid_file)
        result = await asyncio.wait_for(task, timeout=2)
    finally:
        if child_pid is not None:
            await _stop_recorded_child(child_pid, stop_file)
        if not task.done():
            task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await task

    elapsed = time.perf_counter() - started
    assert elapsed < 2
    assert result == "Error: foreground failed"


@pytest.mark.asyncio
async def test_run_shell_command_tail_parameter(tmp_path: Path) -> None:
    """Only the last N lines should be returned."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["bash", "-c", "for i in $(seq 1 20); do echo line$i; done"], tail=3)
    lines = result.strip().split("\n")
    assert lines == ["line18", "line19", "line20"]


@pytest.mark.asyncio
async def test_run_shell_command_truncates_large_output_by_bytes(tmp_path: Path) -> None:
    """A small line count with very large lines should not return an unbounded result."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    script = "for i in range(120): print(f'{i:03d}:' + 'x' * 1000)"
    result = await entrypoint([sys.executable, "-c", script], tail=120)

    assert "Output truncated to the last" in result
    assert "119:" in result
    assert "000:" not in result
    assert len(result.encode("utf-8")) <= _MAX_OUTPUT_BYTES + 200


@pytest.mark.asyncio
async def test_run_shell_command_truncates_large_stderr_by_bytes(tmp_path: Path) -> None:
    """Large stderr from failed commands should be byte-capped too."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    script = "import sys\nfor i in range(120): print(f'{i:03d}:' + 'x' * 1000, file=sys.stderr)\nsys.exit(1)\n"
    result = await entrypoint([sys.executable, "-c", script])

    assert result.startswith("Error: [Output truncated to the last")
    assert "119:" in result
    assert "000:" not in result
    assert len(result.encode("utf-8")) <= _MAX_OUTPUT_BYTES + 200


@pytest.mark.asyncio
async def test_run_shell_command_truncates_many_short_lines_by_rendered_bytes(tmp_path: Path) -> None:
    """Rendered newline separators should be included in the output byte cap."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    script = "for _ in range(10000): print('xxxxxx')"
    result = await entrypoint([sys.executable, "-c", script], tail=10000)

    assert result.startswith("[Output truncated to the last")
    assert len(result.encode("utf-8")) <= _MAX_OUTPUT_BYTES + 200


@pytest.mark.asyncio
async def test_run_shell_command_truncates_oversized_single_stdout_line(tmp_path: Path) -> None:
    """Output without newlines should still be retained and byte-capped."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint([sys.executable, "-c", "import sys; sys.stdout.write('A' * 100000)"])

    assert result.startswith("[Output truncated to the last")
    assert "A" * 100 in result
    assert len(result.encode("utf-8")) <= _MAX_OUTPUT_BYTES + 200


@pytest.mark.asyncio
async def test_run_shell_command_truncates_oversized_single_stderr_line(tmp_path: Path) -> None:
    """Oversized stderr lines should be surfaced instead of skipped."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(
        [sys.executable, "-c", "import sys; sys.stderr.write('B' * 100000); sys.exit(1)"],
    )

    assert result.startswith("Error: [Output truncated to the last")
    assert "B" * 100 in result
    assert len(result.encode("utf-8")) <= _MAX_OUTPUT_BYTES + 200


@pytest.mark.asyncio
async def test_run_shell_command_returns_handle_on_timeout(tmp_path: Path) -> None:
    """A workspace-prefixed timeout result should expose a usable handle identifier."""
    runtime_paths = _make_runtime_paths(tmp_path)
    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        disable_sandbox_proxy=True,
        worker_target=None,
        tool_init_overrides={"base_dir": str(tmp_path)},
    )
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert entrypoint is not None
    assert check_fn is not None

    result = await entrypoint(["sleep", "300"], timeout=1)

    assert result.startswith(f"[cwd: {tmp_path}]\n")
    assert "timed out" in result.lower()
    assert "Handle: shell:" in result
    assert "check_shell_command" in result

    # Extract handle and clean up
    handle = result.split("Handle: ")[1].split("\n")[0]
    assert "RUNNING" in check_fn(handle)
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    kill_fn(handle, force=True)
    await asyncio.sleep(0.1)


@pytest.mark.asyncio
async def test_run_shell_command_error_on_bad_command(tmp_path: Path) -> None:
    """Non-existent command should return an error."""
    tool = _get_toolkit(tmp_path)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["__nonexistent_command_xyz__"])
    assert result.startswith("Error:")


@pytest.mark.asyncio
async def test_run_shell_command_respects_base_dir(tmp_path: Path) -> None:
    """Shell should execute in base_dir when configured."""
    runtime_paths = _make_runtime_paths(tmp_path)
    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        disable_sandbox_proxy=True,
        worker_target=None,
        tool_init_overrides={"base_dir": str(tmp_path)},
    )
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["pwd"])
    assert result.splitlines() == [f"[cwd: {tmp_path}]", str(tmp_path)]


# ---------------------------------------------------------------------------
# check_shell_command
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_check_shell_command_running(tmp_path: Path) -> None:
    """Checking a running process should show RUNNING status."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert run_fn is not None
    assert check_fn is not None

    result = await run_fn(["sleep", "300"], timeout=0)
    handle = result.split("Handle: ")[1].split("\n")[0]

    status = check_fn(handle)
    assert "RUNNING" in status

    # Clean up
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    kill_fn(handle, force=True)
    await asyncio.sleep(0.1)


@pytest.mark.asyncio
async def test_check_shell_command_finished(tmp_path: Path) -> None:
    """Checking a finished process should show FINISHED status and output."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert run_fn is not None
    assert check_fn is not None

    # Force backgrounding immediately, but let the command live long enough for
    # the monitor path to observe its output and final status.
    result = await run_fn(["bash", "-c", "echo done-output; sleep 0.05"], timeout=0)
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    # Wait for the process to finish
    for _ in range(50):
        status = check_fn(handle)
        if "FINISHED" in status:
            break
        await asyncio.sleep(0.02)
    assert "FINISHED" in status
    assert "done-output" in status


@pytest.mark.asyncio
async def test_check_shell_command_unknown_handle(tmp_path: Path) -> None:
    """Checking an unknown handle should return an error."""
    tool = _get_toolkit(tmp_path)
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert check_fn is not None

    result = check_fn("shell:nonexistent")
    assert "Error:" in result
    assert "Unknown handle" in result


@pytest.mark.asyncio
async def test_check_shell_command_partial_output(tmp_path: Path) -> None:
    """Checking a running process should show partial output."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert run_fn is not None
    assert check_fn is not None

    result = await run_fn(
        ["bash", "-c", "for i in 1 2 3; do echo partial-line-$i; done; sleep 300"],
        timeout=0,
    )
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    for _ in range(50):
        status = check_fn(handle)
        if "partial-line" in status:
            break
        await asyncio.sleep(0.02)
    assert "RUNNING" in status
    assert "lines buffered" in status
    assert "lines so far" not in status
    assert "partial-line" in status

    # Clean up
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    kill_fn(handle, force=True)
    await asyncio.sleep(0.1)


# ---------------------------------------------------------------------------
# kill_shell_command
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_kill_shell_command(tmp_path: Path) -> None:
    """Killing a running process should succeed."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert run_fn is not None
    assert kill_fn is not None
    assert check_fn is not None

    result = await run_fn(["sleep", "300"], timeout=0)
    handle = result.split("Handle: ")[1].split("\n")[0]

    kill_result = kill_fn(handle)
    assert "Terminated" in kill_result or "SIGTERM" in kill_result

    # Wait for process to actually exit
    for _ in range(30):
        status = check_fn(handle)
        if "FINISHED" in status:
            break
        await asyncio.sleep(0.02)

    assert "FINISHED" in status


@pytest.mark.asyncio
async def test_kill_shell_command_force(tmp_path: Path) -> None:
    """Force-killing should send SIGKILL."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    assert run_fn is not None
    assert kill_fn is not None

    result = await run_fn(["sleep", "300"], timeout=0)
    handle = result.split("Handle: ")[1].split("\n")[0]

    kill_result = kill_fn(handle, force=True)
    assert "Force-killed" in kill_result or "SIGKILL" in kill_result
    await asyncio.sleep(0.1)


@pytest.mark.asyncio
async def test_kill_shell_command_already_finished(tmp_path: Path) -> None:
    """Killing an already-finished process should say so."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert run_fn is not None
    assert kill_fn is not None
    assert check_fn is not None

    # Force backgrounding: command finishes in ~0.2s but timeout is very short
    result = await run_fn(["bash", "-c", "echo fast; sleep 0.2"], timeout=0)
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    # Wait for it to finish
    for _ in range(30):
        status = check_fn(handle)
        if "FINISHED" in status:
            break
        await asyncio.sleep(0.1)
    assert "FINISHED" in status

    kill_result = kill_fn(handle)
    assert "already finished" in kill_result.lower()


@pytest.mark.asyncio
async def test_kill_shell_command_unknown_handle(tmp_path: Path) -> None:
    """Killing an unknown handle should return error."""
    tool = _get_toolkit(tmp_path)
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    assert kill_fn is not None

    result = kill_fn("shell:nonexistent")
    assert "Error:" in result
    assert "Unknown handle" in result


# ---------------------------------------------------------------------------
# Tool registration
# ---------------------------------------------------------------------------


def test_shell_registers_three_tools(tmp_path: Path) -> None:
    """Shell toolkit should register run, check, and kill functions."""
    tool = _get_toolkit(tmp_path)

    assert "run_shell_command" in tool.async_functions
    assert "check_shell_command" in tool.functions
    assert "kill_shell_command" in tool.functions


def test_shell_disabled_registers_no_tools(tmp_path: Path) -> None:
    """Shell toolkit with enable_run_shell_command=False should have no tools."""
    runtime_paths = _make_runtime_paths(tmp_path)

    cls = shell_tools()
    toolkit = cls(enable_run_shell_command=False, runtime_paths=runtime_paths)
    assert len(toolkit.functions) == 0
    assert len(toolkit.async_functions) == 0


# ---------------------------------------------------------------------------
# Stale record cleanup
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_sweep_stale_records(tmp_path: Path) -> None:
    """Stale finished records should be cleaned up based on finished_at."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert run_fn is not None
    assert check_fn is not None

    # Force backgrounding with timeout=0, command finishes quickly
    result = await run_fn(["bash", "-c", "echo quick; sleep 0.2"], timeout=0)
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    # Wait for process to finish
    for _ in range(30):
        status = check_fn(handle)
        if "FINISHED" in status:
            break
        await asyncio.sleep(0.1)
    assert "FINISHED" in status

    # Record should still exist — aging started_at should NOT cause sweep
    # because sweep now uses finished_at
    record = _process_registry.get(handle)
    assert record is not None
    record.started_at = time.monotonic() - 700  # 700 seconds ago

    # Trigger sweep — record should survive because finished_at is recent
    await run_fn(["echo", "trigger-sweep"])
    assert handle in _process_registry

    # Now age finished_at — record should be swept
    record.finished_at = time.monotonic() - 700
    await run_fn(["echo", "trigger-sweep-2"])
    assert handle not in _process_registry


# ---------------------------------------------------------------------------
# Env passthrough
# ---------------------------------------------------------------------------


def test_workspace_home_identity_env_builds_shell_identity_fragment(tmp_path: Path) -> None:
    """The shared workspace HOME fragment should use stable workspace-relative XDG paths."""
    workspace = tmp_path / "workspace"

    assert workspace_home_identity_env(workspace) == {
        "HOME": str(workspace),
        "MINDROOM_AGENT_WORKSPACE": str(workspace),
        "XDG_CONFIG_HOME": str(workspace / ".config"),
        "XDG_DATA_HOME": str(workspace / ".local" / "share"),
        "XDG_STATE_HOME": str(workspace / ".local" / "state"),
    }


def test_workspace_home_contract_env_requires_full_identity_fragment(tmp_path: Path) -> None:
    """Shell subprocesses should forward the workspace contract only when identity is coherent."""
    workspace = tmp_path / "workspace"
    base_env = {
        **workspace_home_identity_env(workspace),
        "XDG_CACHE_HOME": str(tmp_path / "cache"),
        "PIP_CACHE_DIR": str(tmp_path / "cache" / "pip"),
        "UV_CACHE_DIR": str(tmp_path / "cache" / "uv"),
        "PYTHONPYCACHEPREFIX": str(tmp_path / "cache" / "pycache"),
        "VIRTUAL_ENV": str(tmp_path / "venv"),
    }

    assert _workspace_home_contract_env_from_process_env(base_env) == base_env

    mismatched_env = dict(base_env)
    mismatched_env["XDG_DATA_HOME"] = str(tmp_path / "other-data")
    assert _workspace_home_contract_env_from_process_env(mismatched_env) == {}


def test_shell_subprocess_env_exports_workspace_from_base_dir_without_home_contract(tmp_path: Path) -> None:
    """Local execution keeps the host HOME but still exports the workspace env var."""
    workspace = tmp_path / "workspace"

    env = _shell_subprocess_env({}, base_process_env={"HOME": "/home/host-user"}, workspace_dir=workspace)

    assert env["MINDROOM_AGENT_WORKSPACE"] == str(workspace.resolve())
    assert env["HOME"] == "/home/host-user"


def test_shell_subprocess_env_replaces_incomplete_workspace_contract(tmp_path: Path) -> None:
    """An unvalidated process workspace must not override the local agent workspace."""
    workspace = tmp_path / "workspace"
    stale_workspace = tmp_path / "stale-workspace"
    base_env = {
        "HOME": "/home/host-user",
        "MINDROOM_AGENT_WORKSPACE": str(stale_workspace),
    }

    env = _shell_subprocess_env(
        {"MINDROOM_AGENT_WORKSPACE": str(stale_workspace)},
        base_process_env=base_env,
        workspace_dir=workspace,
    )

    assert env["MINDROOM_AGENT_WORKSPACE"] == str(workspace.resolve())
    assert env["HOME"] == "/home/host-user"


def test_shell_subprocess_env_prefers_worker_home_contract_over_base_dir(tmp_path: Path) -> None:
    """The worker workspace HOME contract stays authoritative over the base_dir fallback."""
    contract_workspace = tmp_path / "contract-workspace"
    base_env = {
        **workspace_home_identity_env(contract_workspace),
        "XDG_CACHE_HOME": str(tmp_path / "cache"),
        "PIP_CACHE_DIR": str(tmp_path / "cache" / "pip"),
        "UV_CACHE_DIR": str(tmp_path / "cache" / "uv"),
        "PYTHONPYCACHEPREFIX": str(tmp_path / "cache" / "pycache"),
        "VIRTUAL_ENV": str(tmp_path / "venv"),
    }

    env = _shell_subprocess_env({}, base_process_env=base_env, workspace_dir=tmp_path / "other-base-dir")

    assert env["MINDROOM_AGENT_WORKSPACE"] == str(contract_workspace)
    assert env["HOME"] == str(contract_workspace)


@pytest.mark.asyncio
async def test_local_shell_replaces_stale_agent_workspace_env(tmp_path: Path) -> None:
    """Local shell execution should replace a stale process workspace with base_dir."""
    workspace = tmp_path / "agent-workspace"
    workspace.mkdir()
    runtime_paths = _make_runtime_paths(
        tmp_path,
        process_env={
            "HOME": str(tmp_path / "host-home"),
            "MINDROOM_AGENT_WORKSPACE": str(tmp_path / "stale-workspace"),
        },
    )
    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        disable_sandbox_proxy=True,
        worker_target=None,
        tool_init_overrides={"base_dir": str(workspace)},
    )
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint('printf %s "$MINDROOM_AGENT_WORKSPACE"')
    assert result.splitlines() == [f"[cwd: {workspace}]", str(workspace.resolve())]


def test_run_shell_command_description_uses_portable_workspace_paths(tmp_path: Path) -> None:
    """The shell description should give workspace guidance valid in every execution mode."""
    workspace = tmp_path / "workspace"
    runtime_paths = _make_runtime_paths(tmp_path, process_env={"HOME": "/home/host-user"})
    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        disable_sandbox_proxy=True,
        worker_target=None,
        tool_init_overrides={"base_dir": str(workspace)},
    )

    description = tool.async_functions["run_shell_command"].description
    assert description is not None
    assert "Always use relative paths" in description
    assert "$MINDROOM_AGENT_WORKSPACE" in description
    assert "worker-routed execution maps `~` to the workspace" in description
    assert "local execution maps it to the host home" in description


def test_proxied_run_shell_command_description_does_not_claim_host_home(tmp_path: Path) -> None:
    """The model-facing proxy description must not misidentify worker execution as local."""
    workspace = tmp_path / "workspace"
    runtime_paths = _make_runtime_paths(
        tmp_path,
        process_env={
            "HOME": "/home/host-user",
            "MINDROOM_SANDBOX_EXECUTION_MODE": "all",
        },
    )
    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        worker_target=None,
        tool_init_overrides={"base_dir": str(workspace)},
    )

    description = tool.async_functions["run_shell_command"].description
    assert description is not None
    assert "worker-routed execution maps `~` to the workspace" in description
    assert "`~` and `$HOME` point at the host home" not in description


def test_run_shell_command_description_has_no_workspace_note_without_base_dir(tmp_path: Path) -> None:
    """Without a workspace there is no cwd contract to describe."""
    tool = _get_toolkit(tmp_path)

    description = tool.async_functions["run_shell_command"].description
    assert description is not None
    assert "[cwd:" not in description


@pytest.mark.asyncio
async def test_env_passthrough_preserved(tmp_path: Path) -> None:
    """Runtime env values from .env should be visible in shell commands."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (tmp_path / ".env").write_text("MY_TEST_VAR=async-shell-works\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={},
    )
    tool = get_tool_by_name("shell", runtime_paths, disable_sandbox_proxy=True, worker_target=None)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["bash", "-lc", "printf '%s' \"$MY_TEST_VAR\""])
    assert result == "async-shell-works"


@pytest.mark.asyncio
async def test_login_bash_preserves_runtime_path_after_profile_reset(tmp_path: Path) -> None:
    """MindRoom shell calls should keep runtime PATH even when login startup rewrites it."""
    fake_bin = tmp_path / "fake-bin"
    fake_bin.mkdir()
    fake_bash = fake_bin / "bash"
    fake_bash.write_text(
        "#!/bin/sh\n"
        'if [ "$1" = "-lc" ]; then\n'
        "  export PATH=/profile/default\n"
        '  exec /bin/sh -c "$2"\n'
        "fi\n"
        'exec /bin/sh "$@"\n',
        encoding="utf-8",
    )
    fake_bash.chmod(0o755)

    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={"PATH": "/worker-env/bin:/usr/bin:/bin"},
    )
    tool = get_tool_by_name("shell", runtime_paths, disable_sandbox_proxy=True, worker_target=None)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint([str(fake_bash), "-lc", "printf '%s' \"$PATH\""])

    assert result == "/worker-env/bin:/usr/bin:/bin"


# ---------------------------------------------------------------------------
# Handle persistence across toolkit instances (sandbox runner path)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_handle_persists_across_toolkit_instances(tmp_path: Path) -> None:
    """Handles created by one toolkit instance should be visible to a new one.

    This simulates the sandbox runner path where _resolve_entrypoint creates
    a fresh MindRoomShellTools per request.
    """
    tool1 = _get_toolkit(tmp_path)
    run_fn = tool1.async_functions["run_shell_command"].entrypoint
    assert run_fn is not None

    result = await run_fn(["sleep", "300"], timeout=0)
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    # Create a second toolkit instance (simulates sandbox runner re-creation)
    tool2 = _get_toolkit(tmp_path)
    check_fn = tool2.functions["check_shell_command"].entrypoint
    assert check_fn is not None

    status = check_fn(handle)
    assert "RUNNING" in status

    # Kill via the second instance too
    kill_fn = tool2.functions["kill_shell_command"].entrypoint
    assert kill_fn is not None
    kill_result = kill_fn(handle, force=True)
    assert "Force-killed" in kill_result or "already exited" in kill_result.lower()
    await asyncio.sleep(0.1)


@pytest.mark.asyncio
async def test_handle_check_then_kill_across_instances(tmp_path: Path) -> None:
    """Full run→check→kill lifecycle across separate toolkit instances."""
    tool_run = _get_toolkit(tmp_path)
    run_fn = tool_run.async_functions["run_shell_command"].entrypoint
    assert run_fn is not None

    result = await run_fn(["bash", "-c", "for i in 1 2 3; do echo line-$i; done; sleep 300"], timeout=0)
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    # Check from a fresh instance
    tool_check = _get_toolkit(tmp_path)
    check_fn = tool_check.functions["check_shell_command"].entrypoint
    assert check_fn is not None
    for _ in range(50):
        status = check_fn(handle)
        if "line-" in status:
            break
        await asyncio.sleep(0.02)
    assert "RUNNING" in status
    assert "line-" in status

    # Kill from yet another fresh instance
    tool_kill = _get_toolkit(tmp_path)
    kill_fn = tool_kill.functions["kill_shell_command"].entrypoint
    assert kill_fn is not None
    kill_fn(handle, force=True)

    for _ in range(30):
        tool_final = _get_toolkit(tmp_path)
        final_check = tool_final.functions["check_shell_command"].entrypoint
        assert final_check is not None
        status = final_check(handle)
        if "FINISHED" in status:
            break
        await asyncio.sleep(0.02)

    assert "FINISHED" in status


@pytest.mark.asyncio
async def test_handle_isolation_blocks_cross_runtime_access(tmp_path: Path) -> None:
    """A handle from one runtime should not be visible to a different runtime root."""
    runtime_a = tmp_path / "runtime-a"
    runtime_b = tmp_path / "runtime-b"
    runtime_a.mkdir()
    runtime_b.mkdir()

    tool_a = _get_toolkit(runtime_a)
    tool_b = _get_toolkit(runtime_b)
    run_a = tool_a.async_functions["run_shell_command"].entrypoint
    check_b = tool_b.functions["check_shell_command"].entrypoint
    kill_b = tool_b.functions["kill_shell_command"].entrypoint
    kill_a = tool_a.functions["kill_shell_command"].entrypoint
    assert run_a is not None
    assert check_b is not None
    assert kill_b is not None
    assert kill_a is not None

    result = await run_a(["sleep", "300"], timeout=0)
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    try:
        check_result = check_b(handle)
        assert "Unknown handle" in check_result

        kill_result = kill_b(handle, force=True)
        assert "Unknown handle" in kill_result
    finally:
        kill_a(handle, force=True)
        await asyncio.sleep(0.1)


# ---------------------------------------------------------------------------
# _MAX_BACKGROUNDED limit
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_max_backgrounded_limit(tmp_path: Path) -> None:
    """Exceeding _MAX_BACKGROUNDED should return an error and kill the excess process."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    assert run_fn is not None

    handles: list[str] = []

    # Patch to a small limit
    with patch("mindroom.shell_execution._MAX_BACKGROUNDED", 2):
        # Fill up to the limit
        for _ in range(2):
            result = await run_fn(["sleep", "300"], timeout=0)
            assert "Handle:" in result
            handles.append(result.split("Handle: ")[1].split("\n")[0])

        # Third one should fail
        result = await run_fn(["sleep", "300"], timeout=0)
        assert "Too many backgrounded processes" in result

    # Clean up
    kill_fn = tool.functions["kill_shell_command"].entrypoint
    for h in handles:
        kill_fn(h, force=True)
    await asyncio.sleep(0.1)


# ---------------------------------------------------------------------------
# Long-running command retention (finished_at not started_at)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_long_running_command_not_swept_on_finish(tmp_path: Path) -> None:
    """A command that ran for >10min should not be swept immediately on finish."""
    tool = _get_toolkit(tmp_path)
    run_fn = tool.async_functions["run_shell_command"].entrypoint
    check_fn = tool.functions["check_shell_command"].entrypoint
    assert run_fn is not None
    assert check_fn is not None

    # Force backgrounding with timeout=0
    result = await run_fn(["bash", "-c", "echo long-output; sleep 0.2"], timeout=0)
    assert "Handle:" in result
    handle = result.split("Handle: ")[1].split("\n")[0]

    # Wait for finish
    for _ in range(30):
        status = check_fn(handle)
        if "FINISHED" in status:
            break
        await asyncio.sleep(0.1)
    assert "FINISHED" in status

    # Simulate a process that started 15 minutes ago (longer than _STALE_RECORD_SECONDS)
    record = _process_registry[handle]
    record.started_at = time.monotonic() - 900

    # Trigger sweep — record should survive because finished_at is recent
    await run_fn(["echo", "trigger"])
    assert handle in _process_registry
    assert "long-output" in check_fn(handle)
