"""Unit tests for GooseExecutor (headless Goose ACP / JSON-RPC 2.0 mode).

Covers construction defaults, provider-env overrides, tool-call extraction and
permission-outcome mapping from Goose's ACP ``session/request_permission`` shape,
usage mapping, prompt-block folding, the permission → policy/elicitation
round-trip, run_turn streaming, and the harness wrap. Protocol shapes match a
verified Goose 1.38 ``goose acp`` session.
"""

from __future__ import annotations

import asyncio
import json
import os
from unittest.mock import AsyncMock, MagicMock

import pytest

from omnigent.inner.executor import ExecutorError, TextChunk, TurnComplete
from omnigent.inner.goose_executor import GooseExecutor

# ---------------------------------------------------------------------------
# Construction / attribute defaults
# ---------------------------------------------------------------------------


def test_executor_default_attributes() -> None:
    executor = GooseExecutor(goose_path="goose")
    assert executor._goose_path == "goose"
    assert executor._model is None
    assert executor._provider is None
    assert executor._builtins == ("developer",)
    assert executor._proc is None
    assert executor._session_id is None
    assert executor._initialized is False
    assert executor._rpc_id == 0
    assert executor.max_context_tokens() is None


def test_executor_custom_model_provider_builtins() -> None:
    executor = GooseExecutor(
        model="claude-x", provider="anthropic", builtins=("developer", "computercontroller")
    )
    assert executor._model == "claude-x"
    assert executor._provider == "anthropic"
    assert executor._builtins == ("developer", "computercontroller")


def test_executor_cwd_defaults_and_explicit() -> None:
    assert GooseExecutor()._cwd == os.getcwd()
    assert GooseExecutor(cwd="/tmp")._cwd == "/tmp"


def test_provider_env_only_sets_when_present() -> None:
    assert GooseExecutor()._provider_env() == {}
    assert GooseExecutor(provider="anthropic")._provider_env() == {"GOOSE_PROVIDER": "anthropic"}
    assert GooseExecutor(model="claude-x")._provider_env() == {"GOOSE_MODEL": "claude-x"}
    assert GooseExecutor(provider="anthropic", model="claude-x")._provider_env() == {
        "GOOSE_PROVIDER": "anthropic",
        "GOOSE_MODEL": "claude-x",
    }


# ---------------------------------------------------------------------------
# Tool-call extraction (Goose ACP shapes)
# ---------------------------------------------------------------------------


def test_extract_tool_call_uses_title_and_raw_input() -> None:
    """Goose's permission ``toolCall`` names the tool via ``title`` + ``rawInput``."""
    params = {
        "toolCall": {
            "kind": "other",
            "status": "pending",
            "title": "shell",
            "rawInput": {"command": "echo hi"},
        }
    }
    name, args = GooseExecutor._extract_tool_call(params)
    assert name == "shell"
    assert args == {"command": "echo hi"}


def test_extract_tool_call_prefers_meta_tool_name() -> None:
    """When the precise ``_meta.goose.toolCall.toolName`` is present, prefer it."""
    params = {
        "toolCall": {
            "kind": "other",
            "title": "shell · echo hi",
            "rawInput": {"command": "echo hi"},
            "_meta": {"goose": {"toolCall": {"toolName": "developer__shell"}}},
        }
    }
    name, args = GooseExecutor._extract_tool_call(params)
    assert name == "developer__shell"
    assert args == {"command": "echo hi"}


def test_extract_tool_call_falls_back_to_kind_then_tool() -> None:
    assert GooseExecutor._extract_tool_call({"toolCall": {"kind": "execute"}}) == ("execute", {})
    assert GooseExecutor._extract_tool_call({}) == ("tool", {})


# ---------------------------------------------------------------------------
# Permission outcome mapping (Goose option kinds)
# ---------------------------------------------------------------------------

_GOOSE_OPTIONS = [
    {"optionId": "allow_always", "name": "allow_always", "kind": "allow_always"},
    {"optionId": "allow_once", "name": "allow_once", "kind": "allow_once"},
    {"optionId": "reject_once", "name": "reject_once", "kind": "reject_once"},
    {"optionId": "reject_always", "name": "reject_always", "kind": "reject_always"},
]


def test_permission_outcome_allow_prefers_once() -> None:
    out = GooseExecutor._permission_outcome({"options": _GOOSE_OPTIONS}, allow=True)
    assert out == {"outcome": "selected", "optionId": "allow_once"}


def test_permission_outcome_deny_prefers_reject_once() -> None:
    out = GooseExecutor._permission_outcome({"options": _GOOSE_OPTIONS}, allow=False)
    assert out == {"outcome": "selected", "optionId": "reject_once"}


def test_permission_outcome_cancels_when_no_matching_option() -> None:
    # allow requested but only reject options offered → cancelled (fail-safe).
    only_reject = [{"optionId": "r", "kind": "reject_once"}]
    assert GooseExecutor._permission_outcome({"options": only_reject}, allow=True) == {
        "outcome": "cancelled"
    }
    assert GooseExecutor._permission_outcome({"options": []}, allow=False) == {
        "outcome": "cancelled"
    }


# ---------------------------------------------------------------------------
# Usage mapping
# ---------------------------------------------------------------------------


def test_usage_from_result_maps_goose_keys() -> None:
    result = {
        "stopReason": "end_turn",
        "usage": {"totalTokens": 100, "inputTokens": 80, "outputTokens": 20},
    }
    assert GooseExecutor._usage_from_result(result) == {
        "input_tokens": 80,
        "output_tokens": 20,
        "total_tokens": 100,
    }


def test_usage_from_result_none_when_absent() -> None:
    assert GooseExecutor._usage_from_result({"stopReason": "end_turn"}) is None
    assert GooseExecutor._usage_from_result({"usage": "nope"}) is None


# ---------------------------------------------------------------------------
# Prompt-block folding
# ---------------------------------------------------------------------------


def test_text_from_blocks_text_and_file() -> None:
    blocks = [
        {"type": "input_text", "text": "do the thing"},
        {"type": "input_file", "filename": "a.txt", "file_data": "data:text/plain;base64,aGk="},
        {
            "type": "input_file",
            "filename": "b.pdf",
            "file_data": "data:application/pdf;base64,AAA=",
        },
    ]
    text = GooseExecutor._text_from_blocks(blocks)
    assert "do the thing" in text
    assert "--- attached file: a.txt ---\nhi\n--- end of a.txt ---" in text
    assert "[attached file: b.pdf]" in text  # binary → marker, not inlined


# ---------------------------------------------------------------------------
# Permission round-trip (agent → client request)
# ---------------------------------------------------------------------------


def _perm_request(req_id: int = 9) -> dict:
    return {
        "jsonrpc": "2.0",
        "id": req_id,
        "method": "session/request_permission",
        "params": {
            "sessionId": "20260623_1",
            "options": _GOOSE_OPTIONS,
            "toolCall": {
                "kind": "other",
                "status": "pending",
                "title": "shell",
                "rawInput": {"command": "rm -f victim.txt"},
            },
        },
    }


@pytest.mark.asyncio
async def test_respond_to_permission_allows_when_no_gates_wired() -> None:
    executor = GooseExecutor()
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]
    await executor._respond_to_agent_request(_perm_request())
    assert sent[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "allow_once"}


@pytest.mark.asyncio
async def test_respond_to_permission_denied_by_policy() -> None:
    executor = GooseExecutor()
    executor._policy_evaluator = AsyncMock(  # type: ignore[attr-defined]
        return_value=MagicMock(action="POLICY_ACTION_DENY")
    )
    executor._elicitation_handler = AsyncMock(return_value=True)  # type: ignore[attr-defined]
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]

    await executor._respond_to_agent_request(_perm_request())

    assert sent[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "reject_once"}
    executor._elicitation_handler.assert_not_called()  # DENY short-circuits
    phase, data = executor._policy_evaluator.call_args.args
    assert phase == "PHASE_TOOL_CALL"
    assert data == {"name": "shell", "arguments": {"command": "rm -f victim.txt"}}


@pytest.mark.asyncio
async def test_respond_to_permission_elicitation_allow_and_deny() -> None:
    # Accept → allow_once.
    allow_exec = GooseExecutor()
    allow_exec._elicitation_handler = AsyncMock(return_value=True)  # type: ignore[attr-defined]
    sent_a: list[dict] = []
    allow_exec._send = AsyncMock(side_effect=lambda m: sent_a.append(m))  # type: ignore[method-assign]
    await allow_exec._respond_to_agent_request(_perm_request())
    assert sent_a[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "allow_once"}
    allow_exec._elicitation_handler.assert_awaited_once_with(
        "shell", {"command": "rm -f victim.txt"}
    )

    # Deny → reject_once.
    deny_exec = GooseExecutor()
    deny_exec._elicitation_handler = AsyncMock(return_value=False)  # type: ignore[attr-defined]
    sent_d: list[dict] = []
    deny_exec._send = AsyncMock(side_effect=lambda m: sent_d.append(m))  # type: ignore[method-assign]
    await deny_exec._respond_to_agent_request(_perm_request())
    assert sent_d[0]["result"]["outcome"] == {"outcome": "selected", "optionId": "reject_once"}


@pytest.mark.asyncio
async def test_respond_to_unknown_method_returns_jsonrpc_error() -> None:
    executor = GooseExecutor()
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]
    await executor._respond_to_agent_request(
        {"jsonrpc": "2.0", "id": 11, "method": "terminal/create", "params": {}}
    )
    assert sent[0]["error"]["code"] == -32601


# ---------------------------------------------------------------------------
# Filesystem delegation (fs/read_text_file, fs/write_text_file)
# ---------------------------------------------------------------------------


class _FakeOSEnv:
    """Minimal OSEnvironment stand-in capturing read/write calls."""

    def __init__(self, read_result: dict | None = None, write_result: dict | None = None) -> None:
        self._read_result = read_result if read_result is not None else {}
        self._write_result = write_result if write_result is not None else {}
        self.read_calls: list[tuple] = []
        self.write_calls: list[tuple] = []
        self.closed = False

    async def read(self, path: str, offset: int = 1, limit: int | None = None) -> dict:
        self.read_calls.append((path, offset, limit))
        return self._read_result

    async def write(self, path: str, content: str) -> dict:
        self.write_calls.append((path, content))
        return self._write_result

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


def test_fs_delegation_flag_tracks_os_env() -> None:
    """Delegation is on with an os_env, off without one or for a fork env."""
    from omnigent.inner.datamodel import OSEnvSpec

    assert GooseExecutor()._fs_delegation is False
    assert GooseExecutor(os_env=OSEnvSpec(type="caller_process"))._fs_delegation is True
    assert (
        GooseExecutor(os_env=OSEnvSpec(type="caller_process", fork=True))._fs_delegation is False
    )


@pytest.mark.asyncio
async def test_initialize_advertises_fs_capability_per_delegation() -> None:
    """initialize advertises clientCapabilities.fs matching the delegation flag."""
    from omnigent.inner.datamodel import OSEnvSpec

    init_result = {"result": {"agentCapabilities": {"promptCapabilities": {}}}}

    on = GooseExecutor(os_env=OSEnvSpec(type="caller_process"))
    on._rpc = AsyncMock(return_value=init_result)  # type: ignore[method-assign]
    await on._ensure_initialized()
    assert on._rpc.call_args.args[1]["clientCapabilities"]["fs"] == {
        "readTextFile": True,
        "writeTextFile": True,
    }

    off = GooseExecutor()
    off._rpc = AsyncMock(return_value=init_result)  # type: ignore[method-assign]
    await off._ensure_initialized()
    assert off._rpc.call_args.args[1]["clientCapabilities"]["fs"] == {
        "readTextFile": False,
        "writeTextFile": False,
    }


@pytest.mark.asyncio
async def test_fs_read_returns_content_and_maps_window() -> None:
    """fs/read_text_file reads through the OSEnvironment; line/limit → offset/limit."""
    from omnigent.inner.datamodel import OSEnvSpec

    executor = GooseExecutor(os_env=OSEnvSpec(type="caller_process"))
    fake = _FakeOSEnv(read_result={"content": "hi\n", "encoding": "utf-8"})
    executor._os_environment = fake  # type: ignore[assignment]
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]

    await executor._respond_to_agent_request(
        {
            "jsonrpc": "2.0",
            "id": 3,
            "method": "fs/read_text_file",
            "params": {"path": "a.txt", "line": 2, "limit": 5},
        }
    )

    assert sent[0]["result"] == {"content": "hi\n"}
    assert fake.read_calls == [("a.txt", 2, 5)]


@pytest.mark.asyncio
async def test_fs_read_missing_file_maps_to_enoent() -> None:
    """A 'no such file' read error maps to the ENOENT code (-32002)."""
    from omnigent.inner.datamodel import OSEnvSpec

    executor = GooseExecutor(os_env=OSEnvSpec(type="caller_process"))
    executor._os_environment = _FakeOSEnv(  # type: ignore[assignment]
        read_result={"error": "[Errno 2] No such file or directory: 'gone.txt'"}
    )
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]

    await executor._respond_to_agent_request(
        {"jsonrpc": "2.0", "id": 5, "method": "fs/read_text_file", "params": {"path": "gone.txt"}}
    )

    assert sent[0]["error"]["code"] == -32002


@pytest.mark.asyncio
async def test_fs_read_binary_file_is_rejected() -> None:
    """A non-utf-8 (binary) file is refused rather than returned as bytes."""
    from omnigent.inner.datamodel import OSEnvSpec

    executor = GooseExecutor(os_env=OSEnvSpec(type="caller_process"))
    executor._os_environment = _FakeOSEnv(  # type: ignore[assignment]
        read_result={"content": "AAAA", "encoding": "base64"}
    )
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]

    await executor._respond_to_agent_request(
        {"jsonrpc": "2.0", "id": 6, "method": "fs/read_text_file", "params": {"path": "img.png"}}
    )

    assert sent[0]["error"]["code"] == -32603


@pytest.mark.asyncio
async def test_fs_write_writes_through_os_env() -> None:
    """fs/write_text_file writes via the OSEnvironment and returns an empty result."""
    from omnigent.inner.datamodel import OSEnvSpec

    executor = GooseExecutor(os_env=OSEnvSpec(type="caller_process"))
    fake = _FakeOSEnv(write_result={"path": "out.txt"})
    executor._os_environment = fake  # type: ignore[assignment]
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]

    await executor._respond_to_agent_request(
        {
            "jsonrpc": "2.0",
            "id": 8,
            "method": "fs/write_text_file",
            "params": {"path": "out.txt", "content": "abc"},
        }
    )

    assert sent[0]["result"] == {}
    assert fake.write_calls == [("out.txt", "abc")]


@pytest.mark.asyncio
async def test_fs_unsupported_when_delegation_off() -> None:
    """Without an os_env, fs/* is method-not-found (delegation not advertised)."""
    executor = GooseExecutor()  # no os_env
    assert executor._fs_delegation is False
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]

    await executor._respond_to_agent_request(
        {"jsonrpc": "2.0", "id": 7, "method": "fs/read_text_file", "params": {"path": "/x"}}
    )

    assert sent[0]["error"]["code"] == -32601


@pytest.mark.asyncio
async def test_close_releases_fs_os_environment() -> None:
    """close() tears down a lazily-created fs-delegation OSEnvironment."""
    from omnigent.inner.datamodel import OSEnvSpec

    executor = GooseExecutor(os_env=OSEnvSpec(type="caller_process"))
    fake = _FakeOSEnv()
    executor._os_environment = fake  # type: ignore[assignment]

    await executor.close()

    assert fake.closed is True
    assert executor._os_environment is None


# ---------------------------------------------------------------------------
# run_turn streaming
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_turn_streams_text_and_usage() -> None:
    """run_turn yields TextChunk for agent_message_chunk and a TurnComplete with
    usage parsed from the final session/prompt result."""
    executor = GooseExecutor()
    executor._initialized = True
    executor._session_id = "20260623_1"
    executor._proc = MagicMock()
    executor._proc.returncode = None
    loop = asyncio.get_event_loop()

    async def fake_send(msg: dict) -> None:
        if msg.get("method") == "session/prompt":
            req_id = msg["id"]
            await executor._queue.put(
                {
                    "jsonrpc": "2.0",
                    "method": "session/update",
                    "params": {
                        "update": {
                            "sessionUpdate": "agent_message_chunk",
                            "content": {"type": "text", "text": "Done"},
                        }
                    },
                }
            )

            def _resolve() -> None:
                fut = executor._pending.get(req_id)
                if fut and not fut.done():
                    fut.set_result(
                        {
                            "jsonrpc": "2.0",
                            "id": req_id,
                            "result": {
                                "stopReason": "end_turn",
                                "usage": {"totalTokens": 10, "inputTokens": 7, "outputTokens": 3},
                            },
                        }
                    )

            loop.call_soon(_resolve)

    executor._send = fake_send  # type: ignore[method-assign]

    events = [
        e async for e in executor.run_turn([{"role": "user", "content": "hi"}], [], "be nice")
    ]
    text_chunks = [e for e in events if isinstance(e, TextChunk)]
    completes = [e for e in events if isinstance(e, TurnComplete)]

    assert [c.text for c in text_chunks] == ["Done"]
    assert len(completes) == 1
    assert completes[0].response == "Done"
    assert completes[0].usage == {"input_tokens": 7, "output_tokens": 3, "total_tokens": 10}


def test_history_prefix_serializes_prior_turns() -> None:
    """_history_prefix renders prior turns as labeled role: content lines."""
    prior = [
        {"role": "user", "content": "what is 2+2"},
        {"role": "assistant", "content": [{"type": "output_text", "text": "4"}]},
    ]
    out = GooseExecutor._history_prefix(prior)
    assert out.startswith("Conversation so far:")
    assert "user: what is 2+2" in out
    assert "assistant: 4" in out
    assert out.rstrip().endswith("using the conversation above as context.")


@pytest.mark.asyncio
async def test_run_turn_replays_history_on_fresh_session() -> None:
    """A fresh Goose session folds prior turns into the prompt (e.g. /model respawn).

    Goose normally only sees the latest user turn; on a brand-new subprocess
    that would drop everything before the switch, so the first turn replays
    the transcript to keep context.
    """
    executor = GooseExecutor()
    executor._initialized = True
    executor._session_id = "20260623_fresh"
    executor._proc = MagicMock()
    executor._proc.returncode = None
    loop = asyncio.get_event_loop()

    sent_prompts: list[str] = []

    async def fake_send(msg: dict) -> None:
        if msg.get("method") == "session/prompt":
            sent_prompts.append(msg["params"]["prompt"][0]["text"])
            req_id = msg["id"]

            def _resolve() -> None:
                fut = executor._pending.get(req_id)
                if fut and not fut.done():
                    fut.set_result(
                        {"jsonrpc": "2.0", "id": req_id, "result": {"stopReason": "end_turn"}}
                    )

            loop.call_soon(_resolve)

    executor._send = fake_send  # type: ignore[method-assign]

    messages = [
        {"role": "user", "content": "remember 42"},
        {"role": "assistant", "content": "ok, 42"},
        {"role": "user", "content": "what number?"},
    ]
    async for _ in executor.run_turn(messages, [], "SYS"):
        pass

    prompt = sent_prompts[0]
    assert prompt.startswith("SYS\n\n")
    assert "Conversation so far:" in prompt
    assert "user: remember 42" in prompt
    assert "assistant: ok, 42" in prompt
    assert prompt.rstrip().endswith("user: what number?")


@pytest.mark.asyncio
async def test_run_turn_no_replay_on_continuing_session() -> None:
    """A continuing Goose session sends only the latest turn (it retains context)."""
    executor = GooseExecutor()
    executor._initialized = True
    executor._session_id = "20260623_cont"
    executor._system_prompt_sent = True  # not a fresh session
    executor._proc = MagicMock()
    executor._proc.returncode = None
    loop = asyncio.get_event_loop()

    sent_prompts: list[str] = []

    async def fake_send(msg: dict) -> None:
        if msg.get("method") == "session/prompt":
            sent_prompts.append(msg["params"]["prompt"][0]["text"])
            req_id = msg["id"]

            def _resolve() -> None:
                fut = executor._pending.get(req_id)
                if fut and not fut.done():
                    fut.set_result(
                        {"jsonrpc": "2.0", "id": req_id, "result": {"stopReason": "end_turn"}}
                    )

            loop.call_soon(_resolve)

    executor._send = fake_send  # type: ignore[method-assign]

    messages = [
        {"role": "user", "content": "earlier"},
        {"role": "assistant", "content": "ok"},
        {"role": "user", "content": "latest"},
    ]
    async for _ in executor.run_turn(messages, [], "SYS"):
        pass

    assert sent_prompts[0] == "latest"


@pytest.mark.asyncio
async def test_close_with_no_process_is_a_noop() -> None:
    await GooseExecutor().close()  # must not raise


# ---------------------------------------------------------------------------
# close() / process lifecycle
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_close_terminates_process() -> None:
    executor = GooseExecutor()
    mock_proc = MagicMock()
    mock_proc.stdin = MagicMock()
    mock_proc.returncode = None

    async def fake_wait() -> int:
        return 0

    mock_proc.wait = fake_wait
    executor._proc = mock_proc
    await executor.close()
    mock_proc.terminate.assert_called_once()
    assert executor._proc is None


@pytest.mark.asyncio
async def test_close_kills_when_terminate_raises() -> None:
    executor = GooseExecutor()
    mock_proc = MagicMock()
    mock_proc.stdin = MagicMock()
    mock_proc.terminate.side_effect = OSError("gone")
    mock_proc.returncode = None
    executor._proc = mock_proc
    await executor.close()  # must not propagate
    mock_proc.kill.assert_called_once()


# ---------------------------------------------------------------------------
# ACP transport: _rpc / _read_stdout / _read_stderr
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_rpc_id_increments_monotonically() -> None:
    executor = GooseExecutor()
    sent: list[dict] = []

    async def fake_send(msg: dict) -> None:
        sent.append(msg)
        fut = executor._pending.get(msg["id"])
        if fut and not fut.done():
            fut.set_result({"jsonrpc": "2.0", "id": msg["id"], "result": {}})

    executor._send = fake_send  # type: ignore[method-assign]
    await executor._rpc("initialize", {"protocolVersion": 1})
    await executor._rpc("session/new", {"cwd": "/", "mcpServers": []})
    assert [m["id"] for m in sent] == [1, 2]


def _stdout_proc(*lines: str) -> MagicMock:
    """A fake proc whose stdout yields *lines* then EOF."""
    mock_stdout = AsyncMock()
    mock_stdout.readline = AsyncMock(
        side_effect=[(line + "\n").encode() for line in lines] + [b""]
    )
    proc = MagicMock()
    proc.stdout = mock_stdout
    return proc


@pytest.mark.asyncio
async def test_read_stdout_resolves_pending_future() -> None:
    executor = GooseExecutor()
    fut: asyncio.Future = asyncio.get_event_loop().create_future()
    executor._pending[42] = fut
    executor._proc = _stdout_proc(json.dumps({"jsonrpc": "2.0", "id": 42, "result": {"ok": True}}))
    await executor._read_stdout()
    assert fut.done() and fut.result()["result"]["ok"] is True


@pytest.mark.asyncio
async def test_read_stdout_puts_notifications_on_queue() -> None:
    executor = GooseExecutor()
    executor._proc = _stdout_proc(
        json.dumps(
            {
                "jsonrpc": "2.0",
                "method": "session/update",
                "params": {"update": {"sessionUpdate": "agent_message_chunk"}},
            }
        )
    )
    await executor._read_stdout()
    assert executor._queue.get_nowait()["method"] == "session/update"


@pytest.mark.asyncio
async def test_read_stdout_colliding_request_is_queued_not_resolved() -> None:
    """A server request (has ``method``) whose id collides with a pending _rpc
    routes to the queue, never resolving our future with a result."""
    executor = GooseExecutor()
    fut: asyncio.Future = asyncio.get_event_loop().create_future()
    executor._pending[2] = fut
    executor._proc = _stdout_proc(
        json.dumps(
            {"jsonrpc": "2.0", "id": 2, "method": "session/request_permission", "params": {}}
        )
    )
    await executor._read_stdout()
    # EOF wakes the still-pending future with EOFError (never a result).
    assert isinstance(fut.exception(), EOFError)
    assert executor._queue.get_nowait()["method"] == "session/request_permission"


@pytest.mark.asyncio
async def test_read_stdout_wakes_pending_futures_on_eof() -> None:
    executor = GooseExecutor()
    fut: asyncio.Future = asyncio.get_event_loop().create_future()
    executor._pending[7] = fut
    executor._proc = _stdout_proc()  # immediate EOF
    await executor._read_stdout()
    assert isinstance(fut.exception(), EOFError)


@pytest.mark.asyncio
async def test_read_stderr_drains_without_raising() -> None:
    executor = GooseExecutor()
    mock_stderr = AsyncMock()
    mock_stderr.readline = AsyncMock(side_effect=[b"goose: warming up\n", b""])
    proc = MagicMock()
    proc.stderr = mock_stderr
    executor._proc = proc
    await executor._read_stderr()  # must drain to EOF without raising


# ---------------------------------------------------------------------------
# Handshake / session lifecycle
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_ensure_initialized_learns_image_capability() -> None:
    executor = GooseExecutor()
    executor._rpc = AsyncMock(  # type: ignore[method-assign]
        return_value={"result": {"agentCapabilities": {"promptCapabilities": {"image": True}}}}
    )
    await executor._ensure_initialized()
    assert executor._initialized is True
    assert executor._image_supported is True
    # Second call is a no-op (latched).
    executor._rpc.reset_mock()
    await executor._ensure_initialized()
    executor._rpc.assert_not_called()


@pytest.mark.asyncio
async def test_ensure_session_uses_server_assigned_id_and_caches() -> None:
    executor = GooseExecutor()
    executor._rpc = AsyncMock(return_value={"result": {"sessionId": "20260623_7"}})  # type: ignore[method-assign]
    sid = await executor._ensure_session()
    assert sid == "20260623_7" and executor._session_id == "20260623_7"
    executor._rpc.reset_mock()
    assert await executor._ensure_session() == "20260623_7"
    executor._rpc.assert_not_called()  # cached


@pytest.mark.asyncio
async def test_ensure_session_raises_on_missing_session_id() -> None:
    executor = GooseExecutor()
    executor._rpc = AsyncMock(return_value={"result": {}})  # type: ignore[method-assign]
    with pytest.raises(RuntimeError, match="session/new"):
        await executor._ensure_session()


# ---------------------------------------------------------------------------
# Spawn / sandbox
# ---------------------------------------------------------------------------


def test_sandbox_launch_path_bare_when_no_sandbox() -> None:
    from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec

    # os_env=None → bare binary.
    assert GooseExecutor(goose_path="goose")._sandbox_launch_path(()) == "goose"
    # os_env present but sandbox explicitly disabled → bare binary.
    disabled = GooseExecutor(
        goose_path="/usr/bin/goose", os_env=OSEnvSpec(sandbox=OSEnvSandboxSpec(type="none"))
    )
    assert disabled._sandbox_launch_path(("PATH",)) == "/usr/bin/goose"


def test_sandbox_launch_path_wraps_active_policy(monkeypatch, tmp_path) -> None:
    """An active sandbox wraps goose in a launcher with its config/state dirs as
    write roots and our spawn env names allowlisted."""
    from omnigent.inner import sandbox as sandbox_mod
    from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec
    from omnigent.inner.sandbox import SandboxPolicy

    captured: dict = {}

    def _fake_resolve(_os_env, cwd) -> SandboxPolicy:
        return SandboxPolicy(
            backend_type="linux_bwrap",
            active=True,
            read_roots=[cwd.resolve(strict=False)],
            write_roots=[cwd.resolve(strict=False)],
            write_files=[],
            allow_network=True,
        )

    def _fake_launcher(target: str, sandbox: SandboxPolicy) -> str:
        captured["target"] = target
        captured["policy"] = sandbox
        return "/fake/launcher"

    monkeypatch.setattr(sandbox_mod, "resolve_sandbox", _fake_resolve)
    monkeypatch.setattr(sandbox_mod, "create_exec_launcher", _fake_launcher)

    executor = GooseExecutor(
        cwd=str(tmp_path),
        goose_path="/usr/bin/goose",
        os_env=OSEnvSpec(sandbox=OSEnvSandboxSpec(type="linux_bwrap")),
    )
    path = executor._sandbox_launch_path(("PATH", "GOOSE_PROVIDER"))

    assert path == "/fake/launcher"
    assert captured["target"] == "/usr/bin/goose"
    policy = captured["policy"]
    # goose's config dir is a write root so it can start inside the jail.
    assert any(str(p).endswith(".config/goose") for p in policy.write_roots)
    assert policy.spawn_env_allowlist is not None
    assert "PATH" in policy.spawn_env_allowlist
    assert "GOOSE_PROVIDER" in policy.spawn_env_allowlist


def test_sandbox_launch_path_falls_back_when_backend_unavailable(monkeypatch, tmp_path) -> None:
    """A backend failure degrades to the bare binary, never blocks startup."""
    from omnigent.inner import sandbox as sandbox_mod
    from omnigent.inner.datamodel import OSEnvSandboxSpec, OSEnvSpec

    def _boom(_os_env, _cwd) -> None:
        raise NotImplementedError("no bwrap here")

    monkeypatch.setattr(sandbox_mod, "resolve_sandbox", _boom)
    executor = GooseExecutor(
        cwd=str(tmp_path),
        goose_path="/usr/bin/goose",
        os_env=OSEnvSpec(sandbox=OSEnvSandboxSpec(type="linux_bwrap")),
    )
    assert executor._sandbox_launch_path(("PATH",)) == "/usr/bin/goose"


@pytest.mark.asyncio
async def test_start_process_resets_handshake_state(monkeypatch) -> None:
    """A (re)start clears the one-way handshake latch and spawns goose acp."""
    executor = GooseExecutor(goose_path="goose", builtins=("developer",))
    executor._initialized = True
    executor._image_supported = True

    captured: dict = {}

    async def fake_exec(*args, **kwargs):
        captured["argv"] = args
        return _stdout_proc()  # stdout EOF immediately so the reader exits fast

    monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec)
    await executor._start_process()
    try:
        assert executor._initialized is False  # latch reset
        assert executor._image_supported is False
        assert captured["argv"][1:] == ("acp", "--with-builtin", "developer")
    finally:
        await executor.close()


# ---------------------------------------------------------------------------
# run_turn error + usage paths
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_run_turn_boot_failure_yields_error(monkeypatch) -> None:
    executor = GooseExecutor()

    async def boom() -> None:
        raise FileNotFoundError("goose not found")

    monkeypatch.setattr(executor, "_start_process", boom)
    events = [e async for e in executor.run_turn([{"role": "user", "content": "hi"}], [], "")]
    assert len(events) == 1 and isinstance(events[0], ExecutorError)
    assert events[0].retryable is False


@pytest.mark.asyncio
async def test_run_turn_acp_error_resets_session(monkeypatch) -> None:
    """An ACP ``Session not found`` error resets the session and yields a
    retryable error (next turn re-creates the session + re-sends system prompt)."""
    executor = GooseExecutor()
    executor._initialized = True
    executor._session_id = "stale"
    executor._system_prompt_sent = True
    executor._proc = MagicMock()
    executor._proc.returncode = None
    loop = asyncio.get_event_loop()

    async def fake_send(msg: dict) -> None:
        if msg.get("method") == "session/prompt":
            rid = msg["id"]
            loop.call_soon(
                lambda: executor._pending[rid].set_result(
                    {"id": rid, "error": {"message": "Session not found: stale"}}
                )
            )

    executor._send = fake_send  # type: ignore[method-assign]
    events = [e async for e in executor.run_turn([{"role": "user", "content": "hi"}], [], "sys")]
    assert any(isinstance(e, ExecutorError) and e.retryable for e in events)
    assert executor._session_id is None  # reset
    assert executor._system_prompt_sent is False


@pytest.mark.asyncio
async def test_run_turn_tracks_context_window_from_usage_update() -> None:
    executor = GooseExecutor()
    executor._initialized = True
    executor._session_id = "s"
    executor._proc = MagicMock()
    executor._proc.returncode = None
    loop = asyncio.get_event_loop()

    async def fake_send(msg: dict) -> None:
        if msg.get("method") == "session/prompt":
            rid = msg["id"]
            await executor._queue.put(
                {
                    "jsonrpc": "2.0",
                    "method": "session/update",
                    "params": {
                        "update": {"sessionUpdate": "usage_update", "used": 5, "size": 200000}
                    },
                }
            )
            loop.call_soon(
                lambda: executor._pending[rid].set_result(
                    {"id": rid, "result": {"stopReason": "end_turn"}}
                )
            )

    executor._send = fake_send  # type: ignore[method-assign]
    [e async for e in executor.run_turn([{"role": "user", "content": "hi"}], [], "")]
    assert executor.max_context_tokens() == 200000


# ---------------------------------------------------------------------------
# Attachment / image helpers
# ---------------------------------------------------------------------------


def test_inline_text_file_data_variants() -> None:
    from omnigent.inner.goose_executor import _inline_text_file_data

    assert _inline_text_file_data("plain text") == "plain text"  # non-data-URI passthrough
    assert _inline_text_file_data("") == ""
    assert _inline_text_file_data(123) == ""  # non-str
    assert _inline_text_file_data("data:image/png;base64,AAA=") == ""  # binary → not inlined
    assert _inline_text_file_data("data:text/plain;base64,aGk=") == "hi"  # text decoded


def test_image_blocks_from_content_parses_and_skips() -> None:
    blocks = [
        {"type": "input_text", "text": "ignore"},
        {"type": "input_image", "image_url": "data:image/png;base64,AAAB"},
        {"type": "input_image", "image_url": "https://x/y.png"},  # external → skipped (SSRF)
    ]
    assert GooseExecutor._image_blocks_from_content(blocks) == [
        {"type": "image", "mimeType": "image/png", "data": "AAAB"}
    ]
    assert GooseExecutor._image_blocks_from_content("not a list") == []


def test_text_from_blocks_image_marker_toggle() -> None:
    blocks = [{"type": "input_image", "filename": "pic.png"}]
    assert "[attached image: pic.png]" in GooseExecutor._text_from_blocks(
        blocks, emit_image_marker=True
    )
    assert GooseExecutor._text_from_blocks(blocks, emit_image_marker=False) == ""


@pytest.mark.asyncio
async def test_run_turn_forwards_image_block_when_supported() -> None:
    """With image capability on, an input_image is sent as a real ACP image block
    alongside the text block."""
    executor = GooseExecutor()
    executor._initialized = True
    executor._session_id = "s"
    executor._image_supported = True
    executor._proc = MagicMock()
    executor._proc.returncode = None
    loop = asyncio.get_event_loop()
    prompts: list = []

    async def fake_send(msg: dict) -> None:
        if msg.get("method") == "session/prompt":
            prompts.append(msg["params"]["prompt"])
            rid = msg["id"]
            loop.call_soon(
                lambda: executor._pending[rid].set_result(
                    {"id": rid, "result": {"stopReason": "end_turn"}}
                )
            )

    executor._send = fake_send  # type: ignore[method-assign]
    content = [
        {"type": "input_text", "text": "look at this"},
        {"type": "input_image", "image_url": "data:image/png;base64,AAAB"},
    ]
    [e async for e in executor.run_turn([{"role": "user", "content": content}], [], "")]
    prompt = prompts[0]
    assert any(b.get("type") == "image" for b in prompt)
    assert any(b.get("type") == "text" for b in prompt)


# ---------------------------------------------------------------------------
# _decide_permission branch coverage
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_decide_permission_no_gates_allows() -> None:
    assert await GooseExecutor()._decide_permission(_perm_request()["params"]) is True


@pytest.mark.asyncio
async def test_decide_permission_policy_ask_without_handler_denies() -> None:
    executor = GooseExecutor()
    executor._policy_evaluator = AsyncMock(  # type: ignore[attr-defined]
        return_value=MagicMock(action="POLICY_ACTION_ASK")
    )  # no elicitation handler wired → fail closed
    assert await executor._decide_permission(_perm_request()["params"]) is False


@pytest.mark.asyncio
async def test_decide_permission_policy_exception_falls_through_to_elicit() -> None:
    executor = GooseExecutor()

    async def _boom(*_a) -> object:
        raise RuntimeError("policy backend down")

    executor._policy_evaluator = _boom  # type: ignore[attr-defined]
    executor._elicitation_handler = AsyncMock(return_value=True)  # type: ignore[attr-defined]
    assert await executor._decide_permission(_perm_request()["params"]) is True


@pytest.mark.asyncio
async def test_respond_to_agent_request_exception_yields_error_reply() -> None:
    executor = GooseExecutor()

    async def _boom(_params) -> bool:
        raise RuntimeError("kaboom")

    executor._decide_permission = _boom  # type: ignore[method-assign]
    sent: list[dict] = []
    executor._send = AsyncMock(side_effect=lambda m: sent.append(m))  # type: ignore[method-assign]
    await executor._respond_to_agent_request(_perm_request())
    assert sent[0]["error"]["code"] == -32603


# ---------------------------------------------------------------------------
# Harness wrap (goose_harness)
# ---------------------------------------------------------------------------


def test_resolve_os_env_default(monkeypatch) -> None:
    from omnigent.inner import goose_harness

    monkeypatch.delenv("HARNESS_GOOSE_OS_ENV", raising=False)
    spec = goose_harness._resolve_os_env()
    assert spec.type == "caller_process"
    assert spec.sandbox is not None and spec.sandbox.type == "none"


def test_resolve_os_env_from_json(monkeypatch) -> None:
    from omnigent.inner import goose_harness

    monkeypatch.setenv(
        "HARNESS_GOOSE_OS_ENV",
        json.dumps(
            {
                "type": "caller_process",
                "cwd": "/w",
                "sandbox": {"type": "linux_bwrap"},
                "fork": True,
            }
        ),
    )
    spec = goose_harness._resolve_os_env()
    assert spec.cwd == "/w"
    assert spec.sandbox is not None and spec.sandbox.type == "linux_bwrap"
    assert spec.fork is True


def test_resolve_os_env_malformed_json_falls_back(monkeypatch) -> None:
    from omnigent.inner import goose_harness

    monkeypatch.setenv("HARNESS_GOOSE_OS_ENV", "{not valid json")
    spec = goose_harness._resolve_os_env()
    assert spec.type == "caller_process"
    assert spec.sandbox is not None and spec.sandbox.type == "none"


def test_build_goose_executor_reads_env(monkeypatch) -> None:
    from omnigent.inner import goose_harness

    monkeypatch.setenv("HARNESS_GOOSE_MODEL", "claude-x")
    monkeypatch.setenv("HARNESS_GOOSE_PROVIDER", "anthropic")
    monkeypatch.setenv("HARNESS_GOOSE_CWD", "/work")
    monkeypatch.setenv("HARNESS_GOOSE_PATH", "/bin/goose")
    monkeypatch.setenv("HARNESS_GOOSE_BUILTINS", "developer, computercontroller")
    monkeypatch.delenv("HARNESS_GOOSE_OS_ENV", raising=False)
    ex = goose_harness._build_goose_executor()
    assert ex._model == "claude-x"
    assert ex._provider == "anthropic"
    assert ex._cwd == "/work"
    assert ex._goose_path == "/bin/goose"
    assert ex._builtins == ("developer", "computercontroller")


def test_build_goose_executor_defaults(monkeypatch) -> None:
    from omnigent.inner import goose_harness

    for var in (
        "HARNESS_GOOSE_MODEL",
        "HARNESS_GOOSE_PROVIDER",
        "HARNESS_GOOSE_CWD",
        "HARNESS_GOOSE_PATH",
        "HARNESS_GOOSE_BUILTINS",
        "OMNIGENT_RUNNER_WORKSPACE",
    ):
        monkeypatch.delenv(var, raising=False)
    ex = goose_harness._build_goose_executor()
    assert ex._model is None and ex._provider is None
    assert ex._goose_path == "goose"
    assert ex._builtins == ("developer",)


def test_create_app_returns_fastapi() -> None:
    from fastapi import FastAPI

    from omnigent.inner import goose_harness

    assert isinstance(goose_harness.create_app(), FastAPI)