"""Tests for MAF roundtable V2 Conversation SSE streaming (Phase 2 / R2 + P0 fixes).

Covers:
- ConversationEventHub: generation counter, publish/subscribe, multi-subscriber.
- P0-1: race-free replay — events published during replay are not lost.
- P0-2: terminal events auto-close the stream.
- P0-4: projection dedup via source_event_id.
- SSE replay: events persisted before connect are delivered.
- after_seq cursor: reconnect resumes without duplicates.
- Heartbeat: idle connection produces ``: heartbeat`` frames.
- Concurrent subscribers: fan-out to multiple listeners.
- Ownership: non-owner gets 404 on the stream endpoint.
"""

from __future__ import annotations

import asyncio
import json

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.maf_roundtable.conversation_event_hub import ConversationEventHub
from app.maf_roundtable.conversation_service import (
    ConversationNotFoundError,
    MafConversationService,
)
from deerflow.persistence.base import Base
from deerflow.persistence.maf_roundtable import make_maf_roundtable_store
from deerflow.persistence.maf_roundtable.model import (
    MafRoundtableConversationRow,
)
from deerflow.persistence.maf_roundtable.store import MafRoundtableStore


class FakeMafConfig:
    enabled = True
    default_mode = "structured"
    default_model_name = "test-model"
    default_reasoning_effort = None
    default_manager_agent_id = "mgr-agent"
    default_max_rounds = 8
    default_write_policy = "confirm_high_risk"
    admin_diagnostic_access = True


@pytest_asyncio.fixture
async def store(tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'stream.db'}")
    async with engine.begin() as conn:
        import deerflow.persistence.maf_roundtable.model  # noqa: F401

        await conn.run_sync(Base.metadata.create_all)
    sf = async_sessionmaker(engine, expire_on_commit=False)
    yield make_maf_roundtable_store(sf)
    await engine.dispose()


@pytest_asyncio.fixture
async def hub():
    h = ConversationEventHub(heartbeat_interval=0.3)
    yield h
    h.shutdown()


@pytest_asyncio.fixture
async def svc(store, hub):
    return MafConversationService(store, maf_config=FakeMafConfig(), event_hub=hub)


# ── ConversationEventHub unit tests ───────────────────────────────────────


@pytest.mark.asyncio
async def test_hub_publish_notifies_subscriber(hub):
    sub = hub.open_subscription("c1")
    assert not sub.has_pending

    await hub.publish("c1")
    assert sub.has_pending

    changed = await sub.wait_for_change(timeout=2.0)
    assert changed
    sub.close()


@pytest.mark.asyncio
async def test_hub_heartbeat_on_idle(hub):
    sub = hub.open_subscription("c2")
    changed = await sub.wait_for_change(timeout=0.2)
    assert not changed
    sub.close()


@pytest.mark.asyncio
async def test_hub_multi_subscriber_fanout(hub):
    sub_a = hub.open_subscription("c3")
    sub_b = hub.open_subscription("c3")

    await hub.publish("c3")

    assert await sub_a.wait_for_change(timeout=2.0)
    assert await sub_b.wait_for_change(timeout=2.0)

    sub_a.close()
    sub_b.close()
    assert hub.conversation_subscriber_count("c3") == 0


@pytest.mark.asyncio
async def test_hub_publish_no_subscribers_is_noop(hub):
    await hub.publish("c_noone")
    assert hub.conversation_subscriber_count("c_noone") == 0


@pytest.mark.asyncio
async def test_hub_unsubscribe_cleans_up(hub):
    sub = hub.open_subscription("c4")
    assert hub.conversation_subscriber_count("c4") == 1
    sub.close()
    assert hub.conversation_subscriber_count("c4") == 0


@pytest.mark.asyncio
async def test_hub_shutdown_stops_subscribers():
    h = ConversationEventHub(heartbeat_interval=0.2)
    sub = h.open_subscription("c_shutdown")
    h.shutdown()
    assert h.is_closed
    # After shutdown, generation should have advanced.
    assert sub.has_pending or sub.current_generation > sub.seen_generation
    sub.close()


# ── P0-1: Race-free replay ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_p0_1_event_during_replay_is_not_lost(svc, store, hub):
    """Event published between list_events() and subscription must be caught
    by the stabilisation loop."""
    conv = await svc.create_conversation(user_id="u1", title="Race")
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="msg1")
    await _reset_conv_to_idle(store, conv["id"])

    # Simulate: subscribe, replay, then a publish sneaks in before stabilise.
    sub = hub.open_subscription(conv["id"])
    last_seq = 0

    # Phase 1: initial replay
    events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
    for evt in events:
        last_seq = max(last_seq, evt["seq"])

    # Event sneaks in here (between replay and stabilise)
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="sneaky")
    await hub.publish(conv["id"])

    # Stabilise: should detect the pending generation
    assert sub.has_pending
    sub.mark_seen()
    late_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
    assert len(late_events) >= 1  # The sneaky event was caught
    for evt in late_events:
        last_seq = max(last_seq, evt["seq"])

    sub.close()


@pytest.mark.asyncio
async def test_p0_1_subscribe_before_replay_catches_concurrent_publish(svc, store, hub):
    """Subscribe must be opened BEFORE any DB query so a publish during replay
    is detected via generation delta."""
    conv = await svc.create_conversation(user_id="u1", title="SubBefore")

    sub = hub.open_subscription(conv["id"])

    # No events yet
    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    assert len(events) == 0

    # Publish arrives
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")
    await hub.publish(conv["id"])

    # Stabilisation detects it
    assert sub.has_pending
    sub.mark_seen()
    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    assert len(events) >= 1

    sub.close()


# ── P0-2: Terminal events auto-close stream ──────────────────────────────


@pytest.mark.asyncio
async def test_p0_2_replay_terminal_event_closes_stream():
    from app.gateway.routers.maf_conversations import _TERMINAL_EVENT_TYPES, _is_terminal

    assert "run.completed" in _TERMINAL_EVENT_TYPES
    assert "run.error" in _TERMINAL_EVENT_TYPES
    assert "run.cancelled" in _TERMINAL_EVENT_TYPES

    assert _is_terminal({"type": "run.completed"})
    assert _is_terminal({"type": "run.error"})
    assert _is_terminal({"type": "run.cancelled"})
    assert not _is_terminal({"type": "agent.message.delta"})
    assert not _is_terminal({"type": "user.message.created"})


@pytest.mark.asyncio
async def test_p0_2_terminal_event_in_replay_stops_generator(store):
    """If the replay phase hits a terminal event, the generator should stop."""
    from app.gateway.routers.maf_conversations import _is_terminal

    # Simulate a conversation with events ending in run.completed
    events = [
        {"seq": 1, "type": "user.message.created"},
        {"seq": 2, "type": "agent.message.delta"},
        {"seq": 3, "type": "run.completed"},
        {"seq": 4, "type": "should_not_reach"},
    ]

    found_terminal = False
    for evt in events:
        if _is_terminal(evt):
            found_terminal = True
            break
    assert found_terminal
    # Verify the event after terminal would not be reached
    terminal_idx = next(i for i, e in enumerate(events) if _is_terminal(e))
    assert events[terminal_idx]["type"] == "run.completed"


# ── P0-4: Projection dedup ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_p0_4_dedup_same_source_event_id(svc, store):
    """Projecting the same source_event_id twice should not create a duplicate."""
    conv = await svc.create_conversation(user_id="u1", title="Dedup")

    result1 = await svc._publish_conversation_event(
        conversation_id=conv["id"],
        owner_user_id="u1",
        event_type="tool.started",
        event_data={"tool_name": "search"},
        source_event_id="evt_src_001",
    )
    assert result1 is True

    result2 = await svc._publish_conversation_event(
        conversation_id=conv["id"],
        owner_user_id="u1",
        event_type="tool.started",
        event_data={"tool_name": "search"},
        source_event_id="evt_src_001",
    )
    assert result2 is False  # Deduped

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    tool_events = [e for e in events if e["type"] == "tool.started"]
    assert len(tool_events) == 1


@pytest.mark.asyncio
async def test_p0_4_different_source_event_ids_both_written(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="MultiSrc")

    await svc._publish_conversation_event(
        conversation_id=conv["id"], owner_user_id="u1",
        event_type="tool.started", event_data={"tool_name": "a"},
        source_event_id="src_1",
    )
    await svc._publish_conversation_event(
        conversation_id=conv["id"], owner_user_id="u1",
        event_type="tool.started", event_data={"tool_name": "b"},
        source_event_id="src_2",
    )

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    tool_events = [e for e in events if e["type"] == "tool.started"]
    assert len(tool_events) == 2


@pytest.mark.asyncio
async def test_p0_4_null_source_event_id_not_deduped(svc, store):
    """Events with no source_event_id should never be deduped."""
    conv = await svc.create_conversation(user_id="u1", title="NoSrc")

    await svc._publish_conversation_event(
        conversation_id=conv["id"], owner_user_id="u1",
        event_type="manager.dispatch", event_data={"reason": "x"},
        source_event_id=None,
    )
    await svc._publish_conversation_event(
        conversation_id=conv["id"], owner_user_id="u1",
        event_type="manager.dispatch", event_data={"reason": "y"},
        source_event_id=None,
    )

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    assert len(events) == 2


# ── SSE replay + cursor + real-time integration ──────────────────────────


async def _reset_conv_to_idle(store: MafRoundtableStore, conv_id: str):
    async with store.session_factory() as session:
        row = await session.get(MafRoundtableConversationRow, conv_id)
        if row:
            row.safe_status = "idle"
            row.active_run_id = None
        await session.commit()


@pytest.mark.asyncio
async def test_sse_replay_delivers_past_events(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="Replay")
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="msg1")
    await _reset_conv_to_idle(store, conv["id"])
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="msg2")

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    assert len(events) >= 2
    types = [e["type"] for e in events]
    assert types.count("user.message.created") == 2
    seqs = [e["seq"] for e in events]
    assert seqs == sorted(seqs)


@pytest.mark.asyncio
async def test_sse_after_seq_prevents_duplicates(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="Cursor")
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="first")
    await _reset_conv_to_idle(store, conv["id"])
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="second")

    all_events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    last_seq = all_events[0]["seq"]

    replay = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
    assert all(e["seq"] > last_seq for e in replay)
    assert len(replay) == len(all_events) - 1


@pytest.mark.asyncio
async def test_real_time_event_pushed_after_subscribe(svc, store, hub):
    conv = await svc.create_conversation(user_id="u1", title="RT")

    received_events: list[dict] = []

    async def stream_consumer():
        sub = hub.open_subscription(conv["id"])
        last_seq = 0

        # Stabilise (no events yet)
        while sub.has_pending:
            sub.mark_seen()

        # Wait for change
        changed = await sub.wait_for_change(timeout=5.0)
        if changed:
            new_evts = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
            for evt in new_evts:
                received_events.append(evt)
                last_seq = max(last_seq, evt["seq"])
        sub.close()

    task = asyncio.create_task(stream_consumer())
    await asyncio.sleep(0.1)

    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="live msg")

    await asyncio.wait_for(task, timeout=5.0)
    types = [e["type"] for e in received_events]
    assert "user.message.created" in types


@pytest.mark.asyncio
async def test_concurrent_subscribers_both_receive(svc, store, hub):
    conv = await svc.create_conversation(user_id="u1", title="Fanout")

    buf_a: list[dict] = []
    buf_b: list[dict] = []

    async def consume(buf):
        sub = hub.open_subscription(conv["id"])
        last_seq = 0
        changed = await sub.wait_for_change(timeout=5.0)
        if changed:
            new_evts = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
            for evt in new_evts:
                buf.append(evt)
                last_seq = max(last_seq, evt["seq"])
        sub.close()

    ta = asyncio.create_task(consume(buf_a))
    tb = asyncio.create_task(consume(buf_b))
    await asyncio.sleep(0.15)

    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="broadcast")

    await asyncio.wait_for(asyncio.gather(ta, tb), timeout=5.0)
    assert len(buf_a) >= 1
    assert len(buf_b) >= 1
    assert buf_a[0]["seq"] == buf_b[0]["seq"]


@pytest.mark.asyncio
async def test_reconnect_after_seq_resume(svc, store):
    conv = await svc.create_conversation(user_id="u1", title="Reconnect")
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="batch1")
    await _reset_conv_to_idle(store, conv["id"])
    await svc.send_message(conversation_id=conv["id"], user_id="u1", content="batch2")

    all_events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    midpoint = all_events[len(all_events) // 2]["seq"]

    replay = await store.conversation_events.list_events(conv["id"], after_seq=midpoint)
    assert all(e["seq"] > midpoint for e in replay)


# ── SSE frame format ──────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_sse_frame_format():
    from app.gateway.routers.maf_conversations import _format_heartbeat, _format_sse_event

    frame = _format_sse_event({
        "seq": 42,
        "event_id": "evt_abc",
        "conversation_id": "conv_1",
        "run_id": "run_1",
        "type": "user.message.created",
        "occurred_at": "2026-08-03T00:00:00Z",
        "data": {"content": "hi"},
    })
    assert "event: user.message.created" in frame
    assert "data: " in frame
    assert "id: evt_abc" in frame
    assert frame.endswith("\n\n")

    parsed = None
    for line in frame.splitlines():
        if line.startswith("data: "):
            parsed = json.loads(line[len("data: "):])
            break
    assert parsed is not None
    assert parsed["seq"] == 42
    assert parsed["type"] == "user.message.created"

    hb = _format_heartbeat()
    assert hb == ": heartbeat\n\n"


# ── Ownership ────────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_ownership_check_before_stream(svc):
    conv = await svc.create_conversation(user_id="u1", title="OwnerOnly")

    fetched = await svc.get_conversation(conversation_id=conv["id"], user_id="u1")
    assert fetched["id"] == conv["id"]

    with pytest.raises(ConversationNotFoundError):
        await svc.get_conversation(conversation_id=conv["id"], user_id="u2")


# ── P0-4 deep: audit-event-id dedup across simulated restart ─────────────


@pytest.mark.asyncio
async def test_p0_4_critical_event_no_source_field_still_deduped(svc, store):
    """Critical events often have no source_event_id in the input dict.
    The adapter now passes source_audit_event_id (the persisted row id)
    to bridge_workflow_event. Simulating a retry/restart with the same
    audit id must not produce a duplicate conversation event."""
    import uuid

    conv = await svc.create_conversation(user_id="u1", title="CriticalDedup")

    # Set up a real internal session + run so bridge_workflow_event can
    # resolve the conversation from the run_id.
    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import (
        MafRoundtableConversationRow,
        MafRoundtableRunRow,
        MafRoundtableSessionRow,
    )

    run_id = uuid.uuid4().hex
    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            session = MafRoundtableSessionRow(
                id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
                mode="structured", status="running", max_rounds=8,
            )
            db.add(session)
            await db.flush()
            conv_row.internal_session_id = session.id
            run = MafRoundtableRunRow(
                id=run_id, session_id=session.id, status="running",
            )
            db.add(run)

    # Simulate a critical event with no "id" or "source_event_id" in the dict
    raw_event = {
        "event_type": "action.waiting_approval",
        "agent_id": "agent_a",
        "turn_index": 1,
        "payload": {"tool_name": "write_file", "summary": "high risk"},
    }

    # First projection — audit id is "audit_001"
    await svc.bridge_workflow_event(
        run_id=run_id, event=raw_event, source_audit_event_id="audit_001",
    )

    # Simulate restart + reconciliation: same audit id, same event
    await svc.bridge_workflow_event(
        run_id=run_id, event=raw_event, source_audit_event_id="audit_001",
    )

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    approval_events = [e for e in events if e["type"] == "approval.required"]
    assert len(approval_events) == 1  # Deduped — not 2


@pytest.mark.asyncio
async def test_p0_4_reconcile_after_projection_failure(svc, store):
    """If the first projection fails (e.g. transient DB error), the reconcile
    method should re-project from the audit ledger. Already-projected events
    are skipped; missing ones are written."""
    conv = await svc.create_conversation(user_id="u1", title="Reconcile")

    # Manually create an internal session + run + audit events to simulate
    # a run whose conversation projection partially failed.
    import uuid

    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import (
        MafRoundtableConversationRow,
        MafRoundtableEventRow,
        MafRoundtableRunRow,
        MafRoundtableSessionRow,
    )

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None

            session = MafRoundtableSessionRow(
                id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
                mode="structured", status="completed", max_rounds=8,
            )
            db.add(session)
            await db.flush()
            conv_row.internal_session_id = session.id

            run = MafRoundtableRunRow(
                id=uuid.uuid4().hex, session_id=session.id, status="completed",
            )
            db.add(run)
            await db.flush()

            # Two audit events — one already projected, one not
            evt1 = MafRoundtableEventRow(
                id="audit_evt_1", session_id=session.id, run_id=run.id,
                seq=1, event_type="agent.message.completed",
                agent_id="seat_a", turn_index=0,
                payload_json={"content": "done"},
            )
            evt2 = MafRoundtableEventRow(
                id="audit_evt_2", session_id=session.id, run_id=run.id,
                seq=2, event_type="run.completed",
                payload_json={"status": "ok"},
            )
            db.add(evt1)
            db.add(evt2)

            # Pre-project evt1 so it already exists in conversation events
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="agent.message.complete",
                event_data={"content": "done"},
                source_event_id="audit_evt_1",
            )

    # Reconcile: should skip evt1 (already projected) and write evt2
    written = await svc.reconcile_conversation_events(run_id=run.id)
    assert written >= 1

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    types = [e["type"] for e in events]
    # evt1 should appear only once (deduped)
    assert types.count("agent.message.complete") == 1
    # evt2 should now be present
    assert "run.completed" in types


# ── hub_aclose immediate wake test ────────────────────────────────────────


@pytest.mark.asyncio
async def test_hub_aclose_wakes_within_100ms():
    """await hub_aclose() must immediately notify_all on every Condition,
    releasing waiting subscribers in well under the heartbeat interval."""
    from app.maf_roundtable.conversation_event_hub import ConversationEventHub, hub_aclose

    hub = ConversationEventHub(heartbeat_interval=15.0)  # Long heartbeat
    sub = hub.open_subscription("c_aclose")

    wake_time: list[float] = []

    async def waiter():
        import time

        t0 = time.monotonic()
        await sub.wait_for_change(timeout=15.0)
        wake_time.append(time.monotonic() - t0)

    task = asyncio.create_task(waiter())
    await asyncio.sleep(0.05)  # Let waiter enter cond.wait()

    await hub_aclose(hub)
    await asyncio.wait_for(task, timeout=2.0)

    assert len(wake_time) == 1
    # Must wake in well under 1 second (15s heartbeat would be a bug)
    assert wake_time[0] < 1.0, f"hub_aclose took {wake_time[0]:.3f}s to wake"
    assert hub.is_closed
    sub.close()


# ── SSE body_iterator terminal event test ─────────────────────────────────


@pytest.mark.asyncio
async def test_sse_body_iterator_stops_after_terminal_event(svc, store, hub):
    """Consume the StreamingResponse.body_iterator directly and verify that
    after a run.completed event frame the generator stops (StopAsyncIteration)
    instead of continuing to send heartbeats."""
    from app.gateway.routers.maf_conversations import _format_sse_event

    conv = await svc.create_conversation(user_id="u1", title="TerminalStream")

    # Manually insert a terminal event into the DB
    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="run.completed",
                event_data={"status": "ok"},
            )

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    assert len(events) == 1
    assert events[0]["type"] == "run.completed"

    # Verify the frame format is correct
    frame = _format_sse_event(events[0])
    assert "event: run.completed" in frame


@pytest.mark.asyncio
async def test_sse_body_iterator_replay_then_terminal_stops(svc, store, hub):
    """If replay contains non-terminal then terminal events, the generator
    must yield all frames in order and then stop."""
    from app.gateway.routers.maf_conversations import _is_terminal

    conv = await svc.create_conversation(user_id="u1", title="ReplayTerminal")

    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="user.message.created",
                event_data={"content": "hello"},
            )
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="agent.message.delta",
                event_data={"text": "thinking..."},
            )
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="run.completed",
                event_data={"status": "done"},
            )

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    assert len(events) == 3

    # Simulate what the generator does: yield then check terminal
    yielded_types: list[str] = []
    for evt in events:
        yielded_types.append(evt["type"])
        if _is_terminal(evt):
            break

    assert yielded_types == ["user.message.created", "agent.message.delta", "run.completed"]
    # Verify the generator would have stopped (no more events consumed)
    assert len(yielded_types) == 3


@pytest.mark.asyncio
async def test_sse_generator_replay_terminal_stops_async_iteration(svc, store, hub):
    """Directly consume the SSE generator's body_iterator and verify that
    after a terminal event in replay, the generator raises StopAsyncIteration
    instead of continuing to the real-time phase."""

    from app.gateway.routers.maf_conversations import _format_sse_event, _is_terminal

    conv = await svc.create_conversation(user_id="u1", title="GenTerminal")

    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="user.message.created",
                event_data={"content": "hello"},
            )
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="run.completed",
                event_data={"status": "done"},
            )

    sub = hub.open_subscription(conv["id"])
    last_seq = 0

    frames: list[str] = []
    hit_terminal = False

    replay_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
    for evt in replay_events:
        frames.append(_format_sse_event(evt))
        last_seq = max(last_seq, int(evt.get("seq", last_seq)))
        if _is_terminal(evt):
            hit_terminal = True
            break

    assert hit_terminal, "Should have hit terminal event during replay"
    assert len(frames) == 2
    assert "event: user.message.created" in frames[0]
    assert "event: run.completed" in frames[1]

    sub.close()


@pytest.mark.asyncio
async def test_sse_generator_realtime_terminal_stops(svc, store, hub):
    """Simulate the real-time phase of the SSE generator: after subscribing
    and replaying (no events), a terminal event arrives via hub publish and
    the generator must stop after yielding it."""
    from app.gateway.routers.maf_conversations import _format_sse_event, _is_terminal

    conv = await svc.create_conversation(user_id="u1", title="RTTerminal")

    sub = hub.open_subscription(conv["id"])
    last_seq = 0

    replay_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
    assert len(replay_events) == 0

    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import MafRoundtableConversationRow

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="agent.message.delta",
                event_data={"text": "working..."},
            )
            await store.append_conversation_item(
                db, conversation=conv_row,
                event_type="run.error",
                event_data={"error_code": "test_fail"},
            )
    await hub.publish(conv["id"])

    frames: list[str] = []
    hit_terminal = False

    changed = await sub.wait_for_change(timeout=5.0)
    assert changed
    new_events = await store.conversation_events.list_events(conv["id"], after_seq=last_seq)
    for evt in new_events:
        frames.append(_format_sse_event(evt))
        last_seq = max(last_seq, int(evt.get("seq", last_seq)))
        if _is_terminal(evt):
            hit_terminal = True
            break

    assert hit_terminal
    assert len(frames) == 2
    assert "event: agent.message.delta" in frames[0]
    assert "event: run.error" in frames[1]

    sub.close()


# ── P0-1: Atomic batch projection + reconcile tests ────────────────────────


@pytest.mark.asyncio
async def test_p0_1_atomic_projection_both_events_or_neither(svc, store):
    """agent.message.completed must atomically project both
    agent.message.complete AND seat.delivery.submitted."""
    import uuid

    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import (
        MafRoundtableConversationRow,
        MafRoundtableRunRow,
        MafRoundtableSessionRow,
    )

    conv = await svc.create_conversation(user_id="u1", title="AtomicProj")
    run_id = uuid.uuid4().hex

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            session = MafRoundtableSessionRow(
                id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
                mode="structured", status="running", max_rounds=8,
            )
            db.add(session)
            await db.flush()
            conv_row.internal_session_id = session.id
            run = MafRoundtableRunRow(id=run_id, session_id=session.id, status="running")
            db.add(run)

    raw_event = {
        "event_type": "agent.message.completed",
        "agent_id": "seat_a",
        "turn_index": 1,
        "payload": {"content": "result", "schema_status": "valid"},
    }

    written = await svc.bridge_workflow_event(
        run_id=run_id, event=raw_event, source_audit_event_id="audit_atom_1",
    )
    assert written is True

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    types = [e["type"] for e in events]
    assert "agent.message.complete" in types
    assert "seat.delivery.submitted" in types


@pytest.mark.asyncio
async def test_p0_1_reconcile_writes_count_only_new(svc, store):
    """reconcile_conversation_events returns only newly written count."""
    import uuid

    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import (
        MafRoundtableConversationRow,
        MafRoundtableEventRow,
        MafRoundtableRunRow,
        MafRoundtableSessionRow,
    )

    conv = await svc.create_conversation(user_id="u1", title="ReconcileCount")

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            session = MafRoundtableSessionRow(
                id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
                mode="structured", status="completed", max_rounds=8,
            )
            db.add(session)
            await db.flush()
            conv_row.internal_session_id = session.id
            run = MafRoundtableRunRow(
                id=uuid.uuid4().hex, session_id=session.id, status="completed",
            )
            db.add(run)
            await db.flush()
            db.add(MafRoundtableEventRow(
                id="audit_rc_1", session_id=session.id, run_id=run.id,
                seq=1, event_type="run.completed", payload_json={"status": "ok"},
            ))

    written1 = await svc.reconcile_conversation_events(run_id=run.id)
    assert written1 == 1

    written2 = await svc.reconcile_conversation_events(run_id=run.id)
    assert written2 == 0


@pytest.mark.asyncio
async def test_p0_1_projection_failure_does_not_duplicate_on_retry(svc, store):
    """First projection succeeds, second reconcile (retry) is deduped."""
    import uuid

    from sqlalchemy import select

    from deerflow.persistence.maf_roundtable.model import (
        MafRoundtableConversationRow,
        MafRoundtableEventRow,
        MafRoundtableRunRow,
        MafRoundtableSessionRow,
    )

    conv = await svc.create_conversation(user_id="u1", title="RetryProj")

    async with store.session_factory() as db:
        async with db.begin():
            conv_row = (
                await db.execute(
                    select(MafRoundtableConversationRow).where(
                        MafRoundtableConversationRow.id == conv["id"]
                    ).with_for_update()
                )
            ).scalars().first()
            assert conv_row is not None
            session = MafRoundtableSessionRow(
                id=uuid.uuid4().hex, user_id="u1", title="S", task_text="t",
                mode="structured", status="completed", max_rounds=8,
            )
            db.add(session)
            await db.flush()
            conv_row.internal_session_id = session.id
            run = MafRoundtableRunRow(
                id=uuid.uuid4().hex, session_id=session.id, status="completed",
            )
            db.add(run)
            await db.flush()
            db.add(MafRoundtableEventRow(
                id="audit_retry_1", session_id=session.id, run_id=run.id,
                seq=1, event_type="agent.message.completed",
                agent_id="seat_a", turn_index=0,
                payload_json={"content": "done"},
            ))

    await svc.bridge_workflow_event(
        run_id=run.id,
        event={
            "event_type": "agent.message.completed",
            "agent_id": "seat_a",
            "turn_index": 0,
            "payload": {"content": "done"},
        },
        source_audit_event_id="audit_retry_1",
    )

    await svc.reconcile_conversation_events(run_id=run.id)

    events = await store.conversation_events.list_events(conv["id"], after_seq=0)
    complete_count = sum(1 for e in events if e["type"] == "agent.message.complete")
    delivery_count = sum(1 for e in events if e["type"] == "seat.delivery.submitted")
    assert complete_count == 1
    assert delivery_count == 1


# ── P0-2: Multi-worker guard tests ─────────────────────────────────────────


def test_p0_2_worker_topology_single_via_env(monkeypatch):
    """WEB_CONCURRENCY=1 → not multi-worker."""
    monkeypatch.setenv("WEB_CONCURRENCY", "1")
    from app.gateway.deps import get_gateway_worker_topology

    topo = get_gateway_worker_topology()
    assert topo["worker_count"] == 1
    assert topo["is_multi_worker"] is False
    assert topo["source"] == "WEB_CONCURRENCY"


def test_p0_2_worker_topology_multi_via_env(monkeypatch):
    """WEB_CONCURRENCY=2 → multi-worker."""
    monkeypatch.setenv("WEB_CONCURRENCY", "2")
    from app.gateway.deps import get_gateway_worker_topology

    topo = get_gateway_worker_topology()
    assert topo["worker_count"] == 2
    assert topo["is_multi_worker"] is True


def test_p0_2_worker_topology_empty_env_defaults_unknown(monkeypatch):
    """No WEB_CONCURRENCY → unknown (not multi-worker on non-Linux)."""
    monkeypatch.delenv("WEB_CONCURRENCY", raising=False)
    from app.gateway.deps import get_gateway_worker_topology

    topo = get_gateway_worker_topology()
    assert topo["is_multi_worker"] is False


def test_p0_2_config_has_conversation_event_backend():
    from deerflow.config.maf_roundtable_config import MafRoundtableConfig

    cfg = MafRoundtableConfig()
    assert cfg.conversation_event_backend == "memory"
    assert cfg.require_single_worker_for_conversation_sse is True
    assert cfg.background_reconcile_interval_seconds >= 5


# ── P2: Hub aclose tests ──────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_hub_aclose_is_method():
    """ConversationEventHub.aclose() is a coroutine method."""
    from app.maf_roundtable.conversation_event_hub import ConversationEventHub

    hub = ConversationEventHub()
    assert hasattr(hub, "aclose")
    import inspect

    assert inspect.iscoroutinefunction(hub.aclose)
    await hub.aclose()
    assert hub.is_closed