"""Unit tests for navi.core.tasks — TaskManager / TaskJob / BoundedEventQueue."""

import asyncio
import time

import pytest

from navi.core import tasks as tasks_mod
from navi.core.events import TaskUpdate
from navi.core.tasks import (
    BoundedEventQueue,
    TaskJob,
    TaskManager,
    args_summary,
    get_task_manager,
)
from navi.tools._internal.base import ToolContext, ToolResult, current_stop_event


def patch_settings(monkeypatch, **overrides):
    """Replace the frozen settings singleton with an overridden copy."""
    import navi.config as config_mod
    from navi.config import Settings

    new_settings = Settings(**overrides)
    monkeypatch.setattr(config_mod, "settings", new_settings)
    return new_settings


@pytest.fixture(autouse=True)
def fresh_manager(monkeypatch):
    """Isolate the module singleton; clean up stray tasks afterwards."""
    manager = TaskManager()
    monkeypatch.setattr(tasks_mod, "_manager", manager)
    yield manager
    for job in list(manager._jobs.values()):
        if job.task is not None and not job.task.done():
            job.task.cancel()
    if manager._sweeper_task is not None and not manager._sweeper_task.done():
        manager._sweeper_task.cancel()


def make_ctx(**overrides):
    defaults = dict(
        session_id="s1",
        event_sink=None,
        stop_event=None,
        model="m",
        user_id="u1",
        user_role="admin",
        user_info=None,
        cwd="/tmp",
    )
    defaults.update(overrides)
    return ToolContext(**defaults)


class TestLifecycle:
    async def test_completed_job(self, fresh_manager):
        async def factory(bg_ctx):
            return ToolResult(success=True, output="done 42")

        job = fresh_manager.submit("s1", "code_exec", {"code": "x"}, factory, make_ctx())
        assert job.status == "running"
        await job.done.wait()
        assert job.status == "completed"
        assert job.result.output == "done 42"
        assert job.finished_at is not None

    async def test_failed_job(self, fresh_manager):
        async def factory(bg_ctx):
            raise ValueError("boom")

        job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        await job.done.wait()
        assert job.status == "failed"
        assert "boom" in job.error
        assert job.result.success is False

    async def test_cancel_running(self, fresh_manager):
        started = asyncio.Event()

        async def factory(bg_ctx):
            started.set()
            await asyncio.sleep(30)

        job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        await started.wait()
        assert fresh_manager.cancel(job) is True
        await job.done.wait()
        assert job.status == "cancelled"
        assert fresh_manager.cancel(job) is False  # already finished

    async def test_task_stop_event_is_isolated(self, fresh_manager):
        """The job sees its own stop_event, not any run-level one."""
        seen = {}

        async def factory(bg_ctx):
            seen["ctx_stop"] = bg_ctx.stop_event
            seen["var_stop"] = current_stop_event.get()
            return ToolResult(success=True, output="ok")

        job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        await job.done.wait()
        assert seen["ctx_stop"] is job.stop_event
        assert seen["var_stop"] is job.stop_event

    async def test_get_is_session_scoped(self, fresh_manager):
        async def factory(bg_ctx):
            return ToolResult(success=True, output="ok")

        job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        await job.done.wait()
        assert fresh_manager.get(job.task_id, "s1") is job
        assert fresh_manager.get(job.task_id, "s2") is None
        assert [j.task_id for j in fresh_manager.list("s1")] == [job.task_id]
        assert fresh_manager.list("s2") == []


class TestCaps:
    async def test_per_session_cap(self, fresh_manager, monkeypatch):
        patch_settings(monkeypatch, tasks_max_per_session=1)

        async def factory(bg_ctx):
            await asyncio.sleep(30)

        first = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        assert not isinstance(first, str)
        second = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        assert isinstance(second, str) and "limit" in second
        fresh_manager.cancel(first)

    async def test_rate_limit(self, fresh_manager, monkeypatch):
        patch_settings(monkeypatch, tasks_rate_limit=1)

        async def factory(bg_ctx):
            return ToolResult(success=True, output="ok")

        first = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        assert not isinstance(first, str)
        second = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        assert isinstance(second, str) and "rate limit" in second

    async def test_spawn_cap_only_limits_spawn_agent(self, fresh_manager, monkeypatch):
        patch_settings(monkeypatch, tasks_max_spawn=1)

        async def factory(bg_ctx):
            await asyncio.sleep(30)

        first = fresh_manager.submit("s1", "spawn_agent", {}, factory, make_ctx())
        assert not isinstance(first, str)
        # another spawn_agent is rejected…
        second = fresh_manager.submit("s1", "spawn_agent", {}, factory, make_ctx())
        assert isinstance(second, str) and "subagent" in second
        # …but a regular tool is not
        third = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        assert not isinstance(third, str)
        fresh_manager.cancel(first)
        fresh_manager.cancel(third)


class TestUpdateCallback:
    async def test_published_on_submit_and_finish(self, fresh_manager):
        updates = []
        fresh_manager.set_update_callback(updates.append)

        async def factory(bg_ctx):
            return ToolResult(success=True, output="ok")

        job = fresh_manager.submit("s1", "terminal", {"command": "ls"}, factory, make_ctx(),
                                   parent_tool_call_id="tc1")
        await job.done.wait()
        await asyncio.sleep(0)  # let the callback fire settle
        statuses = [u.status for u in updates]
        assert statuses[0] == "running"
        assert statuses[-1] == "completed"
        final = updates[-1]
        assert isinstance(final, TaskUpdate)
        assert final.task_id == job.task_id
        assert final.tool == "terminal"
        assert final.parent_tool_call_id == "tc1"
        assert final.to_wire()["type"] == "task_update"

    async def test_broken_callback_does_not_kill_job(self, fresh_manager):
        def bad_callback(update):
            raise RuntimeError("nope")

        fresh_manager.set_update_callback(bad_callback)

        async def factory(bg_ctx):
            return ToolResult(success=True, output="ok")

        job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        await job.done.wait()
        assert job.status == "completed"


class TestBoundedEventQueue:
    async def test_drop_oldest_when_full(self):
        q = BoundedEventQueue(maxsize=2)
        await q.put("a")
        q.put_nowait("b")
        await q.put("c")
        assert list(q._queue) == ["b", "c"]

    async def test_put_never_blocks(self):
        q = BoundedEventQueue(maxsize=1)
        await asyncio.wait_for(q.put("a"), timeout=0.1)
        await asyncio.wait_for(q.put("b"), timeout=0.1)
        assert q.qsize() == 1


class TestReap:
    async def test_reap_drops_expired_finished_jobs(self, fresh_manager, monkeypatch):
        patch_settings(monkeypatch, tasks_ttl_sec=0)

        async def factory(bg_ctx):
            return ToolResult(success=True, output="ok")

        job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        await job.done.wait()
        assert fresh_manager.reap() == 1
        assert fresh_manager.get(job.task_id, "s1") is None

    async def test_reap_keeps_running_jobs(self, fresh_manager, monkeypatch):
        patch_settings(monkeypatch, tasks_ttl_sec=0)

        async def factory(bg_ctx):
            await asyncio.sleep(30)

        job = fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        assert fresh_manager.reap() == 0
        assert fresh_manager.get(job.task_id, "s1") is job
        fresh_manager.cancel(job)


class TestHelpers:
    def test_get_task_manager_singleton(self, monkeypatch):
        monkeypatch.setattr(tasks_mod, "_manager", None)
        assert get_task_manager() is get_task_manager()

    def test_args_summary(self):
        assert args_summary({"a": 1}) == '{"a": 1}'
        assert args_summary({"a": "x" * 500}, limit=10) == '{"a": "xx' + "x"

    async def test_task_job_defaults(self):
        job = TaskJob(task_id="bt-x", session_id="s1", tool="terminal", args={})
        assert job.status == "running"
        assert job.subagent_tokens is None
        assert job.preview() == ""  # running → empty preview

    async def test_time_in_submit_window(self, fresh_manager):
        async def factory(bg_ctx):
            return ToolResult(success=True, output="ok")

        fresh_manager.submit("s1", "code_exec", {}, factory, make_ctx())
        assert len(fresh_manager._submit_times["s1"]) == 1
        assert time.time() - fresh_manager._submit_times["s1"][0] < 5