hermes-webui/tests/test_issue7565_codex_commentary_media.py

"""Regression coverage for issue #7565: session MEDIA authorization
and snapshot capture must also inspect typed public assistant
commentary carried in ``codex_message_items`` (Agent phase:
"commentary"). The Agent producer (``agent/codex_responses_adapter.py``)
persists such items with ``role: assistant`` and a normalized phase;
the WebUI's auth predicate (``api/routes.py::
_session_media_token_allows_path``) and snapshot annotator
(``api/media_snapshots.py::annotate_media_snapshots``) previously
only scanned top-level assistant ``content``, so a safe image
emitted only inside commentary returned 403 and was never
snapshotted at settle time.
"""
from __future__ import annotations

from api.media_snapshots import codex_commentary_text


# ── unit tests for the shared helper ─────────────────────────────────────────


def _commentary_row(text="Synthetic preview is ready.\n\nMEDIA:/opt/artifacts/preview.png",
                    phase="commentary",
                    item_type="message",
                    item_role="assistant",
                    outer_role="assistant"):
    return {
        "role": outer_role,
        "content": "",
        "codex_message_items": [
            {
                "type": item_type,
                "role": item_role,
                "phase": phase,
                "content": [
                    {"type": "output_text", "text": text},
                ],
            }
        ],
    }


def test_commentary_text_returns_text_for_valid_shape():
    """The valid-shape case from the issue body — outer role
    assistant, item type message, item role assistant, phase
    commentary, list of output_text parts with a string text —
    must surface the embedded text so the auth predicate and
    snapshot scan can both find the MEDIA: token."""
    row = _commentary_row(text="MEDIA:/opt/artifacts/preview.png")
    text = codex_commentary_text(row)
    assert "MEDIA:/opt/artifacts/preview.png" in text, (
        "valid commentary shape must surface the textual MEDIA: "
        "token so the auth predicate can resolve it"
    )


def test_commentary_text_rejects_user_outer_role():
    """A user-row sidecar must not mint grants even when the
    sidecar contains a commentary-shaped item, preserving the
    implicit threat model that user-authored content cannot
    mint allow-list entries."""
    row = _commentary_row(outer_role="user", text="MEDIA:/etc/passwd")
    assert codex_commentary_text(row) == "", (
        "user outer role must short-circuit before the sidecar is read"
    )


def test_commentary_text_rejects_wrong_item_type():
    """Item type must be exactly ``message``. Reasoning, function
    call, and other non-message items must not be inspected even
    when their content is shaped like commentary."""
    row = _commentary_row(item_type="reasoning", text="MEDIA:/opt/x.png")
    assert codex_commentary_text(row) == ""


def test_commentary_text_rejects_wrong_item_role():
    """Item role must be exactly ``assistant``. User-sidecar
    assistant commentary (an item with role: user nested inside
    an outer assistant row) must not mint grants."""
    row = _commentary_row(item_role="user", text="MEDIA:/opt/x.png")
    assert codex_commentary_text(row) == ""


def test_commentary_text_rejects_wrong_phase():
    """Phase must normalize to exactly ``commentary``. Analysis,
    final, and arbitrary phases must not be inspected, even when
    the row carries a textual output_text part with a MEDIA:
    token. ``"Commentary"`` with a capital C is accepted because
    the helper lowercases the phase; the others are rejected."""
    for wrong_phase in ("analysis", "final", "tool_call", ""):
        row = _commentary_row(phase=wrong_phase, text="MEDIA:/opt/x.png")
        assert codex_commentary_text(row) == "", (
            f"phase={wrong_phase!r} must not be inspected"
        )


def test_commentary_text_accepts_case_insensitive_commentary():
    """Phase matching is case-insensitive; the Agent producer
    stamps a normalized phase, but the helper must still accept
    variants so a future producer change cannot silently break
    the auth predicate."""
    row = _commentary_row(phase="Commentary", text="MEDIA:/opt/x.png")
    assert codex_commentary_text(row) != ""


def test_commentary_text_rejects_non_output_text_parts():
    """Only textual ``output_text`` parts are read. Tool-call
    parts, image parts, file parts, and arbitrary typed parts
    must not be inspected even if they happen to carry a
    MEDIA-shaped string."""
    row = {
        "role": "assistant",
        "content": "",
        "codex_message_items": [
            {
                "type": "message",
                "role": "assistant",
                "phase": "commentary",
                "content": [
                    {"type": "tool_call", "name": "x", "text": "MEDIA:/opt/x.png"},
                    {"type": "image", "text": "MEDIA:/opt/x.png"},
                    {"type": "input_text", "text": "MEDIA:/opt/x.png"},
                ],
            }
        ],
    }
    assert codex_commentary_text(row) == "", (
        "non-output_text parts must not be inspected even when they "
        "carry a MEDIA-shaped string"
    )


def test_commentary_text_rejects_non_text_output_text_part():
    """``output_text`` parts with a non-string ``text`` value must
    be skipped, not coerced. This pins the fail-closed contract:
    the helper will not stringify a dict, list, or number just
    because the part type is correct."""
    row = {
        "role": "assistant",
        "content": "",
        "codex_message_items": [
            {
                "type": "message",
                "role": "assistant",
                "phase": "commentary",
                "content": [
                    {"type": "output_text", "text": {"nested": "MEDIA:/opt/x.png"}},
                    {"type": "output_text", "text": ["MEDIA:/opt/x.png"]},
                    {"type": "output_text", "text": 42},
                ],
            }
        ],
    }
    assert codex_commentary_text(row) == ""


def test_commentary_text_handles_malformed_messages():
    """The helper must never raise on malformed input. Empty,
    None, list, string, and arbitrary objects must all return
    the safe empty default."""
    assert codex_commentary_text(None) == ""
    assert codex_commentary_text("") == ""
    assert codex_commentary_text([]) == ""
    assert codex_commentary_text({}) == ""
    assert codex_commentary_text({"role": "assistant"}) == ""  # no codex_message_items
    assert codex_commentary_text({"role": "assistant", "codex_message_items": "not a list"}) == ""
    # Malformed item shape: not a dict.
    assert codex_commentary_text({
        "role": "assistant",
        "codex_message_items": [None, "string", 42, ["list"]],
    }) == ""


# ── auth-predicate integration via the public surface ───────────────────────


def test_session_media_token_allows_path_imports_commentary_helper():
    """The authorization predicate must import and use the
    commentary helper so a safe image emitted only inside
    commentary gets the same exact-path grant as a top-level
    one. Without the import, the auth surface silently regresses
    to the top-level-only shape that #7565 reports."""
    src = open("api/routes.py", encoding="utf-8").read()
    assert "from api.media_snapshots import codex_commentary_text" in src
    # The helper is actually consumed (not just imported) inside
    # the predicate body.
    auth_idx = src.find("def _session_media_token_allows_path")
    assert auth_idx != -1
    end_idx = src.find("\n\ndef ", auth_idx + 1)
    if end_idx == -1:
        end_idx = len(src)
    body = src[auth_idx:end_idx]
    assert "codex_commentary_text(message)" in body, (
        "the auth predicate must consult the commentary helper so "
        "tokens emitted only inside commentary are honored"
    )


def test_annotate_media_snapshots_uses_commentary_helper():
    """The snapshot annotator must also use the commentary helper
    so commentary-emitted artifacts are snapshotted at settle time
    rather than only the first time the live file is read."""
    src = open("api/media_snapshots.py", encoding="utf-8").read()
    # The helper is defined at module scope.
    assert "def codex_commentary_text(" in src
    # The annotator consumes it.
    annotate_idx = src.find("def annotate_media_snapshots(")
    assert annotate_idx != -1
    end_idx = src.find("\n\ndef ", annotate_idx + 1)
    if end_idx == -1:
        end_idx = len(src)
    body = src[annotate_idx:end_idx]
    assert "codex_commentary_text(msg)" in body, (
        "the snapshot annotator must consult the commentary helper "
        "so commentary-emitted artifacts are captured at settle time"
    )