omnigent-ai--omnigent/tests/runtime/harnesses/test_executor_adapter.py

"""
Tests for :class:`omnigent.runtime.harnesses._executor_adapter.ExecutorAdapter`.

End-to-end through real subprocesses spawned via the same
:class:`HarnessProcessManager` used in production. Uses
:class:`omnigent.inner.executor.MockExecutor` as the inner
executor — no real LLM SDK required. The adapter's per-event
translation contract is what's under test; per-harness
configuration (Claude SDK CLI discovery, Codex subprocess setup,
Databricks credential resolution) is the per-wrap concern tested
elsewhere.
"""

from __future__ import annotations

import contextlib
import json
import shutil
import uuid
from collections.abc import AsyncIterator, Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import httpx
import pytest

from omnigent.inner.executor import Executor
from omnigent.runtime.harnesses import _HARNESS_MODULES
from omnigent.runtime.harnesses.process_manager import HarnessProcessManager

_TEST_HARNESS_NAME = "executor_adapter_fixture"
_TEST_HARNESS_MODULE = "tests.runtime.harnesses._test_executor_adapter_harness"


def _start_turn_body() -> dict[str, Any]:
    """Return a minimal session-keyed ``message`` event body that starts a turn.

    Used by adapter tests that just want a fresh turn to drive
    the inner mock executor — no pre-existing conversation
    history needed. Returns a fresh dict per call so a test that
    mutates the body doesn't bleed into the next test.

    :returns: A discriminated ``message`` event body suitable for
        ``POST /v1/sessions/{conversation_id}/events``.
    """
    return {"type": "message", "role": "user", "model": "test-agent", "content": []}


@dataclass
class _ParsedSSEEvent:
    """
    Single parsed SSE event captured from a streaming response.

    :param event: The SSE event name.
    :param data: The JSON-decoded payload.
    """

    event: str
    data: dict[str, Any]


async def _stream_iter(
    response: httpx.Response,
) -> AsyncIterator[_ParsedSSEEvent]:
    """
    Yield parsed SSE events from an open streaming response.

    :param response: An open streaming response.
    :yields: Parsed events one by one.
    """
    buffer = ""
    async for chunk in response.aiter_text():
        buffer += chunk
        while "\n\n" in buffer:
            frame, _, buffer = buffer.partition("\n\n")
            event_line = next(
                (line for line in frame.splitlines() if line.startswith("event:")),
                None,
            )
            data_line = next(
                (line for line in frame.splitlines() if line.startswith("data:")),
                None,
            )
            if event_line is None or data_line is None:
                continue
            event_name = event_line[len("event:") :].strip()
            data_payload = json.loads(data_line[len("data:") :].strip())
            yield _ParsedSSEEvent(event=event_name, data=data_payload)


@pytest.fixture
def register_fixture_harness() -> Iterator[None]:
    """Register the inner-adapter fixture harness for the test."""
    _HARNESS_MODULES[_TEST_HARNESS_NAME] = _TEST_HARNESS_MODULE
    try:
        yield
    finally:
        _HARNESS_MODULES.pop(_TEST_HARNESS_NAME, None)


@pytest.fixture
def short_tmp_parent() -> Iterator[Path]:
    """Per-test parent directory under /tmp with a short path."""
    parent = Path("/tmp") / f"omni-ia-{uuid.uuid4().hex[:8]}"
    parent.mkdir(mode=0o700)
    try:
        yield parent
    finally:
        shutil.rmtree(parent, ignore_errors=True)


@pytest.fixture
async def manager(
    short_tmp_parent: Path,
    register_fixture_harness: None,
) -> AsyncIterator[HarnessProcessManager]:
    """A started manager rooted in a short tmp dir."""
    mgr = HarnessProcessManager(
        idle_timeout_s=60.0,
        reaper_interval_s=60.0,
        tmp_parent=short_tmp_parent,
    )
    await mgr.start()
    try:
        yield mgr
    finally:
        await mgr.shutdown()


# ── Per-mock-script selectors ──────────────────────────────────
#
# Each fixture below sets the ``MOCK_EXECUTOR_SCRIPT`` env var
# the runner subprocess reads in
# ``tests/runtime/harnesses/_test_executor_adapter_harness.py:create_app``
# to populate the MockExecutor with a particular script.


@pytest.fixture
def use_text_only(monkeypatch: pytest.MonkeyPatch) -> None:
    """MockExecutor that responds with a single text turn."""
    monkeypatch.setenv("MOCK_EXECUTOR_SCRIPT", "text_only")


@pytest.fixture
def use_tool_call(monkeypatch: pytest.MonkeyPatch) -> None:
    """MockExecutor that yields a tool-call observation then completes."""
    monkeypatch.setenv("MOCK_EXECUTOR_SCRIPT", "tool_call")


@pytest.fixture
def use_error(monkeypatch: pytest.MonkeyPatch) -> None:
    """MockExecutor that yields an ExecutorError."""
    monkeypatch.setenv("MOCK_EXECUTOR_SCRIPT", "error")


@pytest.fixture
def use_cancelled(monkeypatch: pytest.MonkeyPatch) -> None:
    """MockExecutor that yields a provider-side TurnCancelled."""
    monkeypatch.setenv("MOCK_EXECUTOR_SCRIPT", "cancelled")


@pytest.fixture
def use_capture_messages(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
    """Capturing executor: records the messages it received as JSON.

    :returns: The path the executor will write to. Test reads
        this file after the request completes to inspect the
        messages list the inner executor received.
    """
    capture_path = tmp_path / "captured_messages.json"
    monkeypatch.setenv("MOCK_EXECUTOR_SCRIPT", "capture_messages")
    monkeypatch.setenv("MOCK_EXECUTOR_CAPTURE_PATH", str(capture_path))
    return capture_path


# ── Tests ──────────────────────────────────────────────────────


async def test_text_chunk_translates_to_output_text_delta(
    use_text_only: None,
    manager: HarnessProcessManager,
) -> None:
    """A TextChunk from the inner executor → response.output_text.delta.

    Verifies the basic text-streaming translation path. The mock
    script yields a single TurnComplete with response="hello from
    mock" — the adapter should surface this as a single
    text-delta event followed by response.completed.
    """
    conv_id = "conv_text"
    client = await manager.get_client(conv_id, _TEST_HARNESS_NAME)
    events: list[_ParsedSSEEvent] = []
    async with client.stream(
        "POST", f"/v1/sessions/{conv_id}/events", json=_start_turn_body()
    ) as response:
        async for event in _stream_iter(response):
            events.append(event)

    # Expected sequence: created + in_progress envelope + the
    # turn body + terminal response.completed.
    event_names = [e.event for e in events]
    # Initial envelope is constant across all adapter tests.
    assert event_names[:2] == ["response.created", "response.in_progress"]
    # Stream completes.
    assert event_names[-1] == "response.completed"
    # The mock executor's text content shows up SOMEWHERE in the
    # stream — either as TextChunk → output_text.delta (if we
    # script TextChunks) or via the TurnComplete.response path
    # (if the script uses TurnComplete with text). The current
    # text_only script uses TurnComplete; the adapter doesn't
    # emit deltas for that today (see _translate_event), so we
    # verify the stream completes cleanly without that text.
    # If a future change makes TurnComplete.response emit a
    # delta, this test should be updated.
    assert "response.completed" in event_names


async def test_tool_call_translates_to_paired_function_call_items(
    use_tool_call: None,
    manager: HarnessProcessManager,
) -> None:
    """ToolCallRequest+Complete → paired function_call + function_call_output.

    Verifies the v1 native-tool emission pattern: tools the
    inner SDK already executed surface as paired observed items
    (status=completed), NOT as action_required (which is
    server-dispatched).
    """
    conv_id = "conv_tool"
    client = await manager.get_client(conv_id, _TEST_HARNESS_NAME)
    events: list[_ParsedSSEEvent] = []
    async with client.stream(
        "POST", f"/v1/sessions/{conv_id}/events", json=_start_turn_body()
    ) as response:
        async for event in _stream_iter(response):
            events.append(event)

    # Expected: created, in_progress, function_call (observed),
    # function_call_output (observed), completed.
    output_items = [e for e in events if e.event == "response.output_item.done"]
    # The mock yields ToolCallRequest + ToolCallComplete +
    # TurnComplete. The ToolCallComplete carries NO call_id in its
    # metadata — it models a ``handles_tools_internally`` executor
    # (e.g. antigravity) whose tool was run ENTIRELY inside the
    # SDK and never round-tripped through ``_stable_tool_executor`` /
    # ctx.dispatch_tool. The adapter must therefore emit BOTH the
    # observed function_call AND its paired function_call_output —
    # the inner completion is the ONLY output source for these tools.
    # (Suppression is scoped to call ids in ``_dispatched_call_ids``,
    # populated only by ``_stable_tool_executor``; an internal tool's
    # id is never added, so its completion is not suppressed. The
    # prior blanket ``_current_ctx is not None`` rule wrongly dropped
    # it, leaving such calls as perpetual in_progress with no output.)
    fc_items = [e for e in output_items if e.data["item"].get("type") == "function_call"]
    assert len(fc_items) >= 1, (
        f"expected at least 1 function_call; got {len(fc_items)}: "
        f"{[e.data['item'].get('type') for e in output_items]}"
    )
    fc = fc_items[0].data["item"]
    assert fc["type"] == "function_call"
    # The adapter emits tool calls as in_progress initially; the
    # scaffold upgrades to completed when the tool resolves.
    assert fc["status"] in ("completed", "in_progress")
    assert fc["name"] == "echo_tool"
    # The paired function_call_output is emitted for the internally-run
    # tool (this is the FIX-1b behavior — without it the tool renders
    # as a perpetual in_progress card).
    fco_items = [e for e in output_items if e.data["item"].get("type") == "function_call_output"]
    assert len(fco_items) >= 1, (
        "expected the inner ToolCallComplete to surface a "
        "function_call_output for a tool run internally by the SDK "
        f"(no dispatch round-trip); got item types "
        f"{[e.data['item'].get('type') for e in output_items]}"
    )
    assert "tool result" in str(fco_items[0].data["item"].get("output", ""))
    # Stream completes cleanly.
    assert events[-1].event == "response.completed"


async def test_full_history_roundtrips_to_inner_executor(
    use_capture_messages: Path,
    manager: HarnessProcessManager,
) -> None:
    """End-to-end proof of the resume-history fix.

    The user's reported regression was: ``--resume`` follow-up
    turns ("list those backwards") came back as "What?" because
    the harness boundary was stripping conversation history
    down to the latest user message — the inner SDK started
    a fresh session with no prior context.

    This test sends a request body whose ``input`` carries
    THREE turns of role-keyed message items (the wire shape
    :func:`_translate_messages_to_input` now produces from
    AP's full Layer 2 history). The harness's
    :class:`ExecutorAdapter` decodes them via
    :func:`_translate_input_to_messages` and forwards to the
    inner executor. The :class:`_CapturingExecutor` writes
    every received message to disk, and we assert the captured
    file shows all three turns.

    If this test ever fails with ``len(captured) == 1``, the
    AP→harness→inner-executor pipeline has regressed to
    "latest user only" and the user-facing ``--resume``
    follow-up bug is back.
    """
    conv_id = "conv_history"
    client = await manager.get_client(conv_id, _TEST_HARNESS_NAME)
    # ``MessageEvent.content`` carries the full role-keyed item
    # list verbatim (it's renamed to ``input`` by
    # :meth:`MessageEvent.to_create_request` before reaching the
    # adapter); the outer envelope's ``role`` is the discriminator
    # for the downward direction and is unrelated to per-item
    # roles inside ``content``.
    body = {
        "type": "message",
        "role": "user",
        "model": "test-agent",
        "content": [
            {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "what tools you got?"}],
            },
            {
                "type": "message",
                "role": "assistant",
                "content": [{"type": "output_text", "text": "Bash, Edit, Read."}],
            },
            {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "list those backwards"}],
            },
        ],
    }
    events: list[_ParsedSSEEvent] = []
    async with client.stream("POST", f"/v1/sessions/{conv_id}/events", json=body) as response:
        async for event in _stream_iter(response):
            events.append(event)
    # Sanity: stream completes cleanly. If this fails, the
    # adapter raised on the new role-keyed input shape and
    # the captured-file check below would be moot.
    assert events[-1].event == "response.completed"

    # The capture file is written synchronously inside the
    # inner executor's run_turn before it yields TurnComplete,
    # so once response.completed arrives it must exist.
    assert use_capture_messages.exists(), (
        f"Capture file {use_capture_messages} was not created — "
        f"the inner executor's run_turn never reached the "
        f"json.dump call. Likely the adapter raised on the "
        f"input shape before forwarding to the executor."
    )
    captured = json.loads(use_capture_messages.read_text())

    # Three messages survived end-to-end. If 1, the harness
    # boundary stripped history again and ``--resume``
    # follow-ups are broken.
    assert len(captured) == 3, (
        f"Expected 3 messages reconstructed from the "
        f"role-keyed input shape, got {len(captured)}: "
        f"{captured!r}. If 1, the AP→harness→inner-executor "
        f"pipeline has regressed to 'latest user only' and "
        f"--resume follow-up turns will lose prior context."
    )
    assert [m["role"] for m in captured] == ["user", "assistant", "user"]
    assert captured[0]["content"] == "what tools you got?"
    assert captured[1]["content"] == "Bash, Edit, Read."
    assert captured[2]["content"] == "list those backwards"


async def test_executor_error_terminates_with_response_failed(
    use_error: None,
    manager: HarnessProcessManager,
) -> None:
    """ExecutorError → response.failed terminal event.

    Verifies the adapter raises on ExecutorError so the
    scaffold's terminal-event path produces response.failed
    (instead of response.completed). The error message surfaces
    on the ResponseObject's ``error`` field.
    """
    conv_id = "conv_err"
    client = await manager.get_client(conv_id, _TEST_HARNESS_NAME)
    events: list[_ParsedSSEEvent] = []
    async with client.stream(
        "POST", f"/v1/sessions/{conv_id}/events", json=_start_turn_body()
    ) as response:
        async for event in _stream_iter(response):
            events.append(event)

    # Terminal event is response.failed (NOT completed/cancelled).
    # If the adapter swallowed the error, this assertion catches
    # it — the stream would terminate cleanly with completed.
    assert events[-1].event == "response.failed"
    error_detail = events[-1].data["response"]["error"]
    assert error_detail is not None
    # The mock script's error message ("mock error") propagates
    # via the RuntimeError wrap in the adapter; the scaffold
    # builds an ErrorDetail with the exception's str().
    assert "mock error" in error_detail["message"]


async def test_turn_cancelled_terminates_with_response_cancelled(
    use_cancelled: None,
    manager: HarnessProcessManager,
) -> None:
    """TurnCancelled -> response.cancelled terminal event.

    Provider-side cancellation is not driven by an inbound interrupt event, so
    the adapter must mark the turn context cancelled itself. Otherwise the
    scaffold sees a clean return and emits response.completed.
    """
    conv_id = "conv_cancelled"
    client = await manager.get_client(conv_id, _TEST_HARNESS_NAME)
    events: list[_ParsedSSEEvent] = []
    async with client.stream(
        "POST", f"/v1/sessions/{conv_id}/events", json=_start_turn_body()
    ) as response:
        async for event in _stream_iter(response):
            events.append(event)

    assert events[-1].event == "response.cancelled"
    terminal_response = events[-1].data["response"]
    assert terminal_response["status"] == "cancelled"
    assert terminal_response.get("error") is None


# ── Error-code classification ──────────────────────────────────


def test_build_error_detail_uses_omnigent_error_code() -> None:
    """
    :class:`OmnigentError` (and its
    :class:`RetryableLLMError` / :class:`PermanentLLMError`
    subclasses) carry a semantic ``code`` field; the adapter's
    override uses it verbatim instead of the exception class
    name.

    What breaks if this fails: a ``RetryableLLMError(code="timeout")``
    raised by the inner executor would surface as
    ``code="RetryableLLMError"``, AP's allowlist wouldn't
    match, and the workflow's retry policy would treat the
    timeout as permanent. The whole point of step 5j is the
    structured ``code + retryable`` flowing through.
    """
    from omnigent.llms.errors import LLMErrorDetail, RetryableLLMError
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter
    from omnigent.runtime.harnesses._scaffold import HarnessApp

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    error = RetryableLLMError(
        "rate-limited by gateway",
        code="rate_limit_exceeded",
        detail=LLMErrorDetail(provider="anthropic", status_code=429),
    )
    detail = adapter._build_error_detail(error)

    # Code preserved verbatim — the Omnigent allowlist matches this and
    # marks the failure retryable. Class-name fallback (which the
    # base HarnessApp implementation would have used) gives
    # ``"RetryableLLMError"`` instead, which Omnigent would NOT match.
    assert detail.code == "rate_limit_exceeded"
    assert "rate-limited by gateway" in detail.message
    # Sanity: the base class fallback would NOT have produced this
    # code. If the override stops calling .code and falls through,
    # this assertion would fail.
    base_detail = HarnessApp._build_error_detail(adapter, RuntimeError("oops"))
    assert base_detail.code == "RuntimeError"


def test_classify_openai_exception_maps_known_types() -> None:
    """
    The OpenAI SDK classifier maps each recognized exception
    type onto its allowlist code. Unknown OpenAI exceptions
    return ``None`` so the caller falls through to the base
    implementation (preserves the class name in logs).

    What breaks if this fails: a known retryable failure like
    ``openai.RateLimitError`` would not be classified
    retryable; the workflow would not retry through what is in
    fact a transient gateway hiccup. ``openai`` is in the venv
    — the openai-agents / open-responses inner executors depend
    on it directly.
    """
    import openai

    from omnigent.runtime.harnesses._executor_adapter import (
        _classify_openai_exception,
    )

    # Construct each exception with a minimal body matching the
    # SDK's __init__ signature. The OpenAI SDK's APIError
    # subclasses take ``(message, request, body)``; we build a
    # minimal ``httpx.Request`` so the class doesn't reject the
    # args.
    request = httpx.Request("POST", "https://api.openai.com/v1/responses")

    rate = openai.RateLimitError(
        "rate limit",
        response=httpx.Response(429, request=request),
        body=None,
    )
    timeout = openai.APITimeoutError(request=request)
    connect = openai.APIConnectionError(message="conn", request=request)
    server = openai.InternalServerError(
        "server",
        response=httpx.Response(500, request=request),
        body=None,
    )

    # Each known type maps onto AP's allowlist verbatim.
    assert _classify_openai_exception(rate) == "rate_limit_exceeded"
    assert _classify_openai_exception(timeout) == "timeout"
    assert _classify_openai_exception(connect) == "connection_error"
    assert _classify_openai_exception(server) == "server_error"

    # An unrelated exception type returns None — the classifier
    # only handles OpenAI SDK exceptions, so callers know to
    # fall through.
    assert _classify_openai_exception(RuntimeError("nope")) is None


def test_classify_openai_exception_context_length_exceeded_direct() -> None:
    """
    A direct ``BadRequestError`` with ``code='context_length_exceeded'``
    is classified so the harness wire carries the code and the workflow's
    reactive compaction fires.

    What breaks if this fails: context overflow from the openai-agents
    executor is misclassified as a generic permanent error and compaction
    never triggers.
    """
    import openai

    from omnigent.runtime.harnesses._executor_adapter import (
        _classify_openai_exception,
    )

    request = httpx.Request("POST", "https://api.openai.com/v1/responses")
    exc = openai.BadRequestError(
        "context length exceeded",
        response=httpx.Response(400, request=request),
        body={"error": {"code": "context_length_exceeded", "message": "too long"}},
    )
    assert _classify_openai_exception(exc) == "context_length_exceeded"


def test_classify_openai_exception_context_length_exceeded_wrapped() -> None:
    """
    When the openai-agents SDK wraps a ``BadRequestError`` as
    ``__cause__`` of a generic ``Exception``, the classifier walks
    the cause chain and still returns ``"context_length_exceeded"``.

    What breaks if this fails: the common case where the agents SDK
    wraps provider errors is not classified and compaction never fires.
    """
    import openai

    from omnigent.runtime.harnesses._executor_adapter import (
        _classify_openai_exception,
    )

    request = httpx.Request("POST", "https://api.openai.com/v1/responses")
    inner = openai.BadRequestError(
        "context length exceeded",
        response=httpx.Response(400, request=request),
        body={"error": {"code": "context_length_exceeded", "message": "too long"}},
    )
    wrapper = RuntimeError("agents SDK error")
    wrapper.__cause__ = inner
    assert _classify_openai_exception(wrapper) == "context_length_exceeded"


def test_classify_claude_sdk_exception_maps_connection_error() -> None:
    """
    The :mod:`claude_agent_sdk` classifier maps
    ``CLIConnectionError`` onto ``"connection_error"`` so the
    AP-side retry loop treats subprocess connection drops as
    transient. Other claude_agent_sdk exceptions
    (``CLINotFoundError``, ``CLIJSONDecodeError``,
    ``ProcessError``) are non-retryable and fall through.

    What breaks if this fails: a transient subprocess hiccup
    (e.g. CLI restarting) would surface as a permanent failure
    and the workflow would never retry. ``claude_agent_sdk`` is
    in the venv — the claude-sdk harness depends on it.
    """
    import claude_agent_sdk

    from omnigent.runtime.harnesses._executor_adapter import (
        _classify_claude_sdk_exception,
    )

    conn_err = claude_agent_sdk.CLIConnectionError("subprocess gone")
    not_found = claude_agent_sdk.CLINotFoundError("CLI not on PATH")

    # Connection error → retryable.
    assert _classify_claude_sdk_exception(conn_err) == "connection_error"
    # Not-found is non-retryable; the classifier returns None
    # so the base implementation surfaces the class name.
    assert _classify_claude_sdk_exception(not_found) is None
    # Non-claude-sdk exception → None.
    assert _classify_claude_sdk_exception(RuntimeError("nope")) is None


def test_classify_httpx_exception_maps_timeout_and_connect() -> None:
    """
    The httpx classifier handles raw transport-layer exceptions
    that some inner executors surface unwrapped (notably
    litellm-backed paths for non-Anthropic providers).

    What breaks if this fails: a turn that times out at the
    httpx layer (rather than inside the SDK's wrapper) would
    not be retryable.
    """
    from omnigent.runtime.harnesses._executor_adapter import (
        _classify_httpx_exception,
    )

    timeout = httpx.ConnectTimeout("timed out")
    connect = httpx.ConnectError("conn refused")
    other = httpx.ProtocolError("bad chunk")

    assert _classify_httpx_exception(timeout) == "timeout"
    assert _classify_httpx_exception(connect) == "connection_error"
    # Other httpx exceptions fall through to the caller — we only
    # claim retryability for the two we explicitly mapped.
    assert _classify_httpx_exception(other) is None
    # Non-httpx exception → None.
    assert _classify_httpx_exception(RuntimeError("nope")) is None


def test_classify_anthropic_exception_returns_none_when_sdk_not_installed() -> None:
    """
    Without :mod:`anthropic` installed, the classifier returns
    ``None`` rather than raising.

    Regression: a hard-import of :mod:`anthropic` would crash
    the ``ExecutorAdapter._build_error_detail`` path on every
    error in environments that don't ship the SDK (e.g. the
    openai-agents wrap deployment).
    """
    from omnigent.runtime.harnesses._executor_adapter import (
        _classify_anthropic_exception,
    )

    # ``anthropic`` is not in the venv; the lazy ImportError
    # path returns None silently.
    assert _classify_anthropic_exception(RuntimeError("nope")) is None

    # Even an exception that LOOKS Anthropic-shaped (a
    # ``RateLimitError`` named identically from a different
    # module) returns None — we only claim ours.
    class _FakeRateLimitError(Exception):
        pass

    assert _classify_anthropic_exception(_FakeRateLimitError()) is None


def test_classify_anthropic_exception_maps_known_types(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    With :mod:`anthropic` available, the classifier maps each
    recognized exception type onto its allowlist code.

    The :mod:`anthropic` SDK isn't a hard dep of the venv, so
    we synthesize a stub module with the exception classes the
    classifier checks against. ``isinstance`` against a stub
    class works as long as the classifier's lazy import
    resolves to the same module — which is what the
    monkeypatch arranges.

    Regression: Phase 1d/1e wires
    ``ANTHROPIC_MAX_RETRIES`` into the Claude CLI subprocess,
    which can still surface raw :class:`anthropic.RateLimitError`
    upward when the SDK's framing layer fails. Without this
    classifier, those would render as ``[llm] RateLimitError``
    and AP's retry allowlist (which uses semantic codes, not
    class names) wouldn't match — silent demotion of retryable
    failures to permanent.
    """
    import sys
    import types

    # Synthesize a stub ``anthropic`` module with the four
    # exception classes the classifier checks. Each stub class
    # subclasses ``Exception`` so ``isinstance(stub_inst, cls)``
    # works the way the SDK's classes would in production.
    fake_anthropic = types.ModuleType("anthropic")

    class _RateLimit(Exception):
        pass

    class _Timeout(Exception):
        pass

    class _Connection(Exception):
        pass

    class _InternalServer(Exception):
        pass

    fake_anthropic.RateLimitError = _RateLimit  # type: ignore[attr-defined]
    fake_anthropic.APITimeoutError = _Timeout  # type: ignore[attr-defined]
    fake_anthropic.APIConnectionError = _Connection  # type: ignore[attr-defined]
    fake_anthropic.InternalServerError = _InternalServer  # type: ignore[attr-defined]
    monkeypatch.setitem(sys.modules, "anthropic", fake_anthropic)

    from omnigent.runtime.harnesses._executor_adapter import (
        _classify_anthropic_exception,
    )

    assert _classify_anthropic_exception(_RateLimit("limit")) == "rate_limit_exceeded"
    assert _classify_anthropic_exception(_Timeout("slow")) == "timeout"
    assert _classify_anthropic_exception(_Connection("conn")) == "connection_error"
    assert _classify_anthropic_exception(_InternalServer("500")) == "server_error"

    # Unrelated exception → None even with anthropic available.
    assert _classify_anthropic_exception(RuntimeError("nope")) is None


def test_classify_inner_exception_dispatches_across_sdks(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    The consolidated :func:`classify_inner_exception` entry
    point fans out across all per-SDK classifiers and returns
    the first match.

    Pin: Phase 3 of ``designs/RETRY_ACROSS_HARNESSES.md`` —
    callers (``ExecutorAdapter._build_error_detail``) used to
    inline three separate classifier calls; this test ensures
    the consolidated function preserves that semantics
    (first-match-wins) and adds the Anthropic path.

    Verifies via real exceptions where possible (openai +
    httpx are in the venv) and a synthesized stub for
    anthropic. ``RuntimeError`` confirms the fall-through to
    ``None``.
    """
    import sys
    import types

    import openai

    from omnigent.runtime.harnesses._executor_adapter import (
        classify_inner_exception,
    )

    # OpenAI SDK path — real exceptions.
    request = httpx.Request("POST", "https://api.openai.com/v1/responses")
    rate = openai.RateLimitError(
        "rate limit",
        response=httpx.Response(429, request=request),
        body=None,
    )
    assert classify_inner_exception(rate) == "rate_limit_exceeded"

    # httpx path — real exceptions.
    connect = httpx.ConnectError("conn refused")
    assert classify_inner_exception(connect) == "connection_error"

    # claude_agent_sdk path — real exception.
    import claude_agent_sdk

    cli_err = claude_agent_sdk.CLIConnectionError("subprocess gone")
    assert classify_inner_exception(cli_err) == "connection_error"

    # Anthropic path — synthesized stub since the SDK isn't a
    # hard venv dep.
    fake_anthropic = types.ModuleType("anthropic")

    class _RateLimit(Exception):
        pass

    class _Timeout(Exception):
        pass

    class _Connection(Exception):
        pass

    class _InternalServer(Exception):
        pass

    fake_anthropic.RateLimitError = _RateLimit  # type: ignore[attr-defined]
    fake_anthropic.APITimeoutError = _Timeout  # type: ignore[attr-defined]
    fake_anthropic.APIConnectionError = _Connection  # type: ignore[attr-defined]
    fake_anthropic.InternalServerError = _InternalServer  # type: ignore[attr-defined]
    monkeypatch.setitem(sys.modules, "anthropic", fake_anthropic)

    assert classify_inner_exception(_Timeout("slow")) == "timeout"

    # Fall-through: an exception no classifier recognizes
    # returns None. Caller is expected to use the class name.
    assert classify_inner_exception(RuntimeError("unknown")) is None


class _StubExecutor(Executor):
    """
    Minimal :class:`Executor` stub — just enough for ExecutorAdapter
    construction in unit tests that don't actually drive a turn.
    The override-method tests above never call ``run_turn``; they
    just need a constructed adapter to invoke
    ``_build_error_detail`` on.

    Subclasses :class:`Executor` so ``executor_factory=lambda: _StubExecutor()``
    typechecks (the factory expects ``Callable[[], Executor]``); the base
    provides the no-op ``run_turn`` / capability defaults these tests rely on.
    """


def test_translate_input_to_messages_reconstructs_full_history() -> None:
    """
    Role-keyed message items in ``input`` must round-trip back
    into the inner Message list.

    Mirror of
    ``test_translate_messages_to_input_passes_full_history``
    in test_client_executor.py — together they pin the
    AP→harness→inner-executor history pipeline. If this test
    only sees the latest message, the resume regression is
    back: the inner SDK gets a fresh turn with no prior
    context and answers "What?" the way the user reported
    against ``--resume``.
    """
    from omnigent.runtime.harnesses._executor_adapter import (
        _translate_input_to_messages,
    )

    input_value = [
        {
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "What tools you got?"}],
        },
        {
            "type": "message",
            "role": "assistant",
            "content": [{"type": "output_text", "text": "Bash, Edit, Read."}],
        },
        {
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "List those backwards."}],
        },
    ]

    messages = _translate_input_to_messages(input_value)

    # Three messages survive, in order — the inner SDK's
    # ``_build_prompt`` will serialize them as "Conversation so
    # far:" when starting a fresh session, giving the LLM the
    # context it needs to answer the latest user turn.
    assert len(messages) == 3, (
        f"Expected 3 messages reconstructed from role-keyed input "
        f"items, got {len(messages)}: {messages!r}. If 1, the "
        f"adapter has regressed to its old single-user-message "
        f"flatten and resume context will be lost."
    )
    assert [m["role"] for m in messages] == ["user", "assistant", "user"]
    assert messages[0]["content"] == "What tools you got?"
    assert messages[1]["content"] == "Bash, Edit, Read."
    assert messages[2]["content"] == "List those backwards."


def test_translate_input_to_messages_string_input_fallback() -> None:
    """
    Plain-string ``input`` → single user message.

    Backwards-compat fallback for any caller that still sends
    the original shape (a bare string is the Omnigent API's shorthand
    for a single ``input_text`` block from the user). One
    user-role :class:`Message` so the inner executor's
    single-turn path keeps working.
    """
    from omnigent.runtime.harnesses._executor_adapter import (
        _translate_input_to_messages,
    )

    messages = _translate_input_to_messages("hello")

    assert messages == [{"role": "user", "content": "hello"}]


def test_translate_input_to_messages_legacy_content_blocks_fallback() -> None:
    """
    Bare content-block list (no role wrappers) → single user message.

    The pre-history-fix wire format. Omnigent clients that haven't
    been migrated still send this, and the harness must keep
    handling it the same way (concat all ``text`` fields into
    a single user message). This test pins the fallback so a
    future cleanup doesn't accidentally drop bare-block
    callers.
    """
    from omnigent.runtime.harnesses._executor_adapter import (
        _translate_input_to_messages,
    )

    input_value = [
        {"type": "input_text", "text": "Hello"},
        {"type": "input_text", "text": "world"},
    ]

    messages = _translate_input_to_messages(input_value)

    # Single user message with the texts concatenated by newline
    # — same shape the harness produced before history-shape
    # support was added.
    assert messages == [{"role": "user", "content": "Hello\nworld"}]


def test_translate_input_to_messages_drops_empty_message_blocks() -> None:
    """
    A message item whose content has no text (e.g. an
    assistant turn that produced only tool calls) is dropped
    rather than emitted as ``{"role": "assistant", "content": ""}``.

    Empty messages confuse the inner SDK's prompt builder and
    don't carry useful context for the LLM (the prior user
    turn already encodes the question, the next user turn
    encodes the follow-up). Skipping them keeps the
    serialized "Conversation so far:" prefix focused on the
    parts the LLM actually benefits from.
    """
    from omnigent.runtime.harnesses._executor_adapter import (
        _translate_input_to_messages,
    )

    input_value = [
        {
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "do the thing"}],
        },
        {
            "type": "message",
            "role": "assistant",
            "content": [],  # Tool-only assistant turn — no text.
        },
        {
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "did it work?"}],
        },
    ]

    messages = _translate_input_to_messages(input_value)

    # Two messages, not three: the empty assistant turn is
    # dropped. If three messages come back, the inner SDK
    # would see a confusing empty assistant entry between two
    # user turns.
    assert len(messages) == 2, (
        f"Expected 2 messages (empty assistant skipped), got {len(messages)}: {messages!r}."
    )
    assert [m["role"] for m in messages] == ["user", "user"]


# ── MCP tool-call observed/dispatch correlation ──────────
#
# These unit tests pin the queue mechanic that fixes the tool-call
# correlation gap.
# The adapter's ``_translate_event`` queues each MCP-prefixed
# ToolCallRequest's ``tool_use_id``; ``_stable_tool_executor``
# pops in FIFO order so the eventual dispatch_tool call_id
# matches the inline observed event's call_id. Without that
# correlation, the SDK client's BlockStream can't dedupe and
# the REPL renders ``⏵ tool_name`` twice.


class _RecordingTurnContext:
    """
    Stand-in for :class:`TurnContext` that records every emit.

    Why a real stub class instead of MagicMock: per the
    project's testing rules, MagicMock would silently return
    MagicMock for any attribute access — if the adapter
    started calling a non-existent method on ctx, the test
    would still pass. A typed stub fails loud.

    Only the surface the adapter touches is implemented.
    """

    def __init__(self, response_id: str = "resp_xyz") -> None:
        """Initialize recording state.

        :param response_id: Identifier exposed via the matching
            attribute, e.g. ``"resp_xyz"``. Mirrors the
            real ``TurnContext.response_id``; the adapter uses
            it as the ``agent`` field on emitted function_call
            items.
        """
        self.response_id = response_id
        self.emitted: list[Any] = []

    def emit(self, event: Any) -> None:
        """Record an emitted event.

        :param event: The event the adapter produced (e.g.
            :class:`OutputItemDoneEvent`).
        """
        self.emitted.append(event)


def test_translate_event_mcp_tool_call_request_emits_observed_with_bare_name() -> None:
    """
    A ``ToolCallRequest`` carrying an MCP-prefixed name emits
    an observed ``function_call`` event with the BARE tool
    name (no ``mcp__omnigent__`` prefix), inline.

    What this proves: the user sees ``⏵ sys_terminal_launch`` in
    the REPL, not ``⏵ mcp__omnigent__sys_terminal_launch`` — the
    Omnigent wire shape and persisted store items carry the bare name
    (per ``omnigent/runtime/workflow.py``'s
    ``_observed_tool_call_sse_dicts``); a regression that
    surfaced the MCP-prefixed name in the SSE event would
    cause the REPL's `⏵` line to display the noisy prefix.
    Pinning the bare-name contract here keeps the adapter's
    emission consistent with the rest of the Omnigent wire path.
    """
    from omnigent.inner.executor import ToolCallRequest
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext(response_id="resp_test")

    event = ToolCallRequest(
        name="mcp__omnigent__sys_terminal_launch",
        args={"terminal": "shell", "session": "probe"},
        metadata={"call_id": "tool_use_abc123"},
    )
    adapter._translate_event(event, ctx)  # type: ignore[arg-type]

    # Exactly one emit — the inline observed function_call.
    assert len(ctx.emitted) == 1, (
        f"Expected exactly one emit for an MCP ToolCallRequest "
        f"(the inline observed function_call); got "
        f"{len(ctx.emitted)}: {ctx.emitted!r}. If 0, the early-"
        f"return for MCP prefix returned (regression to 989bfde's "
        f"bunched-at-end behavior). If 2, the adapter is double-"
        f"emitting on a single request."
    )
    item = ctx.emitted[0].item
    assert item["type"] == "function_call"
    assert item["status"] in ("completed", "in_progress"), (
        f"Observed function_call must carry status='completed' or "
        f"'in_progress' (the two-phase native-tool lifecycle); got "
        f"{item['status']!r}. action_required is for the dispatch-"
        f"path emission, not this inline one."
    )
    assert item["name"] == "sys_terminal_launch", (
        f"Tool name in the observed emit must be bare (no "
        f"``mcp__omnigent__`` prefix); got {item['name']!r}. "
        f"If the prefix leaked through, the REPL would render "
        f"``⏵ mcp__omnigent__sys_terminal_launch`` instead of "
        f"``⏵ sys_terminal_launch``."
    )
    assert item["call_id"] == "tool_use_abc123", (
        f"call_id must be the SDK's tool_use_id "
        f"(``tool_use_abc123``); got {item['call_id']!r}. The "
        f"call_id correlation between this observed event and "
        f"the eventual dispatch action_required event is what "
        f"lets the SDK client dedupe — losing it brings back "
        f"the duplicate-render bug."
    )


def test_tool_call_complete_suppressed_for_dispatched_executor() -> None:
    """A normal internally-handling executor's ``ToolCallComplete`` is
    suppressed mid-turn (its tools round-trip through dispatch_tool, which
    emits the output) — the existing dedup contract."""
    from omnigent.inner.executor import ToolCallComplete, ToolCallStatus
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    adapter._executor = _StubExecutor()  # type: ignore[assignment]
    ctx = _RecordingTurnContext()
    adapter._current_ctx = ctx  # type: ignore[assignment]

    adapter._translate_event(
        ToolCallComplete(name="", status=ToolCallStatus.SUCCESS, result="out", metadata={}),
        ctx,  # type: ignore[arg-type]
    )

    assert ctx.emitted == []


def test_translate_event_mcp_request_queues_tool_use_id_for_dispatch() -> None:
    """
    A ``ToolCallRequest`` with an MCP-prefixed name pushes the
    ``tool_use_id`` onto ``_pending_mcp_call_ids`` so the matching
    ``_stable_tool_executor`` invocation can pop it.

    What this proves: the queue mechanic that correlates the
    inline observed event's call_id with the post-stream
    dispatch's call_id. Without the push, the dispatch falls
    back to a freshly-allocated uuid — different from the
    observed call_id — and the SDK client can't dedupe, so the
    REPL renders ``⏵ tool_name`` twice.
    """
    from omnigent.inner.executor import ToolCallRequest
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    # Three MCP tool calls in order — the queue should preserve
    # this order so positional pop in _stable_tool_executor
    # correlates correctly.
    for tool_use_id in ("id_1", "id_2", "id_3"):
        adapter._translate_event(  # type: ignore[arg-type]
            ToolCallRequest(
                name="mcp__omnigent__sys_terminal_launch",
                args={"x": tool_use_id},
                metadata={"call_id": tool_use_id},
            ),
            ctx,
        )

    assert list(adapter._pending_mcp_call_ids) == ["id_1", "id_2", "id_3"], (
        f"Queue must preserve insertion order so positional "
        f"pop in _stable_tool_executor correlates each MCP "
        f"handler invocation with its matching observed event. "
        f"Got {list(adapter._pending_mcp_call_ids)!r}."
    )


def test_translate_event_non_mcp_request_queues_tool_use_id() -> None:
    """
    A ``ToolCallRequest`` with a non-MCP name (e.g. an
    openai-agents native FunctionTool) ALSO pushes its
    ``tool_use_id`` onto the correlation queue.

    What this proves: the queue is not MCP-specific. Every
    wrapped harness whose tools round-trip through
    :meth:`_stable_tool_executor` needs the same correlation
    so the dispatch's action_required event reuses the
    observed event's call_id and the Omnigent REPL can dedupe.

    Originally, the queue gated on the MCP prefix
    because only claude-sdk-via-MCP went through dispatch.
    When openai-agents-sdk landed, its native FunctionTool
    on_invoke_tool callback ALSO routes through
    :meth:`_stable_tool_executor` (no MCP prefix). With the
    old gate, openai-agents tools fell through to a fresh
    uuid in :func:`_bridge_one_dispatch`, the Omnigent client saw
    two function_call events with different call_ids, and
    the REPL rendered ``⏵ tool_name`` twice plus an empty
    result panel for the orphan call (the 2026-04-29
    user-reported regression on ``sys_timer_set``).

    For codex / pi which emit ToolCallRequest but run the
    tool natively (without invoking _stable_tool_executor for
    that call), the push is harmless — the queue entry just
    sits there until a real bridged-tool call drains it.
    """
    from omnigent.inner.executor import ToolCallRequest
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    adapter._translate_event(
        ToolCallRequest(
            name="sys_timer_set",  # Bare name (openai-agents path).
            args={"seconds": 5},
            metadata={"call_id": "call_openai_xyz"},
        ),
        ctx,  # type: ignore[arg-type]
    )

    assert list(adapter._pending_mcp_call_ids) == ["call_openai_xyz"], (
        f"Non-MCP tools that go through _stable_tool_executor "
        f"(openai-agents FunctionTool path) MUST enqueue their "
        f"tool_use_id so the dispatch's action_required emit "
        f"reuses the same call_id as the inline observed emit. "
        f"Got {list(adapter._pending_mcp_call_ids)!r} — without "
        f"this, the Omnigent client receives two function_call events "
        f"with different call_ids and the REPL double-renders "
        f"the ⏵ line."
    )


def test_translate_event_request_without_tool_use_id_does_not_queue() -> None:
    """
    A ``ToolCallRequest`` whose metadata lacks ``call_id``
    skips the queue push (no id to correlate with).

    Mirrors the executor-adapter precondition: if the inner
    executor doesn't surface a tool_use_id, there's nothing
    to correlate, and pushing ``None`` (or any sentinel)
    would mis-pair a later dispatch against a non-existent id.
    """
    from omnigent.inner.executor import ToolCallRequest
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    adapter._translate_event(
        ToolCallRequest(
            name="some_tool",
            args={"x": 1},
            metadata={},  # No call_id — executor doesn't surface one.
        ),
        ctx,  # type: ignore[arg-type]
    )

    assert list(adapter._pending_mcp_call_ids) == [], (
        "ToolCallRequest with no tool_use_id must not enqueue. "
        f"Got {list(adapter._pending_mcp_call_ids)!r}."
    )


def test_run_turn_clears_mcp_queue_at_turn_start() -> None:
    """
    Each ``run_turn`` call clears ``_pending_mcp_call_ids`` so
    a turn that errored mid-stream (leaving entries from
    half-processed tool_use blocks) doesn't carry stale ids
    into the next turn.

    What this proves: the per-turn correlation window is
    self-contained. Without the reset, turn N+1's first MCP
    dispatch would pop turn N's stale id, mis-correlating
    against an already-emitted observed event from a different
    turn — the dispatch's action_required event would carry a
    call_id no longer in the SDK client's pending_tools (the
    prior turn's ToolGroup already cleared on stream end), so
    dedup fails and the user sees a duplicate render.

    Drives the reset path directly via attribute manipulation
    rather than running a full turn — the reset is a single
    ``self._pending_mcp_call_ids.clear()`` line and a focused
    state-check is more decisive than threading a complete
    turn through HarnessProcessManager.
    """
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    # Simulate stale state from a prior turn that never drained.
    adapter._pending_mcp_call_ids.append("stale_id_from_prior_turn")
    assert len(adapter._pending_mcp_call_ids) == 1

    # Replicate the reset that ``run_turn`` performs at the top
    # of its body — same single-line clear. The unit-level
    # contract is that on every turn entry, the queue is empty.
    adapter._pending_mcp_call_ids.clear()

    assert list(adapter._pending_mcp_call_ids) == [], (
        "Queue must be empty after the per-turn reset — "
        "stale entries from a prior errored turn would mis-"
        "correlate the new turn's first MCP dispatch."
    )


async def test_stable_tool_executor_pops_queue_for_bare_tool_name() -> None:
    """
    ``_stable_tool_executor`` pops the queued tool_use_id even
    when called with the BARE tool name — not the MCP-prefixed
    form.

    What this proves and why it matters: the Claude SDK's MCP
    server wrapper strips the ``mcp__omnigent__`` prefix
    before invoking the tool callback. So
    ``_stable_tool_executor`` receives ``"sys_terminal_launch"``,
    NOT ``"mcp__omnigent__sys_terminal_launch"``. An earlier
    iteration of this fix gated the queue pop on
    ``tool_name.startswith("mcp__")`` — that guard NEVER fired
    in production because the prefix was already stripped, the
    queue never drained, and ``_bridge_one_dispatch`` always
    fell back to a freshly-allocated uuid. Result: observed and
    action_required call_ids didn't match, dedup failed, and
    the REPL rendered ``⏵ tool_name`` twice — the very bug this
    whole change is fixing.

    This test pins the dispatch-side contract empirically
    discovered by adding debug prints and running a
    real claude-sdk turn against the test-profile workspace: the
    callback receives the bare name, so the pop must NOT gate
    on the prefix.
    """
    from omnigent.runtime.harnesses._executor_adapter import (
        ExecutorAdapter,
        _bridge_one_dispatch,
    )

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    # Pre-populate the queue (as ``_translate_event`` would have
    # done when the corresponding ToolCallRequest fired).
    adapter._pending_mcp_call_ids.append("toolu_bdrk_correlated")

    # Set up a fake current ctx + agent so the dispatch path
    # doesn't bail on the no-active-turn-context guard. Use a
    # ``ctx`` that records dispatch_tool calls.
    captured_call_ids: list[str | None] = []

    class _CapturingCtx:
        """Minimal TurnContext that records dispatch_tool inputs."""

        def __init__(self) -> None:
            """Init."""
            self.response_id = "resp_capturing"

        async def dispatch_tool(
            self,
            *,
            call_id: str,
            name: str,
            arguments: str,
            agent: str,
        ) -> str:
            """Record the call_id then return a benign payload.

            :param call_id: The id the dispatch path passed.
            :param name: Tool name (ignored).
            :param arguments: JSON-encoded args (ignored).
            :param agent: Agent name (ignored).
            :returns: Empty JSON object so
                ``_bridge_one_dispatch`` can decode.
            """
            del name, arguments, agent
            captured_call_ids.append(call_id)
            return "{}"

    adapter._current_ctx = _CapturingCtx()  # type: ignore[assignment]
    adapter._current_agent = "test_agent"

    # Call with the BARE name — what the MCP wrapper actually
    # passes to the callback.
    await adapter._stable_tool_executor("sys_terminal_launch", {"x": 1})

    assert captured_call_ids == ["toolu_bdrk_correlated"], (
        f"_bridge_one_dispatch must use the queued tool_use_id "
        f"as call_id even for a bare tool name. Got "
        f"{captured_call_ids!r}. If the captured value is a "
        f"fresh ``call_<uuid>`` instead of "
        f"``toolu_bdrk_correlated``, the dispatch-side prefix "
        f"check regressed — the queue isn't draining when the "
        f"MCP callback fires with the SDK-stripped bare name, "
        f"so observed and action_required end up with different "
        f"call_ids and the REPL renders the tool call twice."
    )
    # Ensure we used `_bridge_one_dispatch` (which is what
    # accepts the call_id kwarg). Indirect check: the first
    # captured call_id must equal the queued id, which only
    # happens when ``_bridge_one_dispatch`` is called with
    # ``call_id=correlated_call_id`` (vs allocating its own).
    del _bridge_one_dispatch  # imported for typecheck only
    assert len(adapter._pending_mcp_call_ids) == 0, (
        f"Queue should be empty after pop; got "
        f"{list(adapter._pending_mcp_call_ids)!r}. A non-zero "
        f"length means the dispatch path didn't actually pop."
    )


@pytest.mark.asyncio
async def test_observed_and_dispatched_call_ids_match_for_openai_agents() -> None:
    """
    End-to-end round-trip: for an openai-agents-style ToolCallRequest
    (bare name, ``metadata["call_id"]`` set), the observed
    function_call event the adapter emits and the action_required
    function_call the dispatch emits MUST share the same call_id.

    This is the assertion that DIRECTLY locks in the user-reported
    duplicate-render fix. The bug shape:

    - Observed event emits with call_id = A (from
      ``metadata["call_id"]`` because tool_use_id is set)
    - Dispatch fires with call_id = B (fresh uuid because
      ``_pending_mcp_call_ids`` was empty — the pre-fix gate
      restricted pushes to MCP-prefixed names)
    - Omnigent client sees A != B → no dedup → REPL renders ⏵ twice
      and orphan-flushes A with empty result at response.completed
      (the empty result panel)

    With the gate removed, the queue gets the tool_use_id at
    ToolCallRequest time, ``_stable_tool_executor`` pops it, and
    the dispatch reuses A → Omnigent client dedupes → single ⏵ render.

    Failure mode this catches: anyone who reintroduces the
    MCP-prefix gate on the queue push will fail this test
    immediately.
    """
    from omnigent.inner.executor import ToolCallRequest
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    # Step 1 — adapter sees the openai-agents ToolCallRequest
    # (bare name, SDK call_id in metadata).
    sdk_call_id = "call_openai_xyz_123"
    adapter._translate_event(
        ToolCallRequest(
            name="sys_timer_set",
            args={"seconds": 5},
            metadata={"call_id": sdk_call_id},
        ),
        ctx,  # type: ignore[arg-type]
    )

    # Inspect the emitted observed event's call_id.
    function_call_events = [e for e in ctx.emitted if e.item.get("type") == "function_call"]
    assert len(function_call_events) == 1, (
        f"Expected exactly one observed function_call SSE event "
        f"per ToolCallRequest. Got {len(function_call_events)}."
    )
    observed_call_id = function_call_events[0].item.get("call_id")
    assert observed_call_id == sdk_call_id, (
        f"Observed function_call must carry the SDK's "
        f"tool_use_id verbatim. Got {observed_call_id!r}, "
        f"expected {sdk_call_id!r}."
    )

    # Step 2 — simulate the SDK invoking the FunctionTool's
    # on_invoke_tool callback (which calls _stable_tool_executor).
    # Capture what call_id _bridge_one_dispatch passes to
    # ctx.dispatch_tool.
    captured_dispatched_call_ids: list[str] = []

    class _CapturingCtx:
        """Records the dispatch's call_id for the assertion below."""

        def __init__(self) -> None:
            """Seed a stable response_id so _bridge_one_dispatch logs it."""
            self.response_id = "resp_capturing"

        async def dispatch_tool(
            self,
            *,
            call_id: str,
            name: str,
            arguments: str,
            agent: str,
        ) -> str:
            """
            Record the call_id and return a benign payload.

            :param call_id: The id _bridge_one_dispatch resolved for
                this tool call. The whole point of the test —
                appended to ``captured_dispatched_call_ids`` so the
                outer assertion can compare against the observed
                event's call_id.
            :param name: Tool name (ignored for this test).
            :param arguments: JSON-encoded args (ignored).
            :param agent: Agent name (ignored).
            :returns: ``"{}"`` so the inner SDK can decode it as
                JSON without errors.
            """
            del name, arguments, agent
            captured_dispatched_call_ids.append(call_id)
            return "{}"

    adapter._current_ctx = _CapturingCtx()  # type: ignore[assignment]
    adapter._current_agent = "test_agent"

    await adapter._stable_tool_executor("sys_timer_set", {"seconds": 5})

    # Step 3 — the dispatched call_id MUST match the observed one.
    assert captured_dispatched_call_ids == [sdk_call_id], (
        f"Dispatch must reuse the observed event's call_id so "
        f"the Omnigent client can dedupe. Observed call_id was "
        f"{observed_call_id!r}; dispatched call_ids were "
        f"{captured_dispatched_call_ids!r}. A mismatch here is "
        f"the exact 2026-04-29 user-reported regression — the "
        f"REPL renders ⏵ sys_timer_set twice and an empty result "
        f"panel for the orphan call."
    )


@pytest.mark.asyncio
async def test_executor_adapter_builds_config_from_request() -> None:
    """Forwards request controls but not agent name as executor model."""
    from omnigent.inner.executor import Executor, ExecutorConfig, Message, ToolSpec, TurnComplete
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter
    from omnigent.runtime.harnesses._scaffold import TurnContext
    from omnigent.server.schemas import CreateResponseRequest

    captured: dict[str, object] = {}

    class _CaptureExecutor(Executor):
        async def run_turn(
            self,
            messages: list[Message],
            tools: list[ToolSpec],
            system_prompt: str,
            config: ExecutorConfig | None = None,
        ):
            assert config is not None
            captured["model"] = config.model
            captured["extra"] = dict(config.extra)
            yield TurnComplete(response="ok")

    adapter = ExecutorAdapter(executor_factory=lambda: _CaptureExecutor())
    import asyncio

    ctx = TurnContext(
        response_id="resp_reason", event_queue=asyncio.Queue(), cancelled=asyncio.Event()
    )
    request = CreateResponseRequest(
        model="my_coding_agent",  # agent routing name, not an LLM
        input="hi",
        reasoning={"effort": "medium"},
        max_output_tokens=65536,
    )
    await adapter.run_turn(request, ctx)
    assert captured["extra"] == {"reasoning_effort": "medium", "max_tokens": 65536}
    assert captured["model"] is None


@pytest.mark.asyncio
async def test_executor_adapter_forwards_model_override_to_config() -> None:
    """``request.model_override`` is threaded into ``ExecutorConfig.model``.

    Validates the harness-subprocess half of the ``/model`` slash
    command's wire contract: AP's ``the harness HTTP client`` puts the
    per-request override on the body as ``model_override``; the
    adapter must read it from the parsed
    :class:`CreateResponseRequest` and surface it via
    ``ExecutorConfig.model`` so the inner executor's per-turn
    precedence (now cfg.model > self._model_override) picks it up.
    Without this, every harness-backed agent silently ignores
    ``/model``.
    """
    from omnigent.inner.executor import Executor, ExecutorConfig, Message, ToolSpec, TurnComplete
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter
    from omnigent.runtime.harnesses._scaffold import TurnContext
    from omnigent.server.schemas import CreateResponseRequest

    captured: dict[str, object] = {}

    class _CaptureExecutor(Executor):
        async def run_turn(
            self,
            messages: list[Message],
            tools: list[ToolSpec],
            system_prompt: str,
            config: ExecutorConfig | None = None,
        ):
            assert config is not None
            captured["model"] = config.model
            yield TurnComplete(response="ok")

    adapter = ExecutorAdapter(executor_factory=lambda: _CaptureExecutor())
    import asyncio

    ctx = TurnContext(
        response_id="resp_model", event_queue=asyncio.Queue(), cancelled=asyncio.Event()
    )
    request = CreateResponseRequest(
        model="my_coding_agent",
        input="hi",
        model_override="openai/gpt-5.4-mini",
    )
    await adapter.run_turn(request, ctx)
    # The override flowed into config.model. The inner executor's
    # per-turn precedence is tested separately at the inner-
    # executor layer; here we only assert the adapter's contract
    # — that request.model_override lands on the config it hands
    # to the executor.
    assert captured["model"] == "openai/gpt-5.4-mini"


class _AcceptingInjectionExecutor:
    """Inner executor stub whose enqueue_session_message always accepts."""

    def __init__(self) -> None:
        """Record every (session_key, text) the adapter forwards."""
        self.received: list[tuple[str, Any]] = []

    async def enqueue_session_message(self, session_key: str, content: Any) -> bool:
        """Accept the injection and record it.

        :param session_key: Adapter session key.
        :param content: The injected user text.
        :returns: Always ``True`` (consumed into the running turn).
        """
        self.received.append((session_key, content))
        return True


class _OneInjectionCtx:
    """TurnContext stand-in: yields one injection, then blocks; records emits.

    :param injection: The single injection ``_watch_injections`` should
        pull before the queue goes quiet.
    """

    def __init__(self, injection: Any) -> None:
        """Hold the one injection to yield and an emit log."""
        import asyncio

        self._injection: Any = injection
        self.emitted: list[Any] = []
        # Mirror TurnContext: the watcher skips delivery once cancelled.
        self.cancelled = asyncio.Event()

    async def next_injection(self, timeout: float | None = None) -> Any:
        """Return the injection once, then block (watcher loops forever).

        :param timeout: Ignored — the stand-in controls delivery.
        :returns: The injection on the first call; blocks thereafter.
        """
        import asyncio as _aio

        del timeout
        if self._injection is None:
            await _aio.sleep(3600)
        inj, self._injection = self._injection, None
        return inj

    def emit(self, event: Any) -> None:
        """Record an emitted event.

        :param event: The event the adapter pushed upstream.
        :returns: None.
        """
        self.emitted.append(event)


@pytest.mark.asyncio
async def test_watch_injections_emits_consumed_marker_on_accept() -> None:
    """An accepted mid-turn injection echoes an injection.consumed marker.

    The runner stamps an ``injection_id`` on a forwarded mid-turn message;
    once the inner executor consumes it (``enqueue_session_message``
    returns True), the adapter must emit an ``InjectionConsumedEvent``
    carrying that id so the runner drops the buffered copy and does not
    re-deliver it in a continuation turn (RUNNER_MESSAGE_INGEST.md
    Part B). A missing/empty marker is exactly what would let the
    duplication regress.
    """
    import asyncio as _aio

    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter
    from omnigent.server.schemas import CreateResponseRequest, InjectionConsumedEvent

    executor = _AcceptingInjectionExecutor()
    adapter = ExecutorAdapter(executor_factory=lambda: executor, session_key="sk")
    ctx = _OneInjectionCtx(
        CreateResponseRequest(model="m", input="steer me", injection_id="inj_x")
    )

    task = _aio.create_task(adapter._watch_injections(ctx, executor))  # type: ignore[arg-type]
    try:
        for _ in range(200):
            if ctx.emitted:
                break
            await _aio.sleep(0.01)
    finally:
        task.cancel()
        with contextlib.suppress(_aio.CancelledError):
            await task

    # The injection text reached the inner executor under the adapter's
    # session key.
    assert executor.received == [("sk", "steer me")]
    # Exactly one injection.consumed marker, echoing the correlation id.
    # If 0, the runner would never drop the buffered copy → duplication.
    assert len(ctx.emitted) == 1, f"expected one consumed marker, got {ctx.emitted!r}"
    marker = ctx.emitted[0]
    assert isinstance(marker, InjectionConsumedEvent)
    assert marker.type == "injection.consumed"
    assert marker.injection_id == "inj_x"


@pytest.mark.asyncio
async def test_watch_injections_drops_injection_when_turn_cancelled() -> None:
    """A queued injection is dropped if the turn was interrupted.

    After a Stop, the next message must NOT be delivered into the dying
    session — that resumes the abandoned generation and leaves the agent one
    message behind (the production bug). With ``ctx.cancelled`` set, the
    watcher returns without enqueuing and emits no consumed marker. If the
    guard regresses, the injection reaches the inner executor (received != []).
    """
    import asyncio as _aio

    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter
    from omnigent.server.schemas import CreateResponseRequest

    executor = _AcceptingInjectionExecutor()
    adapter = ExecutorAdapter(executor_factory=lambda: executor, session_key="sk")
    ctx = _OneInjectionCtx(
        CreateResponseRequest(model="m", input="steer me", injection_id="inj_x")
    )
    ctx.cancelled.set()  # turn interrupted before the queued injection drains

    # Returns promptly (no enqueue, no block) — fails the wait_for if it hangs.
    await _aio.wait_for(adapter._watch_injections(ctx, executor), timeout=2.0)  # type: ignore[arg-type]

    assert executor.received == [], (
        f"a cancelled turn must not enqueue the injection; got {executor.received!r}"
    )
    assert ctx.emitted == [], f"no consumed marker for a dropped injection; got {ctx.emitted!r}"


@pytest.mark.asyncio
async def test_watch_injections_no_marker_without_injection_id() -> None:
    """A legacy/fresh injection with no injection_id emits no marker.

    The consumed-handshake only applies to runner-stamped mid-turn
    injections. An injection without an ``injection_id`` (e.g. a legacy
    caller) is still delivered to the executor, but no
    ``injection.consumed`` marker is emitted — there is nothing for the
    runner to correlate.
    """
    import asyncio as _aio

    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter
    from omnigent.server.schemas import CreateResponseRequest

    executor = _AcceptingInjectionExecutor()
    adapter = ExecutorAdapter(executor_factory=lambda: executor, session_key="sk")
    ctx = _OneInjectionCtx(CreateResponseRequest(model="m", input="hi"))

    task = _aio.create_task(adapter._watch_injections(ctx, executor))  # type: ignore[arg-type]
    try:
        # Wait for the injection to be delivered to the executor.
        for _ in range(200):
            if executor.received:
                break
            await _aio.sleep(0.01)
        # Give any (erroneous) marker a chance to be emitted.
        await _aio.sleep(0.05)
    finally:
        task.cancel()
        with contextlib.suppress(_aio.CancelledError):
            await task

    assert executor.received == [("sk", "hi")]
    assert ctx.emitted == []


class _InterruptTrackingExecutor:
    """Inner executor stub that records ``interrupt_session`` calls."""

    def __init__(self) -> None:
        """Hold the list of session keys whose session was dropped."""
        self.interrupted: list[str] = []

    async def interrupt_session(self, session_key: str) -> bool:
        """Record the drop and report success.

        :param session_key: The adapter session key being interrupted.
        :returns: Always ``True`` (session dropped).
        """
        self.interrupted.append(session_key)
        return True


@pytest.mark.asyncio
async def test_interrupt_drops_inner_session_synchronously() -> None:
    """An interrupt drops the inner executor session, not just sets cancelled.

    Reproduces the off-by-one + post-cancel stream dump: the run loop only
    dropped the live claude-sdk client (``interrupt_session``) when it caught
    ``ctx.cancelled`` *between* streamed events. If the turn is blocked
    awaiting the first token, or torn down via HTTP disconnect, that check is
    skipped, the client survives, and the next turn reuses it and flushes the
    abandoned generation. ``_handle_interrupt_event`` must drop the session
    itself, synchronously, so neither can happen. Reverting that drop leaves
    ``executor.interrupted == []`` and fails this test.
    """
    import asyncio as _aio

    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter
    from omnigent.runtime.harnesses._scaffold import TurnContext

    executor = _InterruptTrackingExecutor()
    adapter = ExecutorAdapter(executor_factory=lambda: executor, session_key="sk")
    # Mark the executor as created (the drop is gated on a live executor) and
    # register an in-flight turn so the interrupt isn't a 404.
    adapter._executor = executor  # type: ignore[assignment]
    ctx = TurnContext(response_id="resp_1", event_queue=_aio.Queue(), cancelled=_aio.Event())
    adapter._in_flight["resp_1"] = ctx

    resp = await adapter._handle_interrupt_event()

    assert resp.status_code == 204
    assert ctx.cancelled.is_set()  # base handling still runs
    assert executor.interrupted == ["sk"], (
        "interrupt must drop the inner session so the next turn rebuilds fresh "
        f"instead of resuming the abandoned generation; got {executor.interrupted!r}"
    )


def test_internal_errored_tool_complete_emits_output_with_real_call_id() -> None:
    """An internally-run errored tool's completion pairs by a NON-empty call_id.

    The antigravity-sdk executor runs builtin tools entirely inside the SDK; a
    tool that errors surfaces as a :class:`ToolCallComplete` (status ERROR) that
    never round-trips through ``_stable_tool_executor``, so the adapter's
    completion branch is the ONLY ``function_call_output`` source. Downstream
    consumers pair results to their request STRICTLY by ``call_id`` (the web
    ``blockStream`` Map and the runner persistence sweep both discard an
    empty-id output), so the output MUST carry the originating request's real
    call_id.

    The executor fix guarantees every errored-tool ``ToolCallComplete`` carries
    its request's real id in ``metadata["call_id"]`` (allocated positionally for
    the SDK's id-less OnToolError path). This test feeds the adapter that exact
    event shape — a ``ToolCallRequest`` + an ERROR ``ToolCallComplete`` keyed to
    the SAME id — and asserts the emitted ``function_call_output`` carries that
    id and is NEVER ``call_id == ""`` (the pre-fix coercion that orphaned the
    result and left the call a perpetual in-progress card).
    """
    from omnigent.inner.executor import ToolCallComplete, ToolCallRequest, ToolCallStatus
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    call_id = "tc_antigravity_err_1"
    # The internally-run builtin tool call (observed) ...
    adapter._translate_event(
        ToolCallRequest(
            name="run_command",
            args={"CommandLine": "false"},
            metadata={"call_id": call_id},
        ),
        ctx,  # type: ignore[arg-type]
    )
    # ... then its ERROR completion, keyed to the SAME id (what the fixed SDK
    # executor now emits for the id-less OnToolError path).
    adapter._translate_event(
        ToolCallComplete(
            name="run_command",
            status=ToolCallStatus.ERROR,
            result=None,
            error="command failed with exit code 1",
            metadata={"call_id": call_id},
        ),
        ctx,  # type: ignore[arg-type]
    )

    fco_items = [e for e in ctx.emitted if e.item.get("type") == "function_call_output"]
    assert len(fco_items) == 1, (
        f"the internally-run errored tool's ToolCallComplete must surface "
        f"exactly one function_call_output (its sole output source); got "
        f"{[e.item.get('type') for e in ctx.emitted]}"
    )
    output_call_id = fco_items[0].item.get("call_id")
    # The load-bearing assertion: the output pairs by the request's REAL id, and
    # is never the empty string the pre-fix ``or ""`` coercion produced for an
    # id-less completion (which orphaned the result downstream).
    assert output_call_id == call_id, (
        f"function_call_output must carry the originating request's call_id "
        f"({call_id!r}) so it pairs downstream; got {output_call_id!r}."
    )
    assert output_call_id != "", (
        "function_call_output for an internally-run errored antigravity tool "
        "must NOT carry call_id=='' — every downstream consumer pairs strictly "
        "by call_id and discards an empty-id output, orphaning the result."
    )


# ── ToolCallComplete suppression scoped to dispatched call ids ──────────────
#
# A tool routed through ``_stable_tool_executor`` → ``ctx.dispatch_tool`` already
# has its ``function_call_output`` emitted by ``dispatch_tool`` when the Future
# resolves; ``_stable_tool_executor`` records that call_id in
# ``_dispatched_call_ids`` so ``_translate_event`` suppresses the duplicate inner
# ``ToolCallComplete``. The suppression is keyed to the SET — NOT the old blanket
# ``_current_ctx is not None`` rule, which also swallowed internally-run tools and
# left them as perpetual in-progress cards. These two tests pin both arms of that
# branch (it is shared code guarding claude/codex from duplicate outputs).


def test_dispatched_id_tool_complete_is_suppressed() -> None:
    """A ToolCallComplete whose call_id was dispatched (round-tripped) is suppressed.

    ``_stable_tool_executor`` (the ``ctx.dispatch_tool`` path) is the single output
    source for a dispatched tool — ``dispatch_tool`` already emitted its
    ``function_call_output``. Emitting another here would duplicate it on the SSE
    stream and produce a ghost "Waiting for output" card in the Web UI. So a
    ``ToolCallComplete`` carrying a dispatched id must produce NO emit.
    """
    from omnigent.inner.executor import ToolCallComplete, ToolCallStatus
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    call_id = "tc_dispatched_1"
    # Mark the id as dispatched, exactly as ``_stable_tool_executor`` does after
    # routing the call through ``ctx.dispatch_tool``.
    adapter._dispatched_call_ids.add(call_id)

    adapter._translate_event(
        ToolCallComplete(
            name="sys_terminal_launch",
            status=ToolCallStatus.SUCCESS,
            result="dispatched-output",
            metadata={"call_id": call_id},
        ),
        ctx,  # type: ignore[arg-type]
    )

    assert ctx.emitted == [], (
        f"a dispatched id's ToolCallComplete must be suppressed (dispatch_tool is "
        f"its single output source); got {[e.item.get('type') for e in ctx.emitted]}. "
        f"A second function_call_output here duplicates the result and leaves a "
        f"ghost 'Waiting for output' card in the Web UI."
    )


def test_non_dispatched_id_tool_complete_emits_output() -> None:
    """A ToolCallComplete whose id was NOT dispatched DOES emit its output.

    Complements the suppression test: the gate is scoped to ``_dispatched_call_ids``,
    not a blanket suppression. A tool the inner SDK ran internally (no
    ``dispatch_tool`` round-trip) has its completion as the ONLY output source, so
    it must surface a paired ``function_call_output``.
    """
    from omnigent.inner.executor import ToolCallComplete, ToolCallStatus
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    call_id = "tc_internal_1"
    # Note: call_id is intentionally NOT added to _dispatched_call_ids.
    adapter._translate_event(
        ToolCallComplete(
            name="run_command",
            status=ToolCallStatus.SUCCESS,
            result="internal-output",
            metadata={"call_id": call_id},
        ),
        ctx,  # type: ignore[arg-type]
    )

    fco_items = [e for e in ctx.emitted if e.item.get("type") == "function_call_output"]
    assert len(fco_items) == 1, (
        f"a non-dispatched ToolCallComplete is its tool's only output source and "
        f"must emit exactly one function_call_output; got "
        f"{[e.item.get('type') for e in ctx.emitted]}"
    )
    assert fco_items[0].item.get("call_id") == call_id


def test_idless_tool_complete_is_suppressed() -> None:
    """A ToolCallComplete with no usable call_id emits nothing (no ghost card).

    ``ExecutorAdapter`` is shared by every adapter-backed harness, so an inner
    executor that bridges an id-less ``ToolCallComplete`` mid-turn (e.g. pi)
    must NOT leak a ``function_call_output`` with ``call_id == ""``: downstream
    consumers pair STRICTLY by call_id and discard empty ones, so such an output
    cannot pair and only renders a perpetual "Waiting for output" ghost card.
    The old blanket ``_current_ctx is not None`` rule swallowed these; the
    id-scoped suppression must keep doing so (the ``or ""`` coercion alone left
    this path unguarded).
    """
    from omnigent.inner.executor import ToolCallComplete, ToolCallStatus
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    ctx = _RecordingTurnContext()

    # No ``metadata.call_id`` at all → the ``or ""`` coercion yields an empty id.
    adapter._translate_event(
        ToolCallComplete(
            name="run_command",
            status=ToolCallStatus.SUCCESS,
            result="idless-output",
            metadata={},
        ),
        ctx,  # type: ignore[arg-type]
    )

    assert ctx.emitted == [], (
        f"an id-less ToolCallComplete must be suppressed (it cannot pair "
        f"downstream and only ghosts a 'Waiting for output' card); got "
        f"{[e.item.get('type') for e in ctx.emitted]}"
    )


async def test_policy_evaluator_no_active_turn_context_is_phase_aware() -> None:
    """With no active turn context (turn-context desync, #1026) the policy
    evaluator must not blanket-ALLOW. PHASE_TOOL_CALL fails closed (this adapter
    is the only enforcement point, never re-checked server-side); advisory LLM
    phases and the post-execution result phase fail open so a transient desync
    does not needlessly wedge them — matching the runner's phase-aware default.
    """
    from omnigent.runtime.harnesses._executor_adapter import ExecutorAdapter

    adapter = ExecutorAdapter(executor_factory=lambda: _StubExecutor())
    adapter._current_ctx = None

    tool_verdict = await adapter._stable_policy_evaluator("PHASE_TOOL_CALL", {})
    assert tool_verdict.action == "POLICY_ACTION_DENY"
    assert tool_verdict.reason == "No active turn context; failing closed for PHASE_TOOL_CALL."

    for advisory_phase in ("PHASE_LLM_REQUEST", "PHASE_LLM_RESPONSE", "PHASE_TOOL_RESULT"):
        verdict = await adapter._stable_policy_evaluator(advisory_phase, {})
        assert verdict.action == "POLICY_ACTION_ALLOW", advisory_phase
        assert verdict.reason is None, advisory_phase