"""Commands must return after timeout even when descendants inherit output."""

import ctypes
import io
import os
import shlex
import signal
import subprocess
import sys
import time
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock

import pytest
from pydantic import ValidationError

from deerflow.community.aio_sandbox import aio_sandbox as aio_module
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.sandbox.local.local_sandbox import LocalSandbox


@pytest.fixture
def python_sandbox(monkeypatch):
    sandbox = LocalSandbox("command-liveness-test")
    # A Python parent stands in for a shell, keeping these process-tree tests
    # independent of bash installation and PowerShell quoting differences.
    monkeypatch.setattr(sandbox, "_get_shell", lambda: sys.executable)
    monkeypatch.setattr(sandbox, "_command_timeout_seconds", lambda: 10)
    return sandbox


def _is_running(pid: int) -> bool:
    if os.name == "nt":
        kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
        kernel32.OpenProcess.restype = ctypes.c_void_p
        kernel32.OpenProcess.argtypes = [ctypes.c_uint, ctypes.c_bool, ctypes.c_uint]
        kernel32.WaitForSingleObject.argtypes = [ctypes.c_void_p, ctypes.c_uint]
        kernel32.CloseHandle.argtypes = [ctypes.c_void_p]
        handle = kernel32.OpenProcess(0x00100000, False, pid)  # SYNCHRONIZE
        if not handle:
            return False
        try:
            return kernel32.WaitForSingleObject(handle, 0) == 0x102  # WAIT_TIMEOUT
        finally:
            kernel32.CloseHandle(handle)
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    # A killed grandchild can briefly remain as a zombie until its new parent
    # reaps it. It cannot execute work or retain pipe handles in this state.
    stat = Path(f"/proc/{pid}/stat")
    if stat.exists() and stat.read_text().rsplit(")", 1)[1].split()[0] == "Z":
        return False
    return True


def test_command_preserves_stdout_stderr_and_exit_code(python_sandbox):
    output = python_sandbox.execute_command("import sys; print('stdout marker'); print('stderr marker', file=sys.stderr); sys.exit(7)")
    assert "stdout marker" in output
    assert "Std Error:\nstderr marker" in output
    assert "Exit Code: 7" in output


def test_command_stdin_is_closed(python_sandbox):
    output = python_sandbox.execute_command("import sys; print(repr(sys.stdin.read()))")
    assert output.strip() == "''"


def test_command_timeout_terminates_child_that_inherits_output(python_sandbox, tmp_path, monkeypatch):
    monkeypatch.setattr(python_sandbox, "_command_timeout_seconds", lambda: 2)
    pid_file = tmp_path / "child.pid"
    child = "import time; time.sleep(60)"
    command = f"import pathlib,subprocess,sys,time; child=subprocess.Popen([sys.executable,'-c',{child!r}]); pathlib.Path({str(pid_file)!r}).write_text(str(child.pid)); print('started', flush=True); time.sleep(60)"
    started = time.monotonic()
    child_pid = None
    try:
        output = python_sandbox.execute_command(command)
        assert time.monotonic() - started < 8
        assert output.startswith("Error: Command timed out after 2 seconds")
        assert "started" in output
        child_pid = int(pid_file.read_text())
        deadline = time.monotonic() + 2
        while _is_running(child_pid) and time.monotonic() < deadline:
            time.sleep(0.02)
        assert not _is_running(child_pid), "Timed-out skill child was left running"
    finally:
        if child_pid is None and pid_file.exists():
            child_pid = int(pid_file.read_text())
        if child_pid is not None and _is_running(child_pid):
            os.kill(child_pid, signal.SIGTERM)


def test_shell_exit_does_not_wait_for_descendants_output_handles(python_sandbox, tmp_path):
    pid_file = tmp_path / "background-child.pid"
    command = f"import pathlib,subprocess,sys; child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(60)']); pathlib.Path({str(pid_file)!r}).write_text(str(child.pid)); print('parent done')"
    try:
        started = time.monotonic()
        output = python_sandbox.execute_command(command)
        assert time.monotonic() - started < 8
        assert output.strip() == "parent done"
    finally:
        if pid_file.exists():
            child_pid = int(pid_file.read_text())
            if _is_running(child_pid):
                os.kill(child_pid, signal.SIGTERM)


def test_large_command_output_is_bounded_before_tool_truncation(python_sandbox):
    output = python_sandbox.execute_command("import sys; sys.stdout.write('HEAD' + 'x' * (3 * 1024 * 1024) + 'TAIL')")
    assert output.startswith("HEAD") and output.endswith("TAIL")
    assert "command output truncated" in output
    assert len(output) < LocalSandbox._COMMAND_OUTPUT_MAX_BYTES + 100


def test_runaway_output_is_stopped_before_command_timeout(python_sandbox, monkeypatch):
    monkeypatch.setattr(python_sandbox, "_COMMAND_CAPTURE_MAX_BYTES", 8192)
    output = python_sandbox.execute_command("import sys,time\nwhile True:\n sys.stdout.write('x' * 4096); sys.stdout.flush(); time.sleep(0.01)")
    assert output.startswith("Error: Command output exceeded 8192 bytes")


def test_command_timeout_setting_and_default(python_sandbox, monkeypatch):
    from deerflow import config

    settings = SandboxConfig(use="test", command_timeout_seconds=12.5)
    monkeypatch.setattr(config, "get_app_config", lambda: SimpleNamespace(sandbox=settings))
    assert LocalSandbox._command_timeout_seconds(python_sandbox) == 12.5
    assert SandboxConfig(use="test").command_timeout_seconds == 600
    for invalid in (0, -1, float("inf"), float("nan")):
        with pytest.raises(ValidationError):
            SandboxConfig(use="test", command_timeout_seconds=invalid)


@pytest.mark.parametrize("operation", ["read_file", "read_file_range"])
def test_local_file_read_rejects_pipe_before_reading(monkeypatch, operation):
    read_fd, write_fd = os.pipe()
    try:
        monkeypatch.setattr(os, "open", lambda *_args: os.dup(read_fd))
        with pytest.raises(OSError, match="Only regular files"):
            getattr(LocalSandbox("test"), operation)("/virtual/fifo")
    finally:
        os.close(read_fd)
        os.close(write_fd)


def test_local_range_reader_stops_at_budget_without_draining_long_line(monkeypatch):
    class LongLine(io.StringIO):
        reads = 0

        def readline(self, size=-1):
            self.reads += 1
            if self.reads > 1:
                raise AssertionError("Read the rest of the long line after filling the result")
            return "x" * 4097

    sandbox = LocalSandbox("test")
    monkeypatch.setattr(sandbox, "_open_regular_text", lambda *_args, **_kwargs: LongLine())
    assert sandbox.read_file_range("/virtual/long-line", max_chars=100) == ("x" * 100, 2, True)


@pytest.mark.parametrize(
    ("operation", "args"),
    [
        ("execute_command", ("echo test",)),
        ("read_file_range", ("/tmp/skill.md",)),
        ("list_dir", ("/tmp",)),
        ("write_file", ("/tmp/output", "test")),
        ("update_file", ("/tmp/output", b"test")),
    ],
)
def test_aio_busy_session_fails_without_executing_or_releasing_other_request(monkeypatch, operation, args):
    client = Mock()
    monkeypatch.setattr(aio_module, "AioSandboxClient", Mock(return_value=client))
    sandbox = aio_module.AioSandbox("test", "http://127.0.0.1:1")
    sandbox._LOCK_WAIT_TIMEOUT = 0.01
    sandbox._lock.acquire()
    try:
        started = time.monotonic()
        with pytest.raises(TimeoutError, match="waiting for the previous operation"):
            getattr(sandbox, operation)(*args)
        assert time.monotonic() - started < 1
        assert sandbox._lock.locked()
        assert client.mock_calls == []
    finally:
        sandbox._lock.release()
    client.shell.exec_command.return_value = SimpleNamespace(data=SimpleNamespace(output="next command"))
    assert sandbox.execute_command("echo next") == "next command"
    assert not sandbox._lock.locked()


def test_aio_range_reader_preserves_continuation_and_rejects_non_files(monkeypatch, tmp_path):
    def execute_script(*, command, no_change_timeout):
        # Run only the generated bounded-reader program against test fixtures;
        # no container or external sandbox service is contacted.
        args = shlex.split(command)
        completed = subprocess.run([sys.executable, *args[1:]], capture_output=True, encoding="utf-8", timeout=5)
        return SimpleNamespace(data=SimpleNamespace(output=completed.stdout + completed.stderr))

    client = Mock()
    client.shell.exec_command.side_effect = execute_script
    monkeypatch.setattr(aio_module, "AioSandboxClient", Mock(return_value=client))
    sandbox = aio_module.AioSandbox("test", "http://127.0.0.1:1")
    source = tmp_path / "long.txt"
    source.write_text("x" * 10_000 + "\nsecond\nthird", encoding="utf-8")
    assert sandbox.read_file_range(str(source), max_chars=100) == ("x" * 100, 2, True)
    assert sandbox.read_file_range(str(source), start_line=2, end_line=2, max_chars=100) == ("second", None, False)
    with pytest.raises(OSError, match="bounded sandbox read failed"):
        sandbox.read_file_range(str(tmp_path))