"""Integration tests for the scheduler background loop."""

import asyncio
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock

import pytest

from navi.core.events import StreamEnd
from navi.core.orchestrator import AgentSessionOrchestrator
from navi.core.scheduler import Recall, _fire_recall, recall_scheduler_loop


@pytest.fixture
def fake_orchestrator():
    container = MagicMock()
    container.profile_registry = None
    container.tool_registry = None
    container.backend_registry = None
    container.cp_registry = None
    container.workers = []
    container.memory_store = None
    container.mcp_manager = None
    return AgentSessionOrchestrator(container)


@pytest.fixture(autouse=True)
def patch_scheduler_deps(monkeypatch):
    """Prevent real _fire_recall from triggering heavy dependency initialization."""
    monkeypatch.setattr("navi.api.deps.get_workers", lambda: [])
    monkeypatch.setattr("navi.api.deps.get_memory_store", lambda: None)
    monkeypatch.setattr("navi.api.deps.get_mcp_manager", AsyncMock(return_value=None))


class TestSchedulerLoop:
    @pytest.mark.anyio
    async def test_loop_fires_past_due_recalls(self, monkeypatch):
        scheduler = AsyncMock()
        scheduler.get_pending_recalls.return_value = [
            Recall(
                id="r1", session_id="s1", call_type="once",
                trigger_at=datetime.now(timezone.utc) - timedelta(minutes=5),
                interval_seconds=None, internal_comment=None,
                additional_context_message="ctx", status="pending",
                created_at=datetime.now(timezone.utc),
                updated_at=datetime.now(timezone.utc),
            )
        ]
        scheduler.get_next_trigger_at.return_value = None
        scheduler.mark_fired.return_value = None

        store = AsyncMock()
        store.get.return_value = AsyncMock()
        store.get.return_value.messages = []
        store.get.return_value.context = []

        # Patch _fire_recall to avoid full Agent construction
        fire_calls = []
        async def _fake_fire(semaphore, recall, scheduler, store, orchestrator):
            fire_calls.append(recall.id)
            await scheduler.mark_fired(recall.id)

        monkeypatch.setattr("navi.core.scheduler._fire_recall", _fake_fire)

        # Run loop for one iteration then cancel
        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, AsyncMock()))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

        assert "r1" in fire_calls

    @pytest.mark.anyio
    async def test_loop_respects_semaphore(self, monkeypatch, fake_orchestrator):
        scheduler = AsyncMock()
        scheduler.get_pending_recalls.return_value = [
            Recall(
                id=f"r{i}", session_id=f"s{i}", call_type="once",
                trigger_at=datetime.now(timezone.utc) - timedelta(minutes=5),
                interval_seconds=None, internal_comment=None,
                additional_context_message="ctx", status="pending",
                created_at=datetime.now(timezone.utc),
                updated_at=datetime.now(timezone.utc),
            )
            for i in range(5)
        ]
        scheduler.get_next_trigger_at.return_value = None

        store = AsyncMock()

        running = asyncio.Semaphore(0)
        max_concurrent = 0
        current = 0

        async def _slow_fire(semaphore, recall, scheduler, store, orchestrator):
            nonlocal max_concurrent, current
            async with semaphore:
                current += 1
                max_concurrent = max(max_concurrent, current)
                await asyncio.sleep(0.1)
                current -= 1
                await scheduler.mark_fired(recall.id)

        monkeypatch.setattr("navi.core.scheduler._fire_recall", _slow_fire)

        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, fake_orchestrator))
        await asyncio.sleep(0.15)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

        assert max_concurrent <= 3

    @pytest.mark.anyio
    async def test_loop_defers_when_session_busy(self, monkeypatch, fake_orchestrator):
        scheduler = AsyncMock()
        scheduler.get_pending_recalls.return_value = [
            Recall(
                id="r1", session_id="s1", call_type="once",
                trigger_at=datetime.now(timezone.utc) - timedelta(minutes=5),
                interval_seconds=None, internal_comment=None,
                additional_context_message="ctx", status="pending",
                created_at=datetime.now(timezone.utc),
                updated_at=datetime.now(timezone.utc),
            )
        ]
        scheduler.get_next_trigger_at.return_value = None
        scheduler.reschedule.return_value = None

        store = AsyncMock()

        # Simulate an active websocket run in the orchestrator
        fake_orchestrator.create_run("s1")

        async def _fake_fire(semaphore, recall, scheduler, store, orchestrator):
            await _fire_recall(semaphore, recall, scheduler, store, orchestrator)

        monkeypatch.setattr("navi.core.scheduler._fire_recall", _fake_fire)

        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, fake_orchestrator))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        finally:
            fake_orchestrator._sessions.pop("s1", None)

        scheduler.reschedule.assert_called_once()

    @pytest.mark.anyio
    async def test_loop_cancels_when_session_missing(self, monkeypatch, fake_orchestrator):
        scheduler = AsyncMock()
        scheduler.get_pending_recalls.return_value = [
            Recall(
                id="r1", session_id="s1", call_type="once",
                trigger_at=datetime.now(timezone.utc) - timedelta(minutes=5),
                interval_seconds=None, internal_comment=None,
                additional_context_message="ctx", status="pending",
                created_at=datetime.now(timezone.utc),
                updated_at=datetime.now(timezone.utc),
            )
        ]
        scheduler.get_next_trigger_at.return_value = None
        scheduler.mark_cancelled.return_value = None

        store = AsyncMock()
        store.get.return_value = None

        async def _fake_fire(semaphore, recall, scheduler, store, orchestrator):
            await _fire_recall(semaphore, recall, scheduler, store, orchestrator)

        monkeypatch.setattr("navi.core.scheduler._fire_recall", _fake_fire)

        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, fake_orchestrator))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

        scheduler.mark_cancelled.assert_called_once_with("r1")

    @pytest.mark.anyio
    async def test_recurring_rescheduled_on_success(self, monkeypatch, fake_orchestrator):
        scheduler = AsyncMock()
        recall = Recall(
            id="r1", session_id="s1", call_type="recurring",
            trigger_at=datetime.now(timezone.utc) - timedelta(minutes=5),
            interval_seconds=3600, internal_comment=None,
            additional_context_message="ctx", status="pending",
            created_at=datetime.now(timezone.utc),
            updated_at=datetime.now(timezone.utc),
        )
        scheduler.get_pending_recalls.return_value = [recall]
        scheduler.get_next_trigger_at.return_value = None
        scheduler.reschedule.return_value = None

        store = AsyncMock()
        store.get.return_value = AsyncMock()
        store.get.return_value.messages = []
        store.get.return_value.context = []

        # Mock Agent as a plain object with an async-generator run_stream
        class FakeAgent:
            async def run_stream(self, *a, **kw):
                yield StreamEnd(full_content="done")

        monkeypatch.setattr("navi.core.agent.Agent", lambda *a, **kw: FakeAgent())

        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, fake_orchestrator))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

        scheduler.reschedule.assert_called_once()

    @pytest.mark.anyio
    async def test_recurring_rescheduled_on_failure(self, monkeypatch, fake_orchestrator):
        scheduler = AsyncMock()
        recall = Recall(
            id="r1", session_id="s1", call_type="recurring",
            trigger_at=datetime.now(timezone.utc) - timedelta(minutes=5),
            interval_seconds=3600, internal_comment=None,
            additional_context_message="ctx", status="pending",
            created_at=datetime.now(timezone.utc),
            updated_at=datetime.now(timezone.utc),
        )
        scheduler.get_pending_recalls.return_value = [recall]
        scheduler.get_next_trigger_at.return_value = None
        scheduler.reschedule.return_value = None

        store = AsyncMock()
        store.get.return_value = AsyncMock()
        store.get.return_value.messages = []
        store.get.return_value.context = []

        class FakeAgent:
            async def run_stream(self, *a, **kw):
                raise RuntimeError("boom")
                yield  # makes it an async generator

        monkeypatch.setattr("navi.core.agent.Agent", lambda *a, **kw: FakeAgent())

        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, fake_orchestrator))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

        scheduler.reschedule.assert_called_once()

    @pytest.mark.anyio
    async def test_one_time_cancelled_on_failure(self, monkeypatch, fake_orchestrator):
        scheduler = AsyncMock()
        recall = Recall(
            id="r1", session_id="s1", call_type="once",
            trigger_at=datetime.now(timezone.utc) - timedelta(minutes=5),
            interval_seconds=None, internal_comment=None,
            additional_context_message="ctx", status="pending",
            created_at=datetime.now(timezone.utc),
            updated_at=datetime.now(timezone.utc),
        )
        scheduler.get_pending_recalls.return_value = [recall]
        scheduler.get_next_trigger_at.return_value = None
        scheduler.mark_cancelled.return_value = None

        store = AsyncMock()
        store.get.return_value = AsyncMock()
        store.get.return_value.messages = []
        store.get.return_value.context = []

        class FakeAgent:
            async def run_stream(self, *a, **kw):
                raise RuntimeError("boom")
                yield  # makes it an async generator

        monkeypatch.setattr("navi.core.agent.Agent", lambda *a, **kw: FakeAgent())

        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, fake_orchestrator))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

        scheduler.mark_cancelled.assert_called_once_with("r1")

    @pytest.mark.anyio
    async def test_loop_picks_up_after_restart(self, monkeypatch, fake_orchestrator):
        scheduler = AsyncMock()
        scheduler.get_pending_recalls.return_value = [
            Recall(
                id="r1", session_id="s1", call_type="once",
                trigger_at=datetime.now(timezone.utc) - timedelta(hours=2),
                interval_seconds=None, internal_comment=None,
                additional_context_message="ctx", status="pending",
                created_at=datetime.now(timezone.utc),
                updated_at=datetime.now(timezone.utc),
            )
        ]
        scheduler.get_next_trigger_at.return_value = None
        scheduler.mark_fired.return_value = None

        store = AsyncMock()

        fire_calls = []
        async def _fake_fire(semaphore, recall, scheduler, store, orchestrator):
            fire_calls.append(recall.id)
            await scheduler.mark_fired(recall.id)

        monkeypatch.setattr("navi.core.scheduler._fire_recall", _fake_fire)

        task = asyncio.create_task(recall_scheduler_loop(scheduler, store, fake_orchestrator))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

        assert "r1" in fire_calls