roboco/roboco/services/a2a.py

"""
A2A (Agent-to-Agent) Protocol Service

Provides business logic for A2A protocol operations including:
- Agent discovery and card generation
- Task lifecycle management via A2A semantics
- Message handling and routing
"""

from datetime import UTC, datetime
from typing import Any, Final, cast
from uuid import UUID

import structlog
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from roboco.agents_config import (
    A2A_ALLOWED_PAIRS,
    ALL_AGENTS,
    get_agent_skills,
    get_agent_team,
)
from roboco.config import settings
from roboco.db.tables import (
    A2AConversationTable,
    A2AMessageTable,
    AgentTable,
    TaskTable,
)
from roboco.enforcement import A2AAccessDeniedError, validate_a2a_access
from roboco.events import Event, EventType, get_event_bus
from roboco.models.a2a import (
    A2AAdminPairSummary,
    A2AArtifact,
    A2AChatMessage,
    A2AConversation,
    A2AConversationAdminSummary,
    A2AConversationStatus,
    A2AConversationSummary,
    A2AInboxSummary,
    A2AMessage,
    A2AMessageKind,
    A2APair,
    A2ATask,
    A2ATaskStatus,
    AgentCapabilities,
    AgentCard,
    AgentProvider,
    AgentSkill,
    SecurityScheme,
    SendMessageRequest,
    TextPart,
    task_status_to_a2a_state,
)
from roboco.models.base import Team
from roboco.seeds.initial_data import AGENT_UUIDS

logger = structlog.get_logger()

# The A2A_MESSAGE_SENT WS frame (operator live view) carries a briefing-sized
# excerpt only — the full body remains readable via the existing REST message
# endpoints, so the live stream doesn't balloon on long A2A bodies.
_LIVE_VIEW_EXCERPT_CHARS: Final[int] = 240


def _excerpt(text: str, limit: int = _LIVE_VIEW_EXCERPT_CHARS) -> str:
    """Truncate ``text`` to ``limit`` chars, appending an ellipsis marker
    only when truncation actually happened."""
    return text if len(text) <= limit else text[:limit].rstrip() + "…"


class A2AService:
    """
    Service layer for A2A protocol operations.

    Provides methods for:
    - Building Agent Cards for discovery
    - Converting between RoboCo tasks and A2A tasks
    - Processing A2A messages
    """

    def __init__(self, session: AsyncSession):
        """Initialize with database session."""
        self.session = session

    @staticmethod
    def get_service_endpoint() -> str:
        """Build service endpoint URL from settings.

        When the API binds to the all-interfaces address (dev default),
        we can't use it for outbound callbacks — dial loopback instead.
        `is_unspecified` covers both 0.0.0.0 and ::, and avoids a bare
        literal that trips bandit B104.
        """
        import ipaddress

        try:
            is_any_iface = ipaddress.ip_address(settings.host).is_unspecified
        except ValueError:
            is_any_iface = False
        connect_host = "127.0.0.1" if is_any_iface else settings.host
        return f"http://{connect_host}:{settings.port}"

    @staticmethod
    def build_system_agent_card() -> AgentCard:
        """
        Build the system-level Agent Card for RoboCo.

        This card represents the entire RoboCo system and is served
        at /.well-known/agent.json
        """
        return AgentCard(
            id="roboco-system",
            name="RoboCo System",
            description=(
                "RoboCo is an AI Agentic Company - a virtual organization of "
                "AI agents designed to operate as a complete software "
                "development workforce."
            ),
            provider=AgentProvider(
                organization="RoboCo",
                url="https://github.com/roboco",
            ),
            protocol_version="1.0",
            service_endpoint=f"{A2AService.get_service_endpoint()}/api/a2a",
            version=settings.app_version,
            capabilities=AgentCapabilities(
                streaming=True,
                push_notifications=False,
                state_transition_history=True,
            ),
            default_input_modes=["text/plain", "application/json"],
            default_output_modes=["text/plain", "application/json"],
            skills=[
                AgentSkill(
                    id="software-development",
                    name="Software Development",
                    description="Full-stack software development with AI agents",
                    tags=["development", "coding", "qa", "documentation"],
                ),
                AgentSkill(
                    id="task-management",
                    name="Task Management",
                    description="Create and manage development tasks",
                    tags=["tasks", "kanban", "planning"],
                ),
                AgentSkill(
                    id="code-review",
                    name="Code Review",
                    description="Review and quality assurance of code",
                    tags=["qa", "review", "testing"],
                ),
            ],
            documentation_url="https://github.com/roboco/docs",
            security_schemes={
                "bearerAuth": SecurityScheme(type="http", scheme="bearer"),
            },
            security=[{"bearerAuth": []}],
        )

    async def build_agent_card(self, agent_id: str) -> AgentCard | None:
        """
        Build an Agent Card for a specific agent.

        Args:
            agent_id: Either a UUID string or agent slug

        Returns:
            AgentCard for the agent, or None if not found
        """
        # Try to parse as UUID first
        try:
            uuid = UUID(agent_id)
            result = await self.session.execute(
                select(AgentTable).where(AgentTable.id == uuid)
            )
        except ValueError:
            # Not a UUID, try slug lookup
            result = await self.session.execute(
                select(AgentTable).where(AgentTable.slug == agent_id)
            )

        agent = result.scalar_one_or_none()
        if agent is None:
            return None

        return self._agent_to_card(agent)

    def _agent_to_card(self, agent: AgentTable) -> AgentCard:
        """Convert an AgentTable row to an AgentCard."""
        agent_id = str(agent.id)
        agent_slug = agent.slug

        # Map role to skills
        role_skills: dict[str, list[AgentSkill]] = {
            "developer": [
                AgentSkill(
                    id="coding",
                    name="Code Development",
                    description="Write and implement code",
                    tags=["development", "coding"],
                ),
                AgentSkill(
                    id="debugging",
                    name="Debugging",
                    description="Debug and fix code issues",
                    tags=["debugging", "troubleshooting"],
                ),
            ],
            "qa": [
                AgentSkill(
                    id="testing",
                    name="Testing",
                    description="Test code and verify quality",
                    tags=["qa", "testing"],
                ),
                AgentSkill(
                    id="review",
                    name="Code Review",
                    description="Review code for quality and issues",
                    tags=["qa", "review"],
                ),
            ],
            "documenter": [
                AgentSkill(
                    id="documentation",
                    name="Documentation",
                    description="Write technical documentation",
                    tags=["documentation", "writing"],
                ),
            ],
            "cell_pm": [
                AgentSkill(
                    id="coordination",
                    name="Task Coordination",
                    description="Coordinate tasks within the cell",
                    tags=["management", "coordination"],
                ),
            ],
            "main_pm": [
                AgentSkill(
                    id="planning",
                    name="Project Planning",
                    description="Plan and coordinate across cells",
                    tags=["management", "planning"],
                ),
            ],
        }

        skills = role_skills.get(agent.role, [])

        return AgentCard(
            id=agent_id,
            name=agent.name,
            description=f"{agent.name} - {agent.role} agent in RoboCo",
            provider=AgentProvider(
                organization="RoboCo",
                url="https://github.com/roboco",
            ),
            protocol_version="1.0",
            service_endpoint=f"{self.get_service_endpoint()}/api/a2a",
            version=settings.app_version,
            capabilities=AgentCapabilities(
                streaming=True,
                push_notifications=False,
                state_transition_history=True,
            ),
            default_input_modes=["text/plain", "application/json"],
            default_output_modes=["text/plain", "application/json"],
            skills=skills,
            metadata={
                "slug": agent_slug,
                "role": agent.role,
                "team": agent.team,
            },
            security_schemes={
                "bearerAuth": SecurityScheme(type="http", scheme="bearer"),
            },
            security=[{"bearerAuth": []}],
        )

    def task_to_a2a(self, task: TaskTable) -> A2ATask:
        """
        Convert a RoboCo TaskTable to A2A Task.

        This is the canonical conversion that maintains semantic
        mapping between RoboCo's internal task model and A2A.
        """
        task_id = str(task.id)

        # Get status value as string
        if hasattr(task.status, "value"):
            status_value = task.status.value
        else:
            status_value = str(task.status)

        a2a_state = task_status_to_a2a_state(status_value)

        # Build status message from dev_notes if present
        status_message = None
        if task.dev_notes:
            status_message = A2AMessage(
                role="agent",
                parts=[TextPart(text=task.dev_notes)],
                task_id=task_id,
            )

        a2a_status = A2ATaskStatus(
            state=a2a_state,
            message=status_message,
            timestamp=task.updated_at or task.created_at,
        )

        # Build artifacts (placeholder — per-task file outputs are not tracked)
        artifacts: list[A2AArtifact] = []

        # Build metadata
        metadata: dict[str, str | int] = {
            "roboco_status": status_value,
            "priority": task.priority,
            "team": str(task.team),
        }
        if task.assigned_to:
            metadata["assigned_to"] = str(task.assigned_to)
        if task.parent_task_id:
            metadata["parent_task_id"] = str(task.parent_task_id)

        return A2ATask(
            id=task_id,
            context_id=task_id,
            status=a2a_status,
            artifacts=artifacts,
            history=[],
            metadata=metadata,
        )

    async def get_task(self, task_id: str) -> A2ATask | None:
        """
        Get a task by ID and return as A2A Task.

        Args:
            task_id: Task UUID string

        Returns:
            A2ATask or None if not found
        """
        try:
            task_uuid = UUID(task_id)
        except ValueError:
            return None

        result = await self.session.execute(
            select(TaskTable).where(TaskTable.id == task_uuid)
        )
        task = result.scalar_one_or_none()

        if task is None:
            return None

        return self.task_to_a2a(task)

    async def list_tasks(
        self,
        page_size: int = 20,
        offset: int = 0,
        order_by: str | None = None,
    ) -> tuple[list[A2ATask], bool]:
        """
        List tasks with pagination.

        Args:
            page_size: Number of results to return
            offset: Starting offset
            order_by: Sort order ("created_at desc" or "created_at asc")

        Returns:
            Tuple of (tasks, has_more)
        """
        query = select(TaskTable)

        # Apply ordering
        if order_by == "created_at asc":
            query = query.order_by(TaskTable.created_at.asc())
        else:
            query = query.order_by(TaskTable.created_at.desc())

        # Apply pagination (fetch one extra to detect more)
        query = query.offset(offset).limit(page_size + 1)

        result = await self.session.execute(query)
        tasks = list(result.scalars().all())

        has_more = len(tasks) > page_size
        if has_more:
            tasks = tasks[:page_size]

        return [self.task_to_a2a(t) for t in tasks], has_more

    @staticmethod
    def _status_value_of(task: Any) -> str:
        """Status as a comparable string (enum value or raw str)."""
        return task.status.value if hasattr(task.status, "value") else str(task.status)

    async def _apply_cancel_note(
        self, task: Any, actor_slug: str | None, reason: str | None
    ) -> None:
        """Append an actor-attributed cancellation note to dev_notes.

        Kept out of route handlers so the audit trail records who cancelled and
        why (the route passes the authenticated slug).
        """
        note_parts: list[str] = []
        if actor_slug:
            note_parts.append(f"Cancelled via A2A by {actor_slug}")
        if reason:
            note_parts.append(f"reason: {reason}")
        cancellation_note = "; ".join(note_parts) if note_parts else None
        if cancellation_note:
            if task.dev_notes:
                task.dev_notes = f"{task.dev_notes}\n\n{cancellation_note}"
            else:
                task.dev_notes = cancellation_note
            await self.session.flush()

    async def cancel_task(
        self,
        task_id: str,
        reason: str | None = None,
        agent_role: str | None = None,
        actor_slug: str | None = None,
    ) -> A2ATask:
        """
        Cancel a task and all non-terminal descendants.

        Args:
            task_id: Task UUID string
            reason: Optional cancellation reason
            agent_role: The authenticated caller's role, threaded into the
                cascade role gate (TaskService.cancel). Defaults to cell_pm
                when unset for back-compat with non-route callers.
            actor_slug: The authenticated caller's slug, recorded in the
                cancellation note so the audit trail attributes the cancel to
                the real actor (the route enforces PM/management; non-route
                callers may omit it).

        Returns:
            Updated A2ATask

        Raises:
            ValueError: If task not found or already in terminal state
        """
        # Import here to avoid circular imports
        from roboco.services.task import TaskService

        try:
            task_uuid = UUID(task_id)
        except ValueError as e:
            raise ValueError(f"Invalid task ID: {task_id}") from e

        # Check task exists and is cancellable before using service
        result = await self.session.execute(
            select(TaskTable).where(TaskTable.id == task_uuid)
        )
        task = result.scalar_one_or_none()

        if task is None:
            raise ValueError(f"Task not found: {task_id}")

        # Check if cancellable
        status_value = self._status_value_of(task)
        if status_value in ("completed", "cancelled"):
            raise ValueError(f"Task already in terminal state: {status_value}")

        # Attribute the cancel to the real actor (the route passes the
        # authenticated slug) in the audit note — kept out of route handlers.
        await self._apply_cancel_note(task, actor_slug, reason)

        # Use TaskService for consistent cancel behavior (cascades to descendants).
        # Thread the caller's role into the cascade role gate so a non-PM
        # caller can't cascade-cancel descendants the role can't cancel.
        task_service = TaskService(self.session)
        task = await task_service.cancel(task_uuid, agent_role=agent_role or "cell_pm")

        if task is None:
            raise ValueError(f"Failed to cancel task: {task_id}")

        logger.info(
            "Cancelled task via A2A",
            task_id=task_id,
            reason=reason,
            actor=actor_slug,
            role=agent_role,
        )

        return self.task_to_a2a(task)

    async def discover_agents(
        self,
        role: str | None = None,
        team: str | None = None,
        skill_tag: str | None = None,
    ) -> list[AgentCard]:
        """
        Discover agents matching criteria.

        Args:
            role: Filter by agent role
            team: Filter by team
            skill_tag: Filter by skill tag (future)

        Returns:
            List of matching AgentCards
        """
        query = select(AgentTable)

        if role:
            query = query.where(AgentTable.role == role)
        if team:
            query = query.where(AgentTable.team == team)

        result = await self.session.execute(query)
        agents = result.scalars().all()

        cards = [self._agent_to_card(agent) for agent in agents]

        # Filter by skill tag if specified
        if skill_tag:
            cards = [
                card
                for card in cards
                if any(skill_tag in skill.tags for skill in card.skills)
            ]

        return cards

    # =========================================================================
    # MESSAGE ROUTING
    # =========================================================================

    @staticmethod
    def get_team_from_agent(agent_slug: str) -> Team:
        """Get Team enum from agent slug."""
        team_str = get_agent_team(agent_slug)
        team_map = {
            "backend": Team.BACKEND,
            "frontend": Team.FRONTEND,
            "ux_ui": Team.UX_UI,
        }
        return team_map.get(team_str or "", Team.BACKEND)

    @staticmethod
    def resolve_target_agent(metadata: dict[str, Any]) -> str | None:
        """
        Resolve target agent from A2A request metadata.

        Returns agent slug or None if not specified.
        """
        # Check for explicit target
        target = metadata.get("target_agent")
        if target and target in ALL_AGENTS:
            return cast("str", target)

        # Check for skill-based routing
        skill = metadata.get("skill")
        if skill:
            for agent_slug in ALL_AGENTS:
                agent_skills = get_agent_skills(agent_slug)
                skill_ids = [s.get("id", "") for s in agent_skills]
                if skill in skill_ids:
                    return agent_slug

        return None

    # =========================================================================
    # MESSAGE HANDLING
    # =========================================================================

    @staticmethod
    def extract_message_text(message: A2AMessage) -> tuple[str, str, str]:
        """Extract title, description, and full text from message parts."""
        text_parts = [p for p in message.parts if p.type == "text"]
        if not text_parts:
            return "A2A Task", "", ""

        text_part = text_parts[0]
        if not hasattr(text_part, "text"):
            return "A2A Task", "", ""

        message_text = text_part.text
        lines = message_text.split("\n", 1)
        title = lines[0][:200]
        description = lines[1] if len(lines) > 1 else message_text
        return title, description, message_text

    @staticmethod
    def update_task_with_message(task: TaskTable, message: A2AMessage) -> None:
        """Append A2A-protocol message text to the task's A2A log (dev_notes).

        NOTE: this is the *legacy A2A-protocol* message store — A2A-protocol
        tasks carry their request/response thread in ``dev_notes`` (keyed by the
        ``"A2A Request"`` marker that ``_notify_original_requester`` checks).
        It is NOT the gateway agent flow (those use the A2AConversation tables),
        so it does not pollute normal delivery tasks' developer notes.
        """
        text_parts = [p for p in message.parts if p.type == "text"]
        if not text_parts:
            return

        text_part = text_parts[0]
        if not hasattr(text_part, "text"):
            return

        new_text = text_part.text
        task.dev_notes = (
            f"{task.dev_notes}\n\n{new_text}" if task.dev_notes else new_text
        )

    async def resolve_creator_agent(
        self, from_agent_id: str | None
    ) -> AgentTable | None:
        """Resolve the creator agent from ID or fall back to main PM."""
        if from_agent_id and from_agent_id in ALL_AGENTS:
            from_uuid = AGENT_UUIDS.get(from_agent_id)
            if from_uuid:
                result = await self.session.execute(
                    select(AgentTable).where(AgentTable.id == UUID(from_uuid))
                )
                return result.scalar_one_or_none()

        # Fall back to main PM
        result = await self.session.execute(
            select(AgentTable).where(AgentTable.role == "main_pm").limit(1)
        )
        return result.scalar_one_or_none()

    async def create_a2a_notification(
        self,
        request: SendMessageRequest,
    ) -> dict[str, Any]:
        """
        Create an A2A notification for peer-to-peer communication.

        Does NOT create tasks - A2A is messaging only.
        task_id is REQUIRED - A2A is communication about existing tasks.

        Returns dict with notification_id, status, and target_agent.
        """
        from roboco.services.notification import NotificationService

        message = request.message
        metadata = request.metadata or {}
        config = request.configuration

        # task_id is REQUIRED for A2A
        task_id = message.task_id
        if not task_id:
            raise ValueError("A2A requests must reference a task_id")

        from_agent = metadata.get("from_agent")
        target_agent = self.resolve_target_agent(metadata)
        skill = metadata.get("skill", "general")

        # Enforce A2A hierarchy permissions — UNCONDITIONALLY. The conversation
        # path (validate_a2a_access, below) requires both ends present and
        # rejects self-A2A with a typed A2AAccessDeniedError + route_hint. This
        # legacy notification path used to gate on `if from_agent and
        # target_agent:`, so an unattributed (from_agent falsy) or untargeted
        # (target unresolvable) request slipped past the hierarchy matrix and
        # dispatched with from_agent='unknown' / to_agent='' — and a hierarchy
        # denial came back as a bare ValueError indistinguishable from the
        # missing-task_id ValueError above. Require both resolved, then validate
        # via the shared typed path so both A2A surfaces enforce the same
        # who-may-talk-to-whom invariant.
        if not from_agent:
            raise ValueError(
                "A2A notification requires a 'from_agent' in metadata — an "
                "unattributed request would bypass the hierarchy gate"
            )
        if not target_agent:
            raise ValueError(
                "A2A notification could not resolve a target agent — provide an "
                "explicit 'target_agent' (a known agent slug) or a 'skill' that "
                "matches an agent's capability"
            )
        validate_a2a_access(from_agent, target_agent)
        # Priority parsing: full tristate (NORMAL/HIGH/URGENT) survives
        # end-to-end. Resolution rules live in
        # foundation.policy.communications.parse_priority.
        from roboco.foundation.policy.communications import parse_priority

        legacy_urgent = bool(
            (config and config.urgent) or metadata.get("urgent", False)
        )
        priority = parse_priority(metadata.get("priority"), legacy_urgent)

        # Extract message content
        _, _, message_text = self.extract_message_text(message)

        logger.info(
            "Creating A2A notification (fallback)",
            task_id=task_id,
            from_agent=from_agent,
            target_agent=target_agent,
            skill=skill,
            priority=priority.value,
        )

        # Create notification - orchestrator dispatcher will handle spawning
        notification_service = NotificationService()
        await notification_service.send_a2a_notification(
            task_id=task_id,
            a2a_context={
                "from_agent": from_agent or "unknown",
                "to_agent": target_agent or "",
                "skill": skill,
                "message": message_text,
                "priority": priority,
            },
        )

        return {
            "status": "sent",
            "target_agent": target_agent,
            "task_id": task_id,
        }

    async def update_task_from_message(
        self,
        task_id: str,
        message: A2AMessage,
        responder_agent: str | None = None,
    ) -> TaskTable:
        """
        Update an existing task with a new message (response).

        When a response is received, notifies the original requester
        and spawns them if offline (bidirectional A2A).

        Args:
            task_id: Task UUID string
            message: A2A message to append
            responder_agent: Agent sending the response (for routing back)

        Returns:
            Updated TaskTable

        Raises:
            ValueError: If task not found or invalid ID
        """
        try:
            task_uuid = UUID(task_id)
        except ValueError as e:
            raise ValueError(f"Invalid task ID: {task_id}") from e

        result = await self.session.execute(
            select(TaskTable).where(TaskTable.id == task_uuid)
        )
        task = result.scalar_one_or_none()

        if task is None:
            raise ValueError(f"Task not found: {task_id}")

        self.update_task_with_message(task, message)

        # Notify original requester of the response (bidirectional A2A)
        await self._notify_original_requester(task, responder_agent)

        return task

    @staticmethod
    def _lookup_requester_slug(created_by: Any) -> str | None:
        """Find the agent slug for a task creator UUID."""
        from roboco.seeds.initial_data import AGENT_UUIDS

        created_by_str = str(created_by)
        for slug, uuid_str in AGENT_UUIDS.items():
            if uuid_str == created_by_str:
                return slug
        return None

    @staticmethod
    async def _publish_a2a_response_event(
        task: TaskTable,
        created_by: Any,
        requester_slug: str,
        responder_agent: str | None,
    ) -> None:
        """Publish a TASK_ASSIGNED event to notify/spawn the requester."""
        try:
            bus = get_event_bus()
            if not bus.is_connected():
                return
            await bus.publish(
                Event(
                    type=EventType.TASK_ASSIGNED,
                    data={
                        "task_id": str(task.id),
                        "assigned_to": str(created_by),
                        "agent_slug": requester_slug,
                        "skill": "a2a_response",
                        "message": f"Response received for A2A task {task.id}",
                        "source": "a2a_response",
                        "urgent": False,
                        "from_agent": responder_agent or "agent",
                    },
                )
            )
        except Exception:
            pass  # Don't fail if event bus unavailable

    async def _notify_original_requester(
        self,
        task: TaskTable,
        responder_agent: str | None = None,
    ) -> None:
        """
        Notify the original A2A requester of a response.

        If the requester is offline, triggers spawn via event.
        This enables bidirectional A2A where both parties can be
        spawned as needed until they're both online.
        """
        dev_notes = task.dev_notes or ""
        if "A2A Request" not in dev_notes:
            return  # Not an A2A task

        created_by = task.created_by
        if not created_by:
            return

        requester_slug = self._lookup_requester_slug(created_by)
        if not requester_slug:
            return

        # Don't notify if responder is the same as requester
        if responder_agent and responder_agent == requester_slug:
            return

        await self._publish_a2a_response_event(
            task, created_by, requester_slug, responder_agent
        )

    # =========================================================================
    # PERSISTENT CONVERSATION MANAGEMENT
    # =========================================================================
    # These methods handle persistent A2A conversations stored in the database.
    # They complement the existing A2A protocol methods above.

    @staticmethod
    def _canonical_pair(agent_a: str, agent_b: str) -> tuple[str, str]:
        """Return agents in canonical order (lexically smaller first)."""
        return (agent_a, agent_b) if agent_a < agent_b else (agent_b, agent_a)

    async def get_or_create_conversation(
        self,
        agent_a: str,
        agent_b: str,
        topic: str | None = None,
        task_id: UUID | None = None,
    ) -> A2AConversation:
        """
        Get existing conversation or create new one.

        Args:
            agent_a: First agent slug
            agent_b: Second agent slug
            topic: Optional conversation topic
            task_id: Optional task to link

        Returns:
            A2AConversation model

        Raises:
            A2AAccessDeniedError: If A2A not permitted between agents
        """
        # Validate permissions
        validate_a2a_access(agent_a, agent_b)

        # Canonical ordering
        a, b = self._canonical_pair(agent_a, agent_b)

        # Try to find existing
        query = select(A2AConversationTable).where(
            A2AConversationTable.agent_a == a,
            A2AConversationTable.agent_b == b,
        )
        if topic:
            query = query.where(A2AConversationTable.topic == topic)
        else:
            query = query.where(A2AConversationTable.topic.is_(None))

        result = await self.session.execute(query)
        existing = result.scalar_one_or_none()

        if existing:
            return self._conv_to_model(existing)

        # Create new conversation
        conv = A2AConversationTable(
            agent_a=a,
            agent_b=b,
            topic=topic,
            task_id=task_id,
            status=A2AConversationStatus.ACTIVE,
        )
        self.session.add(conv)
        await self.session.flush()
        await self.session.refresh(conv)

        logger.info(
            "Created A2A conversation",
            conversation_id=str(conv.id),
            agent_a=a,
            agent_b=b,
            topic=topic,
        )

        return self._conv_to_model(conv)

    async def get_conversation(
        self,
        conversation_id: UUID,
        agent_slug: str,
    ) -> A2AConversation | None:
        """
        Get conversation by ID if agent is a participant.

        Args:
            conversation_id: Conversation UUID
            agent_slug: Agent requesting (must be participant)

        Returns:
            A2AConversation or None if not found/not authorized
        """
        result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.id == conversation_id
            )
        )
        conv = result.scalar_one_or_none()

        if conv is None:
            return None

        # Verify agent is participant
        if agent_slug not in (conv.agent_a, conv.agent_b):
            return None

        return self._conv_to_model(conv)

    async def get_conversation_admin(
        self,
        conversation_id: UUID,
    ) -> A2AConversation | None:
        """Get a conversation by ID with NO participant check.

        The CEO's live view needs to look up (and reply into) any
        conversation, including ones it is not itself a party to — unlike
        ``get_conversation``, which gates on membership.
        """
        result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.id == conversation_id
            )
        )
        conv = result.scalar_one_or_none()
        if conv is None:
            return None
        return self._conv_to_model(conv)

    async def list_conversations(
        self,
        agent_slug: str,
        status: A2AConversationStatus | None = None,
        with_agent: str | None = None,
        task_id: UUID | None = None,
        limit: int = 50,
    ) -> list[A2AConversationSummary]:
        """
        List conversations for an agent.

        Args:
            agent_slug: Agent to list for
            status: Filter by status
            with_agent: Filter by other participant
            task_id: Filter by linked task
            limit: Max results

        Returns:
            List of conversation summaries
        """
        from sqlalchemy import or_

        query = select(A2AConversationTable).where(
            or_(
                A2AConversationTable.agent_a == agent_slug,
                A2AConversationTable.agent_b == agent_slug,
            )
        )

        if status:
            query = query.where(A2AConversationTable.status == status)

        if with_agent:
            a, b = self._canonical_pair(agent_slug, with_agent)
            query = query.where(
                A2AConversationTable.agent_a == a,
                A2AConversationTable.agent_b == b,
            )

        if task_id:
            query = query.where(A2AConversationTable.task_id == task_id)

        query = query.order_by(A2AConversationTable.updated_at.desc()).limit(limit)

        result = await self.session.execute(query)
        conversations = result.scalars().all()

        summaries = []
        for conv in conversations:
            # Get last message preview
            msg_query = (
                select(A2AMessageTable)
                .where(A2AMessageTable.conversation_id == conv.id)
                .order_by(A2AMessageTable.created_at.desc())
                .limit(1)
            )
            msg_result = await self.session.execute(msg_query)
            last_msg = msg_result.scalar_one_or_none()

            other = conv.agent_b if agent_slug == conv.agent_a else conv.agent_a
            unread = (
                conv.unread_by_a if agent_slug == conv.agent_a else conv.unread_by_b
            )

            summaries.append(
                A2AConversationSummary(
                    id=str(conv.id),
                    other_agent=other,
                    topic=conv.topic,
                    task_id=str(conv.task_id) if conv.task_id else None,
                    status=conv.status,
                    message_count=conv.message_count,
                    unread_count=unread,
                    last_message_at=conv.last_message_at,
                    last_message_preview=(last_msg.content[:100] if last_msg else None),
                )
            )

        return summaries

    async def _last_message(self, conversation_id: UUID) -> A2AMessageTable | None:
        """Most recent message row in a conversation, or None."""
        result = await self.session.execute(
            select(A2AMessageTable)
            .where(A2AMessageTable.conversation_id == conversation_id)
            .order_by(A2AMessageTable.created_at.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def list_conversations_admin(
        self, limit: int = 50
    ) -> list[A2AConversationAdminSummary]:
        """ALL conversations across every agent pair, most-recent-first, bounded.

        No participant filter — this is the CEO's org-wide live view, not a
        per-agent inbox. Reuses the same last-message-preview lookup as
        ``list_conversations``.
        """
        query = (
            select(A2AConversationTable)
            .order_by(A2AConversationTable.updated_at.desc())
            .limit(limit)
        )
        result = await self.session.execute(query)
        conversations = result.scalars().all()

        summaries = []
        for conv in conversations:
            last_msg = await self._last_message(cast("UUID", conv.id))
            summaries.append(
                A2AConversationAdminSummary(
                    id=str(conv.id),
                    agent_a=conv.agent_a,
                    agent_b=conv.agent_b,
                    topic=conv.topic,
                    task_id=str(conv.task_id) if conv.task_id else None,
                    status=conv.status,
                    message_count=conv.message_count,
                    last_message_at=conv.last_message_at,
                    last_message_preview=(last_msg.content[:100] if last_msg else None),
                    created_at=conv.created_at,
                    updated_at=conv.updated_at,
                )
            )

        return summaries

    async def list_admin_pairs(self) -> list[A2AAdminPairSummary]:
        """CEO switchboard: every allowed agent pair (static matrix, see
        ``agents_config.A2A_ALLOWED_PAIRS``) joined with its representative
        conversation when one exists.

        One bulk query over the bounded static pair count — never N+1. When a
        pair has more than one conversation (distinct topics), the most
        recently updated one is treated as "the" conversation for that pair.
        """
        from sqlalchemy import tuple_

        canonical_keys = [(p.agent_a, p.agent_b) for p in A2A_ALLOWED_PAIRS]
        conv_by_pair: dict[tuple[str, str], A2AConversationTable] = {}
        if canonical_keys:
            result = await self.session.execute(
                select(A2AConversationTable).where(
                    tuple_(
                        A2AConversationTable.agent_a, A2AConversationTable.agent_b
                    ).in_(canonical_keys)
                )
            )
            for conv in result.scalars().all():
                key = (conv.agent_a, conv.agent_b)
                current = conv_by_pair.get(key)
                if current is None or conv.updated_at > current.updated_at:
                    conv_by_pair[key] = conv

        summaries: list[A2AAdminPairSummary] = []
        for p in A2A_ALLOWED_PAIRS:
            rep = conv_by_pair.get((p.agent_a, p.agent_b))
            summaries.append(
                A2AAdminPairSummary(
                    agent_a=p.agent_a,
                    role_a=p.role_a,
                    team_a=p.team_a,
                    agent_b=p.agent_b,
                    role_b=p.role_b,
                    team_b=p.team_b,
                    group_key=p.group_key,
                    conversation_id=str(rep.id) if rep else None,
                    last_message_at=rep.last_message_at if rep else None,
                    message_count=rep.message_count if rep else 0,
                )
            )
        return summaries

    async def close_conversation(
        self,
        conversation_id: UUID,
        agent_slug: str,
        resolution: str | None = None,
    ) -> None:
        """Close a conversation."""
        result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.id == conversation_id
            )
        )
        conv = result.scalar_one_or_none()

        if conv is None:
            raise ValueError(f"Conversation not found: {conversation_id}")

        if agent_slug not in (conv.agent_a, conv.agent_b):
            raise ValueError("Not a participant in this conversation")

        conv.status = A2AConversationStatus.CLOSED
        conv.resolution = resolution
        await self.session.flush()

        logger.info(
            "Closed A2A conversation",
            conversation_id=str(conversation_id),
            by_agent=agent_slug,
        )

    async def _enforce_ceo_reply_budget(
        self,
        conv: A2AConversationTable,
        conversation_id: UUID,
        from_agent: str,
    ) -> None:
        """Reply-then-wait budget on the CEO's inbox — the one stateful gate
        the stateless ``can_a2a_direct`` matrix can't see (it only blocks
        conversation *creation*, unconditionally, as defense-in-depth).

        An agent may message the CEO only inside a conversation the CEO
        itself opened, and only up to the CEO's own message count there:
        reject when the agent's message count >= the CEO's message count.
        No-op for CEO-authored sends or conversations the CEO isn't part of.
        """
        other = conv.agent_b if from_agent == conv.agent_a else conv.agent_a
        if other != "ceo" or from_agent == "ceo":
            return

        from sqlalchemy import func

        agent_count = await self.session.scalar(
            select(func.count())
            .select_from(A2AMessageTable)
            .where(
                A2AMessageTable.conversation_id == conversation_id,
                A2AMessageTable.from_agent == from_agent,
            )
        )
        ceo_count = await self.session.scalar(
            select(func.count())
            .select_from(A2AMessageTable)
            .where(
                A2AMessageTable.conversation_id == conversation_id,
                A2AMessageTable.from_agent == "ceo",
            )
        )
        if (agent_count or 0) >= (ceo_count or 0):
            raise A2AAccessDeniedError(
                from_agent=from_agent,
                to_agent="ceo",
                reason=(
                    "you have already replied to the CEO's last message — "
                    "wait for the CEO to respond before sending again"
                ),
                route_hint="Wait for the CEO to post again in this conversation.",
            )

    async def send_chat_message(
        self,
        conversation_id: UUID,
        from_agent: str,
        content: str,
        options: dict[str, Any] | None = None,
    ) -> A2AChatMessage:
        """
        Send message in conversation.

        Args:
            conversation_id: Target conversation
            from_agent: Sender slug
            content: Message content
            options: Optional dict with message_kind, response_to_id, requires_response

        Returns:
            Created A2AChatMessage

        Raises:
            ValueError: If conversation_id is nil, conversation not found,
                or sender not participant
        """
        if conversation_id.int == 0:
            raise ValueError(
                "conversation_id must not be the nil UUID; "
                "call get_or_create_conversation() first"
            )

        from datetime import UTC, datetime

        opts = options or {}
        message_kind = opts.get("message_kind", A2AMessageKind.MESSAGE)
        response_to_id = opts.get("response_to_id")
        requires_response = opts.get("requires_response", False)
        skill = opts.get("skill")

        result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.id == conversation_id
            )
        )
        conv = result.scalar_one_or_none()

        if conv is None:
            raise ValueError(f"Conversation not found: {conversation_id}")

        if from_agent not in (conv.agent_a, conv.agent_b):
            raise ValueError("Not a participant in this conversation")

        # Purpose-dedup: if an identical message from this sender is still
        # unread in this conversation, the sender is re-saying the same thing
        # (a respawn re-emitting, or a retry) — don't stack another copy on the
        # recipient's inbox or re-bump the unread count. Keyed on
        # (conversation, sender, kind, content) while unread, so genuinely
        # different messages are never collapsed.
        dup = await self.session.scalar(
            select(A2AMessageTable)
            .where(
                A2AMessageTable.conversation_id == conversation_id,
                A2AMessageTable.from_agent == from_agent,
                A2AMessageTable.message_kind == message_kind,
                A2AMessageTable.content == content,
                A2AMessageTable.read_at.is_(None),
            )
            .limit(1)
        )
        if dup is not None:
            logger.info(
                "Suppressed duplicate unread A2A message",
                conversation_id=str(conversation_id),
                from_agent=from_agent,
                existing_message_id=str(dup.id),
            )
            return self._msg_to_model(dup)

        await self._enforce_ceo_reply_budget(conv, conversation_id, from_agent)

        # Create message
        msg = A2AMessageTable(
            conversation_id=conversation_id,
            from_agent=from_agent,
            content=content,
            message_kind=message_kind,
            response_to_id=response_to_id,
            requires_response=requires_response,
            skill=skill,
        )
        self.session.add(msg)

        # Update conversation stats
        conv.message_count += 1
        conv.last_message_at = datetime.now(UTC)

        # Update unread count for the OTHER agent
        if from_agent == conv.agent_a:
            conv.unread_by_b += 1
        else:
            conv.unread_by_a += 1

        await self.session.flush()
        await self.session.refresh(msg)

        logger.info(
            "Sent A2A chat message",
            conversation_id=str(conversation_id),
            message_id=str(msg.id),
            from_agent=from_agent,
        )

        model = self._msg_to_model(msg)
        # Single chokepoint for the operator live view: every persisted A2A
        # message emits A2A_MESSAGE_SENT here, so the direct REST send paths
        # (conversation-create + post-message) light up the /a2a view too, not
        # just the gateway send() wrapper. Suppressed duplicates return above
        # and deliberately don't re-emit.
        to_agent = conv.agent_b if from_agent == conv.agent_a else conv.agent_a
        task_id = str(conv.task_id) if conv.task_id else None
        await self._publish_a2a_message_sent(
            model, task_id, from_agent, to_agent, skill
        )
        return model

    async def get_messages(
        self,
        conversation_id: UUID,
        agent_slug: str,
        limit: int = 100,
        before: datetime | None = None,
    ) -> list[A2AChatMessage]:
        """Get messages in conversation."""
        # Verify access
        conv_result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.id == conversation_id
            )
        )
        conv = conv_result.scalar_one_or_none()

        if conv is None:
            return []

        if agent_slug not in (conv.agent_a, conv.agent_b):
            return []

        query = (
            select(A2AMessageTable)
            .where(A2AMessageTable.conversation_id == conversation_id)
            .order_by(A2AMessageTable.created_at.desc())
            .limit(limit)
        )

        if before:
            query = query.where(A2AMessageTable.created_at < before)

        result = await self.session.execute(query)
        messages = result.scalars().all()

        # Return in chronological order
        return [self._msg_to_model(m) for m in reversed(list(messages))]

    async def get_messages_admin(
        self,
        conversation_id: UUID,
        limit: int = 100,
        before: datetime | None = None,
    ) -> list[A2AChatMessage]:
        """Like ``get_messages`` but WITHOUT the participant check — the CEO
        can read any conversation's transcript for the live view. Returns
        ``[]`` only when the conversation truly doesn't exist.
        """
        conv_result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.id == conversation_id
            )
        )
        conv = conv_result.scalar_one_or_none()

        if conv is None:
            return []

        query = (
            select(A2AMessageTable)
            .where(A2AMessageTable.conversation_id == conversation_id)
            .order_by(A2AMessageTable.created_at.desc())
            .limit(limit)
        )

        if before:
            query = query.where(A2AMessageTable.created_at < before)

        result = await self.session.execute(query)
        messages = result.scalars().all()

        return [self._msg_to_model(m) for m in reversed(list(messages))]

    async def mark_read(
        self,
        conversation_id: UUID,
        agent_slug: str,
    ) -> None:
        """Mark all unread incoming messages in conversation as read by agent.

        Collect-then-mark: only the unread rows seen at call time are stamped,
        so a message arriving mid-call stays unread rather than being silently
        consumed. The unread counter is recomputed from the DB after the stamp.
        """
        from datetime import UTC, datetime

        from sqlalchemy import update

        result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.id == conversation_id
            )
        )
        conv = result.scalar_one_or_none()

        if conv is None:
            return

        if agent_slug not in (conv.agent_a, conv.agent_b):
            return

        rows = await self.session.execute(
            select(A2AMessageTable.id).where(
                A2AMessageTable.conversation_id == conversation_id,
                A2AMessageTable.from_agent != agent_slug,
                A2AMessageTable.read_at.is_(None),
            )
        )
        msg_ids = [r for (r,) in rows.all()]
        if msg_ids:
            await self.session.execute(
                update(A2AMessageTable)
                .where(A2AMessageTable.id.in_(msg_ids))
                .values(read_at=datetime.now(UTC))
            )
        await self._reset_unread_counter(conversation_id, agent_slug)
        await self.session.flush()

    async def mark_all_read(self, agent_id: UUID) -> int:
        """Mark every conversation with unread-for-this-agent as read.

        Agent-keyed bulk form of ``mark_read``: stamps ``read_at`` on the
        inbound messages across all its conversations and recomputes each
        counter from the DB. Returns the number cleared. Lets an agent satisfy
        ``i_am_idle``'s unread-A2A soft-block in one call.

        Collect-then-mark (mirrors ``get_unread_messages``): only the unread
        rows seen at call time are stamped, so a message arriving mid-call
        stays unread rather than being silently consumed.
        """
        from datetime import UTC, datetime

        from sqlalchemy import or_, update

        slug = await self._resolve_slug_from_id(agent_id)
        result = await self.session.execute(
            select(A2AConversationTable).where(
                or_(
                    (A2AConversationTable.agent_a == slug)
                    & (A2AConversationTable.unread_by_a > 0),
                    (A2AConversationTable.agent_b == slug)
                    & (A2AConversationTable.unread_by_b > 0),
                )
            )
        )
        convs = list(result.scalars().all())
        if not convs:
            return 0
        conv_ids = [c.id for c in convs]
        rows = await self.session.execute(
            select(A2AMessageTable.id).where(
                A2AMessageTable.conversation_id.in_(conv_ids),
                A2AMessageTable.from_agent != slug,
                A2AMessageTable.read_at.is_(None),
            )
        )
        msg_ids = [r for (r,) in rows.all()]
        if msg_ids:
            await self.session.execute(
                update(A2AMessageTable)
                .where(A2AMessageTable.id.in_(msg_ids))
                .values(read_at=datetime.now(UTC))
            )
        for cid in conv_ids:
            await self._reset_unread_counter(cast("UUID", cid), slug)
        await self.session.flush()
        return len(convs)

    async def _reset_unread_counter(self, conversation_id: UUID, slug: str) -> None:
        """Recompute a conversation's unread-for-``slug`` counter from the rows
        still unread, so a message arriving mid-drain is preserved, not zeroed."""
        from sqlalchemy import func

        conv = await self.session.get(A2AConversationTable, conversation_id)
        if conv is None:
            return
        remaining = await self.session.scalar(
            select(func.count())
            .select_from(A2AMessageTable)
            .where(
                A2AMessageTable.conversation_id == conversation_id,
                A2AMessageTable.from_agent != slug,
                A2AMessageTable.read_at.is_(None),
            )
        )
        if conv.agent_a == slug:
            conv.unread_by_a = remaining or 0
        else:
            conv.unread_by_b = remaining or 0

    async def get_unread_messages(self, agent_id: UUID) -> list[dict[str, Any]]:
        """Return the agent's unread INCOMING A2A messages and mark them read.

        Delivers the actual message bodies (not just counts) so the agent can
        reason about what was said to it. Only inbound messages (``from_agent``
        != caller) are returned — the agent's own sends are never echoed back.
        Collect-then-mark is atomic within the session: only the exact rows
        returned are stamped read, so a message arriving mid-call stays unread
        rather than being silently cleared.
        """
        from datetime import UTC, datetime

        from sqlalchemy import or_, update

        slug = await self._resolve_slug_from_id(agent_id)
        conv_ids = select(A2AConversationTable.id).where(
            or_(
                A2AConversationTable.agent_a == slug,
                A2AConversationTable.agent_b == slug,
            )
        )
        result = await self.session.execute(
            select(A2AMessageTable)
            .where(
                A2AMessageTable.conversation_id.in_(conv_ids),
                A2AMessageTable.from_agent != slug,
                A2AMessageTable.read_at.is_(None),
            )
            .order_by(A2AMessageTable.created_at.asc())
        )
        msgs = list(result.scalars().all())
        if not msgs:
            return []

        await self.session.execute(
            update(A2AMessageTable)
            .where(A2AMessageTable.id.in_([m.id for m in msgs]))
            .values(read_at=datetime.now(UTC))
        )
        # Recompute each affected conversation's unread counter (see helper) —
        # a message that arrived mid-call is preserved, not zeroed.
        for cid in {cast("UUID", m.conversation_id) for m in msgs}:
            await self._reset_unread_counter(cid, slug)
        await self.session.flush()

        return [
            {
                "conversation_id": str(m.conversation_id),
                "from_agent": m.from_agent,
                "content": m.content,
                "created_at": m.created_at.isoformat() if m.created_at else None,
            }
            for m in msgs
        ]

    async def get_inbox_summary(self, agent_slug: str) -> A2AInboxSummary:
        """Get summary of pending A2A for agent."""
        from sqlalchemy import func, or_

        # Get conversations with unread
        conv_query = select(A2AConversationTable).where(
            or_(
                A2AConversationTable.agent_a == agent_slug,
                A2AConversationTable.agent_b == agent_slug,
            )
        )
        conv_result = await self.session.execute(conv_query)
        conversations = conv_result.scalars().all()

        total_unread = 0
        conversations_with_unread = 0

        for conv in conversations:
            unread = (
                conv.unread_by_a if agent_slug == conv.agent_a else conv.unread_by_b
            )
            if unread > 0:
                conversations_with_unread += 1
                total_unread += unread

        # Count pending responses (messages I sent that require response)
        pending_query = (
            select(func.count())
            .select_from(A2AMessageTable)
            .where(
                A2AMessageTable.from_agent == agent_slug,
                A2AMessageTable.requires_response.is_(True),
            )
        )
        pending_result = await self.session.execute(pending_query)
        pending_responses = pending_result.scalar() or 0

        # Count unanswered requests (messages to me that require response)
        unanswered_query = (
            select(func.count())
            .select_from(A2AMessageTable)
            .join(A2AConversationTable)
            .where(
                or_(
                    A2AConversationTable.agent_a == agent_slug,
                    A2AConversationTable.agent_b == agent_slug,
                ),
                A2AMessageTable.from_agent != agent_slug,
                A2AMessageTable.requires_response.is_(True),
            )
        )
        unanswered_result = await self.session.execute(unanswered_query)
        unanswered_requests = unanswered_result.scalar() or 0

        return A2AInboxSummary(
            total_unread=total_unread,
            conversations_with_unread=conversations_with_unread,
            pending_responses=pending_responses,
            unanswered_requests=unanswered_requests,
        )

    async def list_pairs(self, agent_slug: str) -> list[A2APair]:
        """List unique agent pairs for frontend display."""
        from sqlalchemy import or_

        query = (
            select(A2AConversationTable)
            .where(
                or_(
                    A2AConversationTable.agent_a == agent_slug,
                    A2AConversationTable.agent_b == agent_slug,
                )
            )
            .order_by(A2AConversationTable.updated_at.desc())
        )

        result = await self.session.execute(query)
        conversations = result.scalars().all()

        # Group by pair
        pairs: dict[tuple[str, str], A2APair] = {}
        for conv in conversations:
            pair_key = (conv.agent_a, conv.agent_b)
            if pair_key not in pairs:
                pairs[pair_key] = A2APair(
                    agent_a=conv.agent_a,
                    agent_b=conv.agent_b,
                    conversation_count=0,
                    total_unread=0,
                    last_activity=None,
                )

            pairs[pair_key].conversation_count += 1

            unread = (
                conv.unread_by_a if agent_slug == conv.agent_a else conv.unread_by_b
            )
            pairs[pair_key].total_unread += unread

            current_activity = pairs[pair_key].last_activity
            if current_activity is None or (
                conv.updated_at is not None and conv.updated_at > current_activity
            ):
                pairs[pair_key].last_activity = conv.updated_at

        return list(pairs.values())

    # =========================================================================
    # MODEL CONVERSIONS
    # =========================================================================

    def _conv_to_model(self, conv: A2AConversationTable) -> A2AConversation:
        """Convert table row to Pydantic model."""
        return A2AConversation(
            id=str(conv.id),
            agent_a=conv.agent_a,
            agent_b=conv.agent_b,
            topic=conv.topic,
            task_id=str(conv.task_id) if conv.task_id else None,
            status=conv.status,
            resolution=conv.resolution,
            message_count=conv.message_count,
            unread_by_a=conv.unread_by_a,
            unread_by_b=conv.unread_by_b,
            created_at=conv.created_at,
            updated_at=conv.updated_at,
            last_message_at=conv.last_message_at,
        )

    def _msg_to_model(self, msg: A2AMessageTable) -> A2AChatMessage:
        """Convert table row to Pydantic model."""
        return A2AChatMessage(
            id=str(msg.id),
            conversation_id=str(msg.conversation_id),
            from_agent=msg.from_agent,
            content=msg.content,
            message_kind=msg.message_kind,
            skill=msg.skill,
            response_to_id=str(msg.response_to_id) if msg.response_to_id else None,
            requires_response=msg.requires_response,
            read_at=msg.read_at,
            created_at=msg.created_at,
            edited_at=msg.edited_at,
            edit_history=msg.edit_history or [],
        )

    # =========================================================================
    # GATEWAY (CHOREOGRAPHER + CONTENT_ACTIONS) BACKFILL
    # =========================================================================

    async def _resolve_slug_from_id(self, agent_id: UUID) -> str:
        """Look up an agent's slug from its UUID; raise ValueError if missing."""
        result = await self.session.execute(
            select(AgentTable.slug).where(AgentTable.id == agent_id)
        )
        slug = result.scalar_one_or_none()
        if not slug:
            raise ValueError(f"Agent not found for id {agent_id}")
        return str(slug)

    async def _get_conversation_for_reply_to_ceo(
        self, from_slug: str, to_slug: str
    ) -> A2AConversation:
        """Resolve the conversation for an agent replying to the CEO.

        Agents can never CREATE a CEO conversation (the matrix blocks
        initiation unconditionally), so an existing pair conversation's mere
        presence proves the CEO opened it. Looked up directly here —
        bypassing ``get_or_create_conversation``'s validate-first gate,
        which would otherwise deny even a legitimate reply.
        """
        a, b = self._canonical_pair(from_slug, to_slug)
        result = await self.session.execute(
            select(A2AConversationTable).where(
                A2AConversationTable.agent_a == a,
                A2AConversationTable.agent_b == b,
                A2AConversationTable.topic.is_(None),
            )
        )
        conv = result.scalar_one_or_none()
        if conv is None:
            raise A2AAccessDeniedError(
                from_agent=from_slug,
                to_agent=to_slug,
                reason=(
                    "CEO is human. You may only reply inside a conversation "
                    "the CEO opened — use notify() otherwise."
                ),
                route_hint="Wait for the CEO to open an A2A conversation with you.",
            )
        return self._conv_to_model(conv)

    async def send(
        self,
        *,
        from_agent: UUID,
        to_agent: UUID | str,
        task_id: UUID,
        body: str,
        skill: str | None = None,
    ) -> A2AChatMessage:
        """Gateway adapter — send a directed A2A message between two agents.

        Recipient may be either a UUID (choreographer call shape) or a
        slug string (content_actions call shape). The sender is always a
        UUID; both ends are resolved to slugs because the
        conversation/message tables key on slug.

        Resolves to:
        1. `get_or_create_conversation(sender_slug, recipient_slug, task_id=...)`
        2. `send_chat_message(conversation.id, sender_slug, content=body, ...)`

        `skill` is persisted on the message row so the receiver (and the
        inbox) learns which capability is being requested.

        The recipient "ceo" is special-cased: an agent can never CREATE a
        CEO conversation (the matrix blocks it unconditionally), so calling
        ``get_or_create_conversation`` would deny even a legitimate reply.
        Instead the existing pair conversation is looked up directly — its
        mere existence proves the CEO opened it — and the reply proceeds to
        ``send_chat_message``, where the reply budget applies.
        """
        from_slug = await self._resolve_slug_from_id(from_agent)
        to_slug = (
            await self._resolve_slug_from_id(to_agent)
            if isinstance(to_agent, UUID)
            else to_agent
        )

        if to_slug == "ceo" and from_slug != "ceo":
            conv = await self._get_conversation_for_reply_to_ceo(from_slug, to_slug)
        else:
            conv = await self.get_or_create_conversation(
                agent_a=from_slug,
                agent_b=to_slug,
                task_id=task_id,
            )
        options: dict[str, Any] = {}
        if skill is not None:
            options["skill"] = skill
        msg = await self.send_chat_message(
            conversation_id=UUID(conv.id),
            from_agent=from_slug,
            content=body,
            options=options or None,
        )
        return msg

    async def interject_as_ceo(
        self,
        conversation_id: UUID,
        to_agent: str,
        content: str,
        skill: str | None = None,
    ) -> A2AChatMessage:
        """CEO interjection: post a message directly into an existing
        agent<->agent conversation, addressed to one of its participants.

        One-directional and NOT a participant send: unlike
        ``send_chat_message`` (which requires the sender to be a party to
        the conversation), the CEO here is watching and interjecting into
        someone else's thread, not conversing in its own — so the
        participant check on the sender is deliberately bypassed rather
        than weakened for every other caller. ``to_agent`` still must be
        one of the conversation's two real participants.

        Only ``to_agent``'s unread counter is bumped (a ping to whoever
        it's addressed to); the other participant still sees the row via
        the shared transcript / ``read_a2a``, just without a ping.

        Direction is encoded as an ``@{to_agent}: `` content prefix —
        ponytail: no ``to_agent`` column yet; add one (and stop parsing
        the prefix) if the panel ever needs to render/filter by recipient
        directly instead.
        """
        conv = await self.session.get(A2AConversationTable, conversation_id)
        if conv is None:
            raise ValueError(f"Conversation not found: {conversation_id}")
        if to_agent not in (conv.agent_a, conv.agent_b):
            raise ValueError(
                f"{to_agent} is not a participant in this conversation "
                f"(participants: {conv.agent_a}, {conv.agent_b})"
            )

        msg = A2AMessageTable(
            conversation_id=conversation_id,
            from_agent="ceo",
            content=f"@{to_agent}: {content}",
            message_kind=A2AMessageKind.MESSAGE,
            skill=skill,
        )
        self.session.add(msg)

        conv.message_count += 1
        conv.last_message_at = datetime.now(UTC)
        if to_agent == conv.agent_a:
            conv.unread_by_a += 1
        else:
            conv.unread_by_b += 1

        await self.session.flush()
        await self.session.refresh(msg)

        model = self._msg_to_model(msg)
        task_id = str(conv.task_id) if conv.task_id else None
        await self._publish_a2a_message_sent(model, task_id, "ceo", to_agent, skill)
        return model

    @staticmethod
    async def _publish_a2a_message_sent(
        msg: A2AChatMessage,
        task_id: str | None,
        from_slug: str,
        to_slug: str,
        skill: str | None,
    ) -> None:
        """Best-effort publish of A2A_MESSAGE_SENT for the operator live view.

        A bus outage is logged and never rolls back the already-persisted
        message.
        """
        try:
            bus = get_event_bus()
            if bus.is_connected():
                timestamp = (
                    msg.created_at.isoformat()
                    if msg.created_at
                    else datetime.now(UTC).isoformat()
                )
                await bus.publish(
                    Event(
                        type=EventType.A2A_MESSAGE_SENT,
                        data={
                            "conversation_id": msg.conversation_id,
                            "message_id": msg.id,
                            "task_id": task_id,
                            "from_agent": from_slug,
                            "to_agent": to_slug,
                            "skill": skill,
                            "body_excerpt": _excerpt(msg.content),
                            "timestamp": timestamp,
                        },
                    )
                )
        except Exception as e:
            logger.warning("Failed to publish A2A message event", error=str(e))