tests_python/test_thread_inspector.py

"""Tests for the standalone Thread inspector seam."""

from dataclasses import dataclass, field

import pytest

from arena.thread_inspector import (
    AgentRunFailed,
    FileAccessFailed,
    ThreadInspector,
    ThreadNotFound,
)


class FakeResult:
    def __init__(self, *, error: str | None = None, entries=None, file_data=None) -> None:
        self.error = error
        self.entries = entries
        self.file_data = file_data


class FakeBackend:
    def __init__(self) -> None:
        self.listed: list[str] = []
        self.reads: list[str] = []

    def ls(self, path: str) -> FakeResult:
        self.listed.append(path)
        return FakeResult(entries=[{"path": f"{path}/hi.txt", "is_dir": False}])

    def read(self, path: str) -> FakeResult:
        self.reads.append(path)
        return FakeResult(file_data={"content": f"contents of {path}"})


class FakeAgent:
    def __init__(self) -> None:
        self.invocations = 0

    def invoke(self, payload, config):
        self.invocations += 1
        return {"messages": [{"role": "assistant", "content": f"reply #{self.invocations}"}]}


@dataclass
class FakeSession:
    thread_id: str
    provider: str
    backend: FakeBackend
    agent: FakeAgent
    messages: list[dict[str, str]] = field(default_factory=list)
    closed: bool = False

    def close(self) -> None:
        self.closed = True


def make_inspector():
    backend = FakeBackend()
    agent = FakeAgent()

    def factory(thread_id: str | None = None):
        return FakeSession(
            thread_id=thread_id or "thread-1",
            provider="local",
            backend=backend,
            agent=agent,
        )

    return ThreadInspector(session_factory=factory), backend, agent


def test_inspector_create_thread_records_session() -> None:
    inspector, _, _ = make_inspector()
    response = inspector.create_thread()

    assert response.thread_id == "thread-1"
    assert response.sandbox_provider == "local"
    assert "thread-1" in inspector.sessions


def test_inspector_send_message_appends_user_and_assistant_turns() -> None:
    inspector, _, _ = make_inspector()
    inspector.create_thread()

    response = inspector.send_message("thread-1", "hello")

    session = inspector.sessions["thread-1"]
    assert session.messages[0]["role"] == "user"
    assert session.messages[0]["content"] == "hello"
    assert session.messages[1]["role"] == "assistant"
    assert response.content == "reply #1"


def test_inspector_send_message_wraps_agent_failures() -> None:
    inspector, _, _ = make_inspector()
    inspector.create_thread()

    original_invoke = inspector.sessions["thread-1"].agent.invoke

    def boom(payload, config):
        original_invoke(payload, config)
        raise RuntimeError("kaboom")

    inspector.sessions["thread-1"].agent.invoke = boom

    with pytest.raises(AgentRunFailed):
        inspector.send_message("thread-1", "hello")


def test_inspector_list_files_uses_backend_path() -> None:
    inspector, backend, _ = make_inspector()
    inspector.create_thread()

    entries = inspector.list_files("thread-1", path="/work")

    assert [entry.path for entry in entries] == ["/work/hi.txt"]
    assert backend.listed == ["/work"]


def test_inspector_read_file_returns_decoded_content() -> None:
    inspector, backend, _ = make_inspector()
    inspector.create_thread()

    payload = inspector.read_file("thread-1", "/work/hi.txt")

    assert payload == {"path": "/work/hi.txt", "content": "contents of /work/hi.txt"}
    assert backend.reads == ["/work/hi.txt"]


def test_inspector_list_files_wraps_backend_error() -> None:
    inspector, _, _ = make_inspector()
    inspector.create_thread()

    inspector.sessions["thread-1"].backend.ls = lambda path: FakeResult(error="nope")

    with pytest.raises(FileAccessFailed):
        inspector.list_files("thread-1", path="/work")


def test_inspector_delete_thread_drops_session_and_closes_backend() -> None:
    inspector, _, _ = make_inspector()
    inspector.create_thread()
    session = inspector.sessions["thread-1"]

    result = inspector.delete_thread("thread-1")

    assert result == {"deleted": True}
    assert "thread-1" not in inspector.sessions
    assert session.closed is True


def test_inspector_unknown_thread_raises_thread_not_found() -> None:
    inspector, _, _ = make_inspector()

    with pytest.raises(ThreadNotFound):
        inspector.send_message("missing", "hello")
    with pytest.raises(ThreadNotFound):
        inspector.list_files("missing", path="/")
    with pytest.raises(ThreadNotFound):
        inspector.read_file("missing", "/x")
    with pytest.raises(ThreadNotFound):
        inspector.delete_thread("missing")