plyngent/tests/test_cli/test_repl_commands.py

from __future__ import annotations

import asyncio
import base64
from typing import TYPE_CHECKING, Literal, cast, overload

import pytest
import tomlkit

from plyngent.agent import ChatAgent, Notice
from plyngent.cli.mentions import resolve_mentions
from plyngent.cli.repl import _send_turn
from plyngent.cli.slash import handle_slash
from plyngent.cli.state import ReplState
from plyngent.config.models import DatabaseConfig, ModelConfig, OpenAIProvider
from plyngent.config.store import ConfigStore
from plyngent.lmproto.openai_compatible.model import (
    AssistantChatMessage,
    ChatCompletionChoice,
    ChatCompletionChunk,
    ChatCompletionResponse,
    ChatCompletionsParam,
    ChatImagePart,
    ChatTextPart,
    UserChatMessage,
)
from plyngent.memory import MemoryStore
from plyngent.tools import image_input_refusal
from plyngent.tools.context import bind_instance

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Sequence
    from pathlib import Path


# A real 1x1 PNG: ``/image`` sniffs the format from these bytes.
_PNG = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)


def _accept_images(state: ReplState) -> None:
    """Mark the fixture's model as taking pictures (the config default is text-only)."""
    state.provider.models[state.model] = ModelConfig(image_in=True)


class DummyClient:
    @overload
    async def chat_completions(
        self, param: ChatCompletionsParam, *, stream: Literal[False] = False
    ) -> ChatCompletionResponse: ...

    @overload
    async def chat_completions(
        self, param: ChatCompletionsParam, *, stream: Literal[True]
    ) -> AsyncIterator[ChatCompletionChunk]: ...

    async def chat_completions(
        self, param: ChatCompletionsParam, *, stream: bool = False
    ) -> ChatCompletionResponse | AsyncIterator[ChatCompletionChunk]:
        del param
        if stream:

            async def empty() -> AsyncIterator[ChatCompletionChunk]:
                empty_chunks: list[ChatCompletionChunk] = []
                for chunk in empty_chunks:
                    yield chunk

            return empty()
        return ChatCompletionResponse(
            id="1",
            object="chat.completion",
            created=0,
            model="m",
            choices=[
                ChatCompletionChoice(
                    index=0,
                    message=AssistantChatMessage(content="ok"),
                    logprobs={},
                    finish_reason="stop",
                )
            ],
            system_fingerprint="",
            usage={},
        )


@pytest.fixture
async def state(tmp_path: Path) -> AsyncIterator[ReplState]:
    memory = await MemoryStore.open(DatabaseConfig())
    provider = OpenAIProvider(access_key_or_token="sk-test")
    config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
    config.providers = {"local": provider}
    st = ReplState(
        config=config,
        memory=memory,
        workspace=tmp_path,
        provider_name="local",
        provider=provider,
        model="gpt-test",
        tools_enabled=False,
    )
    st.client = DummyClient()
    st.agent = ChatAgent(st.client, model=st.model, memory=st.memory, session_id=None)
    await st.new_session("t")
    yield st
    await memory.close()


async def test_help(state: ReplState) -> None:
    assert await handle_slash(state, "/help") is True


async def test_a_cancelled_slash_command_keeps_the_repl(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """A command's own work being cancelled must not cancel the REPL's task."""
    from plyngent.cli import slash as slash_module

    def cancelled(_args: Sequence[str], _state: object) -> None:
        raise asyncio.CancelledError

    monkeypatch.setattr(slash_module, "_run_slash_argv", cancelled)
    assert await slash_module.handle_slash(cast("ReplState", cast("object", None)), "/btw hello") is True
    assert "cancelled" in capsys.readouterr().out


async def test_auto_continue_limits_installs_on_limit(tmp_path: Path) -> None:
    from plyngent.cli.limits import auto_continue_enabled, reset_auto_continue

    memory = await MemoryStore.open(DatabaseConfig())
    provider = OpenAIProvider(access_key_or_token="sk-test")
    config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
    config.providers = {"local": provider}
    try:
        st = ReplState(
            config=config,
            memory=memory,
            workspace=tmp_path,
            provider_name="local",
            provider=provider,
            model="gpt-test",
            tools_enabled=False,
            interactive_limits=False,
            auto_continue_limits=True,
        )
        assert auto_continue_enabled() is True
        assert st.agent.on_limit is not None
    finally:
        reset_auto_continue()
        await memory.close()


async def test_help_command_usage_line(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    assert await handle_slash(state, "/help compact") is True
    out = capsys.readouterr().out
    assert "Usage: /compact" in out
    assert "help [COMMAND] compact" not in out
    assert "Soft-compact" in out
    assert "--help" not in out
    assert "Options:" not in out


async def test_help_history_no_fake_options(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    assert await handle_slash(state, "/help history") is True
    out = capsys.readouterr().out
    assert "Usage: /history" in out
    # Real flags are documented; Click's meta --help is not.
    assert "--verbose" in out
    assert "--preview" in out
    assert "  --help" not in out


async def test_help_stream_clearer(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    assert await handle_slash(state, "/help stream") is True
    out = capsys.readouterr().out
    assert "Usage: /stream [on|off]" in out
    assert "ENABLED" not in out
    assert "tokens arrive" in out
    assert "default" in out


async def test_history_rejects_help_flag(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    assert await handle_slash(state, "/history --help") is True
    captured = capsys.readouterr()
    combined = captured.out + captured.err
    assert "No such option" in combined


async def test_quit(state: ReplState) -> None:
    assert await handle_slash(state, "/quit") is False


async def test_edit_sets_pending(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    monkeypatch.setattr(
        "plyngent.cli.editor.edit_text_in_editor",
        lambda *a, **k: "composed message",
    )
    assert await handle_slash(state, "/edit") is True
    assert state.pending_user_text == "composed message"
    assert "characters ready" in capsys.readouterr().out


async def test_new_and_sessions(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    first = state.session_id
    assert await handle_slash(state, "/new other") is True
    assert state.session_id != first
    assert await handle_slash(state, "/sessions") is True
    out = capsys.readouterr().out
    assert str(state.session_id) in out


async def test_tools_toggle(state: ReplState) -> None:
    assert await handle_slash(state, "/tools on") is True
    assert state.tools_enabled is True
    assert await handle_slash(state, "/tools off") is True
    assert state.tools_enabled is False


async def test_tools_list(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    assert await handle_slash(state, "/tools --list") is True
    out_off = capsys.readouterr().out
    assert "tools=off" in out_off or "no registry" in out_off

    assert await handle_slash(state, "/tools on") is True
    assert state.tools_enabled is True
    assert state.agent.tools is not None
    _ = capsys.readouterr()
    assert await handle_slash(state, "/tools --list") is True
    out = capsys.readouterr().out
    assert "tools=on count=" in out
    assert "read_file" in out
    assert "LOCAL" in out
    assert "builtin" in out


async def test_rename_slash(state: ReplState) -> None:
    sid = state.session_id
    assert sid is not None
    assert await handle_slash(state, "/rename my-chat") is True
    row = await state.memory.get_session(sid)
    assert row is not None
    assert row.name == "my-chat"


async def test_config_slash_reloads(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    path = state.config.path
    assert path is not None
    # Fixture ConfigStore may not have written the file yet.
    state.config.write()

    def fake_open(p: object, **_k: object) -> None:
        assert str(p) == str(path)

    monkeypatch.setattr("plyngent.cli.editor.open_in_editor", fake_open)
    assert await handle_slash(state, "/config") is True
    out = capsys.readouterr().out
    assert "config reloaded" in out
    assert state.provider_name


async def test_model_switch_persists(state: ReplState) -> None:
    from plyngent.config.models import ModelConfig

    state.provider.models = {
        "m1": ModelConfig(),
        "m2": ModelConfig(),
    }
    state.model = "m1"
    await state.persist_llm_selection()
    assert await handle_slash(state, "/model m2") is True
    assert state.model == "m2"
    assert state.session_id is not None
    row = await state.memory.get_session(state.session_id)
    assert row is not None
    assert row.model == "m2"


async def test_provider_switch_prompts_when_model_missing(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from plyngent.config.models import ModelConfig, OpenAICompatibleProvider, OpenAIProvider

    state.config.providers = {
        "a": OpenAIProvider(
            access_key_or_token="sk",
            models={"shared": ModelConfig(), "only-a": ModelConfig()},
        ),
        "b": OpenAICompatibleProvider(
            access_key_or_token="sk",
            url="https://x/v1",
            models={"shared": ModelConfig(), "only-b": ModelConfig()},
        ),
    }
    state.provider_name = "a"
    state.provider = state.config.providers["a"]
    state.model = "only-a"
    state.rebuild_client()

    # No real GET /models — catalog is config-only for unit tests.
    async def _no_remote(*, refresh: bool = False) -> list[str]:
        del refresh
        return []

    monkeypatch.setattr(state, "ensure_remote_models", _no_remote)

    # When switching to b, only-a is missing → select_model is invoked interactively.
    monkeypatch.setattr(
        "plyngent.cli.slash.select_model",
        lambda provider, preferred=None, interactive=True, choices=None: "only-b",
    )
    assert await handle_slash(state, "/provider b") is True
    assert state.provider_name == "b"
    assert state.model == "only-b"
    sid = state.session_id
    assert sid is not None
    row = await state.memory.get_session(sid)
    assert row is not None
    assert row.provider_name == "b"
    assert row.model == "only-b"


async def test_provider_switch_keeps_shared_model(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from plyngent.config.models import ModelConfig, OpenAICompatibleProvider, OpenAIProvider

    state.config.providers = {
        "a": OpenAIProvider(
            access_key_or_token="sk",
            models={"shared": ModelConfig()},
        ),
        "b": OpenAICompatibleProvider(
            access_key_or_token="sk",
            url="https://x/v1",
            models={"shared": ModelConfig()},
        ),
    }
    state.provider_name = "a"
    state.provider = state.config.providers["a"]
    state.model = "shared"
    state.rebuild_client()

    async def _no_remote(*, refresh: bool = False) -> list[str]:
        del refresh
        return []

    monkeypatch.setattr(state, "ensure_remote_models", _no_remote)
    assert await handle_slash(state, "/provider b") is True
    assert state.provider_name == "b"
    assert state.model == "shared"


async def test_delete_slash_confirm(
    state: ReplState,
    capsys: pytest.CaptureFixture[str],
) -> None:
    from plyngent.prompting import temporary_backend
    from tests.test_prompting import ScriptedBackend

    # Delete a non-current session so SQLite cannot reuse the same sid as "current".
    victim = state.session_id
    assert victim is not None
    assert await handle_slash(state, "/new keep") is True
    current = state.session_id
    assert current != victim

    with temporary_backend(ScriptedBackend([], confirms=[False])):
        assert await handle_slash(state, f"/delete {victim}") is True
    assert await state.memory.get_session(victim) is not None
    assert "cancelled" in capsys.readouterr().out

    with temporary_backend(ScriptedBackend([], confirms=[True])):
        assert await handle_slash(state, f"/delete {victim}") is True
    assert await state.memory.get_session(victim) is None
    assert state.session_id == current
    out = capsys.readouterr().out
    assert "deleted" in out
    assert "new session" not in out


async def test_export_slash(state: ReplState, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    from plyngent.lmproto.openai_compatible.model import AssistantChatMessage, UserChatMessage

    assert state.session_id is not None
    _ = await state.memory.append_message(state.session_id, UserChatMessage(content="hi"))
    _ = await state.memory.append_message(state.session_id, AssistantChatMessage(content="yo"))
    path = tmp_path / "out.md"
    assert await handle_slash(state, f"/export md {path}") is True
    text = path.read_text(encoding="utf-8")
    assert "Session" in text
    assert "hi" in text
    assert "yo" in text
    assert str(path.resolve()) in capsys.readouterr().out

    jpath = tmp_path / "out.json"
    assert await handle_slash(state, f"/export json {jpath}") is True
    raw = jpath.read_text(encoding="utf-8")
    assert '"session_id"' in raw
    assert "hi" in raw


async def test_stream_toggle(state: ReplState) -> None:
    assert state.agent.stream is True
    assert await handle_slash(state, "/stream off") is True
    assert state.agent.stream is False
    assert state.stream_enabled is False
    assert await handle_slash(state, "/stream on") is True
    assert state.agent.stream is True


async def test_verbose_toggle(state: ReplState) -> None:
    from plyngent.cli.display import get_verbose_tool_results

    assert state.verbose is False
    assert await handle_slash(state, "/verbose on") is True
    assert state.verbose is True
    assert get_verbose_tool_results() is True
    assert await handle_slash(state, "/verbose off") is True
    assert state.verbose is False
    assert get_verbose_tool_results() is False


async def test_model_persist_writes_config(
    state: ReplState, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
    # Seed a TOML-backed store so write() has a real path.
    path = tmp_path / "plyngent.toml"
    _ = path.write_text(
        """
[providers.local]
preset = "openai"
access_key_or_token = "sk-test"

[providers.local.models.gpt-test]
""",
        encoding="utf-8",
    )
    from plyngent.config import load as load_config

    state.config = load_config(path)
    state.provider_name = "local"
    state.provider = state.config.providers["local"]
    state.model = "gpt-new-id"
    assert await handle_slash(state, "/model --persist") is True
    out = capsys.readouterr().out
    assert "persisted model" in out
    assert "gpt-new-id" in out
    state.config.reload()
    assert "gpt-new-id" in state.config.providers["local"].models
    assert "gpt-test" in state.config.providers["local"].models


async def test_yolo_toggle(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    assert state.effective_yolo() == "off"
    assert state.soft_confirm_enabled() is True
    assert await handle_slash(state, "/yolo") is True
    assert "yolo=off" in capsys.readouterr().out

    assert await handle_slash(state, "/yolo on") is True
    assert state.effective_yolo() == "on"
    assert state.soft_confirm_enabled() is False
    assert "yolo=on" in capsys.readouterr().out

    assert await handle_slash(state, "/yolo once") is True
    assert state.effective_yolo() == "once"
    assert state.soft_confirm_enabled() is False

    state.expire_yolo_once(quiet=True)
    assert state.effective_yolo() == "off"
    assert state.soft_confirm_enabled() is True

    assert await handle_slash(state, "/yolo once") is True
    state.expire_yolo_once()
    err = capsys.readouterr().err
    assert "yolo=off (once expired)" in err
    assert state.effective_yolo() == "off"


async def test_yolo_rebuilds_tool_registry(tmp_path: Path) -> None:
    memory = await MemoryStore.open(DatabaseConfig())
    provider = OpenAIProvider(access_key_or_token="sk-test")
    config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
    config.providers = {"local": provider}
    st = ReplState(
        config=config,
        memory=memory,
        workspace=tmp_path,
        provider_name="local",
        provider=provider,
        model="gpt-test",
        tools_enabled=True,
    )
    try:
        assert st.agent.tools is not None
        # Soft-confirm hooks stay installed; YOLO mode only auto-approves the
        # confirms of YOLO-tagged tools, so the others still prompt.
        assert st.agent.tools.soft_confirm is True
        assert st.agent.tools.yolo is False
        st.set_yolo("on")
        assert st.agent.tools is not None
        assert st.agent.tools.soft_confirm is True
        assert st.agent.tools.yolo is True
        st.set_yolo("once")
        assert st.agent.tools is not None
        assert st.agent.tools.yolo is True
        st.set_yolo("off")
        assert st.agent.tools is not None
        assert st.agent.tools.soft_confirm is True
        assert st.agent.tools.yolo is False
    finally:
        await memory.close()


async def test_resume(state: ReplState) -> None:
    sid = state.session_id
    assert sid is not None
    state.agent.messages.clear()
    assert await handle_slash(state, f"/resume {sid}") is True


async def test_rounds(state: ReplState) -> None:
    assert await handle_slash(state, "/rounds 40") is True
    assert state.max_rounds == 40
    assert state.agent.max_rounds == 40


async def test_status_shows_context_tokens(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    from plyngent.agent.usage import TokenUsage
    from plyngent.lmproto.openai_compatible.model import UserChatMessage

    state.agent.messages = [UserChatMessage(content="hello")]
    assert await handle_slash(state, "/status") is True
    out = capsys.readouterr().out
    assert "context_tokens=" in out
    assert "(est)" in out  # no API usage yet
    assert "context_chars=" in out
    assert "tool_result_max=" in out
    assert "yolo=off" in out
    assert str(state.workspace) in out

    state.agent.last_request_usage = TokenUsage(
        prompt_tokens=1234,
        completion_tokens=10,
        total_tokens=1244,
        source="api",
    )
    assert await handle_slash(state, "/status") is True
    out2 = capsys.readouterr().out
    assert "context_tokens=1234/" in out2
    assert "(api)" in out2


async def test_repl_ctrl_c_from_turn_returns_to_prompt(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A Ctrl+C escaping a turn lands back at the prompt, not out of the REPL."""
    from plyngent.cli import repl as repl_mod
    from plyngent.cli.repl import run_repl

    calls: dict[str, int] = {"reads": 0}

    def fake_read_entry() -> str:
        calls["reads"] += 1
        if calls["reads"] >= 2:
            raise EOFError
        return "hello"

    async def interrupted_turn(_agent: object, _text: str, **_kwargs: object) -> bool:
        raise KeyboardInterrupt

    monkeypatch.setattr(repl_mod, "setup_readline", lambda _state: None)
    monkeypatch.setattr(repl_mod, "read_repl_entry", fake_read_entry)
    monkeypatch.setattr(repl_mod, "run_user_text_with_retries", interrupted_turn)

    await run_repl(state)  # must return normally (EOF), never raise KeyboardInterrupt
    assert calls["reads"] == 2


async def test_repl_second_ctrl_c_after_cancel_returns_to_prompt(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """A second Ctrl+C landing outside the turn call (e.g. while echoing the
    next user input, right after a cancel) must return to the prompt, not exit."""
    from plyngent.cli import repl as repl_mod
    from plyngent.cli.repl import run_repl

    reads = ["first", "second", "third"]
    calls: dict[str, int] = {"reads": 0, "echoes": 0}

    def fake_read_entry() -> str:
        calls["reads"] += 1
        if calls["reads"] > len(reads):
            raise EOFError
        return reads[calls["reads"] - 1]

    async def cancel_turn(_agent: object, _text: str, **_kwargs: object) -> bool:
        raise KeyboardInterrupt  # first Ctrl+C: cancels every turn

    def echo_user(text: str) -> None:
        calls["echoes"] += 1
        if calls["echoes"] == 2:
            # second Ctrl+C lands while echoing user text — outside the turn
            # call, previously unguarded (used to exit the whole REPL).
            raise KeyboardInterrupt

    monkeypatch.setattr(repl_mod, "setup_readline", lambda _state: None)
    monkeypatch.setattr(repl_mod, "read_repl_entry", fake_read_entry)
    monkeypatch.setattr(repl_mod, "run_user_text_with_retries", cancel_turn)
    monkeypatch.setattr(repl_mod, "_echo_user", echo_user)

    await run_repl(state)  # must return normally (EOF), never raise KeyboardInterrupt
    assert calls["reads"] == 4  # three entries + EOF
    assert calls["echoes"] == 3  # the interrupting echo still counted, loop survived
    assert "interrupted" in capsys.readouterr().out


async def test_btw_default_read_falls_back_when_tools_off(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    """Bare /btw (default --tools=read) still works under /tools off: no tools."""
    assert state.tools_enabled is False
    state.agent.stream = False  # DummyClient only completes via non-stream path
    assert await handle_slash(state, "/btw hello") is True
    out = capsys.readouterr().out
    assert "btw: hello" in out
    assert "btw done" in out
    assert "error" not in out


async def test_btw_explicit_tools_errors_when_off(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    """Explicitly requesting read/full tools while they are off is an error."""
    state.agent.stream = False  # DummyClient only completes via non-stream path
    assert await handle_slash(state, "/btw --tools read q") is True
    assert "error: --tools requires tools on" in capsys.readouterr().out
    assert await handle_slash(state, "/btw --tools full q") is True
    assert "error: --tools requires tools on" in capsys.readouterr().out
    # Explicit no is fine.
    assert await handle_slash(state, "/btw --tools no q") is True
    out = capsys.readouterr().out
    assert "btw done" in out
    assert "error" not in out


async def test_btw_read_mode_wires_read_only_clone(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """--tools=read passes a read-only clone + read-only context + main session."""

    from plyngent.agent import ToolRegistry, ToolTag, tool
    from plyngent.tools.context import is_read_only_context

    @tool(tags=ToolTag.LOCAL | ToolTag.READ_ONLY, register=False)
    async def ro_tool() -> str:
        return "ro"

    @tool(tags=ToolTag.LOCAL, register=False)
    async def w_tool() -> str:
        return "w"

    state.tools_enabled = True
    state.agent.tools = ToolRegistry([ro_tool, w_tool], auto_bind_state=True)
    captured: dict[str, object] = {}

    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
        captured["tools"] = kwargs.get("tools")
        captured["session_state"] = kwargs.get("session_state")
        captured["read_only"] = is_read_only_context()
        if False:
            yield None

    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)

    assert await handle_slash(state, "/btw --tools read hi") is True
    assert "btw done" in capsys.readouterr().out

    tools = captured["tools"]
    assert isinstance(tools, ToolRegistry)
    assert {t.name for t in tools.definitions()} == {"ro_tool"}
    assert captured["session_state"] is state.session_state
    assert captured["read_only"] is True


async def test_btw_full_mode_clones_full_registry(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """--tools=full keeps the whole registry with a fresh session, no context."""

    from plyngent.agent import ToolRegistry, ToolTag, tool
    from plyngent.tools.context import is_read_only_context

    @tool(tags=ToolTag.LOCAL | ToolTag.READ_ONLY, register=False)
    async def ro_tool() -> str:
        return "ro"

    @tool(tags=ToolTag.LOCAL, register=False)
    async def w_tool() -> str:
        return "w"

    state.tools_enabled = True
    state.agent.tools = ToolRegistry([ro_tool, w_tool], auto_bind_state=True)
    captured: dict[str, object] = {}

    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
        captured["tools"] = kwargs.get("tools")
        captured["session_state"] = kwargs.get("session_state")
        captured["read_only"] = is_read_only_context()
        if False:
            yield None

    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)

    assert await handle_slash(state, "/btw --tools full hi") is True
    assert "btw done" in capsys.readouterr().out

    assert captured["tools"] is True
    assert captured["session_state"] is not state.session_state
    assert captured["read_only"] is False


async def test_btw_sends_notice_and_shows_it(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    """The aside request carries a notice; it never enters the main transcript."""
    captured: dict[str, object] = {}

    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
        captured.update(kwargs)
        if False:
            yield None

    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)
    before = list(state.agent.messages)

    assert await handle_slash(state, "/btw --tools no hello") is True

    notice = captured["notice"]
    assert isinstance(notice, Notice)
    assert notice.kind == "aside"
    assert "Tools are disabled" in notice.body
    # Side turns are not persisted, so nothing is pushed into the main history.
    assert state.agent.messages == before
    out = capsys.readouterr().out
    assert "[notice] aside:" in out
    assert "btw: hello" in out


async def test_btw_read_notice_states_read_only_scope(
    state: ReplState,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    from plyngent.agent import ToolRegistry, ToolTag, tool

    @tool(tags=ToolTag.LOCAL | ToolTag.READ_ONLY, register=False)
    async def ro_tool() -> str:
        return "ro"

    state.tools_enabled = True
    state.agent.tools = ToolRegistry([ro_tool], auto_bind_state=True)
    captured: dict[str, object] = {}

    async def fake_run_aside(_text: str, **kwargs: object) -> AsyncIterator[object]:
        captured.update(kwargs)
        if False:
            yield None

    monkeypatch.setattr(state.agent, "run_aside", fake_run_aside)

    assert await handle_slash(state, "/btw hi") is True

    notice = captured["notice"]
    assert isinstance(notice, Notice)
    assert "read-only" in notice.body
    assert "not saved" in notice.body
    assert "[notice] aside: read tools" in capsys.readouterr().out


async def test_todos_update_takes_a_listing_number(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
    assert await handle_slash(state, "/todos push first; second") is True
    out = capsys.readouterr().out
    assert "pushed group depth=1 items=[first, second]" in out
    assert "  [ ] 1. first" in out
    assert "  [ ] 2. second" in out
    assert "t1" not in out

    assert await handle_slash(state, "/todos done 2") is True
    out = capsys.readouterr().out
    assert "updated 2 → done" in out
    assert "  [x] 2. second" in out

    # The stored id is not an address: only listing numbers are.
    assert await handle_slash(state, "/todos done t1") is True
    assert "takes an item number from the list" in capsys.readouterr().out
    assert await handle_slash(state, "/todos done 7") is True
    assert "no todo item 7" in capsys.readouterr().out

    assert await handle_slash(state, "/todos pop") is True
    assert "popped TOP group (first, second)" in capsys.readouterr().out


async def test_a_mention_attaches_the_picture_to_the_turn(state: ReplState, tmp_path: Path) -> None:
    _accept_images(state)
    shot = tmp_path / "shot.png"
    _ = shot.write_bytes(_PNG)
    state.agent.stream = False

    payload = resolve_mentions(state, f"look at @{shot}")
    assert payload.notes == []
    assert len(payload.images) == 1
    assert payload.text == f"look at @{shot}"

    await _send_turn(state, payload.text)
    user = next(m for m in state.agent.messages if isinstance(m, UserChatMessage))
    assert isinstance(user.content, list)
    text_part, image_part = user.content
    assert isinstance(text_part, ChatTextPart)
    assert text_part.text == f"look at @{shot}"
    assert isinstance(image_part, ChatImagePart)
    assert image_part.image_url.url.startswith("data:image/png;base64,")


async def test_a_mention_inlines_a_text_file(state: ReplState, tmp_path: Path) -> None:
    notes = tmp_path / "notes.txt"
    _ = notes.write_text("the deploy needs acme", encoding="utf-8")

    payload = resolve_mentions(state, f"read @{notes}")

    assert payload.images == []
    assert payload.notes == []
    assert payload.text.startswith(f"read @{notes}")
    assert f"--- @{notes} (21 chars) ---" in payload.text
    assert payload.text.endswith("the deploy needs acme")


async def test_a_mention_of_a_missing_file_is_a_note(state: ReplState, tmp_path: Path) -> None:
    payload = resolve_mentions(state, "@nope.png is gone")

    assert payload.images == []
    assert payload.notes == ["@nope.png: file not found"]
    assert payload.text == "@nope.png is gone"


async def test_a_picture_mention_is_skipped_for_a_text_only_model(
    state: ReplState,
    tmp_path: Path,
) -> None:
    """``image_in`` is the gate: a text-only model never receives a picture."""
    shot = tmp_path / "shot.png"
    _ = shot.write_bytes(_PNG)

    payload = resolve_mentions(state, f"@{shot}")

    assert payload.images == []
    note = payload.notes[0]
    assert "does not accept image input" in note
    assert "image_in = true" in note


async def test_the_tool_image_gate_follows_the_selected_model(state: ReplState) -> None:
    """The image-carrying tools read the same verdict the mention gate uses."""
    policy = state.instance_state.workspace

    state.sync_image_capability()
    assert policy.image_in is False
    assert policy.image_model == (state.provider_name, state.model)

    state.provider.models[state.model] = ModelConfig(image_in=True)
    state.sync_image_capability()
    assert policy.image_in is True

    state.provider.models[state.model] = ModelConfig(image_in=False)
    state.sync_image_capability()
    with bind_instance(state.instance_state):
        refusal = image_input_refusal()
    assert refusal is not None
    assert f'[providers.{state.provider_name}.models."{state.model}"]' in refusal