roboco/tests/integration/test_notification_delivery_phantom.py

"""F107 — Redis bus publish must be deferred until the DB commit lands.

`NotificationDeliveryService.deliver` historically published
``NOTIFICATION_SENT`` to the Redis event bus *before* the caller committed
the notification row. A commit failure (DB hiccup, constraint, asyncpg error)
rolled the row back but left the bus event behind — connected WebSocket
clients received a push for an id that no longer existed (a phantom
notification). The fix defers the bus publish to the session's
``after_commit`` so a rollback drops it; the row is durable by the time the
event fires.

These tests need a real ``AsyncSession`` (the deferral uses SQLAlchemy
session commit/rollback events) plus a recording bus stand-in, so they are
integration tests against the migrated Postgres test DB.
"""

from __future__ import annotations

import asyncio
from typing import TYPE_CHECKING
from uuid import UUID, uuid4

import pytest
from roboco.db.tables import AgentTable, NotificationTable
from roboco.events import Event, EventType
from roboco.models import AgentRole, AgentStatus, NotificationPriority, NotificationType
from roboco.models.base import Team
from roboco.services.notification_delivery import get_notification_delivery_service

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession


class _RecordingBus:
    """Stand-in for StreamEventBus that records every published event.

    Mirrors the real bus surface used by ``deliver``: ``is_connected()``
    gates the publish path and ``publish`` is async. Recording lets the
    tests assert exactly when (and whether) the NOTIFICATION_SENT event
    fired — without a Redis stack.
    """

    def __init__(self) -> None:
        self.published: list[Event] = []

    def is_connected(self) -> bool:
        return True

    async def publish(self, event: Event) -> str:
        self.published.append(event)
        return "recorded"


def _drain_tasks(session: AsyncSession) -> list[asyncio.Task[object]]:
    """Pending deferred-publish drain tasks stashed on the session.

    The deferral helper stores the ``asyncio.create_task`` handles here so a
    test can await them deterministically instead of racing the event loop.
    """
    return list(session.info.get("_roboco_drain_tasks", []))


async def _await_drain(session: AsyncSession) -> None:
    """Wait for any scheduled deferred-publish tasks to finish."""
    tasks = _drain_tasks(session)
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)


async def _seed_agents_and_notification(
    db: AsyncSession, *, recipients: int
) -> tuple[UUID, NotificationTable]:
    """Create a sender + N recipient agents and one flushed (uncommitted)
    notification addressed to them. Returns ``(notification_id, row)``.

    Flushed only — the row lives in the session's open transaction, matching
    the real pre-commit state ``deliver`` runs against.
    """
    sender = AgentTable(
        id=uuid4(),
        name="Sender",
        slug=f"sender-{uuid4().hex[:8]}",
        role=AgentRole.DEVELOPER,
        team=Team.BACKEND,
        status=AgentStatus.ACTIVE,
        model_config={},
        system_prompt="sender",
        capabilities=[],
        permissions={},
        metrics={},
    )
    db.add(sender)
    await db.flush()

    recipient_ids: list[UUID] = []
    for i in range(recipients):
        r = AgentTable(
            id=uuid4(),
            name=f"Recipient {i}",
            slug=f"recipient-{i}-{uuid4().hex[:8]}",
            role=AgentRole.QA,
            team=Team.BACKEND,
            status=AgentStatus.ACTIVE,
            model_config={},
            system_prompt="recipient",
            capabilities=[],
            permissions={},
            metrics={},
        )
        db.add(r)
        recipient_ids.append(r.id)
    await db.flush()

    notification = NotificationTable(
        type=NotificationType.REVIEW_REQUEST,
        priority=NotificationPriority.NORMAL,
        from_agent=sender.id,
        to_agents=recipient_ids,
        subject="Please review",
        body="Body text",
        requires_ack=True,
    )
    db.add(notification)
    await db.flush()
    return notification.id, notification


@pytest.mark.asyncio
async def test_deliver_does_not_publish_before_commit(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The bus event must NOT fire until the session commits (F107).

    Currently RED: ``deliver`` publishes immediately, so the bus is non-empty
    before any commit — the phantom window. With the deferred-publish fix,
    ``deliver`` only schedules; the event fires on commit.
    """
    bus = _RecordingBus()
    monkeypatch.setattr(
        "roboco.services.notification_delivery.get_event_bus", lambda: bus
    )

    notif_id, _ = await _seed_agents_and_notification(db_session, recipients=2)
    service = get_notification_delivery_service(db_session)
    await service.deliver(notif_id)

    # Pre-commit: nothing published yet (the row is not durable).
    assert bus.published == []


@pytest.mark.asyncio
async def test_deliver_publishes_after_commit(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Commit drains the deferred publish — one event per recipient (F107)."""
    bus = _RecordingBus()
    monkeypatch.setattr(
        "roboco.services.notification_delivery.get_event_bus", lambda: bus
    )

    recipient_count = 2
    notif_id, _ = await _seed_agents_and_notification(
        db_session, recipients=recipient_count
    )
    service = get_notification_delivery_service(db_session)
    await service.deliver(notif_id)
    assert bus.published == []  # still nothing before commit

    await db_session.commit()
    await _await_drain(db_session)

    assert len(bus.published) == recipient_count
    assert all(ev.type == EventType.NOTIFICATION_SENT for ev in bus.published)
    assert all(ev.data["notification_id"] == str(notif_id) for ev in bus.published)


@pytest.mark.asyncio
async def test_deliver_rollback_drops_phantom(
    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A rollback instead of commit drops the pending publish — no phantom
    event for a row that never became durable (F107)."""
    bus = _RecordingBus()
    monkeypatch.setattr(
        "roboco.services.notification_delivery.get_event_bus", lambda: bus
    )

    notif_id, _ = await _seed_agents_and_notification(db_session, recipients=1)
    service = get_notification_delivery_service(db_session)
    await service.deliver(notif_id)

    await db_session.rollback()
    await _await_drain(db_session)

    assert bus.published == []