test_codex_native_forwarder.py

"""
Tests for the codex-native forwarder's model-change sync-back
(:mod:`omnigent.codex_native_forwarder`).

For codex-native, ``config.toml``'s ``model`` key is the cost-policy source
of truth (it is what an in-TUI ``/model`` writes). At subscription and at
each ``turn/started`` the forwarder reads it (``_refresh_model_from_config``,
which delegates to the shared ``read_codex_config_model`` in the bridge
module) onto ``_CodexForwarderState.model`` and mirrors it to the Omnigent server
as an ``external_model_change`` event (→ persisted ``conv.model_override``)
so the cost-budget policy resolves the selected model. The startup/spawn
model IS mirrored (so Omnigent learns the session's model even when unchanged);
only an already-mirrored value is not re-posted.
"""

from __future__ import annotations

import asyncio
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock

import httpx
import pytest

from omnigent import codex_native_forwarder as fwd
from omnigent.codex_native_bridge import (
    CodexNativeBridgeState,
    codex_home_for_bridge_dir,
    read_bridge_state,
    write_bridge_state,
)
from omnigent.codex_native_forwarder import _persist_codex_compaction_item


class _RecordingClient:
    """
    Async ``httpx`` client stub that records POSTs and returns HTTP 200.

    Only ``post`` is exercised by ``_post_session_event``; each call is
    recorded so the test can assert exactly what was mirrored.
    """

    def __init__(self) -> None:
        """Initialize with an empty record of posts."""
        self.posts: list[tuple[str, dict]] = []

    async def post(self, url: str, *, json: dict) -> httpx.Response:
        """
        Record ``(url, json)`` and return a 200 response.

        :param url: Request URL, e.g. ``"/v1/sessions/conv_x/events"``.
        :param json: JSON body, e.g.
            ``{"type": "external_model_change", "data": {"model": "gpt-5.4"}}``.
        :returns: A real ``httpx.Response`` with status 200.
        """
        self.posts.append((url, json))
        return httpx.Response(200, request=httpx.Request("POST", url))


def _state(model: str | None, posted_model: str | None) -> fwd._CodexForwarderState:
    """
    Build a forwarder state with the given current + last-mirrored model.

    :param model: Current Codex model, e.g. ``"gpt-5.4"`` or ``None``.
    :param posted_model: Last-mirrored model baseline, e.g. ``"gpt-5.5"``.
    :returns: A ``_CodexForwarderState`` for the sync-back helper.
    """
    state = fwd._CodexForwarderState()
    state.model = model
    state.posted_model = posted_model
    return state


@pytest.mark.asyncio
async def test_sync_model_change_posts_on_change() -> None:
    """A model differing from the baseline posts external_model_change.

    The in-TUI ``/model`` switch (gpt-5.5 → gpt-5.4) must mirror to Omnigent as
    an ``external_model_change`` and advance the baseline so it isn't
    re-posted. A missing post here is exactly the bug a user hit: the
    terminal model changed but the cost policy kept seeing gpt-5.5.
    """
    client = _RecordingClient()
    state = _state(model="gpt-5.4", posted_model="gpt-5.5")

    await fwd._sync_model_change(client, session_id="conv_x", forwarder_state=state)

    # Exactly one mirror post, carrying the new raw codex model id.
    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {"type": "external_model_change", "data": {"model": "gpt-5.4"}},
        )
    ]
    # Baseline advanced → the same model won't re-post on the next update.
    assert state.posted_model == "gpt-5.4"


@pytest.mark.asyncio
async def test_sync_model_change_no_post_when_unchanged() -> None:
    """Model equal to the baseline (seeded spawn default) does not post.

    Prevents the spawn/startup model from being echoed back to Omnigent as a
    spurious "change" (which would also fire on every settings update).
    """
    client = _RecordingClient()
    state = _state(model="gpt-5.5", posted_model="gpt-5.5")

    await fwd._sync_model_change(client, session_id="conv_x", forwarder_state=state)

    assert client.posts == []


@pytest.mark.asyncio
async def test_sync_model_change_no_post_when_model_unknown() -> None:
    """No model observed yet (``None``) → nothing to mirror."""
    client = _RecordingClient()
    state = _state(model=None, posted_model="gpt-5.5")

    await fwd._sync_model_change(client, session_id="conv_x", forwarder_state=state)

    assert client.posts == []


def _write_codex_config(bridge_dir: Path, body: str) -> Path:
    """
    Write a ``config.toml`` into the session's per-session ``CODEX_HOME``.

    :param bridge_dir: The bridge dir whose ``codex-home/config.toml`` is
        written (the path the model reader reads).
    :param body: Raw TOML body, e.g. ``'model = "gpt-5.4"\\n'``.
    :returns: The written ``config.toml`` path.
    """
    home = codex_home_for_bridge_dir(bridge_dir)
    home.mkdir(parents=True, exist_ok=True)
    path = home / "config.toml"
    path.write_text(body)
    return path


def test_refresh_model_from_config_updates_state(tmp_path: Path) -> None:
    """``config.toml``'s model lands on the forwarder state for mirroring.

    This is the exact path the subscription and ``turn/started`` handlers use
    to learn the user's ``/model`` selection: read config.toml (via the
    shared ``read_codex_config_model``) → set ``forwarder_state.model`` →
    ``_sync_model_change`` mirrors it to AP. The config.toml parsing itself
    is covered in ``tests/test_codex_native_bridge.py``; this asserts the
    forwarder wires the read into its state.
    """
    _write_codex_config(tmp_path, 'model = "gpt-5.4"\n')
    state = _state(model="gpt-5.5", posted_model="gpt-5.5")

    fwd._refresh_model_from_config(tmp_path, state)

    # The selected model (gpt-5.4) replaces the prior value, ready to mirror.
    assert state.model == "gpt-5.4"


def test_refresh_prefers_pushed_settings_model_over_stale_config(tmp_path: Path) -> None:
    """An unchanged config.toml must not roll back a live thread-settings model.

    Regression for the routed-model reversion: routing switched the running
    thread via ``thread/settings/update`` (notified as
    ``thread/settings/updated``), but config.toml still held the pinned
    launch model; the next ``turn/started`` re-read the stale file and
    mirrored the default back over ``model_override`` — reverting the routed
    model one turn after it applied.
    """
    _write_codex_config(tmp_path, 'model = "databricks-gpt-5-5"\n')
    state = fwd._CodexForwarderState()
    # Subscription-time read adopts the pinned launch model (baseline).
    fwd._refresh_model_from_config(tmp_path, state)
    assert state.model == "databricks-gpt-5-5"
    # Omnigent pushes a routed model thread-level; the live notification wins.
    state.note_thread_settings_updated({"threadSettings": {"model": "databricks-gpt-5-6-luna"}})

    # turn/started re-read: config.toml is UNCHANGED — the pushed model holds.
    fwd._refresh_model_from_config(tmp_path, state)

    assert state.model == "databricks-gpt-5-6-luna"


def test_refresh_adopts_changed_config_over_settings_model(tmp_path: Path) -> None:
    """A config.toml that changed since the last read wins over settings.

    An in-TUI ``/model`` (or the executor's mirror write) rewrites the file —
    that is the freshest signal and must not be masked by an older
    ``thread/settings/updated`` value.
    """
    _write_codex_config(tmp_path, 'model = "databricks-gpt-5-5"\n')
    state = fwd._CodexForwarderState()
    fwd._refresh_model_from_config(tmp_path, state)
    state.note_thread_settings_updated({"threadSettings": {"model": "databricks-gpt-5-6-luna"}})
    # The user picks a third model in the TUI: /model rewrites config.toml.
    _write_codex_config(tmp_path, 'model = "gpt-5.6-sol"\n')

    fwd._refresh_model_from_config(tmp_path, state)

    assert state.model == "gpt-5.6-sol"


def test_refresh_launch_race_ends_on_routed_model(tmp_path: Path) -> None:
    """Launch-race scenario: the pinned default ends up on the routed model.

    The terminal launch pins the default into config.toml before first-turn
    routing runs. The executor then pushes the routed model thread-level AND
    mirrors it into config.toml (``write_codex_config_model``); the next
    ``turn/started`` re-read must adopt the routed model — with or without
    the mirror write having succeeded.
    """
    from omnigent.codex_native_bridge import write_codex_config_model

    _write_codex_config(tmp_path, 'model = "databricks-gpt-5-5"\n')
    state = fwd._CodexForwarderState()
    fwd._refresh_model_from_config(tmp_path, state)
    # First routed turn: settings push (notification) + executor mirror write.
    state.note_thread_settings_updated({"threadSettings": {"model": "databricks-gpt-5-6-luna"}})
    assert write_codex_config_model(tmp_path, "databricks-gpt-5-6-luna") is True

    fwd._refresh_model_from_config(tmp_path, state)

    assert state.model == "databricks-gpt-5-6-luna"
    # Later turns stay on the routed model (no reversion churn).
    fwd._refresh_model_from_config(tmp_path, state)
    assert state.model == "databricks-gpt-5-6-luna"


def test_note_resume_response_records_model_without_seeding_baseline() -> None:
    """The startup/resume model is recorded but the baseline stays unset.

    Omnigent must learn the session's ACTUAL model — including the spawn default —
    because the cost gate resolves ``conv.model_override or spec.llm.model``
    and for codex the spawn model is frequently NOT ``spec.llm.model``. So
    ``note_resume_response`` records ``model`` but leaves ``posted_model``
    ``None``, so the next ``_sync_model_change`` mirrors the real model. If
    this re-seeded the baseline, an unchanged cheap session would never post
    ``external_model_change`` and the gate would wrongly DENY it.
    """
    state = fwd._CodexForwarderState()

    state.note_resume_response({"result": {"model": "gpt-5.4-mini"}})

    assert state.model == "gpt-5.4-mini"
    # Baseline NOT seeded → the spawn model will be mirrored on the next sync.
    assert state.posted_model is None


@pytest.mark.asyncio
async def test_sync_after_resume_posts_spawn_model() -> None:
    """End-to-end: an unchanged spawn model is mirrored to AP.

    This is the regression for the wrongly-blocked cheap session: codex
    spawned on gpt-5.4-mini, the model never "changed", yet Omnigent must still
    receive it as ``model_override`` so the cost gate sees a cheap model
    instead of falling back to the spec model and DENYing.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState()
    state.note_resume_response({"result": {"model": "gpt-5.4-mini"}})

    await fwd._sync_model_change(client, session_id="conv_x", forwarder_state=state)

    # The spawn model is mirrored (not suppressed as "unchanged").
    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {"type": "external_model_change", "data": {"model": "gpt-5.4-mini"}},
        )
    ]
    assert state.posted_model == "gpt-5.4-mini"


def test_thread_settings_updated_records_effort_and_collaboration_mode() -> None:
    """
    ``thread/settings/updated`` records Codex's live thinking settings.

    App-server sends the public ``ThreadSettings`` shape with ``effort`` and
    ``collaborationMode``. If this parser regresses, the later sync helpers have
    no state to mirror, so Omnigent would keep stale ``reasoning_effort`` and
    mode metadata even though Codex changed them.
    """
    state = fwd._CodexForwarderState()

    state.note_thread_settings_updated(
        {
            "threadSettings": {
                "model": "gpt-5.4-codex",
                "effort": "medium",
                "collaborationMode": {
                    "mode": "plan",
                    "settings": {
                        "model": "gpt-5.4-codex",
                        "reasoning_effort": "medium",
                        "developer_instructions": None,
                    },
                },
            }
        }
    )

    assert state.model == "gpt-5.4-codex"
    assert state.effort == "medium"
    assert state.collaboration_mode == "plan"


@pytest.mark.asyncio
async def test_sync_reasoning_effort_change_posts_and_dedupes() -> None:
    """
    Codex effort changes mirror to Omnigent exactly once per observed value.

    The first sync must POST ``external_reasoning_effort_change`` so the server
    persists ``conversation.reasoning_effort``. The second sync with the same
    value must not re-post; otherwise every repeated settings notification would
    churn the session stream.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState(effort="medium")

    await fwd._sync_reasoning_effort_change(
        client,
        session_id="conv_x",
        forwarder_state=state,
    )
    await fwd._sync_reasoning_effort_change(
        client,
        session_id="conv_x",
        forwarder_state=state,
    )

    # One post proves the new effort reached AP; no second post proves the
    # dedupe baseline advanced after a successful mirror.
    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_reasoning_effort_change",
                "data": {"reasoning_effort": "medium"},
            },
        )
    ]
    assert state.posted_effort == "medium"
    assert state.posted_effort_known is True


@pytest.mark.asyncio
async def test_sync_reasoning_effort_change_posts_clear() -> None:
    """
    Codex clearing effort mirrors JSON null to Omnigent.

    ``None`` is a meaningful observed value (model/default effort), so the
    forwarder must still post it after a prior explicit effort. If this returned
    early on falsey ``None``, Omnigent would keep a stale explicit effort.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState(
        effort=None,
        posted_effort="high",
        posted_effort_known=True,
    )

    await fwd._sync_reasoning_effort_change(
        client,
        session_id="conv_x",
        forwarder_state=state,
    )

    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_reasoning_effort_change",
                "data": {"reasoning_effort": None},
            },
        )
    ]
    assert state.posted_effort is None
    assert state.posted_effort_known is True


@pytest.mark.asyncio
async def test_sync_codex_collaboration_mode_change_posts_and_dedupes() -> None:
    """
    Codex collaboration mode changes mirror to Omnigent labels once.

    The ``mode`` value is the durable "Plan vs Default" signal we can get from
    app-server. Missing this POST would leave the session snapshot without the
    current Codex mode.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState(collaboration_mode="plan")

    await fwd._sync_codex_collaboration_mode_change(
        client,
        session_id="conv_x",
        forwarder_state=state,
    )
    await fwd._sync_codex_collaboration_mode_change(
        client,
        session_id="conv_x",
        forwarder_state=state,
    )

    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_codex_collaboration_mode_change",
                "data": {"mode": "plan"},
            },
        )
    ]
    assert state.posted_collaboration_mode == "plan"


@pytest.mark.asyncio
async def test_sync_codex_approval_mode_change_posts_and_dedupes() -> None:
    """Codex ``/permissions`` changes mirror to terminal_launch_args once."""
    client = _RecordingClient()
    state = fwd._CodexForwarderState()
    state.note_thread_settings_updated(
        {
            "threadSettings": {
                "approvalPolicy": "never",
                "approvalsReviewer": "auto_review",
                "sandboxPolicy": {"type": "danger-full-access"},
                "activePermissionProfile": {"id": "dev", "extends": ":workspace"},
            }
        }
    )

    await fwd._sync_codex_approval_mode_change(
        client,
        session_id="conv_x",
        forwarder_state=state,
    )
    await fwd._sync_codex_approval_mode_change(
        client,
        session_id="conv_x",
        forwarder_state=state,
    )

    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_codex_approval_mode_change",
                "data": {
                    "terminal_launch_args": [
                        "-c",
                        'default_permissions="dev"',
                        "-c",
                        'approval_policy="never"',
                        "-c",
                        'approvals_reviewer="auto_review"',
                    ]
                },
            },
        )
    ]
    assert state.posted_terminal_launch_args == [
        "-c",
        'default_permissions="dev"',
        "-c",
        'approval_policy="never"',
        "-c",
        'approvals_reviewer="auto_review"',
    ]


def test_codex_permission_settings_fall_back_to_legacy_policy_args() -> None:
    """Legacy settings without an active profile keep approval and sandbox."""
    assert fwd._codex_terminal_launch_args_from_settings(
        {
            "approvalPolicy": "on-failure",
            "approvalsReviewer": "user",
            "sandboxPolicy": {"type": "workspace-write"},
        }
    ) == [
        "--sandbox",
        "workspace-write",
        "--ask-for-approval",
        "on-failure",
        "-c",
        'approvals_reviewer="user"',
    ]


@pytest.mark.parametrize(
    "content,expected",
    [
        ([{"type": "image", "url": "data:image/png;base64,AAAA"}], True),
        ([{"type": "input_file", "file_data": "data:application/pdf;base64,AAAA"}], True),
        ([{"type": "text", "text": "hi"}, {"type": "image", "url": "data:x"}], True),
        ([{"type": "text", "text": "only text"}], False),
        ([], False),
        ("not a list", False),
    ],
)
def test_user_message_has_file_content(content: object, expected: bool) -> None:
    """
    Detect a non-text (image/file) block in a Codex ``userMessage``.

    Drives the gate that decides whether a text-less ``userMessage`` is a
    real image-bearing message that must be persisted. ``True`` for any
    block whose ``type`` is not ``"text"``, else ``False``. A wrong result
    re-opens the image-only regression (text-less image skipped → dropped
    bubble + pending-FIFO bleed) or makes text-only messages post twice.
    """
    assert fwd._user_message_has_file_content({"content": content}) is expected


@pytest.mark.asyncio
async def test_post_user_message_image_only_posts_empty_content() -> None:
    """
    An image-only ``userMessage`` is posted with EMPTY Omnigent content.

    Regression guard for the image-only bleed/ordering bug: the forwarder
    must post the user item (so the server drains the pending-input FIFO
    entry and folds the image in by file_id). The posted content is empty
    — the base64 ``data:`` URL Codex echoes must NOT be written into text.
    A bail here would drop the user bubble and leak the pending entry into
    the next message.
    """
    client = _RecordingClient()
    item = {
        "type": "userMessage",
        "content": [{"type": "image", "url": "data:image/png;base64,AAAA"}],
    }

    await fwd._post_user_message(client, "conv_x", {"turnId": "t1"}, item)

    assert len(client.posts) == 1, "image-only userMessage must still be posted"
    _url, body = client.posts[0]
    item_data = body["data"]["item_data"]
    assert item_data["role"] == "user"
    # Empty content: the image is supplied server-side from the pending
    # entry; echoing Codex's base64 url here would re-introduce the freeze.
    assert item_data["content"] == []


@pytest.mark.asyncio
async def test_post_user_message_text_posts_input_text() -> None:
    """A text ``userMessage`` posts an ``input_text`` block (unchanged path)."""
    client = _RecordingClient()
    item = {"type": "userMessage", "content": [{"type": "text", "text": "hello"}]}

    await fwd._post_user_message(client, "conv_x", {"turnId": "t1"}, item)

    assert len(client.posts) == 1
    _url, body = client.posts[0]
    assert body["data"]["item_data"]["content"] == [{"type": "input_text", "text": "hello"}]


@pytest.mark.asyncio
async def test_post_user_message_truly_empty_is_skipped() -> None:
    """
    A ``userMessage`` with neither text nor a file block is not posted.

    Without this guard the forwarder would emit spurious empty user
    bubbles. A failure (a post recorded) means the empty-skip branch broke.
    """
    client = _RecordingClient()
    item = {"type": "userMessage", "content": []}

    await fwd._post_user_message(client, "conv_x", {"turnId": "t1"}, item)

    assert client.posts == []


# ── sub-agent usage pricing: seed the child coalescer's model ──────────


@pytest.mark.asyncio
async def test_usage_coalescer_seeded_model_rides_along_so_child_usage_prices() -> None:
    """A coalescer seeded with a model attaches it to its token post.

    Codex sub-agent (child-thread) usage is recorded on a coalescer created on
    the child-event path, where ``forwarder_state`` (the usual model source) is
    intentionally ``None`` — so ``record()`` receives no model. Without the
    constructor seed the token post carries no ``model``, the server leaves the
    child's ``total_cost_usd`` unpriced (``None``), and the sub-agent's spend
    drops out of the parent's subtree cost — letting it run past the budget.
    The seeded model must ride along on every token post so the server can
    price the cumulative tokens.
    """
    client = _RecordingClient()
    coalescer = fwd._SessionUsageCoalescer(client, "conv_child", model="gpt-5.5")
    # Mirror the child path exactly: a usage frame with NO model in record().
    coalescer.record({"tokenUsage": {"total": {"inputTokens": 46003, "outputTokens": 4141}}})
    await coalescer.flush()

    assert len(client.posts) == 1  # one external_session_usage post
    url, body = client.posts[0]
    assert url == "/v1/sessions/conv_child/events"
    assert body["type"] == "external_session_usage"
    data = body["data"]
    # The seeded model rides along — this is what lets the server price the
    # tokens into the child's total_cost_usd (the whole point of the fix).
    assert data["model"] == "gpt-5.5"
    # The cumulative token counts the server prices from are present.
    assert data["cumulative_input_tokens"] == 46003
    assert data["cumulative_output_tokens"] == 4141


@pytest.mark.asyncio
async def test_usage_coalescer_unseeded_omits_model() -> None:
    """Without a seed (and no model via record), the post carries no model.

    This is the pre-fix behavior that left a sub-agent's cost unpriced; the
    test pins the contrast so a regression dropping the seed is caught (the
    post would silently go back to model-less and the budget gap would return).
    """
    client = _RecordingClient()
    coalescer = fwd._SessionUsageCoalescer(client, "conv_child")  # no model seed
    coalescer.record({"tokenUsage": {"total": {"inputTokens": 100, "outputTokens": 5}}})
    await coalescer.flush()

    assert len(client.posts) == 1
    _url, body = client.posts[0]
    assert "model" not in body["data"]


class _FlakyElicitationClient:
    """
    Elicitation client stub: configurable failures, then HTTP 200.

    Real stub (not MagicMock) so unexpected extra calls surface in
    :attr:`posts`. Each call records ``(url, json)``. The first
    ``transport_failures`` calls raise ``httpx.ReadError`` (a severed
    long-poll) and the next ``gateway_failures`` calls return HTTP 502
    (a proxy gateway error); every later call returns 200 with a
    JSON-RPC result body.

    :param transport_failures: Calls to fail with ``httpx.ReadError``
        before succeeding, e.g. ``1``.
    :param gateway_failures: Calls to answer with HTTP 502 after the
        transport failures, e.g. ``0``.
    """

    def __init__(self, transport_failures: int = 0, gateway_failures: int = 0) -> None:
        self.posts: list[tuple[str, dict]] = []
        self._transport_failures = transport_failures
        self._gateway_failures = gateway_failures

    async def post(self, url: str, *, json: dict, timeout: httpx.Timeout) -> httpx.Response:
        """
        Record the call and fail/succeed per the configured schedule.

        :param url: Request URL, e.g.
            ``"/v1/sessions/conv_x/hooks/codex-elicitation-request"``.
        :param json: Codex JSON-RPC request envelope.
        :param timeout: Per-attempt budget (ignored by the stub).
        :returns: HTTP 502 during the gateway-failure window, else 200
            with a JSON-RPC result body.
        :raises httpx.ReadError: During the transport-failure window.
        """
        self.posts.append((url, json))
        attempt = len(self.posts)
        if attempt <= self._transport_failures:
            raise httpx.ReadError(
                "proxy severed the long-poll",
                request=httpx.Request("POST", url),
            )
        if attempt <= self._transport_failures + self._gateway_failures:
            return httpx.Response(502, request=httpx.Request("POST", url))
        return httpx.Response(
            200,
            json={"action": "accept", "content": {}, "_meta": None},
            request=httpx.Request("POST", url),
        )


async def _instant_retry_sleep(_seconds: float) -> None:
    """
    Drop-in for ``_elicitation_retry_sleep`` that returns at once.

    :param _seconds: Ignored backoff duration.
    :returns: None.
    """
    return


_ELICITATION_EVENT: dict = {
    "id": 7,
    "method": "mcpServer/elicitation/request",
    "params": {"mode": "form", "message": "Pick a date"},
}


@pytest.mark.asyncio
async def test_elicitation_post_reposts_after_transport_cut_with_same_envelope(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    A severed elicitation long-poll is re-POSTed with the identical envelope.

    This is the invisible-stuck bug for codex sub-agents: one transport
    error used to abandon the prompt to the native-TUI path nobody is
    watching. The envelope must be byte-identical on the retry — the
    server derives the deterministic elicitation id from (session,
    method, rpc id), so an identical re-POST re-parks the SAME prompt
    and keeps the approval card alive.
    """
    monkeypatch.setattr(fwd, "_elicitation_retry_sleep", _instant_retry_sleep)
    client = _FlakyElicitationClient(transport_failures=1)

    response = await fwd._post_codex_elicitation_request(
        client,  # type: ignore[arg-type]  # stub implements the one used method
        "conv_x",
        event=_ELICITATION_EVENT,
    )

    assert response is not None
    assert response.status_code == 200
    # 2 = one severed attempt + one successful retry. 1 means the
    # transport error abandoned the prompt (the production bug).
    assert len(client.posts) == 2, f"expected 2 attempts, got {len(client.posts)}"
    # Identical (url, envelope) on the retry is the re-park contract.
    assert client.posts[0] == client.posts[1]
    assert client.posts[0][0] == "/v1/sessions/conv_x/hooks/codex-elicitation-request"


@pytest.mark.asyncio
async def test_elicitation_post_retries_gateway_5xx(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    A 5xx (proxy gateway error on a severed long-poll) is retried.

    The Databricks Apps proxy answers a killed upstream long-poll with
    502/504 rather than a clean transport error; the verdict may still
    be pending server-side, so the forwarder must re-park rather than
    treat it as final.
    """
    monkeypatch.setattr(fwd, "_elicitation_retry_sleep", _instant_retry_sleep)
    client = _FlakyElicitationClient(gateway_failures=1)

    response = await fwd._post_codex_elicitation_request(
        client,  # type: ignore[arg-type]
        "conv_x",
        event=_ELICITATION_EVENT,
    )

    assert response is not None
    assert response.status_code == 200
    # 2 = the 502 attempt + the successful retry; 1 would mean 5xx was
    # treated as a final answer and the prompt abandoned.
    assert len(client.posts) == 2


@pytest.mark.asyncio
async def test_elicitation_post_4xx_is_final(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    A 4xx is a deliberate server rejection — returned without retry.

    Retrying a rejection would hammer the server with a request it
    already refused; the caller logs it and leaves the native request
    unanswered.
    """

    class _RejectingClient:
        """Client stub answering every elicitation POST with HTTP 400."""

        def __init__(self) -> None:
            self.posts: list[tuple[str, dict]] = []

        async def post(self, url: str, *, json: dict, timeout: httpx.Timeout) -> httpx.Response:
            """
            Record the call and reject it.

            :param url: Request URL.
            :param json: Codex JSON-RPC request envelope.
            :param timeout: Per-attempt budget (ignored by the stub).
            :returns: HTTP 400.
            """
            self.posts.append((url, json))
            return httpx.Response(400, request=httpx.Request("POST", url))

    monkeypatch.setattr(fwd, "_elicitation_retry_sleep", _instant_retry_sleep)
    client = _RejectingClient()

    response = await fwd._post_codex_elicitation_request(
        client,  # type: ignore[arg-type]
        "conv_x",
        event=_ELICITATION_EVENT,
    )

    assert response is not None
    assert response.status_code == 400
    # 1 = the rejection was final; 2+ means 4xx is being retried.
    assert len(client.posts) == 1


@pytest.mark.asyncio
async def test_elicitation_post_returns_none_when_budget_exhausted(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    An exhausted retry budget returns ``None`` (caller leaves the
    native request unanswered, matching the old single-attempt outcome).
    """
    # Budget smaller than the first backoff → exactly one attempt.
    monkeypatch.setattr(fwd, "_CODEX_ELICITATION_REQUEST_TIMEOUT_SECONDS", 0.5)
    monkeypatch.setattr(fwd, "_elicitation_retry_sleep", _instant_retry_sleep)
    client = _FlakyElicitationClient(transport_failures=100)

    response = await fwd._post_codex_elicitation_request(
        client,  # type: ignore[arg-type]
        "conv_x",
        event=_ELICITATION_EVENT,
    )

    assert response is None
    # 1 = the deadline check stopped the loop before a second attempt
    # (backoff 1.0s > 0.5s budget); more means the budget is ignored.
    assert len(client.posts) == 1


class _StatusClient:
    """httpx client stub whose ``post`` returns a fixed status code."""

    def __init__(self, status_code: int) -> None:
        """:param status_code: Status to return from every post, e.g. ``400``."""
        self.status_code = status_code
        self.posts = 0

    async def post(self, url: str, *, json: dict) -> httpx.Response:
        """Return the configured status; never raises."""
        del json
        self.posts += 1
        return httpx.Response(self.status_code, request=httpx.Request("POST", url))


def test_forward_failures_escalate_to_degraded_once() -> None:
    """
    Sustained forward failures flip the degraded latch exactly once (#1120).

    Network drops previously surfaced only as scattered per-item warnings;
    the latch turns a real outage into a single loud signal and does not
    re-fire per dropped item.
    """
    fwd._reset_forward_health()

    for _ in range(fwd._FORWARD_DEGRADED_THRESHOLD - 1):
        fwd._note_forward_failure("external_output_text_delta")
    # Below threshold: not yet degraded.
    assert fwd._forward_health.degraded_logged is False

    fwd._note_forward_failure("external_output_text_delta")  # crosses threshold
    assert fwd._forward_health.degraded_logged is True
    assert fwd._forward_health.consecutive_failures == fwd._FORWARD_DEGRADED_THRESHOLD

    # The latch holds — further failures keep counting but don't re-escalate.
    fwd._note_forward_failure("external_output_text_delta")
    assert fwd._forward_health.degraded_logged is True
    assert fwd._forward_health.consecutive_failures == fwd._FORWARD_DEGRADED_THRESHOLD + 1


def test_forward_success_resets_degraded_state() -> None:
    """
    A successful forward clears the failure count and degraded latch.

    Recovery must re-arm the indicator so a later outage escalates again.
    """
    fwd._reset_forward_health()
    for _ in range(fwd._FORWARD_DEGRADED_THRESHOLD):
        fwd._note_forward_failure("external_session_usage")
    assert fwd._forward_health.degraded_logged is True

    fwd._note_forward_success()

    assert fwd._forward_health.consecutive_failures == 0
    assert fwd._forward_health.degraded_logged is False


@pytest.mark.asyncio
async def test_post_session_event_tracks_success_and_failure() -> None:
    """
    _post_session_event classifies each outcome into forward health (#1120).

    A 2xx clears the failure run; a permanent 4xx counts as a failure so a
    sustained outage can escalate.
    """
    fwd._reset_forward_health()

    # A permanent 4xx is a failure.
    await fwd._post_session_event(
        _StatusClient(400), "conv_x", event_type="external_session_status", data={"status": "idle"}
    )
    assert fwd._forward_health.consecutive_failures == 1

    # A 2xx resets the run.
    await fwd._post_session_event(
        _RecordingClient(), "conv_x", event_type="external_session_status", data={"status": "idle"}
    )
    assert fwd._forward_health.consecutive_failures == 0


# ── #1108: turn-error "silent success" → surfaced failed ──────────────
#
# A failed Codex turn arrives as ``turn/completed`` (a clean success boundary)
# with ``turn.status == "failed"`` and a ``turn.error`` object. These tests pin
# the surface-only fix: such turns are forced to ``failed``, the reason is
# surfaced as the status output, auth errors (codexErrorInfo / 401-403) carry a
# re-auth hint, the resume path reaches the same verdict, an empty turn is idle
# (+ WARN), and a genuinely clean turn still reports success.


def _seed_active_turn(bridge_dir: Path, turn_id: str) -> None:
    """
    Seed bridge state so a terminal turn edge clears the active turn.

    ``_terminal_turn_status_edge`` only produces an edge when the terminal
    event clears the recorded active turn id; without this seed it returns
    ``None`` as "stale".

    :param bridge_dir: Native Codex bridge directory (the test ``tmp_path``).
    :param turn_id: Active Codex turn id to record, e.g. ``"turn_123"``.
    :returns: None.
    """
    write_bridge_state(
        bridge_dir,
        CodexNativeBridgeState(
            session_id="conv_x",
            socket_path=str(bridge_dir / "app-server.sock"),
            thread_id="thread_123",
            codex_home=str(bridge_dir / "codex-home"),
            active_turn_id=turn_id,
        ),
    )


def test_classify_codex_error_auth_vs_generic() -> None:
    """The shared classifier flags auth errors and leaves the rest generic.

    This is the single classifier reused by both the live and resume paths;
    if it regresses, an expired-login failure would surface without the
    re-auth hint (or a disk-full error would wrongly demand re-auth). It
    prefers ``codexErrorInfo`` (variant / httpStatusCode) and falls back to
    the message text.
    """
    auth = fwd._CODEX_ERROR_KIND_AUTH
    generic = fwd._CODEX_ERROR_KIND_GENERIC
    # Structured codexErrorInfo: string variant, tagged object, http status.
    assert fwd._classify_codex_error({"codexErrorInfo": "Unauthorized"}, "nope") == auth
    assert fwd._classify_codex_error({"codexErrorInfo": {"type": "Unauthorized"}}, "nope") == auth
    assert fwd._classify_codex_error({"codexErrorInfo": {"httpStatusCode": 401}}, "nope") == auth
    # The real app-server enum serializes lowercase snake_case; it must match
    # via the structured path (message "nope" has no auth substring to fall
    # back on), case-insensitively.
    assert fwd._classify_codex_error({"codexErrorInfo": "unauthorized"}, "nope") == auth
    assert fwd._classify_codex_error({"codexErrorInfo": {"type": "unauthorized"}}, "nope") == auth
    # Message-text fallback when codexErrorInfo is absent.
    assert fwd._classify_codex_error({}, "Please run codex login") == auth
    assert fwd._classify_codex_error({}, "ChatGPT session expired") == auth
    assert fwd._classify_codex_error({"codexErrorInfo": "Other"}, "disk full") == generic


def test_terminal_error_from_turn_reads_and_classifies_turn_error() -> None:
    """``_terminal_error_from_turn`` returns the classified ``turn.error``.

    The helper is the single source of truth for "did this turn fail"; both
    edge builders depend on it, so it must read ``turn.error`` and classify it.
    """
    params = {
        "turn": {
            "id": "turn_123",
            "status": "failed",
            "error": {
                "message": "401 Unauthorized: login expired",
                "codexErrorInfo": "Unauthorized",
            },
        }
    }

    error = fwd._terminal_error_from_turn(params)

    assert error is not None
    assert error.message == "401 Unauthorized: login expired"
    assert error.kind == fwd._CODEX_ERROR_KIND_AUTH
    assert error.is_auth is True


def test_terminal_error_from_turn_falls_back_to_error_item() -> None:
    """With no ``turn.error``, an ``error`` ThreadItem in ``turn.items`` is used.

    Both shapes exist in the app-server type system; the fallback keeps the fix
    correct on the version/path that emits the error as an item rather than as a
    ``turn.error`` object.
    """
    params = {
        "turn": {
            "id": "turn_123",
            "status": "completed",
            "items": [
                {"type": "agentMessage", "id": "a", "text": "working"},
                {"type": "error", "message": "please run codex login"},
            ],
        }
    }

    error = fwd._terminal_error_from_turn(params)

    assert error is not None
    assert error.message == "please run codex login"
    assert error.is_auth is True


def test_terminal_error_from_turn_prefers_turn_error_over_item() -> None:
    """``turn.error`` wins when both it and an ``error`` item are present."""
    params = {
        "turn": {
            "id": "turn_123",
            "status": "failed",
            "error": {"message": "from turn.error"},
            "items": [{"type": "error", "message": "from item"}],
        }
    }

    error = fwd._terminal_error_from_turn(params)

    assert error is not None
    assert error.message == "from turn.error"


def test_terminal_error_from_turn_none_for_clean_turn() -> None:
    """A turn with no ``error`` object or item yields ``None`` (no false positives)."""
    params = {
        "turn": {
            "id": "turn_123",
            "status": "completed",
            "items": [{"type": "agentMessage", "id": "a", "text": "done"}],
        }
    }

    assert fwd._terminal_error_from_turn(params) is None


def test_terminal_turn_status_edge_error_item_forces_failed(tmp_path: Path) -> None:
    """A ``turn/completed`` carrying an ``error`` item (no ``turn.error``) fails.

    The item-fallback path must flip the live edge to ``failed`` just like the
    ``turn.error`` path does.
    """
    _seed_active_turn(tmp_path, "turn_123")
    params = {
        "turn": {
            "id": "turn_123",
            "status": "completed",
            "items": [{"type": "error", "message": "model stream broke"}],
        }
    }

    edge = fwd._terminal_turn_status_edge(tmp_path, "turn/completed", params)

    assert edge is not None
    assert edge.status == "failed"
    assert edge.error is not None
    assert edge.error.message == "model stream broke"
    assert edge.source == "turn/completed:turn-error"


def test_terminal_turn_status_edge_turn_error_forces_failed(tmp_path: Path) -> None:
    """A ``turn/completed`` carrying ``turn.error`` is forced to ``failed``.

    This is the core of #1108: Codex reported a *completed* boundary, but the
    turn actually failed. The edge must be ``failed`` (not the silent ``idle``
    the method alone implies) and carry the classified error.
    """
    _seed_active_turn(tmp_path, "turn_123")
    params = {
        "turn": {
            "id": "turn_123",
            "status": "failed",
            "error": {"message": "model stream broke"},
        }
    }

    edge = fwd._terminal_turn_status_edge(tmp_path, "turn/completed", params)

    assert edge is not None
    assert edge.status == "failed"
    assert edge.turn_id == "turn_123"
    assert edge.error is not None
    assert edge.error.message == "model stream broke"
    assert edge.error.kind == fwd._CODEX_ERROR_KIND_GENERIC
    assert edge.source == "turn/completed:turn-error"


def test_terminal_turn_status_edge_auth_turn_error_classified(tmp_path: Path) -> None:
    """An auth-classified ``turn.error`` rides the failed edge as ``auth``."""
    _seed_active_turn(tmp_path, "turn_123")
    params = {
        "turn": {
            "id": "turn_123",
            "status": "failed",
            "error": {
                "message": "Forbidden",
                "codexErrorInfo": {"type": "Unauthorized", "httpStatusCode": 403},
            },
        }
    }

    edge = fwd._terminal_turn_status_edge(tmp_path, "turn/completed", params)

    assert edge is not None
    assert edge.status == "failed"
    assert edge.error is not None
    assert edge.error.is_auth is True


def test_terminal_turn_status_edge_failed_status_without_error(tmp_path: Path) -> None:
    """A ``turn.status == "failed"`` with no ``error`` object still fails.

    Defends against an app-server version that records the failed status but
    omits the populated ``turn.error`` — the edge must not fall back to ``idle``.
    """
    _seed_active_turn(tmp_path, "turn_123")
    params = {"turn": {"id": "turn_123", "status": "failed"}}

    edge = fwd._terminal_turn_status_edge(tmp_path, "turn/completed", params)

    assert edge is not None
    assert edge.status == "failed"
    assert edge.error is None
    assert edge.source == "turn/completed:turn-failed"


def test_terminal_turn_status_edge_clean_turn_still_idle(tmp_path: Path) -> None:
    """A genuinely clean ``turn/completed`` still maps to ``idle`` (regression).

    The turn-error check must not break the happy path: no error → the edge
    stays ``idle`` with no attached error.
    """
    _seed_active_turn(tmp_path, "turn_123")
    params = {
        "turn": {
            "id": "turn_123",
            "status": "completed",
            "items": [{"type": "agentMessage", "id": "a", "text": "all good"}],
        }
    }

    edge = fwd._terminal_turn_status_edge(tmp_path, "turn/completed", params)

    assert edge is not None
    assert edge.status == "idle"
    assert edge.error is None
    assert edge.source == "turn/completed"


def test_terminal_turn_status_edge_empty_turn_idle_and_warns(
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """A zero-item ``turn/completed`` maps to ``idle`` and emits a WARN.

    An empty turn is not an error, but it is unusual enough to log: it maps to
    ``idle`` (so the session closes) while a WARN records the anomaly.
    """
    _seed_active_turn(tmp_path, "turn_123")
    params = {"turn": {"id": "turn_123", "status": "completed", "items": []}}

    with caplog.at_level("WARNING", logger="omnigent.codex_native_forwarder"):
        edge = fwd._terminal_turn_status_edge(tmp_path, "turn/completed", params)

    assert edge is not None
    assert edge.status == "idle"
    assert edge.error is None
    assert any(
        "empty turn" in record.getMessage() and record.levelname == "WARNING"
        for record in caplog.records
    ), "expected a WARN log for the empty (zero-item) turn"


def test_omnigent_status_from_resume_turn_error_parity() -> None:
    """Resume parity: a completed resume turn carrying ``turn.error`` → ``failed``.

    Without this, a reconnect that backfills from ``thread/resume`` would close
    the session as ``idle`` even though the turn had errored — the resume-path
    half of the silent-success bug.
    """
    turn_with_error = {
        "id": "turn_123",
        "status": "completed",
        "error": {"message": "rate limited"},
    }
    turn_clean = {
        "id": "turn_123",
        "status": "completed",
        "items": [{"type": "agentMessage", "id": "a", "text": "hi"}],
    }

    assert fwd._omnigent_status_from_resume_turn(turn_with_error) == "failed"
    # Parity check: the clean turn still resolves to idle.
    assert fwd._omnigent_status_from_resume_turn(turn_clean) == "idle"


def test_resume_terminal_status_edge_attaches_error(tmp_path: Path) -> None:
    """The resume edge carries the classified error like the live edge does."""
    write_bridge_state(
        tmp_path,
        CodexNativeBridgeState(
            session_id="conv_x",
            socket_path=str(tmp_path / "app-server.sock"),
            thread_id="thread_123",
            codex_home=str(tmp_path / "codex-home"),
            active_turn_id="turn_123",
        ),
    )
    turns = [
        {
            "id": "turn_123",
            "status": "failed",
            "error": {"message": "please sign in again"},
        }
    ]

    edge = fwd._resume_terminal_status_edge_for_latest_turn(tmp_path, "thread_123", turns)

    assert edge is not None
    assert edge.status == "failed"
    assert edge.error is not None
    assert edge.error.is_auth is True
    assert edge.source == "thread/resume:turn-error"
    # The active turn id is cleared once the terminal edge is derived.
    state = read_bridge_state(tmp_path)
    assert state is not None
    assert state.active_turn_id is None


@pytest.mark.asyncio
async def test_post_turn_status_edge_surfaces_generic_error_output() -> None:
    """A failed edge with a generic error surfaces the message as output.

    The reason must reach the server (as ``output``) rather than being dropped;
    a generic error carries no re-auth flag.
    """
    client = _RecordingClient()
    edge = fwd._CodexTurnStatusEdge(
        status="failed",
        turn_id="turn_123",
        source="turn/completed:turn-error",
        error=fwd._CodexTerminalError(
            message="model stream broke",
            kind=fwd._CODEX_ERROR_KIND_GENERIC,
        ),
    )

    await fwd._post_turn_status_edge(client, "conv_x", edge)

    assert len(client.posts) == 1
    _url, body = client.posts[0]
    assert body["type"] == "external_session_status"
    data = body["data"]
    assert data["status"] == "failed"
    assert data["output"] == "model stream broke"
    # Generic errors do not demand re-auth.
    assert "reauth_required" not in data


@pytest.mark.asyncio
async def test_post_turn_status_edge_auth_error_includes_reauth_hint() -> None:
    """A failed edge with an auth error flags re-auth and appends the hint."""
    client = _RecordingClient()
    edge = fwd._CodexTurnStatusEdge(
        status="failed",
        turn_id="turn_123",
        source="turn/completed:turn-error",
        error=fwd._CodexTerminalError(
            message="401 Unauthorized",
            kind=fwd._CODEX_ERROR_KIND_AUTH,
        ),
    )

    await fwd._post_turn_status_edge(client, "conv_x", edge)

    assert len(client.posts) == 1
    _url, body = client.posts[0]
    data = body["data"]
    assert data["status"] == "failed"
    assert data["reauth_required"] is True
    assert "401 Unauthorized" in data["output"]
    assert fwd._CODEX_REAUTH_HINT in data["output"]


@pytest.mark.asyncio
async def test_post_turn_status_edge_clean_idle_has_no_output() -> None:
    """A normal idle edge (no error) posts status only — the success path."""
    client = _RecordingClient()
    edge = fwd._CodexTurnStatusEdge(status="idle", turn_id="turn_123", source="turn/completed")

    await fwd._post_turn_status_edge(client, "conv_x", edge)

    assert len(client.posts) == 1
    _url, body = client.posts[0]
    data = body["data"]
    assert data["status"] == "idle"
    assert "output" not in data
    assert "reauth_required" not in data


@pytest.mark.asyncio
async def test_compaction_status_posts_and_dedupes_consecutive() -> None:
    """
    Compaction status mirrors as external_compaction_status, deduped (#1255).

    Codex may signal completion via both a ``contextCompaction`` item and a
    ``thread/compacted`` notification; consecutive identical statuses must
    not double-post (the spinner would flicker).
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState()

    await fwd._post_compaction_status(client, "conv_x", "in_progress", forwarder_state=state)
    await fwd._post_compaction_status(client, "conv_x", "completed", forwarder_state=state)
    # Duplicate completion (e.g. item then notification) is suppressed.
    await fwd._post_compaction_status(client, "conv_x", "completed", forwarder_state=state)

    assert [post[1] for post in client.posts] == [
        {"type": "external_compaction_status", "data": {"status": "in_progress"}},
        {"type": "external_compaction_status", "data": {"status": "completed"}},
    ]
    assert state.compaction_status_posted == "completed"


@pytest.mark.asyncio
async def test_completed_context_compaction_item_clears_spinner() -> None:
    """
    A completed ``contextCompaction`` item posts compaction-completed.

    It is a status edge, not transcript history, so it must clear the
    spinner without being appended as a conversation item.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState()
    state.compaction_status_posted = "in_progress"

    await fwd._handle_completed_item(
        client,
        "conv_x",
        {
            "threadId": "thread_1",
            "turnId": "turn_1",
            "item": {"type": "contextCompaction", "id": "item_c"},
        },
        forwarder_state=state,
    )

    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {"type": "external_compaction_status", "data": {"status": "completed"}},
        )
    ]


@pytest.mark.asyncio
async def test_reasoning_delta_opens_block_then_continues() -> None:
    """
    Codex reasoning deltas mirror as external_output_reasoning_delta (#1254).

    The first delta of a reasoning item opens the block (``started=True``
    → ``response.reasoning.started``); subsequent deltas for the same item
    continue it (``started=False``). Reasoning was previously dropped — only
    the effort *level* synced, never the thinking text.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState()

    await fwd._handle_reasoning_delta(
        client,
        "conv_x",
        {"turnId": "turn_1", "itemId": "item_r", "delta": "Let me "},
        state,
    )
    await fwd._handle_reasoning_delta(
        client,
        "conv_x",
        {"turnId": "turn_1", "itemId": "item_r", "delta": "think."},
        state,
    )

    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_output_reasoning_delta",
                "data": {"delta": "Let me ", "started": True},
            },
        ),
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_output_reasoning_delta",
                "data": {"delta": "think.", "started": False},
            },
        ),
    ]


@pytest.mark.asyncio
async def test_reasoning_delta_new_item_reopens_block() -> None:
    """
    A reasoning delta for a new item id opens a fresh block.

    Multi-step turns (reason → tool → reason) emit a second reasoning item;
    its first delta must re-open the block so the web UI starts a new
    "thinking" section rather than appending to the prior one.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState()

    await fwd._handle_reasoning_delta(
        client, "conv_x", {"itemId": "item_a", "delta": "first"}, state
    )
    await fwd._handle_reasoning_delta(
        client, "conv_x", {"itemId": "item_b", "delta": "second"}, state
    )

    started_flags = [post[1]["data"]["started"] for post in client.posts]
    assert started_flags == [True, True]


@pytest.mark.asyncio
async def test_reasoning_delta_skips_empty_non_opening_delta() -> None:
    """
    An empty delta that does not open a block is dropped (no noise post).

    The block-opening delta is always posted (even empty, to emit
    ``response.reasoning.started``); a later empty continuation carries
    nothing to render and must not POST.
    """
    client = _RecordingClient()
    state = fwd._CodexForwarderState()

    # Opening delta (empty) still posts to open the block.
    await fwd._handle_reasoning_delta(client, "conv_x", {"itemId": "item_r", "delta": ""}, state)
    # Empty continuation for the same item is dropped.
    await fwd._handle_reasoning_delta(client, "conv_x", {"itemId": "item_r", "delta": ""}, state)

    assert len(client.posts) == 1
    assert client.posts[0][1]["data"] == {"delta": "", "started": True}


# ---------------------------------------------------------------------------
# _persist_codex_compaction_item
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_persist_codex_compaction_item_posts_event() -> None:
    """Compaction event is posted with last_item_id and Codex summary."""
    get_resp = MagicMock()
    get_resp.json.return_value = {"data": [{"id": "item_codex"}]}
    get_resp.raise_for_status = MagicMock()

    client = MagicMock()
    client.get = AsyncMock(return_value=get_resp)

    post_resp = MagicMock()
    post_resp.raise_for_status = MagicMock()
    client.post = AsyncMock(return_value=post_resp)

    await _persist_codex_compaction_item(client, session_id="conv_codex")

    client.post.assert_called_once()
    _url, kwargs = client.post.call_args
    body = kwargs["json"]
    assert body["type"] == "compaction"
    assert body["data"]["last_item_id"] == "item_codex"
    assert "Codex" in body["data"]["summary"]
    # Codex can't read post-compaction state, so no compacted_messages
    assert "compacted_messages" not in body["data"]


@pytest.mark.asyncio
async def test_persist_codex_compaction_item_empty_items_fallback() -> None:
    """When no items exist, last_item_id falls back to compact_boundary_ prefix."""
    empty_resp = MagicMock()
    empty_resp.json.return_value = {"data": []}
    empty_resp.raise_for_status = MagicMock()

    client = MagicMock()
    client.get = AsyncMock(return_value=empty_resp)

    post_resp = MagicMock()
    post_resp.raise_for_status = MagicMock()
    client.post = AsyncMock(return_value=post_resp)

    await _persist_codex_compaction_item(client, session_id="conv_codex")

    client.post.assert_called_once()
    _url, kwargs = client.post.call_args
    body = kwargs["json"]
    assert body["data"]["last_item_id"].startswith("compact_boundary_")
    assert "compacted_messages" not in body["data"]


def test_read_compacted_history_extracts_replacement_history_and_window_id(
    tmp_path: Path,
) -> None:
    """_read_compacted_history returns replacement_history and window_id."""
    import json as _json

    rollout = tmp_path / "rollout.jsonl"
    lines = [
        _json.dumps({"type": "session_meta", "payload": {"id": "abc"}}),
        _json.dumps({"type": "response_item", "payload": {"type": "message", "role": "user"}}),
        _json.dumps(
            {
                "type": "compacted",
                "payload": {
                    "message": "summary",
                    "replacement_history": [
                        {
                            "type": "message",
                            "role": "user",
                            "content": [{"type": "input_text", "text": "hi"}],
                        },
                        {
                            "type": "compaction",
                            "encrypted_content": "gAAAA_test_token",
                        },
                    ],
                    "window_id": 2,
                },
            }
        ),
    ]
    rollout.write_text("\n".join(lines) + "\n")

    result = fwd._read_compacted_history(rollout)

    assert result is not None
    assert result["window_id"] == 2
    assert len(result["replacement_history"]) == 2
    assert result["replacement_history"][0]["type"] == "message"
    assert result["replacement_history"][0]["role"] == "user"
    assert result["replacement_history"][1]["type"] == "compaction"
    assert result["replacement_history"][1]["encrypted_content"] == "gAAAA_test_token"


def test_read_compacted_history_returns_none_for_no_compacted_entry(
    tmp_path: Path,
) -> None:
    """_read_compacted_history returns None when no Compacted entry exists."""
    import json as _json

    rollout = tmp_path / "rollout.jsonl"
    rollout.write_text(_json.dumps({"type": "session_meta", "payload": {"id": "abc"}}) + "\n")

    assert fwd._read_compacted_history(rollout) is None


@pytest.mark.asyncio
async def test_post_session_event_dead_letters_durable_event_on_permanent_failure(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """
    A permanently-failed durable event is dead-lettered to disk (#1120).

    Drives ``_post_session_event`` (the health-tracking wrapper) with a stubbed
    inner that returns an HTTP 500, and asserts the dropped
    ``external_conversation_item`` payload is appended to
    ``{bridge_dir}/dead_letter.jsonl``.

    :param tmp_path: Pytest temp dir standing in for the bridge dir.
    :param monkeypatch: Pytest patcher (auto-restores the stubbed inner).
    """
    import json as _json

    fwd._reset_forward_health()

    async def _failing_inner(client, session_id, *, event_type, data):
        return fwd._PostResult(
            response=httpx.Response(500, request=httpx.Request("POST", "http://test"))
        )

    monkeypatch.setattr(fwd, "_post_session_event_inner", _failing_inner)
    token = fwd._dead_letter_dir.set(tmp_path)
    try:
        data = {"item_type": "message", "item_data": {"role": "assistant"}}
        await fwd._post_session_event(
            MagicMock(),
            "conv_codex1",
            event_type="external_conversation_item",
            data=data,
        )
    finally:
        fwd._dead_letter_dir.reset(token)
        fwd._reset_forward_health()

    dl_path = tmp_path / "dead_letter.jsonl"
    lines = dl_path.read_text(encoding="utf-8").splitlines()
    assert len(lines) == 1
    record = _json.loads(lines[0])
    assert record["session_id"] == "conv_codex1"
    assert record["event_type"] == "external_conversation_item"
    assert record["payload"] == data


@pytest.mark.asyncio
async def test_post_session_event_does_not_dead_letter_ephemeral_event(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """
    An ephemeral (non-durable) event is NOT dead-lettered on failure (#1120).

    :param tmp_path: Pytest temp dir standing in for the bridge dir.
    :param monkeypatch: Pytest patcher (auto-restores the stubbed inner).
    """
    fwd._reset_forward_health()

    async def _failing_inner(client, session_id, *, event_type, data):
        return fwd._PostResult(
            response=httpx.Response(500, request=httpx.Request("POST", "http://test"))
        )

    monkeypatch.setattr(fwd, "_post_session_event_inner", _failing_inner)
    token = fwd._dead_letter_dir.set(tmp_path)
    try:
        await fwd._post_session_event(
            MagicMock(),
            "conv_codex1",
            event_type="external_output_text_delta",
            data={"delta": "hi"},
        )
    finally:
        fwd._dead_letter_dir.reset(token)
        fwd._reset_forward_health()

    assert not (tmp_path / "dead_letter.jsonl").exists()


@pytest.mark.asyncio
async def test_post_session_event_dead_letters_usage_on_permanent_failure(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """
    A permanently-failed ``external_session_usage`` event is dead-lettered (#1120).

    Usage is the other durable type alongside conversation items, so its
    transcript/usage data must also be recoverable on a sustained outage.

    :param tmp_path: Pytest temp dir standing in for the bridge dir.
    :param monkeypatch: Pytest patcher (auto-restores the stubbed inner).
    """
    import json as _json

    fwd._reset_forward_health()

    async def _failing_inner(client, session_id, *, event_type, data):
        return fwd._PostResult(
            response=httpx.Response(500, request=httpx.Request("POST", "http://test"))
        )

    monkeypatch.setattr(fwd, "_post_session_event_inner", _failing_inner)
    token = fwd._dead_letter_dir.set(tmp_path)
    try:
        data = {"context_tokens": 1234, "model": "databricks-claude-opus-4-7"}
        await fwd._post_session_event(
            MagicMock(),
            "conv_codex_usage",
            event_type="external_session_usage",
            data=data,
        )
    finally:
        fwd._dead_letter_dir.reset(token)
        fwd._reset_forward_health()

    dl_path = tmp_path / "dead_letter.jsonl"
    lines = dl_path.read_text(encoding="utf-8").splitlines()
    assert len(lines) == 1
    record = _json.loads(lines[0])
    assert record["session_id"] == "conv_codex_usage"
    assert record["event_type"] == "external_session_usage"
    assert record["payload"] == data


class _RaisingPostClient:
    """Async client stub whose ``post`` always raises a fixed transport error."""

    def __init__(self, exc: httpx.HTTPError) -> None:
        self._exc = exc
        self.calls = 0

    async def post(self, url: str, json: object) -> httpx.Response:
        self.calls += 1
        raise self._exc


@pytest.mark.asyncio
async def test_post_session_event_inner_classifies_ambiguous_skip() -> None:
    """
    An ambiguous conversation-item transport failure surfaces as ambiguous (#1579).

    The inner used to conflate this with a proven-undelivered failure (both
    returned ``None``); replay must be able to tell them apart.
    """
    client = _RaisingPostClient(
        httpx.ReadTimeout("response lost", request=httpx.Request("POST", "http://test"))
    )
    result = await fwd._post_session_event_inner(
        client,
        "conv_codex1",
        event_type="external_conversation_item",
        data={"item_type": "message"},
    )
    assert result.response is None
    assert result.delivered_ambiguous is True
    assert result.transport_error == "ReadTimeout"
    # Ambiguous items are abandoned immediately — no retries.
    assert client.calls == 1


@pytest.mark.asyncio
async def test_post_session_event_inner_classifies_proven_undelivered(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """
    A connect failure exhausted after retries is proven-undelivered, not ambiguous.
    """
    monkeypatch.setattr(fwd, "_sleep", AsyncMock())
    client = _RaisingPostClient(
        httpx.ConnectError("refused", request=httpx.Request("POST", "http://test"))
    )
    result = await fwd._post_session_event_inner(
        client,
        "conv_codex1",
        event_type="external_conversation_item",
        data={"item_type": "message"},
    )
    assert result.response is None
    assert result.delivered_ambiguous is False
    assert result.transport_error == "ConnectError"
    # Connect failures are safe to retry, so all attempts are spent.
    assert client.calls == fwd._POST_MAX_ATTEMPTS


@pytest.mark.asyncio
async def test_post_session_event_dead_letters_ambiguous_classification(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """
    An ambiguous-skip drop is dead-lettered with ``delivered_ambiguous=True`` (#1579).

    :param tmp_path: Pytest temp dir standing in for the bridge dir.
    :param monkeypatch: Pytest patcher (auto-restores the stubbed inner).
    """
    import json as _json

    fwd._reset_forward_health()

    async def _ambiguous_inner(client, session_id, *, event_type, data):
        return fwd._PostResult(
            response=None, delivered_ambiguous=True, transport_error="ReadTimeout"
        )

    monkeypatch.setattr(fwd, "_post_session_event_inner", _ambiguous_inner)
    token = fwd._dead_letter_dir.set(tmp_path)
    try:
        await fwd._post_session_event(
            MagicMock(),
            "conv_codex1",
            event_type="external_conversation_item",
            data={"item_type": "message"},
        )
    finally:
        fwd._dead_letter_dir.reset(token)
        fwd._reset_forward_health()

    record = _json.loads((tmp_path / "dead_letter.jsonl").read_text().splitlines()[0])
    assert record["delivered_ambiguous"] is True
    assert record["http_status"] is None
    assert record["transport_error"] == "ReadTimeout"


@pytest.mark.asyncio
async def test_post_session_event_dead_letters_records_http_status(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """
    A status-bearing failure records ``http_status`` and is not ambiguous (#1579).

    :param tmp_path: Pytest temp dir standing in for the bridge dir.
    :param monkeypatch: Pytest patcher (auto-restores the stubbed inner).
    """
    import json as _json

    fwd._reset_forward_health()

    async def _failing_inner(client, session_id, *, event_type, data):
        return fwd._PostResult(
            response=httpx.Response(503, request=httpx.Request("POST", "http://test"))
        )

    monkeypatch.setattr(fwd, "_post_session_event_inner", _failing_inner)
    token = fwd._dead_letter_dir.set(tmp_path)
    try:
        await fwd._post_session_event(
            MagicMock(),
            "conv_codex1",
            event_type="external_conversation_item",
            data={"item_type": "message"},
        )
    finally:
        fwd._dead_letter_dir.reset(token)
        fwd._reset_forward_health()

    record = _json.loads((tmp_path / "dead_letter.jsonl").read_text().splitlines()[0])
    assert record["http_status"] == 503
    assert record["delivered_ambiguous"] is False
    assert record["transport_error"] is None


@pytest.mark.asyncio
async def test_replay_dead_letters_on_startup_reposts_proven_undelivered(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """
    On startup, a proven-undelivered record is re-POSTed and removed (#1579).

    :param tmp_path: Pytest temp dir standing in for the bridge dir.
    :param monkeypatch: Pytest patcher (auto-restores the stubbed inner).
    """
    fwd.append_dead_letter(
        tmp_path,
        session_id="conv_codex1",
        event_type="external_conversation_item",
        payload={"item_type": "message"},
        reason="proven-undelivered transport failure after retries",
        delivered_ambiguous=False,
        http_status=None,
        transport_error="ConnectError",
    )

    posted: list[dict] = []

    async def _ok_inner(client, session_id, *, event_type, data, max_attempts, timeout):
        posted.append(
            {
                "session_id": session_id,
                "event_type": event_type,
                "data": data,
                "max_attempts": max_attempts,
                "timeout": timeout,
            }
        )
        return fwd._PostResult(
            response=httpx.Response(200, request=httpx.Request("POST", "http://test"))
        )

    monkeypatch.setattr(fwd, "_post_session_event_inner", _ok_inner)
    await fwd._replay_dead_letters_on_startup(MagicMock(), tmp_path)

    assert len(posted) == 1
    assert posted[0]["session_id"] == "conv_codex1"
    assert posted[0]["event_type"] == "external_conversation_item"
    assert posted[0]["data"] == {"item_type": "message"}
    # Replay re-POSTs with a single attempt and a short timeout so a large file
    # or a hung server cannot stall startup.
    assert posted[0]["max_attempts"] == 1
    assert posted[0]["timeout"] == fwd._REPLAY_POST_TIMEOUT_SECONDS
    # Delivered → record removed.
    assert not (tmp_path / "dead_letter.jsonl").exists()


@pytest.mark.asyncio
async def test_replay_dead_letters_on_startup_skips_ambiguous(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """
    On startup, an ambiguous record is never re-POSTed and is retained (#1579).

    :param tmp_path: Pytest temp dir standing in for the bridge dir.
    :param monkeypatch: Pytest patcher (auto-restores the stubbed inner).
    """
    fwd.append_dead_letter(
        tmp_path,
        session_id="conv_codex1",
        event_type="external_conversation_item",
        payload={"item_type": "message"},
        reason="ambiguous transport failure (may already be committed)",
        delivered_ambiguous=True,
    )

    called = False

    async def _inner(client, session_id, *, event_type, data, **_kwargs):
        nonlocal called
        called = True
        return fwd._PostResult(
            response=httpx.Response(200, request=httpx.Request("POST", "http://test"))
        )

    monkeypatch.setattr(fwd, "_post_session_event_inner", _inner)
    await fwd._replay_dead_letters_on_startup(MagicMock(), tmp_path)

    assert called is False
    # Ambiguous record retained as a forensic record.
    assert (tmp_path / "dead_letter.jsonl").exists()


class _RecordingPostClient:
    """Async client stub that records each ``post`` call's kwargs."""

    def __init__(self, response: httpx.Response) -> None:
        self._response = response
        self.calls: list[dict] = []

    async def post(self, url: str, **kwargs: object) -> httpx.Response:
        self.calls.append(kwargs)
        return self._response


@pytest.mark.asyncio
async def test_post_session_event_inner_single_attempt_and_timeout() -> None:
    """
    ``max_attempts=1`` makes one POST (no retry) and ``timeout`` is threaded through.

    Replay relies on both so a hung server fails fast and startup is bounded (#1579).
    """
    client = _RecordingPostClient(
        httpx.Response(503, request=httpx.Request("POST", "http://test"))
    )
    result = await fwd._post_session_event_inner(
        client,
        "conv_codex1",
        event_type="external_conversation_item",
        data={"item_type": "message"},
        max_attempts=1,
        timeout=5.0,
    )
    # A single attempt even though 503 is normally retryable.
    assert len(client.calls) == 1
    assert client.calls[0]["timeout"] == 5.0
    assert result.response is not None
    assert result.response.status_code == 503


async def test_post_session_event_records_connectivity_failure_for_watchdog(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Codex's exhausted-retry POST failure is recorded for the idle watchdog.

    Writer half of issue #1119 for the codex forwarder: when every attempt to
    POST a session event raises a connect error, ``_post_session_event`` must
    record the failure in ``_native_forwarder_health`` (via
    ``_log_post_transport_failure``) so the harness idle-turn watchdog can name
    the connectivity cause instead of a generic "wedged LLM" reason.
    """
    from omnigent import _native_forwarder_health as health

    class _AlwaysConnectError:
        """Stub client whose every POST fails to connect."""

        async def post(self, url: str, *, json: object) -> httpx.Response:
            """Raise a connect error mimicking an unreachable server."""
            del json
            raise httpx.ConnectError("No route to host", request=httpx.Request("POST", url))

    async def _no_sleep(_seconds: float) -> None:
        """No-op sleep so the retry loop doesn't add real delay."""

    monkeypatch.setattr(fwd, "_sleep", _no_sleep)

    health.clear()
    try:
        result = await fwd._post_session_event(
            _AlwaysConnectError(),  # type: ignore[arg-type]
            "conv_x",
            event_type="external_session_status",
            data={},
        )
        assert result is None
        detail = health.recent_post_failure(60.0)
        assert detail is not None
        assert "external_session_status" in detail
        assert "No route to host" in detail
    finally:
        health.clear()


# ── MCP startup status mirroring (issue #2058) ─────────────────────────


def _mcp_startup_event(
    name: str,
    status: str,
    *,
    error: str | None = None,
    thread_id: str | None = None,
) -> dict:
    """
    Build a Codex ``mcpServer/startupStatus/updated`` envelope.

    :param name: MCP server name, e.g. ``"safe"``.
    :param status: Startup state, e.g. ``"starting"``.
    :param error: Optional failure detail.
    :param thread_id: Optional ``threadId`` param.
    :returns: Notification envelope dict.
    """
    params: dict = {"name": name, "status": status}
    if error is not None:
        params["error"] = error
    if thread_id is not None:
        params["threadId"] = thread_id
    return {"method": "mcpServer/startupStatus/updated", "params": params}


@pytest.mark.asyncio
async def test_mcp_startup_event_records_and_posts(tmp_path: Path) -> None:
    """
    An MCP startup notification updates the bridge map and mirrors it to AP.

    The bridge write is what unblocks the executor's first-turn gate; the
    ``external_mcp_startup`` post is what makes the web session show
    per-server startup state instead of appearing hung.
    """
    client = _RecordingClient()

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event=_mcp_startup_event("safe", "starting", thread_id="thread_1"),
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    from omnigent.codex_native_bridge import read_mcp_startup

    assert read_mcp_startup(tmp_path) == {"safe": {"status": "starting", "error": None}}
    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_mcp_startup",
                "data": {"servers": {"safe": {"status": "starting", "error": None}}},
            },
        )
    ]


@pytest.mark.asyncio
async def test_mcp_startup_event_carries_failure_error(tmp_path: Path) -> None:
    """
    A ``failed`` update retains the error detail through bridge and post.

    The error text is what the web UI (and the executor's timeout text)
    surface for a server that never came up.
    """
    client = _RecordingClient()

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event=_mcp_startup_event("safe", "failed", error="handshake failed"),
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    from omnigent.codex_native_bridge import read_mcp_startup

    assert read_mcp_startup(tmp_path) == {
        "safe": {"status": "failed", "error": "handshake failed"}
    }
    assert client.posts[0][1]["data"]["servers"]["safe"]["error"] == "handshake failed"


@pytest.mark.asyncio
async def test_mcp_startup_event_for_other_thread_is_ignored(tmp_path: Path) -> None:
    """
    An MCP startup update naming a different thread is dropped.

    A child thread's (or stale thread's) startup must not overwrite the
    parent bridge map or flash the parent session's startup band.
    """
    client = _RecordingClient()

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event=_mcp_startup_event("safe", "starting", thread_id="thread_other"),
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    from omnigent.codex_native_bridge import read_mcp_startup

    assert read_mcp_startup(tmp_path) == {}
    assert client.posts == []


@pytest.mark.asyncio
async def test_mcp_startup_event_with_unknown_status_is_ignored(tmp_path: Path) -> None:
    """
    An unknown status value neither records nor posts.

    Guards against a future Codex enum change feeding a bogus status into
    the executor gate or the web UI.
    """
    client = _RecordingClient()

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event=_mcp_startup_event("safe", "exploded"),
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    from omnigent.codex_native_bridge import read_mcp_startup

    assert read_mcp_startup(tmp_path) == {}
    assert client.posts == []


def _write_session_config(tmp_path: Path, body: str) -> None:
    """
    Write the session's private ``config.toml`` under the bridge dir.

    :param tmp_path: Bridge directory.
    :param body: Raw TOML body, e.g. ``"[mcp_servers.safe]\ncommand='x'"``.
    """
    home = codex_home_for_bridge_dir(tmp_path)
    home.mkdir(parents=True, exist_ok=True)
    (home / "config.toml").write_text(body)


@pytest.mark.asyncio
async def test_seed_mcp_startup_round_posts_config_servers(tmp_path: Path) -> None:
    """
    Seeding records the enabled config servers as ``starting`` and posts them.

    Codex delivers per-server startup edges only to the thread-owning
    connection, so this synthesized round is the only way the web session
    learns startup is in flight (issue #2058). Disabled servers must be
    excluded — codex does not boot them, and a permanently-"starting"
    band entry would never resolve.
    """
    from omnigent.codex_native_bridge import read_mcp_startup

    client = _RecordingClient()
    _write_session_config(
        tmp_path,
        '[mcp_servers.safe]\ncommand = "x"\n'
        '[mcp_servers.off]\ncommand = "y"\nenabled = false\n'
        '[mcp_servers.omnigent]\ncommand = "z"\n',
    )

    timer = await fwd._seed_mcp_startup_round(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
    )

    try:
        assert read_mcp_startup(tmp_path) == {
            "safe": {"status": "starting", "error": None},
            "omnigent": {"status": "starting", "error": None},
        }
        assert client.posts == [
            (
                "/v1/sessions/conv_x/events",
                {
                    "type": "external_mcp_startup",
                    "data": {
                        "servers": {
                            "safe": {"status": "starting", "error": None},
                            "omnigent": {"status": "starting", "error": None},
                        }
                    },
                },
            )
        ]
        assert timer is not None
    finally:
        if timer is not None:
            timer.cancel()


@pytest.mark.asyncio
async def test_seed_mcp_startup_round_skips_when_state_exists(tmp_path: Path) -> None:
    """
    A bridge dir that already carries round state is not reseeded.

    ``supervise_forwarder`` restarts on reconnect mid-session; reseeding
    then would flash a false "starting" band for servers that finished
    booting long ago. Only ``clear_bridge_state`` (each app-server
    launch) resets the map.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    _write_session_config(tmp_path, '[mcp_servers.safe]\ncommand = "x"\n')
    update_mcp_server_startup(tmp_path, "safe", "cancelled")

    timer = await fwd._seed_mcp_startup_round(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
    )

    assert timer is None
    assert client.posts == []
    assert read_mcp_startup(tmp_path) == {"safe": {"status": "cancelled", "error": None}}


@pytest.mark.asyncio
async def test_seed_rearms_settle_timer_when_round_still_pending(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """
    A reconnect that finds the round still pending re-arms the settle window.

    The previous forwarder's settle timer dies with its connection; if the
    reconnect only skipped reseeding, a missed idle edge would leave the
    band stuck on "starting" for the rest of the session. The re-armed
    timer must actually resolve the round: once it fires, the pending
    entries are dropped and the settled map is posted.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    update_mcp_server_startup(tmp_path, "safe", "starting")
    update_mcp_server_startup(tmp_path, "storage-console", "cancelled")

    async def _no_sleep(_seconds: float) -> None:
        """Collapse the settle window so the test observes the timer firing."""

    monkeypatch.setattr(fwd, "_sleep", _no_sleep)

    timer = await fwd._seed_mcp_startup_round(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
    )

    # No reseed/repost on reconnect — the recorded map is left intact...
    assert timer is not None
    assert client.posts == []
    assert read_mcp_startup(tmp_path)["safe"]["status"] == "starting"

    # ...but the re-armed timer settles the round when the window elapses.
    await timer
    assert read_mcp_startup(tmp_path) == {
        "storage-console": {"status": "cancelled", "error": None}
    }
    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_mcp_startup",
                "data": {"servers": {"storage-console": {"status": "cancelled", "error": None}}},
            },
        )
    ]


@pytest.mark.asyncio
async def test_thread_idle_settles_synthesized_round(tmp_path: Path) -> None:
    """
    An idle ``thread/status/changed`` resolves the synthesized round.

    Codex defers turn execution until MCP startup settles, so a thread
    going idle after a turn proves the round ended. Unresolved
    ``starting`` entries are dropped (their real outcomes are only
    delivered to the thread owner) while locally-recorded ``cancelled``
    states survive, and the settled map is posted so the band clears.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    update_mcp_server_startup(tmp_path, "safe", "starting")
    update_mcp_server_startup(tmp_path, "storage-console", "cancelled")

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event={
            "method": "thread/status/changed",
            "params": {"threadId": "thread_1", "status": {"type": "idle"}},
        },
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    assert read_mcp_startup(tmp_path) == {
        "storage-console": {"status": "cancelled", "error": None}
    }
    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {
                "type": "external_mcp_startup",
                "data": {"servers": {"storage-console": {"status": "cancelled", "error": None}}},
            },
        )
    ]


@pytest.mark.asyncio
async def test_thread_active_status_does_not_settle_round(tmp_path: Path) -> None:
    """
    An ``active`` status edge must not settle the round.

    Codex flips the thread active the moment a turn is ACCEPTED — which
    happens mid-startup, before the round ends — so settling there would
    clear the band exactly when it matters most.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    update_mcp_server_startup(tmp_path, "safe", "starting")

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event={
            "method": "thread/status/changed",
            "params": {"threadId": "thread_1", "status": {"type": "active", "activeFlags": []}},
        },
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    assert read_mcp_startup(tmp_path) == {"safe": {"status": "starting", "error": None}}
    assert client.posts == []


@pytest.mark.asyncio
async def test_model_output_item_settles_synthesized_round(tmp_path: Path) -> None:
    """
    The first model-produced item resolves the synthesized round mid-turn.

    Codex defers turn execution until MCP startup settles, so an
    assistant-side item proves the round is over while the turn is still
    running. Without this signal, a server stuck ``starting`` (e.g. a
    misconfigured command that never handshakes) pins the "Starting MCP
    servers" band under a visibly working agent until the turn ends or
    the config-derived window elapses.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    update_mcp_server_startup(tmp_path, "safe", "starting")

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event={
            "method": "item/started",
            "params": {
                "threadId": "thread_1",
                "item": {"id": "item_1", "type": "agentMessage", "text": ""},
            },
        },
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    assert read_mcp_startup(tmp_path) == {}
    assert client.posts == [
        (
            "/v1/sessions/conv_x/events",
            {"type": "external_mcp_startup", "data": {"servers": {}}},
        )
    ]


@pytest.mark.asyncio
async def test_model_output_settles_the_round_only_once(tmp_path: Path) -> None:
    """
    Only the first model item pays the settle path; later items skip it.

    The round is seeded once per forwarder connection (never on thread
    rotation), so once model output settles it the outcome cannot change
    — every later item in the session would otherwise re-read the bridge
    file for nothing. The state flag short-circuits them, which this
    pins by re-populating the map behind the flag: a second item must
    leave it untouched.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    state = fwd._CodexForwarderState()
    update_mcp_server_startup(tmp_path, "safe", "starting")

    item_event = {
        "method": "item/completed",
        "params": {
            "threadId": "thread_1",
            "item": {"id": "item_1", "type": "agentMessage", "text": "hi"},
        },
    }

    async def deliver() -> None:
        """Feed the model-output item through the forwarder dispatch."""
        await fwd._handle_event(
            client,  # type: ignore[arg-type]
            session_id="conv_x",
            bridge_dir=tmp_path,
            event=item_event,
            usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
            elicitation_tracker=fwd._CodexElicitationTaskTracker(),
            expected_thread_id="thread_1",
            forwarder_state=state,
        )

    await deliver()
    assert state.mcp_startup_settled is True
    settle_posts = [p for p in client.posts if p[1]["type"] == "external_mcp_startup"]
    assert len(settle_posts) == 1

    # Re-populate the map behind the flag: without the short-circuit the
    # next item would read it, settle again, and post a second time.
    update_mcp_server_startup(tmp_path, "safe", "starting")
    await deliver()

    assert read_mcp_startup(tmp_path) == {"safe": {"status": "starting", "error": None}}
    assert len([p for p in client.posts if p[1]["type"] == "external_mcp_startup"]) == 1


@pytest.mark.asyncio
async def test_user_message_item_does_not_settle_round(tmp_path: Path) -> None:
    """
    The turn's ``userMessage`` item must not settle the round.

    Like the thread-active edge, the user message materializes when a
    turn is merely ACCEPTED — which happens mid-startup — so settling on
    it would clear the band during the genuine pre-turn wait.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    update_mcp_server_startup(tmp_path, "safe", "starting")

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event={
            "method": "item/started",
            "params": {
                "threadId": "thread_1",
                "item": {"id": "item_0", "type": "userMessage", "text": "hi"},
            },
        },
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    assert read_mcp_startup(tmp_path) == {"safe": {"status": "starting", "error": None}}
    assert client.posts == []


@pytest.mark.asyncio
async def test_other_thread_item_does_not_settle_round(tmp_path: Path) -> None:
    """
    An item on an unrecognized thread must not settle the parent round.

    MCP startup is bridge-level state surfaced on the parent session;
    another thread's activity proves nothing about this round.
    """
    from omnigent.codex_native_bridge import read_mcp_startup, update_mcp_server_startup

    client = _RecordingClient()
    update_mcp_server_startup(tmp_path, "safe", "starting")

    await fwd._handle_event(
        client,  # type: ignore[arg-type]
        session_id="conv_x",
        bridge_dir=tmp_path,
        event={
            "method": "item/started",
            "params": {
                "threadId": "thread_other",
                "item": {"id": "item_1", "type": "agentMessage", "text": ""},
            },
        },
        usage_coalescer=fwd._SessionUsageCoalescer(client, "conv_x"),  # type: ignore[arg-type]
        elicitation_tracker=fwd._CodexElicitationTaskTracker(),
        expected_thread_id="thread_1",
    )

    assert read_mcp_startup(tmp_path) == {"safe": {"status": "starting", "error": None}}
    assert client.posts == []


def test_settle_timeout_tracks_slowest_configured_server(tmp_path: Path) -> None:
    """
    The settle window follows the slowest ``startup_timeout_sec`` in config.

    Codex bounds each server's spawn+handshake by that budget, so the
    round cannot outlive it; a config without budgets falls back to the
    codex default, and an absurd budget is capped.
    """
    _write_session_config(
        tmp_path,
        '[mcp_servers.fast]\ncommand = "x"\n'
        '[mcp_servers.slow]\ncommand = "y"\nstartup_timeout_sec = 120\n',
    )
    assert fwd._mcp_startup_settle_timeout_seconds(tmp_path) == 135.0

    _write_session_config(tmp_path, '[mcp_servers.fast]\ncommand = "x"\n')
    assert fwd._mcp_startup_settle_timeout_seconds(tmp_path) == 25.0

    _write_session_config(
        tmp_path,
        '[mcp_servers.slow]\ncommand = "y"\nstartup_timeout_sec = 100000\n',
    )
    assert fwd._mcp_startup_settle_timeout_seconds(tmp_path) == 240.0

    # A disabled server's budget must not stretch the window: codex never
    # boots it, and the seed excludes it from the synthesized round.
    _write_session_config(
        tmp_path,
        '[mcp_servers.fast]\ncommand = "x"\n'
        '[mcp_servers.off]\ncommand = "y"\nenabled = false\nstartup_timeout_sec = 120\n',
    )
    assert fwd._mcp_startup_settle_timeout_seconds(tmp_path) == 25.0


def _coalescer(client: object) -> fwd._OutputTextDeltaCoalescer:
    """Build a delta coalescer wired to ``client``."""
    return fwd._OutputTextDeltaCoalescer(
        client=client,
        session_id="conv_idle",
        flush_interval_seconds=0.05,
        flush_char_threshold=64,
    )


@pytest.mark.asyncio
async def test_delta_coalescer_close_returns_when_its_worker_is_cancelled() -> None:
    """``close()`` must not park on a marker its cancelled worker can never resolve.

    ``asyncio.run`` cancels every task before resuming any, so when the forwarder
    resumes first its ``finally`` calls ``close()`` while the worker is cancelled but
    not yet ``done()``. The marker is queued with nobody left to complete it.
    """
    coalescer = _coalescer(_RecordingClient())
    coalescer._ensure_worker()
    await asyncio.sleep(0)
    worker = coalescer._worker_task
    assert worker is not None

    worker.cancel()
    # Deliberately NOT awaited: this reproduces the teardown ordering where the
    # worker is cancelled but has not run yet, which no `done()` check can catch.
    assert not worker.done()

    loop = asyncio.get_running_loop()
    started = loop.time()
    await asyncio.wait_for(coalescer.close(), timeout=fwd._DELTA_MARKER_TIMEOUT_SECONDS + 5.0)

    assert coalescer._worker_task is None
    assert loop.time() - started < fwd._DELTA_MARKER_TIMEOUT_SECONDS


@pytest.mark.asyncio
async def test_delta_coalescer_close_gives_up_on_a_wedged_worker(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A worker still alive but stuck mid-post must not hold ``close()`` forever.

    This is the one case the bound exists for: the worker owns the marker, so it is
    off the queue, and the worker is running, so racing it does not help either.
    """

    class _HangingClient:
        """Client whose post never returns, wedging the worker inside a flush."""

        def __init__(self) -> None:
            self.entered = asyncio.Event()

        async def post(self, *args: object, **kwargs: object) -> None:
            self.entered.set()
            await asyncio.sleep(3600)

    monkeypatch.setattr(fwd, "_DELTA_MARKER_TIMEOUT_SECONDS", 0.1)
    client = _HangingClient()
    coalescer = _coalescer(client)
    coalescer._ensure_worker()
    coalescer._queue.put_nowait(fwd._DeltaChunk(message_id="m1", tool_call_id=None, delta="x"))
    await asyncio.wait_for(client.entered.wait(), timeout=5.0)
    worker = coalescer._worker_task
    assert worker is not None

    await asyncio.wait_for(coalescer.close(), timeout=fwd._DELTA_MARKER_TIMEOUT_SECONDS + 5.0)

    # close() gave up on the bound rather than waiting out a worker still stuck in its post.
    assert coalescer._worker_task is None
    assert not worker.done()
    worker.cancel()


@pytest.mark.asyncio
async def test_delta_coalescer_worker_survives_an_already_settled_marker() -> None:
    """Resolving a marker whose future is already settled must not kill the worker.

    ``set_result`` on a settled future raises ``InvalidStateError``, and
    ``_ensure_worker`` only replaces a ``None`` task, so a worker lost this way is
    never restarted and every later delta is dropped without a word.
    """
    client = _RecordingClient()
    coalescer = _coalescer(client)
    coalescer._ensure_worker()
    await asyncio.sleep(0)

    settled: asyncio.Future[None] = asyncio.get_running_loop().create_future()
    settled.cancel()
    coalescer._queue.put_nowait(fwd._DeltaFlushBarrier(done=settled))
    await asyncio.sleep(0.1)

    assert coalescer._worker_task is not None
    assert not coalescer._worker_task.done()

    posts_before = len(client.posts)
    coalescer._queue.put_nowait(
        fwd._DeltaChunk(message_id="m1", tool_call_id=None, delta="still here")
    )
    await asyncio.wait_for(coalescer.flush(), timeout=5.0)
    assert len(client.posts) > posts_before


@pytest.mark.asyncio
async def test_delta_coalescer_survives_a_cancelled_flush_caller() -> None:
    """A cancelled ``flush()`` caller must not kill the worker.

    The caller's cancellation settles its own future. Resolving it again raises
    ``InvalidStateError`` inside the worker, and ``_ensure_worker`` only replaces a
    ``None`` task, so the dead worker was never restarted and every later delta was
    silently dropped.
    """
    client = _RecordingClient()
    coalescer = _coalescer(client)
    coalescer._ensure_worker()
    await asyncio.sleep(0)

    pending = asyncio.ensure_future(coalescer.flush())
    await asyncio.sleep(0)
    pending.cancel()
    await asyncio.gather(pending, return_exceptions=True)
    await asyncio.sleep(0.1)

    assert coalescer._worker_task is not None
    assert not coalescer._worker_task.done()

    posts_before = len(client.posts)
    coalescer._queue.put_nowait(
        fwd._DeltaChunk(message_id="m1", tool_call_id=None, delta="still here")
    )
    await asyncio.wait_for(coalescer.flush(), timeout=5.0)
    assert len(client.posts) > posts_before