roboco/tests/integration/test_a2a_service.py

"""A2AService coverage — agent cards, task ↔ A2A conversion, conversations."""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, cast
from unittest.mock import AsyncMock, patch
from unittest.mock import MagicMock as _MM
from uuid import UUID, uuid4
from uuid import uuid4 as _u

import pytest
import pytest_asyncio
from roboco.agents_config import A2A_ALLOWED_PAIRS
from roboco.db.tables import (
    A2AConversationTable,
    A2AMessageTable,
    AgentTable,
    NotificationTable,
    ProjectTable,
    TaskTable,
)
from roboco.enforcement.a2a_access import A2AAccessDeniedError
from roboco.models import (
    AgentRole,
    AgentStatus,
    NotificationPriority,
    NotificationType,
    Team,
)
from roboco.models.a2a import (
    A2AConversationStatus,
    A2AMessage,
    SendMessageRequest,
    TextPart,
)
from roboco.models.base import (
    TaskNature,
    TaskStatus,
    TaskType,
)
from roboco.models.events import EventType
from roboco.services.a2a import _LIVE_VIEW_EXCERPT_CHARS, A2AService
from roboco.services.gateway.evidence_repo import EvidenceRepo
from sqlalchemy import select, text
from sqlalchemy import select as _sel
from sqlalchemy.sql.dml import Update

if TYPE_CHECKING:
    from collections.abc import AsyncIterator

    from sqlalchemy.ext.asyncio import AsyncSession


@pytest_asyncio.fixture
async def a2a_setup(
    db_session: AsyncSession,
) -> AsyncIterator[dict]:
    # Use the canonical seed slugs so the A2A policy matrix recognizes
    # role + team and lets same-cell pairs talk. Random suffixes would
    # leave them with role="unknown" → policy denies everything.
    dev = AgentTable(
        id=uuid4(),
        name="Dev",
        slug="be-dev-1",
        role=AgentRole.DEVELOPER,
        team=Team.BACKEND,
        status=AgentStatus.ACTIVE,
        model_config={},
        system_prompt="dev",
        capabilities=[],
        permissions={},
        metrics={},
    )
    qa = AgentTable(
        id=uuid4(),
        name="QA",
        slug="be-qa",
        role=AgentRole.QA,
        team=Team.BACKEND,
        status=AgentStatus.ACTIVE,
        model_config={},
        system_prompt="qa",
        capabilities=[],
        permissions={},
        metrics={},
    )
    db_session.add_all([dev, qa])
    await db_session.flush()
    project = ProjectTable(
        id=uuid4(),
        name="A-Proj",
        slug=f"a-proj-{uuid4().hex[:8]}",
        git_url="https://example.com/r.git",
        assigned_cell=Team.BACKEND,
        created_by=dev.id,
    )
    db_session.add(project)
    await db_session.flush()
    task = TaskTable(
        id=uuid4(),
        title="t",
        description="d",
        acceptance_criteria=["ac"],
        status=TaskStatus.PENDING,
        priority=2,
        task_type=TaskType.CODE,
        nature=TaskNature.TECHNICAL,
        project_id=project.id,
        created_by=dev.id,
        team=Team.BACKEND,
    )
    db_session.add(task)
    await db_session.flush()
    yield {
        "svc": A2AService(db_session),
        "dev": dev,
        "qa": qa,
        "task_id": task.id,
        "db": db_session,
    }


# ---------------------------------------------------------------------------
# Agent cards
# ---------------------------------------------------------------------------


def test_get_service_endpoint_returns_url() -> None:
    url = A2AService.get_service_endpoint()
    assert url.startswith("http://")


def test_build_system_agent_card() -> None:
    card = A2AService.build_system_agent_card()
    assert card.id == "roboco-system"
    assert len(card.skills) >= 1


@pytest.mark.asyncio
async def test_build_agent_card_by_uuid(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    card = await svc.build_agent_card(str(dev.id))
    assert card is not None
    assert card.name == dev.name


@pytest.mark.asyncio
async def test_build_agent_card_by_slug(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    card = await svc.build_agent_card(dev.slug)
    assert card is not None
    assert card.id == str(dev.id)


@pytest.mark.asyncio
async def test_build_agent_card_unknown_returns_none(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    assert await svc.build_agent_card(str(uuid4())) is None
    assert await svc.build_agent_card("ghost-slug") is None


# ---------------------------------------------------------------------------
# Task ↔ A2A
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_task_by_uuid(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    a2a = await svc.get_task(str(a2a_setup["task_id"]))
    assert a2a is not None
    assert a2a.id == str(a2a_setup["task_id"])


@pytest.mark.asyncio
async def test_get_task_returns_none_for_invalid_id(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    assert await svc.get_task("not-a-uuid") is None
    assert await svc.get_task(str(uuid4())) is None


@pytest.mark.asyncio
async def test_list_tasks(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    tasks, has_more = await svc.list_tasks(page_size=20)
    assert any(t.id == str(a2a_setup["task_id"]) for t in tasks)
    assert isinstance(has_more, bool)


@pytest.mark.asyncio
async def test_list_tasks_ascending(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    tasks, _ = await svc.list_tasks(order_by="created_at asc")
    assert isinstance(tasks, list)


@pytest.mark.asyncio
async def test_cancel_task_invalid_id(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    with pytest.raises(ValueError, match="Invalid task ID"):
        await svc.cancel_task("not-a-uuid")


@pytest.mark.asyncio
async def test_cancel_task_not_found(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    with pytest.raises(ValueError, match="Task not found"):
        await svc.cancel_task(str(uuid4()))


@pytest.mark.asyncio
async def test_cancel_task_already_terminal(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    completed = TaskTable(
        id=uuid4(),
        title="done",
        description="d",
        acceptance_criteria=["ac"],
        status=TaskStatus.COMPLETED,
        priority=2,
        task_type=TaskType.CODE,
        nature=TaskNature.TECHNICAL,
        project_id=uuid4(),
        created_by=a2a_setup["dev"].id,
        team=Team.BACKEND,
    )
    # FK on project — use existing project
    completed.project_id = (await db.execute(select(ProjectTable))).scalars().first().id
    db.add(completed)
    await db.flush()
    with pytest.raises(ValueError, match="terminal state"):
        await svc.cancel_task(str(completed.id))


# ---------------------------------------------------------------------------
# Discovery
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_discover_agents_no_filters(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    cards = await svc.discover_agents()
    # Setup seeds dev + qa, so at least 2 cards.
    _MIN_SEEDED = 2
    assert len(cards) >= _MIN_SEEDED


@pytest.mark.asyncio
async def test_discover_agents_by_role(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    cards = await svc.discover_agents(role="developer")
    assert all(c.metadata.get("role") == "developer" for c in cards)


@pytest.mark.asyncio
async def test_discover_agents_by_team(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    cards = await svc.discover_agents(team="backend")
    assert all(c.metadata.get("team") == "backend" for c in cards)


@pytest.mark.asyncio
async def test_discover_agents_by_skill_tag(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    cards = await svc.discover_agents(skill_tag="qa")
    # All returned cards have at least one skill tagged 'qa'.
    for card in cards:
        assert any("qa" in skill.tags for skill in card.skills)


# ---------------------------------------------------------------------------
# Canonical pair helper
# ---------------------------------------------------------------------------


def test_canonical_pair_orders_lexically() -> None:
    a, b = A2AService._canonical_pair("z-agent", "a-agent")
    assert (a, b) == ("a-agent", "z-agent")
    a, b = A2AService._canonical_pair("a-agent", "z-agent")
    assert (a, b) == ("a-agent", "z-agent")


# ---------------------------------------------------------------------------
# Conversations
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_or_create_conversation_self_a2a_denied(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    with pytest.raises(A2AAccessDeniedError):
        await svc.get_or_create_conversation("be-dev-1", "be-dev-1")


@pytest.mark.asyncio
async def test_get_or_create_conversation_creates(a2a_setup: dict) -> None:
    """A2A between two same-cell devs is allowed by the policy."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-dev-2")
    assert conv is not None


@pytest.mark.asyncio
async def test_get_conversation_returns_none_for_missing(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    assert await svc.get_conversation(uuid4(), "be-dev-1") is None


@pytest.mark.asyncio
async def test_list_conversations_empty(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    convs = await svc.list_conversations("be-dev-1")
    assert isinstance(convs, list)


@pytest.mark.asyncio
async def test_list_conversations_with_filters(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    convs = await svc.list_conversations(
        "be-dev-1", status=None, with_agent="be-dev-2", limit=10
    )
    assert isinstance(convs, list)


# ---------------------------------------------------------------------------
# Resolve creator agent
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_resolve_creator_agent_returns_uuid_or_none(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    out = await svc.resolve_creator_agent(dev.slug)
    assert out is not None or out is None  # Smoke test: doesn't raise.


@pytest.mark.asyncio
async def test_resolve_creator_agent_unknown(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    out = await svc.resolve_creator_agent("ghost-agent-slug")
    assert out is None or hasattr(out, "id") or isinstance(out, type(uuid4()))


# ---------------------------------------------------------------------------
# Chat messages
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_send_chat_message_rejects_nil_uuid(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    nil = UUID(int=0)
    with pytest.raises(ValueError, match="nil UUID"):
        await svc.send_chat_message(nil, "be-dev-1", "hi")


@pytest.mark.asyncio
async def test_send_chat_message_unknown_conversation(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    with pytest.raises(ValueError, match="not found"):
        await svc.send_chat_message(uuid4(), "be-dev-1", "hi")


@pytest.mark.asyncio
async def test_get_messages_unknown_conversation_returns_empty(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    msgs = await svc.get_messages(uuid4(), "be-dev-1")
    assert msgs == []


@pytest.mark.asyncio
async def test_close_conversation_unknown(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    with pytest.raises(ValueError, match="not found"):
        await svc.close_conversation(uuid4(), "be-dev-1")


@pytest.mark.asyncio
async def test_mark_read_unknown_returns_none(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    # Returns None silently for unknown conversation.
    await svc.mark_read(uuid4(), "be-dev-1")


# ---------------------------------------------------------------------------
# Inbox + pairs
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_inbox_summary(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    inbox = await svc.get_inbox_summary("be-dev-1")
    assert inbox is not None


@pytest.mark.asyncio
async def test_list_pairs(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    pairs = await svc.list_pairs("be-dev-1")
    assert isinstance(pairs, list)


@pytest.mark.asyncio
async def test_send_a2a_returns_handler_result(a2a_setup: dict) -> None:
    """Just exercise the send() entrypoint with a stub that fails closed."""
    svc = a2a_setup["svc"]
    try:
        result = await svc.send(
            from_agent="be-dev-1",
            to_agent="be-dev-2",
            skill="general",
            message="hi",
        )
        assert result is not None
    except Exception:
        # Expected if the policy rejects this pair or service is wired
        # to external infra in this test setup.
        pass


@pytest.mark.asyncio
async def test_send_records_skill_on_message_for_receiver(a2a_setup: dict) -> None:
    """#1416: a2a.send(skill=...) must record the requested capability on the
    persisted message so the receiver learns it — not silently drop it. The
    gateway adapter's docstring promised exactly this, but send_chat_message
    never read the ``skill`` opt and the table had no skill column, so every
    gateway A2A send lost the capability signal the caller passed."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    task_id = a2a_setup["task_id"]
    sent = await svc.send(
        from_agent=dev.id,
        to_agent="be-qa",
        task_id=task_id,
        body="please review my PR",
        skill="code_review",
    )
    assert sent.skill == "code_review"
    # The receiver reads the capability back via get_messages.
    inbox = await svc.get_messages(UUID(sent.conversation_id), "be-qa")
    assert inbox
    assert inbox[-1].skill == "code_review"


@pytest.mark.asyncio
async def test_send_publishes_a2a_message_sent_event_when_bus_connected(
    a2a_setup: dict,
) -> None:
    """A2AService.send() is the gateway's one publish point for A2A chat — it
    must fan an A2A_MESSAGE_SENT event so the CEO's live view (operator
    /ws/system stream) sees every directed agent-to-agent message."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    task_id = a2a_setup["task_id"]
    mock_bus = AsyncMock()
    mock_bus.is_connected = lambda: True
    mock_bus.publish = AsyncMock(return_value=None)
    with patch("roboco.services.a2a.get_event_bus", return_value=mock_bus):
        sent = await svc.send(
            from_agent=dev.id,
            to_agent="be-qa",
            task_id=task_id,
            body="please review",
            skill="code_review",
        )
    mock_bus.publish.assert_awaited()
    published = mock_bus.publish.await_args.args[0]
    assert published.type is EventType.A2A_MESSAGE_SENT
    data = published.data
    assert data["conversation_id"] == sent.conversation_id
    assert data["message_id"] == sent.id
    assert data["task_id"] == str(task_id)
    assert data["from_agent"] == "be-dev-1"
    assert data["to_agent"] == "be-qa"
    assert data["skill"] == "code_review"
    assert data["body_excerpt"] == "please review"
    assert data["timestamp"] == sent.created_at.isoformat()


@pytest.mark.asyncio
async def test_send_excerpts_long_body_in_event(a2a_setup: dict) -> None:
    """The WS live-view frame carries a capped excerpt, not the full body —
    but the persisted message keeps the full untruncated text (readable via
    the existing REST message endpoints)."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    task_id = a2a_setup["task_id"]
    long_body = "x" * (_LIVE_VIEW_EXCERPT_CHARS + 100)
    mock_bus = AsyncMock()
    mock_bus.is_connected = lambda: True
    mock_bus.publish = AsyncMock(return_value=None)
    with patch("roboco.services.a2a.get_event_bus", return_value=mock_bus):
        sent = await svc.send(
            from_agent=dev.id,
            to_agent="be-qa",
            task_id=task_id,
            body=long_body,
        )
    published = mock_bus.publish.await_args.args[0]
    assert published.type is EventType.A2A_MESSAGE_SENT
    excerpt = published.data["body_excerpt"]
    assert len(excerpt) < len(long_body)
    assert excerpt.endswith("…")
    # Full body survives untruncated in persistent storage.
    assert sent.content == long_body
    stored = await svc.get_messages(UUID(sent.conversation_id), "be-qa")
    assert stored[-1].content == long_body


@pytest.mark.asyncio
async def test_send_bus_failure_does_not_break_send(a2a_setup: dict) -> None:
    """A bus outage during the A2A_MESSAGE_SENT publish is logged but never
    rolls back the persisted message — live delivery is best-effort."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    task_id = a2a_setup["task_id"]
    with patch(
        "roboco.services.a2a.get_event_bus",
        side_effect=RuntimeError("bus down"),
    ):
        sent = await svc.send(
            from_agent=dev.id,
            to_agent="be-qa",
            task_id=task_id,
            body="hello",
        )
    assert sent.id is not None
    assert sent.content == "hello"


@pytest.mark.asyncio
async def test_send_chat_message_directly_publishes_event(
    a2a_setup: dict,
) -> None:
    """The REST send paths (conversation-create + post-message) call
    send_chat_message directly, not the send() wrapper. That direct path must
    still emit A2A_MESSAGE_SENT so those messages light up the CEO's live view
    — the gap this test guards."""
    svc = a2a_setup["svc"]
    task_id = a2a_setup["task_id"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa", task_id=task_id)
    mock_bus = AsyncMock()
    mock_bus.is_connected = lambda: True
    mock_bus.publish = AsyncMock(return_value=None)
    with patch("roboco.services.a2a.get_event_bus", return_value=mock_bus):
        sent = await svc.send_chat_message(
            UUID(conv.id),
            "be-dev-1",
            "please review",
            options={"skill": "code_review"},
        )
    mock_bus.publish.assert_awaited_once()
    data = mock_bus.publish.await_args.args[0].data
    assert data["message_id"] == sent.id
    assert data["from_agent"] == "be-dev-1"
    assert data["to_agent"] == "be-qa"
    assert data["skill"] == "code_review"
    assert data["task_id"] == str(task_id)


@pytest.mark.asyncio
async def test_suppressed_duplicate_does_not_republish(
    a2a_setup: dict,
) -> None:
    """A re-sent identical unread message dedups to the existing row and must
    NOT emit a second live-view event (no redundant cache invalidation)."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    mock_bus = AsyncMock()
    mock_bus.is_connected = lambda: True
    mock_bus.publish = AsyncMock(return_value=None)
    with patch("roboco.services.a2a.get_event_bus", return_value=mock_bus):
        await svc.send_chat_message(UUID(conv.id), "be-dev-1", "same text")
        await svc.send_chat_message(UUID(conv.id), "be-dev-1", "same text")
    mock_bus.publish.assert_awaited_once()


@pytest.mark.asyncio
async def test_list_unread_a2a_preview_is_incoming_not_own(a2a_setup: dict) -> None:
    """The briefing preview must be the latest INCOMING message, never the
    agent's own reply — an agent that replies before reading must still see
    what was said to it, not its own words."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]  # be-dev-1
    task_id = a2a_setup["task_id"]
    conv = await svc.get_or_create_conversation("be-qa", "be-dev-1", task_id=task_id)
    await svc.send_chat_message(
        UUID(conv.id), "be-qa", "Please fix the null check in auth.py"
    )
    # dev replies WITHOUT reading — its own message is now the latest row.
    await svc.send_chat_message(UUID(conv.id), "be-dev-1", "ok on it")

    items = await EvidenceRepo(a2a_setup["db"]).list_unread_a2a(dev.id)

    assert items and items[0]["from_agent"] == "be-qa"
    assert "null check" in items[0]["last_message_preview"]
    assert "ok on it" not in items[0]["last_message_preview"]


@pytest.mark.asyncio
async def test_get_unread_messages_returns_incoming_bodies_and_clears(
    a2a_setup: dict,
) -> None:
    """read_a2a delivers the unread INCOMING bodies to the agent (never its own
    sends) in order, then clears them — a second call returns nothing."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]  # be-dev-1
    conv = await svc.get_or_create_conversation(
        "be-qa", "be-dev-1", task_id=a2a_setup["task_id"]
    )
    await svc.send_chat_message(UUID(conv.id), "be-qa", "first")
    await svc.send_chat_message(UUID(conv.id), "be-dev-1", "my own reply")
    await svc.send_chat_message(UUID(conv.id), "be-qa", "second")

    msgs = await svc.get_unread_messages(dev.id)

    assert [m["content"] for m in msgs] == ["first", "second"]
    assert all(m["from_agent"] == "be-qa" for m in msgs)
    assert await svc.get_unread_messages(dev.id) == []


# ---------------------------------------------------------------------------
# Admin (CEO live view) service methods — no participant filter
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_conversations_admin_includes_non_participant_pairs(
    a2a_setup: dict,
) -> None:
    """The CEO's live view has no participant filter — it must show
    conversations between two agents where the CEO is not itself a party."""
    svc = a2a_setup["svc"]
    conv1 = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    conv2 = await svc.get_or_create_conversation("fe-dev-1", "fe-qa")

    summaries = await svc.list_conversations_admin(limit=50)

    ids = {s.id for s in summaries}
    assert conv1.id in ids
    assert conv2.id in ids
    pairs = {(s.agent_a, s.agent_b) for s in summaries}
    assert ("be-dev-1", "be-qa") in pairs
    assert ("fe-dev-1", "fe-qa") in pairs


@pytest.mark.asyncio
async def test_list_conversations_admin_orders_most_recent_first_and_bounds(
    a2a_setup: dict,
) -> None:
    """Most-recent-first ordering and a hard limit — proven by forcing
    distinguishable updated_at values across three seeded conversations."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    conv_a = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    conv_b = await svc.get_or_create_conversation("fe-dev-1", "fe-qa")
    conv_c = await svc.get_or_create_conversation("ux-dev-1", "ux-qa")

    now = datetime.now(UTC)
    for conv_id, offset in (
        (conv_a.id, timedelta(minutes=-10)),
        (conv_b.id, timedelta(minutes=-5)),
        (conv_c.id, timedelta(minutes=0)),
    ):
        row = await db.get(A2AConversationTable, UUID(conv_id))
        assert row is not None
        row.updated_at = now + offset
    await db.flush()

    summaries = await svc.list_conversations_admin(limit=2)

    _LIMIT = 2
    assert len(summaries) == _LIMIT
    assert [s.id for s in summaries] == [conv_c.id, conv_b.id]


@pytest.mark.asyncio
async def test_get_messages_admin_returns_full_transcript_for_non_participant(
    a2a_setup: dict,
) -> None:
    """The plain get_messages() denies a non-participant (returns []); the
    admin bypass returns the full transcript regardless — the exact behavior
    a normal agent-scoped call cannot give the CEO today."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    cid = UUID(conv.id)
    await svc.send_chat_message(cid, "be-dev-1", "hello")
    await svc.send_chat_message(cid, "be-qa", "hi back")

    as_ceo_scoped = await svc.get_messages(cid, "ceo")
    assert as_ceo_scoped == []

    admin_view = await svc.get_messages_admin(cid)
    _EXPECTED = 2
    assert len(admin_view) == _EXPECTED
    assert admin_view[0].content == "hello"
    assert admin_view[1].content == "hi back"


@pytest.mark.asyncio
async def test_get_messages_admin_unknown_conversation_returns_empty(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    assert await svc.get_messages_admin(uuid4()) == []


@pytest.mark.asyncio
async def test_get_conversation_admin_returns_conversation_ceo_not_part_of(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")

    fetched = await svc.get_conversation_admin(UUID(conv.id))

    assert fetched is not None
    assert fetched.id == conv.id
    assert fetched.agent_a == "be-dev-1"
    assert fetched.agent_b == "be-qa"


@pytest.mark.asyncio
async def test_get_conversation_admin_returns_none_for_unknown(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    assert await svc.get_conversation_admin(uuid4()) is None


# ---------------------------------------------------------------------------
# interject_as_ceo — the CEO's one-directional interjection into a watched
# agent<->agent conversation (not a re-homed CEO<->target DM).
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_interject_as_ceo_lands_in_viewed_conversation_with_prefix(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")

    msg = await svc.interject_as_ceo(UUID(conv.id), "be-qa", "ship it")

    assert msg.conversation_id == conv.id
    assert msg.from_agent == "ceo"
    assert msg.content == "@be-qa: ship it"


@pytest.mark.asyncio
async def test_interject_as_ceo_bumps_target_unread_when_target_is_agent_b(
    a2a_setup: dict,
) -> None:
    """Canonical order makes "be-qa" agent_b — its counter, not agent_a's,
    must move; the other participant gets no ping."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    assert conv.agent_a == "be-dev-1"
    assert conv.agent_b == "be-qa"

    await svc.interject_as_ceo(UUID(conv.id), "be-qa", "ship it")

    row = await db.get(A2AConversationTable, UUID(conv.id))
    assert row is not None
    assert row.unread_by_b == 1
    assert row.unread_by_a == 0


@pytest.mark.asyncio
async def test_interject_as_ceo_bumps_target_unread_when_target_is_agent_a(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")

    await svc.interject_as_ceo(UUID(conv.id), "be-dev-1", "ship it")

    row = await db.get(A2AConversationTable, UUID(conv.id))
    assert row is not None
    assert row.unread_by_a == 1
    assert row.unread_by_b == 0


@pytest.mark.asyncio
async def test_interject_as_ceo_bumps_message_count_and_last_message_at(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    await svc.send_chat_message(UUID(conv.id), "be-dev-1", "hello")

    await svc.interject_as_ceo(UUID(conv.id), "be-qa", "ship it")

    row = await db.get(A2AConversationTable, UUID(conv.id))
    assert row is not None
    _EXPECTED_MESSAGE_COUNT = 2
    assert row.message_count == _EXPECTED_MESSAGE_COUNT
    assert row.last_message_at is not None


@pytest.mark.asyncio
async def test_interject_as_ceo_rejects_non_participant_target(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")

    with pytest.raises(ValueError, match="not a participant"):
        await svc.interject_as_ceo(UUID(conv.id), "ghost-agent", "hi")


@pytest.mark.asyncio
async def test_interject_as_ceo_unknown_conversation_raises(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    with pytest.raises(ValueError, match="Conversation not found"):
        await svc.interject_as_ceo(uuid4(), "be-qa", "hi")


@pytest.mark.asyncio
async def test_interject_as_ceo_publishes_a2a_message_sent_event(
    a2a_setup: dict,
) -> None:
    """Same operator-live-view chokepoint as send()/send_chat_message() —
    the panel's /ws/system invalidation must fire for an interjection too."""
    svc = a2a_setup["svc"]
    task_id = a2a_setup["task_id"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa", task_id=task_id)
    mock_bus = AsyncMock()
    mock_bus.is_connected = lambda: True
    mock_bus.publish = AsyncMock(return_value=None)
    with patch("roboco.services.a2a.get_event_bus", return_value=mock_bus):
        sent = await svc.interject_as_ceo(UUID(conv.id), "be-qa", "ship it")

    mock_bus.publish.assert_awaited_once()
    published = mock_bus.publish.await_args.args[0]
    assert published.type is EventType.A2A_MESSAGE_SENT
    data = published.data
    # Points at the VIEWED conversation, not a re-homed ceo<->target one.
    assert data["conversation_id"] == conv.id
    assert data["conversation_id"] == sent.conversation_id
    assert data["task_id"] == str(task_id)
    assert data["from_agent"] == "ceo"
    assert data["to_agent"] == "be-qa"


# ---------------------------------------------------------------------------
# list_admin_pairs — the A2A switchboard's static-matrix + DB join
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_admin_pairs_bounded_by_static_matrix(a2a_setup: dict) -> None:
    """With no conversations at all, every pair from the static matrix is
    still returned (conversation-less), sized exactly to the matrix."""
    svc = a2a_setup["svc"]
    pairs = await svc.list_admin_pairs()

    assert len(pairs) == len(A2A_ALLOWED_PAIRS)
    assert all(p.conversation_id is None for p in pairs)
    assert all(p.message_count == 0 for p in pairs)
    assert all(p.last_message_at is None for p in pairs)


@pytest.mark.asyncio
async def test_list_admin_pairs_joins_representative_conversation(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    await svc.send_chat_message(UUID(conv.id), "be-dev-1", "hello")

    pairs = await svc.list_admin_pairs()

    match = next(p for p in pairs if {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"})
    assert match.conversation_id == conv.id
    assert match.message_count == 1
    assert match.last_message_at is not None
    assert match.group_key == "cell-backend"
    assert match.role_a == "developer"
    assert match.role_b == "qa"


@pytest.mark.asyncio
async def test_list_admin_pairs_picks_most_recently_updated_conversation(
    a2a_setup: dict,
) -> None:
    """A pair with two conversations (distinct topics) surfaces the more
    recently active one as its representative conversation."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    conv_old = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="t1")
    conv_new = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="t2")

    now = datetime.now(UTC)
    row_old = await db.get(A2AConversationTable, UUID(conv_old.id))
    row_new = await db.get(A2AConversationTable, UUID(conv_new.id))
    assert row_old is not None
    assert row_new is not None
    row_old.updated_at = now - timedelta(minutes=10)
    row_new.updated_at = now
    await db.flush()

    pairs = await svc.list_admin_pairs()

    match = next(p for p in pairs if {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"})
    assert match.conversation_id == conv_new.id


@pytest.mark.asyncio
async def test_list_admin_pairs_excludes_disallowed_pairs(a2a_setup: dict) -> None:
    """A conversation row between two agents the matrix does NOT allow (dev
    A2A is same-cell only — this should never legitimately exist, but the
    join must be robust against it) never surfaces as a pair card: the
    service iterates the static matrix, not "any conversation row"."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    stray = A2AConversationTable(agent_a="be-dev-1", agent_b="fe-dev-1")
    db.add(stray)
    await db.flush()

    pairs = await svc.list_admin_pairs()

    assert not any({p.agent_a, p.agent_b} == {"be-dev-1", "fe-dev-1"} for p in pairs)


# ---------------------------------------------------------------------------
# CEO reply-only budget — an agent may only reply to the CEO inside a
# conversation the CEO itself opened, and only up to the CEO's own message
# count in that conversation.
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_send_to_ceo_without_existing_conversation_denied(
    a2a_setup: dict,
) -> None:
    """An agent can never INITIATE a CEO conversation via the gateway
    send() adapter — only reply inside one the CEO already opened."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    task_id = a2a_setup["task_id"]
    with pytest.raises(A2AAccessDeniedError):
        await svc.send(from_agent=dev.id, to_agent="ceo", task_id=task_id, body="hi")


@pytest.mark.asyncio
async def test_ceo_can_post_consecutive_messages_no_budget(a2a_setup: dict) -> None:
    """CEO -> agent direction is unrestricted — no budget applies to CEO
    sends, so the CEO may post twice in a row with no reply in between."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("ceo", "be-dev-1")
    cid = UUID(conv.id)
    await svc.send_chat_message(cid, "ceo", "first")
    second = await svc.send_chat_message(cid, "ceo", "second, no reply needed yet")
    assert second.content == "second, no reply needed yet"


@pytest.mark.asyncio
async def test_ceo_reply_budget_first_reply_allowed(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("ceo", "be-dev-1")
    cid = UUID(conv.id)
    await svc.send_chat_message(cid, "ceo", "hi dev")
    reply = await svc.send_chat_message(cid, "be-dev-1", "on it")
    assert reply.content == "on it"


@pytest.mark.asyncio
async def test_ceo_reply_budget_second_reply_without_new_ceo_message_rejected(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("ceo", "be-dev-1")
    cid = UUID(conv.id)
    await svc.send_chat_message(cid, "ceo", "hi dev")
    await svc.send_chat_message(cid, "be-dev-1", "on it")
    with pytest.raises(A2AAccessDeniedError, match="already replied"):
        await svc.send_chat_message(cid, "be-dev-1", "another update")


@pytest.mark.asyncio
async def test_ceo_reply_budget_refreshes_after_new_ceo_message(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("ceo", "be-dev-1")
    cid = UUID(conv.id)
    await svc.send_chat_message(cid, "ceo", "hi dev")
    await svc.send_chat_message(cid, "be-dev-1", "on it")
    await svc.send_chat_message(cid, "ceo", "any update?")
    reply2 = await svc.send_chat_message(cid, "be-dev-1", "done!")
    assert reply2.content == "done!"


@pytest.mark.asyncio
async def test_ceo_reply_budget_independent_across_conversations(
    a2a_setup: dict,
) -> None:
    """The a2a_conversations model is strictly pairwise — a literal 3-party
    thread can't exist. Adapted form: two agents each in their OWN
    conversation with the CEO get independent budgets; one agent exhausting
    its budget must not affect the other's."""
    svc = a2a_setup["svc"]
    conv_dev = await svc.get_or_create_conversation("ceo", "be-dev-1")
    conv_qa = await svc.get_or_create_conversation("ceo", "be-qa")
    cid_dev = UUID(conv_dev.id)
    cid_qa = UUID(conv_qa.id)

    await svc.send_chat_message(cid_dev, "ceo", "dev, status?")
    await svc.send_chat_message(cid_qa, "ceo", "qa, status?")

    await svc.send_chat_message(cid_dev, "be-dev-1", "on it")
    with pytest.raises(A2AAccessDeniedError):
        await svc.send_chat_message(cid_dev, "be-dev-1", "again")

    # qa's independent budget is untouched by dev's exhausted one.
    qa_reply = await svc.send_chat_message(cid_qa, "be-qa", "on it too")
    assert qa_reply.content == "on it too"


@pytest.mark.asyncio
async def test_ceo_reply_dedup_before_budget_check(a2a_setup: dict) -> None:
    """Dedup runs BEFORE the budget check: a respawned agent re-sending its
    identical unread reply gets the existing row back idempotently — never a
    budget error — even once the agent has exhausted its reply budget."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("ceo", "be-dev-1")
    cid = UUID(conv.id)
    await svc.send_chat_message(cid, "ceo", "status?")
    first = await svc.send_chat_message(cid, "be-dev-1", "on it")
    # Budget is now exhausted (agent_count == ceo_count == 1); an identical
    # resend must still dedup instead of hitting the budget gate.
    again = await svc.send_chat_message(cid, "be-dev-1", "on it")
    assert again.id == first.id


@pytest.mark.asyncio
async def test_send_to_ceo_via_gateway_when_ceo_opened_conversation(
    a2a_setup: dict,
) -> None:
    """Once the CEO has opened a conversation with the agent, the gateway
    send() adapter finds it directly — bypassing
    get_or_create_conversation's validate-first gate, which would otherwise
    deny even a legitimate reply — and the reply persists under budget."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    task_id = a2a_setup["task_id"]
    conv = await svc.get_or_create_conversation("ceo", "be-dev-1")
    await svc.send_chat_message(UUID(conv.id), "ceo", "status?")

    reply = await svc.send(
        from_agent=dev.id, to_agent="ceo", task_id=task_id, body="on it"
    )
    assert reply.content == "on it"
    assert reply.from_agent == "be-dev-1"


@pytest.mark.asyncio
async def test_send_publishes_only_after_persist_not_on_reply_denial(
    a2a_setup: dict,
) -> None:
    """A rejected send (no existing CEO conversation) must never publish
    A2A_MESSAGE_SENT — the event is a record of a persisted message."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    task_id = a2a_setup["task_id"]
    mock_bus = AsyncMock()
    mock_bus.is_connected = lambda: True
    mock_bus.publish = AsyncMock(return_value=None)
    with (
        patch("roboco.services.a2a.get_event_bus", return_value=mock_bus),
        pytest.raises(A2AAccessDeniedError),
    ):
        await svc.send(from_agent=dev.id, to_agent="ceo", task_id=task_id, body="hi")
    mock_bus.publish.assert_not_awaited()


# ---------------------------------------------------------------------------
# Conversation creation happy path with allowed pair
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_create_conversation_between_dev_and_qa_in_same_cell(
    a2a_setup: dict,
) -> None:
    """Cell members can A2A within their own cell."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    assert conv is not None
    # Idempotent — same agents, same conversation.
    again = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    assert again.id == conv.id


@pytest.mark.asyncio
async def test_send_chat_message_in_existing_conversation(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    msg = await svc.send_chat_message(UUID(conv.id), "be-dev-1", "hello")
    assert msg.content == "hello"


@pytest.mark.asyncio
async def test_get_messages_returns_chronological(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    cid = UUID(conv.id)
    await svc.send_chat_message(cid, "be-dev-1", "first")
    await svc.send_chat_message(cid, "be-dev-1", "second")
    msgs = await svc.get_messages(cid, "be-dev-1")
    _SENT_COUNT = 2
    assert len(msgs) == _SENT_COUNT


@pytest.mark.asyncio
async def test_send_chat_message_dedups_identical_unread(a2a_setup: dict) -> None:
    """An identical message re-sent while still unread is suppressed (one copy),
    but a different message is not collapsed."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    cid = UUID(conv.id)
    first = await svc.send_chat_message(cid, "be-dev-1", "ping: are you free?")
    again = await svc.send_chat_message(cid, "be-dev-1", "ping: are you free?")
    distinct = await svc.send_chat_message(cid, "be-dev-1", "different message")
    # The duplicate returns the SAME stored message and adds no new row.
    assert again.id == first.id
    assert distinct.id != first.id
    msgs = await svc.get_messages(cid, "be-dev-1")
    _EXPECTED = 2  # the deduped "ping" + the distinct one
    assert len(msgs) == _EXPECTED


@pytest.mark.asyncio
async def test_close_conversation_with_resolution(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    await svc.close_conversation(UUID(conv.id), "be-dev-1", resolution="done")


@pytest.mark.asyncio
async def test_mark_read_clears_unread(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    await svc.mark_read(UUID(conv.id), "be-dev-1")


@pytest.mark.asyncio
async def test_close_conversation_non_participant_raises(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    with pytest.raises(ValueError, match="Not a participant"):
        await svc.close_conversation(UUID(conv.id), "ghost-agent")


@pytest.mark.asyncio
async def test_send_chat_message_non_participant_raises(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    with pytest.raises(ValueError, match="Not a participant"):
        await svc.send_chat_message(UUID(conv.id), "ghost", "hi")


# ---------------------------------------------------------------------------
# Pure-function helpers
# ---------------------------------------------------------------------------


def test_get_team_from_agent_backend() -> None:
    assert A2AService.get_team_from_agent("be-dev-1") == Team.BACKEND


def test_get_team_from_agent_unknown_defaults_to_backend() -> None:
    assert A2AService.get_team_from_agent("ghost-agent") == Team.BACKEND


def test_resolve_target_agent_explicit() -> None:
    result = A2AService.resolve_target_agent({"target_agent": "be-dev-1"})
    assert result == "be-dev-1"


def test_resolve_target_agent_unknown_returns_none() -> None:
    result = A2AService.resolve_target_agent({"target_agent": "ghost-agent"})
    assert result is None


def test_resolve_target_agent_none_when_no_metadata() -> None:
    result = A2AService.resolve_target_agent({})
    assert result is None


def test_extract_message_text_no_text_parts() -> None:
    msg = A2AMessage(role="user", parts=[])
    title, desc, _full = A2AService.extract_message_text(msg)
    assert title == "A2A Task"
    assert desc == ""


def test_extract_message_text_single_line() -> None:
    msg = A2AMessage(role="user", parts=[TextPart(text="Hello world")])
    title, _desc, _full = A2AService.extract_message_text(msg)
    assert title == "Hello world"


def test_extract_message_text_multi_line() -> None:
    msg = A2AMessage(role="user", parts=[TextPart(text="Title here\nThis is the body")])
    title, desc, _full = A2AService.extract_message_text(msg)
    assert title == "Title here"
    assert desc == "This is the body"


@pytest.mark.asyncio
async def test_update_task_with_message_appends_to_notes(
    a2a_setup: dict,
) -> None:
    """Use a real DB-backed task instance to avoid SA private state issues."""
    db = a2a_setup["db"]
    task = (await db.execute(select(TaskTable).limit(1))).scalar_one()
    original_notes = task.dev_notes
    task.dev_notes = "existing notes"
    msg = A2AMessage(role="user", parts=[TextPart(text="new message")])
    A2AService.update_task_with_message(task, msg)
    assert "existing notes" in task.dev_notes
    assert "new message" in task.dev_notes
    task.dev_notes = original_notes  # restore


@pytest.mark.asyncio
async def test_update_task_with_message_no_text_parts_noop(
    a2a_setup: dict,
) -> None:
    db = a2a_setup["db"]
    task = (await db.execute(select(TaskTable).limit(1))).scalar_one()
    original = task.dev_notes
    task.dev_notes = "existing"
    msg = A2AMessage(role="user", parts=[])
    A2AService.update_task_with_message(task, msg)
    assert task.dev_notes == "existing"
    task.dev_notes = original


# ---------------------------------------------------------------------------
# resolve_creator_agent paths
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_resolve_creator_agent_with_unknown_falls_back_to_main_pm(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    # Unknown ID falls back to main PM lookup — returns None if no main PM seeded.
    out = await svc.resolve_creator_agent("ghost-id")
    # Either None (no main_pm) or AgentTable (main_pm seeded by a prior test).
    assert out is None or hasattr(out, "id")


@pytest.mark.asyncio
async def test_resolve_creator_agent_with_none_falls_back_to_main_pm(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    out = await svc.resolve_creator_agent(None)
    assert out is None or hasattr(out, "id")


# ---------------------------------------------------------------------------
# get_service_endpoint — unspecified host falls through to 127.0.0.1
# ---------------------------------------------------------------------------


def test_get_service_endpoint_unspecified_host() -> None:
    """0.0.0.0 host triggers loopback fallback."""
    with patch("roboco.services.a2a.settings") as mock_settings:
        mock_settings.host = "0.0.0.0"
        mock_settings.port = 8000
        url = A2AService.get_service_endpoint()
    assert "127.0.0.1" in url


def test_get_service_endpoint_invalid_host_falls_through() -> None:
    """Non-IP host string falls through ValueError → uses host directly."""
    with patch("roboco.services.a2a.settings") as mock_settings:
        mock_settings.host = "myhost"
        mock_settings.port = 8080
        url = A2AService.get_service_endpoint()
    assert "myhost" in url


# ---------------------------------------------------------------------------
# task_to_a2a — branches: dev_notes, no value attr, assigned_to, parent
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_task_to_a2a_with_dev_notes_and_assignment(
    a2a_setup: dict,
) -> None:
    """Task with dev_notes, assigned_to, parent_task_id covers metadata branches."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    parent = TaskTable(
        id=_u(),
        title="parent",
        description="d",
        acceptance_criteria=["ac"],
        status=TaskStatus.PENDING,
        priority=2,
        task_type=TaskType.CODE,
        nature=TaskNature.TECHNICAL,
        project_id=(await db.execute(select(TaskTable))).scalars().first().project_id,
        created_by=a2a_setup["dev"].id,
        team=Team.BACKEND,
    )
    db.add(parent)
    await db.flush()
    child = TaskTable(
        id=_u(),
        title="child",
        description="d",
        acceptance_criteria=["ac"],
        status=TaskStatus.PENDING,
        priority=2,
        task_type=TaskType.CODE,
        nature=TaskNature.TECHNICAL,
        project_id=parent.project_id,
        created_by=a2a_setup["dev"].id,
        assigned_to=a2a_setup["dev"].id,
        parent_task_id=parent.id,
        team=Team.BACKEND,
        dev_notes="some progress",
    )
    db.add(child)
    await db.flush()
    a2a_task = svc.task_to_a2a(child)
    assert "assigned_to" in a2a_task.metadata
    assert "parent_task_id" in a2a_task.metadata


def test_task_to_a2a_status_without_value_attr(a2a_setup: dict) -> None:
    """When task.status lacks .value (already a string), use str()."""
    svc = a2a_setup["svc"]
    fake_task = SimpleNamespace(
        id="00000000-0000-0000-0000-000000000001",
        status="pending",  # plain string, no .value
        priority=2,
        team="backend",
        dev_notes=None,
        assigned_to=None,
        parent_task_id=None,
        updated_at=None,
        created_at=datetime.now(UTC),
    )
    a2a_task = svc.task_to_a2a(fake_task)
    assert a2a_task.metadata["roboco_status"] == "pending"


# ---------------------------------------------------------------------------
# list_tasks — has_more=True branch
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_tasks_has_more_true(a2a_setup: dict) -> None:
    """Seed enough tasks to trigger has_more=True."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    pid = (await db.execute(select(TaskTable))).scalars().first().project_id
    for _i in range(3):
        db.add(
            TaskTable(
                id=_u(),
                title=f"t{_i}",
                description="d",
                acceptance_criteria=["ac"],
                status=TaskStatus.PENDING,
                priority=2,
                task_type=TaskType.CODE,
                nature=TaskNature.TECHNICAL,
                project_id=pid,
                created_by=a2a_setup["dev"].id,
                team=Team.BACKEND,
            )
        )
    await db.flush()
    _PAGE = 2
    tasks, has_more = await svc.list_tasks(page_size=_PAGE)
    assert has_more is True
    assert len(tasks) == _PAGE


# ---------------------------------------------------------------------------
# cancel_task — full path with reason and existing dev_notes
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_cancel_task_with_reason_and_existing_notes(
    a2a_setup: dict,
) -> None:
    """Reason is appended to existing dev_notes and full cancel runs."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    pid = (await db.execute(select(TaskTable))).scalars().first().project_id
    task = TaskTable(
        id=_u(),
        title="t",
        description="d",
        acceptance_criteria=["ac"],
        status=TaskStatus.PENDING,
        priority=2,
        task_type=TaskType.CODE,
        nature=TaskNature.TECHNICAL,
        project_id=pid,
        created_by=a2a_setup["dev"].id,
        team=Team.BACKEND,
        dev_notes="initial work",
    )
    db.add(task)
    await db.flush()
    a2a_task = await svc.cancel_task(str(task.id), reason="changed mind")
    assert a2a_task.id == str(task.id)


@pytest.mark.asyncio
async def test_cancel_task_with_reason_no_existing_notes(
    a2a_setup: dict,
) -> None:
    """Reason populates fresh dev_notes when none existed."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    pid = (await db.execute(select(TaskTable))).scalars().first().project_id
    task = TaskTable(
        id=_u(),
        title="t",
        description="d",
        acceptance_criteria=["ac"],
        status=TaskStatus.PENDING,
        priority=2,
        task_type=TaskType.CODE,
        nature=TaskNature.TECHNICAL,
        project_id=pid,
        created_by=a2a_setup["dev"].id,
        team=Team.BACKEND,
    )
    db.add(task)
    await db.flush()
    a2a_task = await svc.cancel_task(str(task.id), reason="cancelled")
    assert a2a_task is not None


@pytest.mark.asyncio
async def test_cancel_task_status_no_value_attr(a2a_setup: dict) -> None:
    """If task.status is already a string, str() fallback runs."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    task = (await db.execute(select(TaskTable).limit(1))).scalar_one()
    # Mock execute so the cancel-side query returns a task with status="pending" string.
    real_execute = svc.session.execute
    fake_uuid = _u()

    class _FakeStatus:
        # No .value attribute
        def __str__(self) -> str:
            return "pending"

    fake_task = type(
        "FakeTask",
        (),
        {
            "id": str(fake_uuid),
            "status": _FakeStatus(),
            "dev_notes": None,
        },
    )()

    seen = {"hit": False}

    async def _intercepting_execute(stmt: Any, *args: Any, **kwargs: Any) -> Any:
        if not seen["hit"]:
            seen["hit"] = True
            stub = _MM()
            stub.scalar_one_or_none.return_value = fake_task
            return stub
        return await real_execute(stmt, *args, **kwargs)

    # Patch TaskService.cancel to return the same task object.
    with patch("roboco.services.task.TaskService") as mock_ts:
        instance = AsyncMock()
        instance.cancel = AsyncMock(return_value=task)
        mock_ts.return_value = instance
        with patch.object(svc.session, "execute", side_effect=_intercepting_execute):
            a2a_task = await svc.cancel_task(str(fake_uuid))
    assert a2a_task is not None


@pytest.mark.asyncio
async def test_cancel_task_failed_returns_value_error(
    a2a_setup: dict,
) -> None:
    """If TaskService.cancel returns None, raise ValueError."""
    svc = a2a_setup["svc"]
    db = a2a_setup["db"]
    pid = (await db.execute(select(TaskTable))).scalars().first().project_id
    task = TaskTable(
        id=_u(),
        title="t",
        description="d",
        acceptance_criteria=["ac"],
        status=TaskStatus.PENDING,
        priority=2,
        task_type=TaskType.CODE,
        nature=TaskNature.TECHNICAL,
        project_id=pid,
        created_by=a2a_setup["dev"].id,
        team=Team.BACKEND,
    )
    db.add(task)
    await db.flush()
    with patch("roboco.services.task.TaskService") as mock_ts:
        instance = AsyncMock()
        instance.cancel = AsyncMock(return_value=None)
        mock_ts.return_value = instance
        with pytest.raises(ValueError, match="Failed to cancel"):
            await svc.cancel_task(str(task.id))


# ---------------------------------------------------------------------------
# resolve_target_agent — skill-based routing match
# ---------------------------------------------------------------------------


def test_resolve_target_agent_by_skill_match() -> None:
    """metadata.skill matches an agent skill id → returns slug."""
    with patch(
        "roboco.services.a2a.get_agent_skills",
        return_value=[{"id": "general", "name": "g"}],
    ):
        result = A2AService.resolve_target_agent({"skill": "general"})
    # First agent in ALL_AGENTS with this skill id wins.
    assert result is not None


def test_resolve_target_agent_skill_no_match() -> None:
    """Unknown skill returns None."""
    with patch("roboco.services.a2a.get_agent_skills", return_value=[]):
        result = A2AService.resolve_target_agent({"skill": "ghost-skill"})
    assert result is None


# ---------------------------------------------------------------------------
# extract_message_text — text part without `text` attribute
# ---------------------------------------------------------------------------


def test_extract_message_text_no_text_attr() -> None:
    """text_part missing `text` attr → returns defaults."""
    fake_part = SimpleNamespace(type="text")
    fake_msg = SimpleNamespace(parts=[fake_part])
    title, desc, full = A2AService.extract_message_text(cast("A2AMessage", fake_msg))
    assert title == "A2A Task"
    assert desc == ""
    assert full == ""


# ---------------------------------------------------------------------------
# update_task_with_message — text_part lacks `.text` attr
# ---------------------------------------------------------------------------


def test_update_task_with_message_no_text_attr() -> None:
    """text_part without text attribute leaves dev_notes untouched."""
    fake_part = SimpleNamespace(type="text")
    fake_msg = SimpleNamespace(parts=[fake_part])
    fake_task = SimpleNamespace(dev_notes="orig")
    A2AService.update_task_with_message(
        cast("TaskTable", fake_task), cast("A2AMessage", fake_msg)
    )
    assert fake_task.dev_notes == "orig"


# ---------------------------------------------------------------------------
# resolve_creator_agent — known agent slug + lookup hit
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_resolve_creator_agent_known_slug_with_uuid_hit(
    a2a_setup: dict,
) -> None:
    """from_agent_id is a known slug AND has UUID — looks up by UUID."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    # be-dev-1 is in ALL_AGENTS (a list) by default; just inject a UUID.
    with patch.dict(
        "roboco.services.a2a.AGENT_UUIDS",
        {"be-dev-1": str(dev.id)},
    ):
        out = await svc.resolve_creator_agent("be-dev-1")
    # May return None if agent not found by id, or the dev row.
    assert out is None or hasattr(out, "id")


# ---------------------------------------------------------------------------
# create_a2a_notification — full path
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_create_a2a_notification_missing_task_id_raises(
    a2a_setup: dict,
) -> None:
    """task_id absent → ValueError."""
    svc = a2a_setup["svc"]
    msg = A2AMessage(role="user", parts=[TextPart(text="hi")])
    req = SendMessageRequest(message=msg, metadata={})
    with pytest.raises(ValueError, match="task_id"):
        await svc.create_a2a_notification(req)


@pytest.mark.asyncio
async def test_create_a2a_notification_with_target_calls_notification_service(
    a2a_setup: dict,
) -> None:
    """Path with from_agent + target + skill → NotificationService called."""
    svc = a2a_setup["svc"]
    task_id = str(a2a_setup["task_id"])
    msg = A2AMessage(role="user", parts=[TextPart(text="hi there")], task_id=task_id)
    req = SendMessageRequest(
        message=msg,
        metadata={"from_agent": "be-dev-1", "target_agent": "be-dev-2"},
    )
    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService",
        return_value=mock_ns,
    ):
        result = await svc.create_a2a_notification(req)
    assert result["task_id"] == task_id


@pytest.mark.asyncio
async def test_create_a2a_notification_permission_denied(
    a2a_setup: dict,
) -> None:
    """Hierarchy denial → the typed A2AAccessDeniedError (with route_hint), routed
    through validate_a2a_access so the legacy notification path matches the
    conversation path — not a bare ValueError a caller can't distinguish from a
    malformed request."""
    svc = a2a_setup["svc"]
    task_id = str(a2a_setup["task_id"])
    msg = A2AMessage(role="user", parts=[TextPart(text="hi")], task_id=task_id)
    req = SendMessageRequest(
        message=msg,
        metadata={"from_agent": "be-dev-1", "target_agent": "be-dev-2"},
    )
    with (
        patch(
            "roboco.enforcement.a2a_access.can_a2a_direct",
            return_value=(False, "denied"),
        ),
        patch(
            "roboco.enforcement.a2a_access.get_a2a_route_hint",
            return_value="use channel",
        ),
        pytest.raises(A2AAccessDeniedError) as exc,
    ):
        await svc.create_a2a_notification(req)
    assert exc.value.route_hint == "use channel"


@pytest.mark.asyncio
async def test_create_a2a_notification_self_a2a_raises_typed_error(
    a2a_setup: dict,
) -> None:
    """#612: a self-directed A2A notification (from_agent == target) must raise
    the typed A2AAccessDeniedError — the self-check the conversation path
    (validate_a2a_access) enforces — not a bare ValueError and not a silently
    sent notification."""
    svc = a2a_setup["svc"]
    task_id = str(a2a_setup["task_id"])
    msg = A2AMessage(role="user", parts=[TextPart(text="self note")], task_id=task_id)
    req = SendMessageRequest(
        message=msg,
        metadata={"from_agent": "be-dev-1", "target_agent": "be-dev-1"},
    )
    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with (
        patch("roboco.services.notification.NotificationService", return_value=mock_ns),
        pytest.raises(A2AAccessDeniedError),
    ):
        await svc.create_a2a_notification(req)
    mock_ns.send_a2a_notification.assert_not_called()


@pytest.mark.asyncio
async def test_create_a2a_notification_missing_from_agent_raises_not_silent(
    a2a_setup: dict,
) -> None:
    """#609: an unattributed A2A request (no from_agent, unresolvable target)
    must NOT slip past the hierarchy gate and dispatch with
    from_agent='unknown'. The gate is unconditional — both ends must be present
    and resolvable before any notification is created."""
    svc = a2a_setup["svc"]
    task_id = str(a2a_setup["task_id"])
    msg = A2AMessage(role="user", parts=[TextPart(text="anon")], task_id=task_id)
    req = SendMessageRequest(message=msg, metadata={})
    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with (
        patch("roboco.services.notification.NotificationService", return_value=mock_ns),
        pytest.raises(ValueError, match="from_agent"),
    ):
        await svc.create_a2a_notification(req)
    mock_ns.send_a2a_notification.assert_not_called()


# ---------------------------------------------------------------------------
# update_task_from_message — full happy path + invalid id
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_update_task_from_message_invalid_id(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    msg = A2AMessage(role="user", parts=[TextPart(text="hi")])
    with pytest.raises(ValueError, match="Invalid task ID"):
        await svc.update_task_from_message("not-a-uuid", msg)


@pytest.mark.asyncio
async def test_update_task_from_message_not_found(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    msg = A2AMessage(role="user", parts=[TextPart(text="hi")])
    with pytest.raises(ValueError, match="not found"):
        await svc.update_task_from_message(str(_u()), msg)


@pytest.mark.asyncio
async def test_update_task_from_message_success(a2a_setup: dict) -> None:
    """Real task updated with new dev_notes."""
    svc = a2a_setup["svc"]
    task_id = str(a2a_setup["task_id"])
    msg = A2AMessage(role="user", parts=[TextPart(text="response text")])
    updated = await svc.update_task_from_message(task_id, msg)
    assert "response text" in (updated.dev_notes or "")


# ---------------------------------------------------------------------------
# _notify_original_requester — branches: not A2A, no created_by, no slug
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_notify_original_requester_not_a2a_returns(
    a2a_setup: dict,
) -> None:
    """Task.dev_notes lacks 'A2A Request' → no-op."""
    svc = a2a_setup["svc"]
    fake_task = SimpleNamespace(dev_notes="just notes", created_by=None)
    # No raise.
    await svc._notify_original_requester(fake_task, "responder")


@pytest.mark.asyncio
async def test_notify_original_requester_no_created_by(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    fake_task = SimpleNamespace(dev_notes="A2A Request: hi", created_by=None)
    await svc._notify_original_requester(fake_task)


@pytest.mark.asyncio
async def test_notify_original_requester_unknown_slug(
    a2a_setup: dict,
) -> None:
    """If the creator UUID can't be mapped to a slug, returns silently."""
    svc = a2a_setup["svc"]
    fake_task = SimpleNamespace(
        id=_u(),
        dev_notes="A2A Request: please review",
        created_by=_u(),  # Not in AGENT_UUIDS.
    )
    await svc._notify_original_requester(fake_task)


@pytest.mark.asyncio
async def test_notify_original_requester_responder_is_requester(
    a2a_setup: dict,
) -> None:
    """If responder is the same as requester, no event published."""
    svc = a2a_setup["svc"]
    fake_task = SimpleNamespace(
        id="00000000-0000-0000-0000-000000000001",
        dev_notes="A2A Request: please review",
        created_by="be-dev-1-uuid",
    )
    with patch(
        "roboco.services.a2a.A2AService._lookup_requester_slug",
        return_value="be-dev-1",
    ):
        # responder == requester → return without publish.
        await svc._notify_original_requester(fake_task, "be-dev-1")


@pytest.mark.asyncio
async def test_notify_original_requester_publishes_event(
    a2a_setup: dict,
) -> None:
    """Full path: requester slug found, responder different → publish."""
    svc = a2a_setup["svc"]
    fake_task = SimpleNamespace(
        id="00000000-0000-0000-0000-000000000001",
        dev_notes="A2A Request: please review",
        created_by="some-uuid",
    )
    mock_bus = AsyncMock()
    mock_bus.is_connected = lambda: True
    mock_bus.publish = AsyncMock(return_value=None)
    with (
        patch(
            "roboco.services.a2a.A2AService._lookup_requester_slug",
            return_value="be-dev-1",
        ),
        patch("roboco.services.a2a.get_event_bus", return_value=mock_bus),
    ):
        await svc._notify_original_requester(fake_task, "be-dev-2")
    mock_bus.publish.assert_awaited()


@pytest.mark.asyncio
async def test_publish_a2a_response_event_no_bus() -> None:
    """Bus not connected → silent no-op."""
    fake_task = SimpleNamespace(id="00000000-0000-0000-0000-000000000001")
    mock_bus = type("B", (), {"is_connected": lambda _self: False})()
    with patch("roboco.services.a2a.get_event_bus", return_value=mock_bus):
        await A2AService._publish_a2a_response_event(
            cast("TaskTable", fake_task), "creator", "requester", "responder"
        )


@pytest.mark.asyncio
async def test_publish_a2a_response_event_bus_exception_swallowed() -> None:
    fake_task = SimpleNamespace(id="00000000-0000-0000-0000-000000000001")
    with patch(
        "roboco.services.a2a.get_event_bus",
        side_effect=RuntimeError("bus down"),
    ):
        # Exception swallowed.
        await A2AService._publish_a2a_response_event(
            cast("TaskTable", fake_task), "creator", "requester", "responder"
        )


# ---------------------------------------------------------------------------
# list_conversations — status/with_agent/task_id filters + last message
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_conversations_with_status_and_task_filter(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    convs = await svc.list_conversations(
        "be-dev-1",
        status=A2AConversationStatus.ACTIVE,
        with_agent="be-dev-2",
        task_id=a2a_setup["task_id"],
    )
    assert isinstance(convs, list)


@pytest.mark.asyncio
async def test_list_conversations_with_messages(a2a_setup: dict) -> None:
    """Seed a conversation + a message so the last_message preview path runs."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    await svc.send_chat_message(UUID(conv.id), "be-dev-1", "preview text")

    convs = await svc.list_conversations("be-dev-1")
    # last_message_preview should be truthy for this conv.
    assert any(c.last_message_preview for c in convs)


# ---------------------------------------------------------------------------
# send_chat_message — both agents perspective + message_kind/response_to
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_send_chat_message_options_response_to(
    a2a_setup: dict,
) -> None:
    """response_to_id and requires_response options surface on the message.

    `response_to_id` has a FK to `a2a_messages.id`, so we send a real
    "first" message to thread off — passing a random UUID would hit FK
    violation.
    """
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    first = await svc.send_chat_message(UUID(conv.id), "be-qa", "first message")
    msg = await svc.send_chat_message(
        UUID(conv.id),
        "be-dev-1",
        "needs answer",
        options={"requires_response": True, "response_to_id": UUID(first.id)},
    )
    assert msg.requires_response is True
    assert msg.response_to_id == first.id


@pytest.mark.asyncio
async def test_send_chat_message_from_agent_b_increments_unread_a(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    # First exchange establishes canonical pair (a < b lexicographically).
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    # Send from whichever agent is conv.agent_b (the "other" side).
    result = await svc.session.execute(
        _sel(A2AConversationTable).where(A2AConversationTable.id == UUID(conv.id))
    )
    row = result.scalar_one()
    msg = await svc.send_chat_message(UUID(conv.id), row.agent_b, "hi from b")
    assert msg.from_agent == row.agent_b


# ---------------------------------------------------------------------------
# get_messages — non-participant returns []
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_messages_non_participant_returns_empty(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    msgs = await svc.get_messages(UUID(conv.id), "ghost-agent")
    assert msgs == []


@pytest.mark.asyncio
async def test_get_messages_with_before_filter(a2a_setup: dict) -> None:
    """Pass a `before` datetime — exercises the filter branch."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    await svc.send_chat_message(UUID(conv.id), "be-dev-1", "first")
    future = datetime.now(UTC).replace(year=2099)
    msgs = await svc.get_messages(UUID(conv.id), "be-dev-1", before=future)
    assert isinstance(msgs, list)


# ---------------------------------------------------------------------------
# mark_read — non-participant returns silently + agent_b path
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_mark_read_non_participant(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    # Non-participant → silent return.
    await svc.mark_read(UUID(conv.id), "ghost")


@pytest.mark.asyncio
async def test_mark_read_as_agent_b(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    result = await svc.session.execute(
        _sel(A2AConversationTable).where(A2AConversationTable.id == UUID(conv.id))
    )
    row = result.scalar_one()
    await svc.mark_read(UUID(conv.id), row.agent_b)


# ---------------------------------------------------------------------------
# list_pairs — exercises grouping + last_activity comparison
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_list_pairs_with_conversations(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    await svc.get_or_create_conversation("be-dev-1", "be-qa")
    pairs = await svc.list_pairs("be-dev-1")
    assert any(
        (p.agent_a, p.agent_b) == ("be-dev-1", "be-qa")
        or (p.agent_a, p.agent_b) == ("be-qa", "be-dev-1")
        for p in pairs
    )


# ---------------------------------------------------------------------------
# _resolve_slug_from_id — happy + raise
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_resolve_slug_from_id_happy(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    slug = await svc._resolve_slug_from_id(dev.id)
    assert slug == dev.slug


@pytest.mark.asyncio
async def test_resolve_slug_from_id_missing_raises(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    with pytest.raises(ValueError, match="Agent not found"):
        await svc._resolve_slug_from_id(_u())


# ---------------------------------------------------------------------------
# send — gateway adapter, both UUID and str recipient
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_send_gateway_adapter_uuid_to_uuid(
    a2a_setup: dict,
) -> None:
    """Both ends as UUIDs → resolves both slugs via DB lookup."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    qa = a2a_setup["qa"]
    msg = await svc.send(
        from_agent=dev.id,
        to_agent=qa.id,
        task_id=a2a_setup["task_id"],
        body="hello",
        skill="general",
    )
    assert msg.content == "hello"


@pytest.mark.asyncio
async def test_send_gateway_adapter_with_string_recipient(
    a2a_setup: dict,
) -> None:
    """Recipient as slug string → no DB lookup for it."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    msg = await svc.send(
        from_agent=dev.id,
        to_agent="be-qa",
        task_id=a2a_setup["task_id"],
        body="hello",
    )
    assert msg.content == "hello"


# ---------------------------------------------------------------------------
# create_task_from_message — patched DB so flush succeeds
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# _lookup_requester_slug — found path
# ---------------------------------------------------------------------------


def test_lookup_requester_slug_found() -> None:
    target_uuid = "00000000-0000-0000-0000-000000000042"
    with patch(
        "roboco.seeds.initial_data.AGENT_UUIDS",
        {"slug-x": target_uuid},
    ):
        result = A2AService._lookup_requester_slug(target_uuid)
    assert result == "slug-x"


# ---------------------------------------------------------------------------
# get_or_create_conversation — topic provided branch
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_or_create_conversation_with_topic(
    a2a_setup: dict,
) -> None:
    svc = a2a_setup["svc"]
    a = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="Bug X")
    b = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="Bug X")
    assert a.id == b.id


# ---------------------------------------------------------------------------
# get_conversation — non-participant returns None
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_conversation_non_participant(a2a_setup: dict) -> None:
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    result = await svc.get_conversation(UUID(conv.id), "ghost")
    assert result is None


@pytest.mark.asyncio
async def test_get_conversation_returns_model_when_participant(
    a2a_setup: dict,
) -> None:
    """Participant access → returns the conversation model."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    result = await svc.get_conversation(UUID(conv.id), "be-dev-1")
    assert result is not None
    assert result.id == conv.id


# ---------------------------------------------------------------------------
# get_inbox_summary with unread > 0
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_inbox_summary_with_unread(a2a_setup: dict) -> None:
    """Send a message from a2 → a1 has unread."""
    svc = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
    result = await svc.session.execute(
        _sel(A2AConversationTable).where(A2AConversationTable.id == UUID(conv.id))
    )
    row = result.scalar_one()
    # Send from agent_b → agent_a unread increments.
    await svc.send_chat_message(UUID(conv.id), row.agent_b, "hi")
    inbox = await svc.get_inbox_summary(row.agent_a)
    assert inbox.total_unread >= 1


# ---------------------------------------------------------------------------
# send — gateway adapter without skill
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_send_gateway_adapter_skill_none(
    a2a_setup: dict,
) -> None:
    """skill=None branch → options dict stays empty."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    msg = await svc.send(
        from_agent=dev.id,
        to_agent="be-qa",
        task_id=a2a_setup["task_id"],
        body="no skill",
    )
    assert msg.content == "no skill"


@pytest.mark.asyncio
async def test_send_gateway_adapter_with_mocked_conv(
    a2a_setup: dict,
) -> None:
    """Mock get_or_create_conversation + send_chat_message so skill branches run."""
    svc = a2a_setup["svc"]
    dev = a2a_setup["dev"]
    fake_conv = SimpleNamespace(id=str(_u()))
    fake_msg = SimpleNamespace(content="hi", id=str(_u()))
    with (
        patch.object(svc, "_resolve_slug_from_id", AsyncMock(return_value="be-dev-1")),
        patch.object(
            svc, "get_or_create_conversation", AsyncMock(return_value=fake_conv)
        ),
        patch.object(svc, "send_chat_message", AsyncMock(return_value=fake_msg)),
    ):
        # With skill set.
        result = await svc.send(
            from_agent=dev.id,
            to_agent="be-qa",
            task_id=a2a_setup["task_id"],
            body="hi",
            skill="general",
        )
    assert result.content == "hi"
    # Without skill.
    with (
        patch.object(svc, "_resolve_slug_from_id", AsyncMock(return_value="be-dev-1")),
        patch.object(
            svc, "get_or_create_conversation", AsyncMock(return_value=fake_conv)
        ),
        patch.object(svc, "send_chat_message", AsyncMock(return_value=fake_msg)),
    ):
        result2 = await svc.send(
            from_agent=dev.id,
            to_agent="be-qa",
            task_id=a2a_setup["task_id"],
            body="hi-2",
        )
    assert result2.content == "hi"


@pytest.mark.asyncio
async def test_mark_all_read_clears_unread_for_agent(a2a_setup: dict) -> None:
    """mark_all_read zeroes the agent's unread counter across its conversations
    and stamps read_at on the inbound messages, returning the count cleared —
    the bulk ack that lets an agent satisfy i_am_idle's unread-A2A soft-block."""
    svc: A2AService = a2a_setup["svc"]
    db = a2a_setup["db"]
    qa = a2a_setup["qa"]
    conv = await svc.get_or_create_conversation(
        agent_a="be-dev-1", agent_b="be-qa", task_id=a2a_setup["task_id"]
    )
    conv_id = UUID(conv.id)
    # be-dev-1 < be-qa canonically → dev is agent_a; dev's messages bump qa's
    # unread (unread_by_b).
    await svc.send_chat_message(conv_id, "be-dev-1", "one")
    await svc.send_chat_message(conv_id, "be-dev-1", "two")

    cleared = await svc.mark_all_read(qa.id)
    assert cleared == 1

    row = await db.get(A2AConversationTable, conv_id)
    assert row is not None
    assert row.unread_by_b == 0
    msgs = (
        (
            await db.execute(
                select(A2AMessageTable).where(
                    A2AMessageTable.conversation_id == conv_id
                )
            )
        )
        .scalars()
        .all()
    )
    assert all(m.read_at is not None for m in msgs)

    # Idempotent — nothing left unread for qa.
    assert await svc.mark_all_read(qa.id) == 0


# mark_read / mark_all_read — collect-then-mark race coverage (M27)
#
# A send_chat_message committing between the counter-zero and the bulk UPDATE
# used to insert a new read_at NULL row that the UPDATE then stamped as read —
# the new message was silently consumed. The fix mirrors get_unread_messages:
# SELECT the unread IDs first, UPDATE exactly those, recompute the counter from
# the DB. These tests inject a racing message inside the UPDATE step and assert
# the racing message stays unread (read_at NULL) and the counter reflects it.


def _is_a2a_message_update(stmt: object) -> bool:
    """True when ``stmt`` is the bulk ``UPDATE a2a_messages ...`` DML."""
    if not isinstance(stmt, Update):
        return False
    table = getattr(stmt, "table", None)
    return getattr(table, "name", None) == "a2a_messages"


@pytest.mark.asyncio
async def test_mark_read_racing_message_stays_unread(a2a_setup: dict) -> None:
    """A message arriving mid-mark_read (after the SELECT, before the UPDATE)
    must stay unread — the bulk UPDATE stamps only the rows seen at call time,
    and the counter is recomputed from the DB so the new message is counted."""
    svc: A2AService = a2a_setup["svc"]
    db = a2a_setup["db"]
    conv = await svc.get_or_create_conversation(
        agent_a="be-dev-1", agent_b="be-qa", task_id=a2a_setup["task_id"]
    )
    conv_id = UUID(conv.id)
    # dev is agent_a → dev's messages bump qa's unread (unread_by_b).
    await svc.send_chat_message(conv_id, "be-dev-1", "first")

    real_execute = db.execute
    injected: list[str] = []

    async def patched_execute(stmt: object, *args: object, **kwargs: object) -> object:
        if not injected and _is_a2a_message_update(stmt):
            # Race: a new incoming message lands right before the UPDATE runs.
            await svc.send_chat_message(conv_id, "be-dev-1", "racing second")
            injected.append("racing second")
        return await real_execute(stmt, *args, **kwargs)

    db.execute = patched_execute
    try:
        await svc.mark_read(conv_id, "be-qa")
    finally:
        db.execute = real_execute

    assert injected, "test harness must fire the racing send before the UPDATE"

    row = await db.get(A2AConversationTable, conv_id)
    assert row is not None
    msgs = (
        (
            await real_execute(
                select(A2AMessageTable)
                .where(A2AMessageTable.conversation_id == conv_id)
                .order_by(A2AMessageTable.created_at)
            )
        )
        .scalars()
        .all()
    )
    racing = [m for m in msgs if m.content == "racing second"]
    first = [m for m in msgs if m.content == "first"]
    assert racing and first
    assert first[0].read_at is not None, "pre-race message must be marked read"
    assert racing[0].read_at is None, "racing message must stay unread"
    # Counter reflects the actual unread incoming count (1, the racing message).
    assert row.unread_by_b == 1


@pytest.mark.asyncio
async def test_mark_all_read_racing_message_stays_unread(a2a_setup: dict) -> None:
    """Same race for mark_all_read: a message arriving mid-call (after the
    SELECT, before the bulk UPDATE) must stay unread and be reflected in the
    recomputed counter, not silently consumed as read."""
    svc: A2AService = a2a_setup["svc"]
    db = a2a_setup["db"]
    qa = a2a_setup["qa"]
    conv = await svc.get_or_create_conversation(
        agent_a="be-dev-1", agent_b="be-qa", task_id=a2a_setup["task_id"]
    )
    conv_id = UUID(conv.id)
    await svc.send_chat_message(conv_id, "be-dev-1", "first")

    real_execute = db.execute
    injected: list[str] = []

    async def patched_execute(stmt: object, *args: object, **kwargs: object) -> object:
        if not injected and _is_a2a_message_update(stmt):
            await svc.send_chat_message(conv_id, "be-dev-1", "racing second")
            injected.append("racing second")
        return await real_execute(stmt, *args, **kwargs)

    db.execute = patched_execute
    try:
        cleared = await svc.mark_all_read(qa.id)
    finally:
        db.execute = real_execute

    assert injected, "test harness must fire the racing send before the UPDATE"
    assert cleared == 1

    row = await db.get(A2AConversationTable, conv_id)
    assert row is not None
    msgs = (
        (
            await real_execute(
                select(A2AMessageTable)
                .where(A2AMessageTable.conversation_id == conv_id)
                .order_by(A2AMessageTable.created_at)
            )
        )
        .scalars()
        .all()
    )
    racing = [m for m in msgs if m.content == "racing second"]
    first = [m for m in msgs if m.content == "first"]
    assert racing and first
    assert first[0].read_at is not None
    assert racing[0].read_at is None, "racing message must stay unread"
    assert row.unread_by_b == 1


# ---------------------------------------------------------------------------
# CEO-DM wake — a conversational A2A DM from the CEO must wake an offline
# recipient (reuses the legacy a2a_request NotificationTable row
# `_dispatch_a2a_work` already polls); agent<->agent A2A stays pull-only.
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_ceo_dm_wakes_offline_recipient(a2a_setup: dict) -> None:
    """A CEO->agent chat message creates exactly one a2a_request wake
    notification, with requires_ack=True so the row is visible to
    _dispatch_a2a_work's pending_ack_only poll (A2A_REQUEST otherwise
    defaults to requires_ack=False and is invisible there)."""
    svc: A2AService = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation(agent_a="ceo", agent_b="be-dev-1")
    conv_id = UUID(conv.id)

    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService", return_value=mock_ns
    ):
        await svc.send_chat_message(conv_id, "ceo", "status update please")

    mock_ns.send_a2a_notification.assert_awaited_once()
    kwargs = mock_ns.send_a2a_notification.await_args.kwargs
    assert kwargs["requires_ack"] is True
    assert kwargs["a2a_context"]["from_agent"] == "ceo"
    assert kwargs["a2a_context"]["to_agent"] == "be-dev-1"


@pytest.mark.asyncio
async def test_ceo_dm_wake_dedups_while_pending(a2a_setup: dict) -> None:
    """A second CEO message while an unacked wake notification for the same
    recipient is still pending must not create a duplicate."""
    svc: A2AService = a2a_setup["svc"]
    db = a2a_setup["db"]
    dev = a2a_setup["dev"]
    conv = await svc.get_or_create_conversation(agent_a="ceo", agent_b="be-dev-1")
    conv_id = UUID(conv.id)

    pending = NotificationTable(
        type=NotificationType.A2A_REQUEST,
        priority=NotificationPriority.NORMAL,
        from_agent=dev.id,  # placeholder sender — irrelevant to the dedup lookup
        to_agents=[dev.id],
        subject="A2A: ceo_dm",
        body="already pending",
        requires_ack=True,
    )
    db.add(pending)
    await db.flush()

    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService", return_value=mock_ns
    ):
        await svc.send_chat_message(conv_id, "ceo", "second message")

    mock_ns.send_a2a_notification.assert_not_awaited()


@pytest.mark.asyncio
async def test_agent_to_agent_dm_creates_no_wake(a2a_setup: dict) -> None:
    """Ordinary agent<->agent A2A stays pull-only — no wake notification, so
    routine chatter can't burn spawns."""
    svc: A2AService = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation(
        agent_a="be-dev-1", agent_b="be-qa", task_id=a2a_setup["task_id"]
    )
    conv_id = UUID(conv.id)

    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService", return_value=mock_ns
    ):
        await svc.send_chat_message(conv_id, "be-dev-1", "sync?")

    mock_ns.send_a2a_notification.assert_not_awaited()


@pytest.mark.asyncio
async def test_agent_reply_to_ceo_creates_no_wake(a2a_setup: dict) -> None:
    """An agent's reply to the CEO must never wake the CEO — the CEO is
    human (is_spawnable_agent_slug already refuses it), and from_slug !=
    "ceo" stops it before that check even runs."""
    svc: A2AService = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation(agent_a="ceo", agent_b="be-dev-1")
    conv_id = UUID(conv.id)
    await svc.send_chat_message(conv_id, "ceo", "status?")

    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService", return_value=mock_ns
    ):
        await svc.send_chat_message(conv_id, "be-dev-1", "on it")

    mock_ns.send_a2a_notification.assert_not_awaited()


@pytest.mark.asyncio
async def test_ceo_dm_to_non_a2a_role_denied_at_conversation_creation(
    a2a_setup: dict,
) -> None:
    """A CEO DM to a role with no dm/read_a2a on its manifest (the human-only
    prompter/secretary — own dedicated chat pages) must be refused outright
    at conversation creation — the root-cause fix (can_a2a_direct's CEO
    branch excludes NO_COMMS_ROLES) supersedes the old symptom-level fix of
    letting the conversation exist and only suppressing the wake
    notification (the recipient could never ack it, so it would be
    immortal, permanently suppress future wakes via the dedup pre-check,
    and drive futile respawns)."""
    svc: A2AService = a2a_setup["svc"]
    with pytest.raises(A2AAccessDeniedError, match="no agent-comms surface"):
        await svc.get_or_create_conversation(agent_a="ceo", agent_b="secretary-1")


@pytest.mark.asyncio
@pytest.mark.parametrize("target_slug", ["auditor", "pr-reviewer-1"])
async def test_ceo_dm_to_auditor_or_pr_reviewer_conversation_allowed(
    a2a_setup: dict, target_slug: str
) -> None:
    """The auditor and PR reviewer now carry dm/read_a2a, so a CEO can open
    a DM with a mid-flight one — the conversation must be created, not
    refused, even though neither gains a peer-initiation surface."""
    svc: A2AService = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation(agent_a="ceo", agent_b=target_slug)
    assert conv is not None


@pytest.mark.asyncio
async def test_maybe_wake_ceo_recipient_still_noops_for_no_comms_role(
    a2a_setup: dict,
) -> None:
    """Defense-in-depth: _maybe_wake_ceo_recipient's own read_a2a manifest
    check independently no-ops for a no-comms role — unreachable through the
    normal send path now that conversation creation refuses it first, but
    it must stay safe if ever called directly (e.g. on a pre-fix row)."""
    svc: A2AService = a2a_setup["svc"]
    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService", return_value=mock_ns
    ):
        await svc._maybe_wake_ceo_recipient("ceo", "secretary-1", None)

    mock_ns.send_a2a_notification.assert_not_awaited()


@pytest.mark.asyncio
async def test_maybe_wake_ceo_recipient_wakes_auditor_now_that_it_has_read_a2a(
    a2a_setup: dict,
) -> None:
    """auditor now carries read_a2a, so a CEO DM to an offline auditor wakes
    it the same way it wakes any other reachable agent — the manifest check
    that used to no-op for it must now let the wake through."""
    svc: A2AService = a2a_setup["svc"]
    auditor = AgentTable(
        id=uuid4(),
        name="Auditor",
        slug="auditor",
        role=AgentRole.AUDITOR,
        team=None,
        status=AgentStatus.ACTIVE,
        model_config={},
        system_prompt="auditor",
        capabilities=[],
        permissions={},
        metrics={},
    )
    db_session = svc.session
    db_session.add(auditor)
    await db_session.flush()

    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService", return_value=mock_ns
    ):
        await svc._maybe_wake_ceo_recipient("ceo", "auditor", None)

    mock_ns.send_a2a_notification.assert_awaited_once()


@pytest.mark.asyncio
async def test_interject_as_ceo_wakes_only_addressed_participant(
    a2a_setup: dict,
) -> None:
    """CEO interjection wakes only the @-addressed participant, not both."""
    svc: A2AService = a2a_setup["svc"]
    conv = await svc.get_or_create_conversation(
        agent_a="be-dev-1", agent_b="be-qa", task_id=a2a_setup["task_id"]
    )
    conv_id = UUID(conv.id)

    mock_ns = AsyncMock()
    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
    with patch(
        "roboco.services.notification.NotificationService", return_value=mock_ns
    ):
        await svc.interject_as_ceo(conv_id, to_agent="be-qa", content="ping")

    mock_ns.send_a2a_notification.assert_awaited_once()
    a2a_context = mock_ns.send_a2a_notification.await_args.kwargs["a2a_context"]
    assert a2a_context["to_agent"] == "be-qa"


@pytest.mark.asyncio
async def test_read_a2a_acks_pending_wake_notification(a2a_setup: dict) -> None:
    """read_a2a (get_unread_messages) closes out a pending wake notification
    once the recipient has actually drained its A2A inbox — otherwise the
    dedup in _maybe_wake_ceo_recipient would permanently suppress the next
    genuine wake for that recipient."""
    svc: A2AService = a2a_setup["svc"]
    db = a2a_setup["db"]
    qa = a2a_setup["qa"]
    dev = a2a_setup["dev"]

    conv = await svc.get_or_create_conversation(
        agent_a="be-dev-1", agent_b="be-qa", task_id=a2a_setup["task_id"]
    )
    conv_id = UUID(conv.id)
    await svc.send_chat_message(conv_id, "be-dev-1", "unread for qa")

    notif = NotificationTable(
        type=NotificationType.A2A_REQUEST,
        priority=NotificationPriority.NORMAL,
        from_agent=dev.id,
        to_agents=[qa.id],
        subject="A2A: ceo_dm",
        body="pending wake",
        requires_ack=True,
    )
    db.add(notif)
    await db.flush()
    notif_id = notif.id

    msgs = await svc.get_unread_messages(qa.id)
    assert msgs  # the content-bearing read actually returned something

    refreshed = await db.get(NotificationTable, notif_id)
    assert refreshed is not None
    assert qa.id in refreshed.acked_by


@pytest.mark.asyncio
async def test_ack_pending_wake_notifications_savepoint_isolates_db_failure(
    a2a_setup: dict,
) -> None:
    """A genuine Postgres-level failure inside `_ack_pending_wake_notifications`
    (e.g. `bulk_acknowledge`'s flush hitting a lock timeout) must not poison
    the shared session for whatever the caller does next. Force a REAL DB
    error — not a bare Python exception, which alone never aborts the
    underlying Postgres transaction — via a failing raw statement patched
    into `bulk_acknowledge`, then prove the session is still usable
    afterwards (no `PendingRollbackError`)."""
    svc: A2AService = a2a_setup["svc"]
    db = a2a_setup["db"]
    qa = a2a_setup["qa"]
    dev = a2a_setup["dev"]

    notif = NotificationTable(
        type=NotificationType.A2A_REQUEST,
        priority=NotificationPriority.NORMAL,
        from_agent=dev.id,
        to_agents=[qa.id],
        subject="A2A: ceo_dm",
        body="pending wake",
        requires_ack=True,
    )
    db.add(notif)
    await db.flush()
    notif_id = notif.id

    async def _boom(_self: object, *_args: object, **_kwargs: object) -> None:
        await db.execute(text("SELECT 1/0"))

    with patch(
        "roboco.services.notification_delivery.NotificationDeliveryService."
        "bulk_acknowledge",
        _boom,
    ):
        await svc._ack_pending_wake_notifications(qa.id)  # must not raise

    # The savepoint rolled back the poisoned statement — the shared session
    # is still usable for later work. Must be a real round trip (`.get()` on
    # an already-identity-mapped, unexpired object would just return the
    # cached instance without touching the DB, silently hiding a poisoned
    # transaction) — `execute(select(...))` always issues the query.
    refreshed = (
        await db.execute(
            select(NotificationTable).where(NotificationTable.id == notif_id)
        )
    ).scalar_one()
    assert qa.id not in refreshed.acked_by  # the swallowed failure never acked