from __future__ import annotations

import json
import os
from pathlib import Path
from types import SimpleNamespace

import pytest

from opensquilla.gateway.approval_queue import get_approval_queue, reset_approval_queue
from opensquilla.sandbox.config import SandboxSettings
from opensquilla.sandbox.integration import configure_runtime, reset_runtime
from opensquilla.tools.builtin import code_exec, filesystem, shell
from opensquilla.tools.builtin.code_exec import execute_code
from opensquilla.tools.builtin.shell_policy import PolicyResult
from opensquilla.tools.types import (
    CallerKind,
    InteractionMode,
    ToolContext,
    ToolError,
    current_tool_context,
)


@pytest.fixture(autouse=True)
def reset_approval_state():
    reset_approval_queue()
    reset_runtime()
    token = current_tool_context.set(
        ToolContext(is_owner=True, caller_kind=CallerKind.CLI, session_key="agent:main:test")
    )
    yield
    current_tool_context.reset(token)
    reset_approval_queue()
    reset_runtime()


def test_audit_command_preserves_long_commands_until_cap() -> None:
    command = "rm " + ("x" * 120)
    assert shell._audit_command(command) == command

    huge = "rm " + ("x" * 5000)
    audited = shell._audit_command(huge)
    assert len(audited) > 80
    assert audited.endswith("...[truncated]")


@pytest.mark.asyncio
async def test_exec_approval_deny_pattern_blocks_shell_command(tmp_path: Path) -> None:
    get_approval_queue().set_settings("prompt", deny_patterns=["rm *"])

    result = await shell.exec_command("rm target.txt", workdir=str(tmp_path))
    payload = json.loads(result)

    assert payload["status"] == "approval_denied"
    assert payload["command"] == "rm target.txt"
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_exec_approval_deny_pattern_blocks_wrapped_shell_command(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    get_approval_queue().set_settings("prompt", deny_patterns=["rm *"])
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda _command: PolicyResult(allowed=True, reason="", needs_approval=False),
    )

    async def fail_host_execution(*args: object, **kwargs: object) -> str:
        raise AssertionError("denied wrapped command should not run")

    monkeypatch.setattr(shell, "_run_host_shell_command", fail_host_execution)

    result = await shell.exec_command(
        'powershell -NoProfile -Command "rm target.txt"',
        workdir=str(tmp_path),
    )
    payload = json.loads(result)

    assert payload["status"] == "approval_denied"
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_exec_approval_deny_pattern_does_not_depend_on_warnlist(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    get_approval_queue().set_settings("prompt", deny_patterns=["git status"])
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda _command: PolicyResult(allowed=True, reason="", needs_approval=False),
    )

    result = await shell.exec_command("git status", workdir=str(tmp_path))
    payload = json.loads(result)

    assert payload["status"] == "approval_denied"
    assert payload["command"] == "git status"


@pytest.mark.asyncio
async def test_full_host_access_warnlisted_exec_skips_approval_and_uses_host(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.run_mode = "full"
    get_approval_queue().set_settings("auto-deny")
    calls: list[str] = []

    async def fake_host_execution(*args: object, **kwargs: object) -> str:
        calls.append("host")
        return "exit_code=0\nhost\n"

    monkeypatch.setattr(shell, "_run_host_shell_command", fake_host_execution)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.exec_command("pip install requests", workdir=str(tmp_path))

    assert result == "exit_code=0\nhost\n"
    assert calls == ["host"]
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_sandbox_disabled_warnlisted_exec_skips_approval_and_uses_host(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.run_mode = "standard"
    ctx.workspace_dir = str(tmp_path)
    configure_runtime(
        SandboxSettings(sandbox=False, security_grading=False),
        workspace=tmp_path,
    )
    get_approval_queue().set_settings("auto-deny")
    calls: list[str] = []

    async def fake_host_execution(*args: object, **kwargs: object) -> str:
        calls.append("host")
        return "exit_code=0\nhost\n"

    monkeypatch.setattr(shell, "_run_host_shell_command", fake_host_execution)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.exec_command("pip install requests", workdir=str(tmp_path))

    assert result == "exit_code=0\nhost\n"
    assert calls == ["host"]
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_warnlisted_exec_allow_pattern_skips_prompt_when_sandbox_enabled(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.workspace_dir = str(tmp_path)
    get_approval_queue().set_settings(
        "prompt",
        allow_patterns=["pip install requests"],
    )
    configure_runtime(
        SandboxSettings(
            sandbox=True,
            security_grading=True,
            backend="noop",
            allow_legacy_mode=True,
        ),
        workspace=tmp_path,
    )
    calls: list[tuple[str, object]] = []

    async def fake_gate_action(**kwargs: object) -> tuple[object, object, object]:
        calls.append(("gate", kwargs))
        policy = SimpleNamespace(network=None)
        request = SimpleNamespace(cwd=tmp_path, action_kind="shell.exec", policy=policy)
        return object(), policy, request

    async def fake_sandbox(request: object, *, runtime: object = None) -> object:
        calls.append(("backend", request))
        return SimpleNamespace(
            returncode=0,
            stdout="sandboxed\n",
            stderr="",
            timed_out=False,
            backend_notes=(),
        )

    monkeypatch.setattr(shell, "gate_action", fake_gate_action)
    monkeypatch.setattr(shell, "run_under_backend", fake_sandbox)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.exec_command("pip install requests", workdir=str(tmp_path))

    assert "sandboxed" in result
    assert [name for name, _ in calls] == ["gate", "backend"]
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_warnlisted_exec_allow_pattern_runs_without_approval_when_no_sandbox_gate(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    get_approval_queue().set_settings(
        "prompt",
        allow_patterns=["pip install requests"],
    )
    calls: list[str] = []

    async def fake_host_execution(*args: object, **kwargs: object) -> str:
        calls.append("host")
        return "exit_code=0\nhost\n"

    monkeypatch.setattr(shell, "_run_host_shell_command", fake_host_execution)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.exec_command("pip install requests", workdir=str(tmp_path))

    assert result == "exit_code=0\nhost\n"
    assert get_approval_queue().list_pending("exec") == []
    assert calls == ["host"]


@pytest.mark.asyncio
async def test_warnlisted_exec_prompt_runs_without_approval_when_no_sandbox_gate(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    get_approval_queue().set_settings("prompt")
    calls: list[str] = []

    async def fake_host_execution(*args: object, **kwargs: object) -> str:
        calls.append("host")
        return "exit_code=0\nhost\n"

    monkeypatch.setattr(shell, "_run_host_shell_command", fake_host_execution)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.exec_command("pip install requests", workdir=str(tmp_path))

    assert result == "exit_code=0\nhost\n"
    assert get_approval_queue().list_pending("exec") == []
    assert calls == ["host"]


@pytest.mark.asyncio
async def test_warnlisted_exec_auto_approve_runs_without_approval_when_no_sandbox_gate(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    get_approval_queue().set_settings("auto-approve")
    calls: list[str] = []

    async def fake_host_execution(*args: object, **kwargs: object) -> str:
        calls.append("host")
        return "exit_code=0\nhost\n"

    monkeypatch.setattr(shell, "_run_host_shell_command", fake_host_execution)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.exec_command("pip install requests", workdir=str(tmp_path))

    assert result == "exit_code=0\nhost\n"
    assert get_approval_queue().list_pending("exec") == []
    assert calls == ["host"]


@pytest.mark.asyncio
async def test_warnlisted_exec_unattended_runs_without_approval_when_no_sandbox_gate(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    get_approval_queue().set_settings("prompt")
    calls: list[str] = []

    async def fake_host_execution(*args: object, **kwargs: object) -> str:
        calls.append("host")
        return "exit_code=0\nhost\n"

    monkeypatch.setattr(shell, "_run_host_shell_command", fake_host_execution)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.exec_command("pip install requests", workdir=str(tmp_path))

    assert result == "exit_code=0\nhost\n"
    assert calls == ["host"]
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_destructive_code_exec_uses_sandbox_gate_when_runtime_enabled(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.workspace_dir = str(tmp_path)
    target = tmp_path / "target.txt"
    target.write_text("delete me", encoding="utf-8")
    configure_runtime(
        SandboxSettings(
            sandbox=True,
            security_grading=True,
            backend="noop",
            allow_legacy_mode=True,
        ),
        workspace=tmp_path,
    )
    calls: list[tuple[str, object]] = []

    async def fake_gate_action(**kwargs):
        calls.append(("gate", kwargs))
        policy = SimpleNamespace(network=None)
        request = SimpleNamespace(cwd=tmp_path, action_kind="code.exec", policy=policy)
        return object(), policy, request

    async def fake_sandbox(request: object, *, runtime: object = None) -> object:
        calls.append(("backend", request))
        return SimpleNamespace(
            returncode=0,
            stdout="sandboxed\n",
            stderr="",
            timed_out=False,
            backend_notes=(),
        )

    monkeypatch.setattr(code_exec, "gate_action", fake_gate_action)
    monkeypatch.setattr(code_exec, "run_under_backend", fake_sandbox)

    result = await execute_code("import os\nos.remove('target.txt')")
    payload = json.loads(result)

    assert payload["exit_code"] == 0
    assert payload["stdout"] == "sandboxed\n"
    assert [name for name, _ in calls] == ["gate", "backend"]
    hints = calls[0][1]["hints"]  # type: ignore[index]
    assert hints.high_impact is True
    assert target.exists()
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_warnlist_background_process_uses_sandbox_gate_when_runtime_enabled(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.workspace_dir = str(tmp_path)
    configure_runtime(
        SandboxSettings(
            sandbox=True,
            security_grading=True,
            backend="noop",
            allow_legacy_mode=True,
        ),
        workspace=tmp_path,
    )
    calls: list[tuple[str, object]] = []

    class _FakeStream:
        async def read(self, size: int) -> bytes:
            return b""

    class _FakeProcess:
        stdout = _FakeStream()
        stdin = None
        returncode = 0

        async def wait(self) -> int:
            return 0

    async def fake_gate_action(**kwargs):
        calls.append(("gate", kwargs))
        policy = SimpleNamespace(network=None, network_proxy=None)
        request = SimpleNamespace(cwd=tmp_path, action_kind="shell.background", policy=policy)
        return object(), policy, request

    async def fake_sandbox_spawn(*args: object, **kwargs: object) -> object:
        calls.append(("backend", kwargs.get("request")))
        return shell._SpawnedBackgroundProcess(process=_FakeProcess())  # type: ignore[arg-type]

    monkeypatch.setattr(shell, "gate_action", fake_gate_action)
    monkeypatch.setattr(shell, "_spawn_sandboxed_background_process", fake_sandbox_spawn)
    monkeypatch.setattr(
        shell,
        "check_safe_bin",
        lambda command: PolicyResult(
            allowed=True,
            reason=f"command requires approval: {command}",
            needs_approval=True,
        ),
    )

    result = await shell.background_process("rm target.txt", workdir=str(tmp_path))

    assert "status: running" in result
    session_id = result.splitlines()[0].split("=", 1)[1]
    session = shell._bg_sessions[session_id]
    assert session.collector_task is not None
    await session.collector_task
    assert [name for name, _ in calls] == ["gate", "backend"]
    hints = calls[0][1]["hints"]  # type: ignore[index]
    assert hints.high_impact is True
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_outside_workspace_write_blocks_without_sandbox_path_approval(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    outside = tmp_path / "outside.txt"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.workspace_dir = str(workspace)

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    payload = json.loads(await write_file(str(outside), "ok"))

    assert payload["status"] == "blocked"
    assert payload["reason"] == "outside_workspace"
    assert not outside.exists()
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_workspace_lockdown_blocks_outside_workspace_write_even_with_bypass(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    outside = tmp_path / "outside.txt"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    with pytest.raises(ToolError, match="workspace lockdown"):
        await write_file(str(outside), "ok")

    assert not outside.exists()
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_workspace_lockdown_allows_configured_scratch_dir_write(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    target = scratch / "debug.py"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    result = await filesystem._gate_out_of_workspace_write(
        "write_file",
        target.resolve(strict=False),
        str(target),
        None,
    )

    assert result is None
    assert not target.exists()
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_workspace_write_deny_globs_block_file_write(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    target = workspace / "blocked" / "generated.txt"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_write_deny_globs = ["blocked/**"]  # type: ignore[attr-defined]

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    with pytest.raises(ToolError, match="workspace write deny policy"):
        await write_file(str(target), "nope")

    assert not target.exists()
    assert get_approval_queue().list_pending("exec") == []


@pytest.mark.asyncio
async def test_workspace_write_deny_globs_block_nested_test_file_write(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    target = workspace / "myproj" / "src" / "test" / "java" / "com" / "example"
    target = target / "SampleTest.java"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_write_deny_globs = ["**/test/**", "**/*Test.java"]  # type: ignore[attr-defined]

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    with pytest.raises(ToolError, match="workspace write deny policy"):
        await write_file(str(target), "nope")

    assert not target.exists()
    assert len(get_approval_queue().list_pending("exec")) == 0


@pytest.mark.asyncio
async def test_workspace_write_deny_globs_allow_configured_scratch_write_file(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    target = scratch / "test_bug.py"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.workspace_dir = str(workspace)
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]
    ctx.workspace_write_deny_globs = ["**/test_*.py"]  # type: ignore[attr-defined]

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    result = await write_file(str(target), "print('scratch')\n")

    assert result.startswith("Written 17 bytes to ")
    assert target.read_text(encoding="utf-8") == "print('scratch')\n"
    assert len(get_approval_queue().list_pending("exec")) == 0


@pytest.mark.asyncio
async def test_configured_scratch_dir_blocks_root_debug_workspace_write_file(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    target = workspace / "debug_regex.php"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    with pytest.raises(ToolError, match="configured scratch directory"):
        await write_file(str(target), "<?php echo 'debug';\n")

    assert not target.exists()


@pytest.mark.asyncio
async def test_configured_scratch_dir_allows_plain_root_test_source_name(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    target = workspace / "test_api.py"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    result = await write_file(str(target), "def test_api():\n    assert True\n")

    assert "Written 32 bytes" in result
    assert str(target) in result
    assert target.exists()


@pytest.mark.asyncio
async def test_write_file_blocks_large_workspace_file_fragment_overwrite(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    target = workspace / "src" / "main.c"
    target.parent.mkdir()
    target.write_text("int keep;\n" * 700, encoding="utf-8")
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)

    await filesystem.read_file(str(target))
    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    with pytest.raises(ToolError, match="write_file refused to overwrite"):
        await write_file(str(target), "int replacement_fragment;\n")

    assert target.read_text(encoding="utf-8").startswith("int keep;\n")


@pytest.mark.asyncio
async def test_write_file_allows_scratch_file_fragment_overwrite(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    target = scratch / "debug.c"
    target.write_text("int keep;\n" * 700, encoding="utf-8")
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]

    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    result = await write_file(str(target), "int replacement_fragment;\n")

    assert result.startswith("Written 26 bytes to ")
    assert target.read_text(encoding="utf-8") == "int replacement_fragment;\n"


@pytest.mark.asyncio
async def test_write_file_allows_small_workspace_file_overwrite(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    target = workspace / "README.md"
    target.write_text("old\n", encoding="utf-8")
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)

    await filesystem.read_file(str(target))
    write_file = filesystem.write_file.__wrapped__.__wrapped__  # type: ignore[attr-defined]
    result = await write_file(str(target), "new\n")

    assert result.startswith("Written 4 bytes to ")
    assert target.read_text(encoding="utf-8") == "new\n"


def test_tool_definitions_include_scratch_guidance_when_configured(tmp_path: Path) -> None:
    from opensquilla.tools.registry import get_default_registry

    scratch = tmp_path / "scratch"
    ctx = ToolContext(is_owner=True, scratch_dir=str(scratch))

    tools = get_default_registry().to_tool_definitions(ctx)
    descriptions = {tool.name: tool.description for tool in tools}

    assert str(scratch) in descriptions["exec_command"]
    assert str(scratch) in descriptions["write_file"]


def test_exec_command_description_names_windows_powershell_shell() -> None:
    from opensquilla.tools.registry import get_default_registry

    ctx = ToolContext(is_owner=True)

    tools = get_default_registry().to_tool_definitions(ctx)
    descriptions = {tool.name: tool.description for tool in tools}

    assert "PowerShell" in descriptions["exec_command"]
    assert "cmd /c" in descriptions["exec_command"]


@pytest.mark.asyncio
async def test_workspace_lockdown_blocks_obvious_outside_shell_redirection(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    outside = tmp_path / "outside.txt"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    result = await shell.exec_command(f"echo ok > {outside}", workdir=str(workspace))
    payload = json.loads(result)

    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_lockdown"


@pytest.mark.asyncio
@pytest.mark.skipif(os.name != "posix", reason="/dev/null redirection is POSIX-specific")
async def test_workspace_lockdown_allows_dev_null_shell_redirection(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    result = await shell.exec_command("printf ok 2>/dev/null", workdir=str(workspace))

    assert result == "exit_code=0\nok"


@pytest.mark.asyncio
@pytest.mark.skipif(os.name != "posix", reason="/dev/null redirection is POSIX-specific")
async def test_workspace_lockdown_dev_null_does_not_hide_sensitive_operand(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    result = await shell.exec_command(
        "cat /dev/sda 2>/dev/null",
        workdir=str(workspace),
    )
    payload = json.loads(result)

    assert payload["status"] == "blocked"
    assert payload["reason"] == "sensitive_path"


@pytest.mark.asyncio
async def test_workspace_write_deny_globs_block_shell_redirection(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_write_deny_globs = ["reports/*.txt"]  # type: ignore[attr-defined]

    result = await shell.exec_command("echo ok > reports/out.txt", workdir=str(workspace))

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_write_deny"
    assert payload["matched_pattern"] == "reports/*.txt"


@pytest.mark.asyncio
@pytest.mark.skipif(os.name != "posix", reason="heredoc redirection is POSIX-specific")
async def test_workspace_write_deny_globs_allow_shell_redirection_after_cd_to_scratch(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    target = scratch / "test_bug.py"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]
    ctx.workspace_write_deny_globs = ["**/test_*.py"]  # type: ignore[attr-defined]

    result = await shell.exec_command(
        f"cd {scratch} && cat > test_bug.py << 'EOF'\nprint('scratch')\nEOF\ncat test_bug.py",
        workdir=str(workspace),
    )

    assert result == "exit_code=0\nprint('scratch')\n"
    assert target.read_text(encoding="utf-8") == "print('scratch')\n"
    assert not (workspace / "test_bug.py").exists()


@pytest.mark.asyncio
async def test_configured_scratch_dir_blocks_root_test_shell_redirection(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    target = workspace / "test_bug.php"
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]
    ctx.workspace_lockdown = True  # type: ignore[attr-defined]

    result = await shell.exec_command(
        "cat > test_bug.php << 'EOF'\n<?php echo 'debug';\nEOF\n",
        workdir=str(workspace),
    )

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_scratch_artifact"
    assert str(scratch) in payload["message"]
    assert not target.exists()


@pytest.mark.asyncio
async def test_workspace_write_deny_globs_inspect_shell_stdin(tmp_path: Path) -> None:
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.interaction_mode = InteractionMode.UNATTENDED
    ctx.elevated = "bypass"
    ctx.workspace_dir = str(workspace)
    ctx.workspace_write_deny_globs = ["reports/*.txt"]  # type: ignore[attr-defined]

    result = await shell.exec_command(
        "sh",
        workdir=str(workspace),
        stdin="mkdir -p reports\necho secret > reports/out.txt\n",
    )

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_write_deny"
    assert payload["matched_pattern"] == "reports/*.txt"


@pytest.mark.asyncio
async def test_bypass_still_blocks_sensitive_shell_targets() -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.elevated = "bypass"

    result = await shell.exec_command("rm ~/.ssh/id_rsa")
    payload = json.loads(result)

    assert payload["status"] == "blocked"
    assert payload["reason"] == "sensitive_path"
    assert shell.full_host_access_active() is False


@pytest.mark.asyncio
async def test_bypass_does_not_override_safe_bin_hard_denies() -> None:
    ctx = current_tool_context.get()
    assert ctx is not None
    ctx.elevated = "bypass"

    with pytest.raises(ToolError, match="command blocked by policy"):
        await shell.exec_command("Clear-Disk")