"""Opt-in workspace write deny levers: host shell, mutator targets, guidance.

Covers OPENSQUILLA_WORKSPACE_WRITE_DENY_HOST_SHELL,
OPENSQUILLA_WORKSPACE_WRITE_DENY_COMMAND_TARGETS and
OPENSQUILLA_WORKSPACE_WRITE_DENY_GUIDANCE (all off by default). Motivation:
deny globs are enforced for filesystem and patch tools, but shell-side
enforcement only recognizes redirection/tee write targets and is skipped
entirely under host execution, and the default block message suggests
recreating the file under the scratch directory — the wrong remediation when
deny globs protect files that must not be modified at all (e.g. test files).
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from opensquilla.gateway.approval_queue import reset_approval_queue
from opensquilla.sandbox.integration import reset_runtime
from opensquilla.tools import write_policy
from opensquilla.tools.builtin import shell
from opensquilla.tools.types import (
    CallerKind,
    InteractionMode,
    SafeToolError,
    ToolContext,
    current_tool_context,
)

_HOST_SHELL_ENV = "OPENSQUILLA_WORKSPACE_WRITE_DENY_HOST_SHELL"
_COMMAND_TARGETS_ENV = "OPENSQUILLA_WORKSPACE_WRITE_DENY_COMMAND_TARGETS"
_GUIDANCE_ENV = "OPENSQUILLA_WORKSPACE_WRITE_DENY_GUIDANCE"


@pytest.fixture(autouse=True)
def _tool_context(monkeypatch: pytest.MonkeyPatch):
    monkeypatch.delenv(_HOST_SHELL_ENV, raising=False)
    monkeypatch.delenv(_COMMAND_TARGETS_ENV, raising=False)
    monkeypatch.delenv(_GUIDANCE_ENV, raising=False)
    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 _configure_ctx(workspace: Path, globs: list[str]) -> ToolContext:
    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 = globs  # type: ignore[attr-defined]
    return ctx


@pytest.mark.parametrize(
    ("command", "expected"),
    [
        ("sed -i 's/a/b/' tests/test_a.py", ["tests/test_a.py"]),
        ("sed -i.bak -e 's/a/b/' tests/test_a.py", ["tests/test_a.py"]),
        (
            "sed -ri 's/a/b/' tests/test_a.py tests/test_b.py",
            ["tests/test_a.py", "tests/test_b.py"],
        ),
        ("sed --in-place=.bak 's/a/b/' tests/test_a.py", ["tests/test_a.py"]),
        # BSD sed: the empty token is -i's backup suffix, not the script.
        ("sed -i '' 's/a/b/' tests/test_a.py", ["tests/test_a.py"]),
        # Script delimiters and separators are not shell operators.
        ("sed -i 's|a|b|' tests/test_a.py", ["tests/test_a.py"]),
        ("sed -i 's/a/b/; s/c/d/' tests/test_a.py", ["tests/test_a.py"]),
        # No in-place flag: read-only sed must not produce write targets.
        ("sed -n '1p' tests/test_a.py", []),
        ("perl -pi -e 's/a/b/' tests/test_a.py", ["tests/test_a.py"]),
        ("perl -i.bak -pe 's/a/b/' tests/test_a.py", ["tests/test_a.py"]),
        ("perl -e 'print 1' tests/test_a.py", []),
        # -I consumes the rest of the token; its 'i' is not the in-place switch.
        ("perl -Ilib -e 'print 1' tests/test_a.py", []),
        ("rm -f tests/test_a.py", ["tests/test_a.py"]),
        # Directory operands are extracted; the gate matches them as dir/**.
        ("rm -rf tests", ["tests"]),
        ("unlink tests/test_a.py", ["tests/test_a.py"]),
        # mv mutates both operands: the source is removed, the dest written.
        ("mv tests/test_a.py /tmp/aside.py", ["tests/test_a.py", "/tmp/aside.py"]),
        ("cp /tmp/patched.py tests/test_a.py", ["tests/test_a.py"]),
        ("cp -t tests src_a.py", ["tests"]),
        ("dd if=/dev/zero of=tests/test_a.py bs=1", ["tests/test_a.py"]),
        ("truncate -s 0 tests/test_a.py", ["tests/test_a.py"]),
        ("git rm tests/test_a.py", ["tests/test_a.py"]),
        # git rm --cached only unstages; the worktree file is untouched.
        ("git rm --cached tests/test_a.py", []),
        ("git -C sub mv tests/test_a.py aside.py", ["tests/test_a.py", "aside.py"]),
        # git checkout/restore recover original content; not treated as writes.
        ("git checkout -- tests/test_a.py", []),
        ("LC_ALL=C sed -i 's/a/b/' tests/test_a.py", ["tests/test_a.py"]),
        (
            "cat tests/test_a.py | grep x && sed -i 's/a/b/' tests/test_b.py",
            ["tests/test_b.py"],
        ),
        # Operators inside quotes are data, not command separators.
        ("git commit -m 'cleanup; rm tests/test_a.py'", []),
        # Heredoc bodies are data, not commands.
        ("cat <<'EOF' > /tmp/notes.txt\nrm tests/test_a.py\nEOF", []),
        ("pytest tests/test_a.py", []),
        ("cat tests/test_a.py", []),
    ],
)
def test_mutating_command_write_target_extraction(command: str, expected: list[str]) -> None:
    assert shell._mutating_command_write_targets(command) == expected


def test_command_targets_lever_default_off_skips_mutator_extraction(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    (workspace / "tests").mkdir(parents=True)
    _configure_ctx(workspace, ["tests/**"])

    block = shell._workspace_write_deny_shell_block(
        "exec_command", "sed -i 's/a/b/' tests/test_a.py", str(workspace)
    )

    assert block is None


def test_command_targets_lever_adds_mutator_targets(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    workspace = tmp_path / "workspace"
    (workspace / "tests").mkdir(parents=True)
    _configure_ctx(workspace, ["tests/**"])

    block = shell._workspace_write_deny_shell_block(
        "exec_command", "sed -i 's/a/b/' tests/test_a.py", str(workspace)
    )

    assert block is not None
    assert block["reason"] == "workspace_write_deny"
    assert block["matched_pattern"] == "tests/**"
    assert block["target"] == "tests/test_a.py"


def test_command_targets_lever_blocks_directory_operands(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    (tests_dir / "test_a.py").write_text("assert a\n", encoding="utf-8")
    _configure_ctx(workspace, ["tests/**"])

    for command in ("rm -rf tests", "cp -t tests src_a.py"):
        block = shell._workspace_write_deny_shell_block(
            "exec_command", command, str(workspace)
        )
        assert block is not None, command
        assert block["reason"] == "workspace_write_deny"
        assert block["matched_pattern"] == "tests/**"
        assert block["target"] == "tests"


def test_command_targets_quoted_operators_and_heredocs_stay_data(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    (tests_dir / "test_a.py").write_text("assert a\n", encoding="utf-8")
    _configure_ctx(workspace, ["tests/**"])

    for command in (
        "git commit -m 'cleanup; rm tests/test_a.py'",
        "cat <<'EOF' > notes.txt\nrm tests/test_a.py\nEOF",
    ):
        block = shell._workspace_write_deny_shell_block(
            "exec_command", command, str(workspace)
        )
        assert block is None, command


def test_match_workspace_write_deny_directory_form_is_opt_in(
    tmp_path: Path,
) -> None:
    # The trailing-slash candidates only apply on request; existing callers
    # (file tools, patch gate) keep byte-identical matching for bare dirs.
    workspace = tmp_path / "workspace"
    (workspace / "tests").mkdir(parents=True)
    _configure_ctx(workspace, ["tests/**"])
    target = workspace / "tests"

    assert write_policy.match_workspace_write_deny(target, workspace=workspace) is None

    match = write_policy.match_workspace_write_deny(
        target, workspace=workspace, as_directory=True
    )
    assert match is not None
    assert match.pattern == "tests/**"


@pytest.mark.asyncio
async def test_exec_command_blocks_mutator_on_denied_path_when_lever_on(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    target = tests_dir / "test_a.py"
    target.write_text("assert a\n", encoding="utf-8")
    _configure_ctx(workspace, ["tests/**"])

    result = await shell.exec_command(
        "sed -i 's/a/b/' tests/test_a.py", workdir=str(workspace)
    )

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_write_deny"
    assert payload["matched_pattern"] == "tests/**"
    assert target.read_text(encoding="utf-8") == "assert a\n"


@pytest.mark.asyncio
async def test_background_process_blocks_mutator_on_denied_path_when_lever_on(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    target = tests_dir / "test_a.py"
    target.write_text("assert a\n", encoding="utf-8")
    _configure_ctx(workspace, ["tests/**"])

    result = await shell.background_process(
        "sed -i 's/a/b/' tests/test_a.py", workdir=str(workspace)
    )

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_write_deny"
    assert payload["tool"] == "background_process"
    assert target.read_text(encoding="utf-8") == "assert a\n"


@pytest.mark.asyncio
async def test_exec_command_scans_stdin_for_mutators_when_lever_on(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    (tests_dir / "test_a.py").write_text("assert a\n", encoding="utf-8")
    _configure_ctx(workspace, ["tests/**"])

    result = await shell.exec_command(
        "sh",
        workdir=str(workspace),
        stdin="sed -i 's/a/b/' tests/test_a.py\n",
    )

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_write_deny"


@pytest.mark.asyncio
async def test_exec_command_mutator_passes_through_by_default(
    tmp_path: Path,
) -> None:
    # Documents the default gap the lever closes: without the lever, in-place
    # mutators are not recognized as writes and execute against denied paths.
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    target = tests_dir / "test_a.py"
    target.write_text("assert a\n", encoding="utf-8")
    _configure_ctx(workspace, ["tests/**"])

    result = await shell.exec_command(
        "sed -i 's/assert a/assert b/' tests/test_a.py", workdir=str(workspace)
    )

    assert result.startswith("exit_code=0")
    assert target.read_text(encoding="utf-8") == "assert b\n"


@pytest.mark.asyncio
async def test_exec_command_reads_stay_unblocked_with_lever_on(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    (tests_dir / "test_a.py").write_bytes(b"assert a\n")
    _configure_ctx(workspace, ["tests/**"])

    result = await shell.exec_command("cat tests/test_a.py", workdir=str(workspace))

    assert result == "exit_code=0\nassert a\n"


@pytest.mark.asyncio
async def test_host_shell_lever_enforces_deny_globs_under_host_execution(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_HOST_SHELL_ENV, "1")
    monkeypatch.setattr(shell, "_host_execution_allowed", lambda: True)
    workspace = tmp_path / "workspace"
    (workspace / "tests").mkdir(parents=True)
    _configure_ctx(workspace, ["tests/**"])

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

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_write_deny"
    assert not (workspace / "tests" / "out.txt").exists()


@pytest.mark.asyncio
async def test_host_shell_default_off_keeps_host_execution_unchecked(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    # Documents the default gap the lever closes: under host execution the
    # whole sandbox policy block — including deny globs — is skipped.
    monkeypatch.setattr(shell, "_host_execution_allowed", lambda: True)
    workspace = tmp_path / "workspace"
    (workspace / "tests").mkdir(parents=True)
    _configure_ctx(workspace, ["tests/**"])

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

    assert result.startswith("exit_code=0")
    assert (workspace / "tests" / "out.txt").exists()


@pytest.mark.asyncio
async def test_host_shell_and_command_targets_levers_compose(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv(_HOST_SHELL_ENV, "1")
    monkeypatch.setenv(_COMMAND_TARGETS_ENV, "1")
    monkeypatch.setattr(shell, "_host_execution_allowed", lambda: True)
    workspace = tmp_path / "workspace"
    tests_dir = workspace / "tests"
    tests_dir.mkdir(parents=True)
    target = tests_dir / "test_a.py"
    target.write_text("assert a\n", encoding="utf-8")
    _configure_ctx(workspace, ["tests/**"])

    result = await shell.exec_command(
        "sed -i 's/a/b/' tests/test_a.py", workdir=str(workspace)
    )

    payload = json.loads(result)
    assert payload["status"] == "blocked"
    assert payload["reason"] == "workspace_write_deny"
    assert target.read_text(encoding="utf-8") == "assert a\n"


def test_deny_guidance_env_overrides_scratch_guidance(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    guidance = "Protected files must not be modified; change the source instead."
    monkeypatch.setenv(_GUIDANCE_ENV, guidance)
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    ctx = _configure_ctx(workspace, ["tests/**"])
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]

    match = write_policy.match_workspace_write_deny(
        workspace / "tests" / "test_a.py", workspace=workspace
    )
    assert match is not None
    block = write_policy.workspace_write_deny_block("write_file", match)

    message = str(block["message"])
    assert message.endswith(guidance)
    assert "scratch directory" not in message


def test_deny_guidance_default_keeps_scratch_guidance(
    tmp_path: Path,
) -> None:
    workspace = tmp_path / "workspace"
    scratch = tmp_path / "scratch"
    workspace.mkdir()
    scratch.mkdir()
    ctx = _configure_ctx(workspace, ["tests/**"])
    ctx.scratch_dir = str(scratch)  # type: ignore[attr-defined]

    match = write_policy.match_workspace_write_deny(
        workspace / "tests" / "test_a.py", workspace=workspace
    )
    assert match is not None
    block = write_policy.workspace_write_deny_block("write_file", match)

    message = str(block["message"])
    assert "must be written under the configured scratch directory" in message
    assert str(scratch) in message


def test_gate_raise_carries_same_message_as_block_envelope(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    guidance = "Fix the source code so the existing checks pass."
    monkeypatch.setenv(_GUIDANCE_ENV, guidance)
    workspace = tmp_path / "workspace"
    workspace.mkdir()
    _configure_ctx(workspace, ["tests/**"])
    target = workspace / "tests" / "test_a.py"

    match = write_policy.match_workspace_write_deny(target, workspace=workspace)
    assert match is not None
    expected = str(write_policy.workspace_write_deny_block("write_file", match)["message"])

    with pytest.raises(SafeToolError) as excinfo:
        write_policy.gate_workspace_write_deny("write_file", target, workspace=workspace)

    assert expected in str(excinfo.value)
    assert guidance in str(excinfo.value)


def test_lever_env_parsing(monkeypatch: pytest.MonkeyPatch) -> None:
    for name in (_HOST_SHELL_ENV, _COMMAND_TARGETS_ENV):
        monkeypatch.delenv(name, raising=False)
        assert shell._write_deny_lever_enabled(name) is False
        monkeypatch.setenv(name, "1")
        assert shell._write_deny_lever_enabled(name) is True
        monkeypatch.setenv(name, "true")
        assert shell._write_deny_lever_enabled(name) is True
        monkeypatch.setenv(name, "0")
        assert shell._write_deny_lever_enabled(name) is False
        monkeypatch.setenv(name, "off")
        assert shell._write_deny_lever_enabled(name) is False