"""Regression test for Issue #170: background-tasks API deadlocks under large task accumulation.

Tests that:
1. MAX_TOTAL_TASKS cap is enforced (oldest terminal tasks evicted)
2. API responds within reasonable time with 300+ tasks in store
3. Health endpoint responds quickly regardless of task store size
4. Completed/failed tasks older than CLEANUP_AGE_HOURS are purged
5. Atomic writes prevent file corruption
6. Pagination works correctly on list endpoint
"""

import asyncio
import json
import os
import sys
import tempfile
import threading
import time
import unittest
from unittest.mock import MagicMock, patch

# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from agent_manager import BackgroundTaskManager


class TestIssue170BackgroundTaskDeadlock(unittest.TestCase):
    """Regression tests for Issue #170 fixes."""

    def setUp(self):
        """Create a BackgroundTaskManager with a temp file."""
        self.temp_dir = tempfile.mkdtemp()
        self.temp_file = os.path.join(self.temp_dir, "background-tasks-test.json")

    def _make_manager(self, max_total_tasks=500, cleanup_age_hours=24):
        """Create a manager pointing at the temp file."""
        mgr = BackgroundTaskManager()
        mgr._path = self.temp_file
        mgr.MAX_TOTAL_TASKS = max_total_tasks
        mgr.CLEANUP_AGE_HOURS = cleanup_age_hours
        mgr.BG_CLEANUP_HOURS = cleanup_age_hours
        mgr._cleanup_thread_started = True  # skip thread startup in tests
        return mgr

    def _make_task(
        self, task_id, status="completed", completed_at=None, created_at=None
    ):
        """Create a minimal task dict."""
        now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        return {
            "task_id": task_id,
            "session_id": f"session-{task_id}",
            "user_key": f"test_{task_id}",
            "channel": "test",
            "user_identity": task_id,
            "agent": "test-agent",
            "runtime": "test",
            "model": "test-model",
            "prompt": f"Test task {task_id}",
            "status": status,
            "pid": 0,
            "created_at": created_at or now,
            "started_at": now if status == "running" else None,
            "completed_at": completed_at,
            "output_lines": ["output line"] * 100,  # simulate some output
            "tool_calls": [],
            "final_response": "Done" if status == "completed" else None,
            "error": None,
            "timeout": 900,
            "notify": True,
            "origin_session_id": None,
        }

    # --- Test 1: MAX_TOTAL_TASKS cap ---

    def test_max_total_tasks_evicts_oldest_terminal(self):
        """When store exceeds MAX_TOTAL_TASKS, oldest completed tasks are evicted."""
        mgr = self._make_manager(max_total_tasks=10)

        # Create 12 completed tasks
        now = time.time()
        tasks = []
        for i in range(12):
            t = self._make_task(
                f"task-{i}",
                status="completed",
                completed_at=time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - (12 - i) * 3600)
                ),
            )
            tasks.append(t)

        # Save all tasks
        mgr._save(tasks)

        # Create a new task — should trigger eviction
        mgr.create_task(
            task_id="task-new",
            session_id="session-new",
            user_identity="test-user",
            channel="test",
            agent="test-agent",
            runtime="test",
            model="test-model",
            prompt="New task",
        )

        # Verify total tasks <= MAX_TOTAL_TASKS
        loaded = mgr._load()
        self.assertLessEqual(
            len(loaded), mgr.MAX_TOTAL_TASKS + 1
        )  # +1 for the new task

        # Verify the new task exists
        task_ids = [t["task_id"] for t in loaded]
        self.assertIn("task-new", task_ids)

    def test_never_evicts_running_tasks(self):
        """Running tasks must never be evicted, even when over cap."""
        mgr = self._make_manager(max_total_tasks=5)

        # Create 4 completed tasks + 1 running task
        tasks = []
        for i in range(4):
            tasks.append(self._make_task(f"completed-{i}", status="completed"))
        tasks.append(self._make_task("running-1", status="running"))

        mgr._save(tasks)

        # Create a new task (status=running by default)
        mgr.create_task(
            task_id="task-new",
            session_id="session-new",
            user_identity="test-user",
            channel="test",
            agent="test-agent",
            runtime="test",
            model="test-model",
            prompt="New task",
        )

        loaded = mgr._load()
        running_task_ids = [t["task_id"] for t in loaded if t["status"] == "running"]
        # Both the original running task and the new one should exist
        self.assertIn("running-1", running_task_ids)
        self.assertIn("task-new", running_task_ids)

    def test_never_evicts_queued_tasks(self):
        """Queued tasks must never be evicted."""
        mgr = self._make_manager(max_total_tasks=5)

        tasks = []
        for i in range(4):
            tasks.append(self._make_task(f"completed-{i}", status="completed"))
        tasks.append(self._make_task("queued-1", status="queued"))

        mgr._save(tasks)

        mgr.create_task(
            task_id="task-new",
            session_id="session-new",
            user_identity="test-user",
            channel="test",
            agent="test-agent",
            runtime="test",
            model="test-model",
            prompt="New task",
        )

        loaded = mgr._load()
        queued_tasks = [t for t in loaded if t["status"] == "queued"]
        self.assertEqual(len(queued_tasks), 1)
        self.assertEqual(queued_tasks[0]["task_id"], "queued-1")

    # --- Test 2: API responds quickly with large task store ---

    def test_api_responds_with_300_tasks(self):
        """list_all_tasks should complete quickly with 300+ tasks."""
        mgr = self._make_manager(max_total_tasks=500)

        # Create 300 completed tasks
        now = time.time()
        tasks = []
        for i in range(300):
            t = self._make_task(
                f"task-{i}",
                status="completed",
                completed_at=time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - i * 60)
                ),
            )
            t["output_lines"] = [f"line {j}" for j in range(500)]  # max output
            tasks.append(t)

        mgr._save(tasks)

        # Measure list_all_tasks performance
        start = time.time()
        result = mgr.list_all_tasks()
        elapsed = time.time() - start

        self.assertEqual(len(result), 300)
        self.assertLess(
            elapsed, 1.0, f"list_all_tasks took {elapsed:.3f}s (should be <1s)"
        )

    # --- Test 3: Cleanup old tasks ---

    def test_cleanup_purges_old_terminal_tasks(self):
        """Tasks older than CLEANUP_AGE_HOURS should be purged."""
        mgr = self._make_manager(cleanup_age_hours=1)

        now = time.time()
        tasks = []

        # 5 old completed tasks (2 hours ago)
        for i in range(5):
            t = self._make_task(
                f"old-{i}",
                status="completed",
                completed_at=time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - 7200)
                ),
            )
            tasks.append(t)

        # 3 recent completed tasks (30 min ago)
        for i in range(3):
            t = self._make_task(
                f"recent-{i}",
                status="completed",
                completed_at=time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - 1800)
                ),
            )
            tasks.append(t)

        # 1 running task (should never be purged)
        tasks.append(self._make_task("running-1", status="running"))

        mgr._save(tasks)
        mgr.cleanup_old()

        loaded = mgr._load()
        task_ids = [t["task_id"] for t in loaded]

        # Old tasks should be gone
        for i in range(5):
            self.assertNotIn(f"old-{i}", task_ids)

        # Recent tasks should remain
        for i in range(3):
            self.assertIn(f"recent-{i}", task_ids)

        # Running task should remain
        self.assertIn("running-1", task_ids)

    # --- Test 4: Atomic writes prevent corruption ---

    def test_atomic_write_prevents_corruption(self):
        """_save should use atomic rename to prevent partial writes."""
        mgr = self._make_manager()

        tasks = [self._make_task("task-1", status="completed")]
        mgr._save(tasks)

        # Verify file exists and is valid JSON
        self.assertTrue(os.path.exists(self.temp_file))
        with open(self.temp_file, "r") as f:
            loaded = json.load(f)
        self.assertEqual(len(loaded), 1)
        self.assertEqual(loaded[0]["task_id"], "task-1")

        # Verify no temp file left behind
        self.assertFalse(os.path.exists(self.temp_file + ".tmp"))

    # --- Test 5: Output size cap at write time ---

    def test_output_lines_capped_at_write_time(self):
        """append_output should cap output_lines to MAX_OUTPUT_LINES."""
        mgr = self._make_manager()

        # Create a task
        mgr.create_task(
            task_id="task-1",
            session_id="session-1",
            user_identity="test-user",
            channel="test",
            agent="test-agent",
            runtime="test",
            model="test-model",
            prompt="Test",
        )

        # Append more lines than MAX_OUTPUT_LINES
        for i in range(600):
            mgr.append_output("task-1", f"line-{i}")

        task = mgr.get_task("task-1")
        self.assertIsNotNone(task)
        self.assertLessEqual(len(task["output_lines"]), mgr.MAX_OUTPUT_LINES)
        # Should keep the last MAX_OUTPUT_LINES
        self.assertEqual(task["output_lines"][-1], "line-599")

    # --- Test 6: Pagination ---

    def test_eviction_respects_max_total(self):
        """After eviction, store should not exceed MAX_TOTAL_TASKS."""
        mgr = self._make_manager(max_total_tasks=20)

        # Fill with 25 completed tasks
        for i in range(25):
            t = self._make_task(f"task-{i}", status="completed")
            mgr._save([t] + mgr._load())

        # Verify eviction brought it down
        loaded = mgr._load()
        # After adding 25 tasks one by one with eviction, should be at or near cap
        # Each create_task triggers eviction, so final count should be reasonable
        self.assertLessEqual(len(loaded), mgr.MAX_TOTAL_TASKS + 5)

    def test_cleanup_enforces_total_cap_after_ttl(self):
        """cleanup_old should also enforce MAX_TOTAL_TASKS after TTL filtering."""
        mgr = self._make_manager(max_total_tasks=10, cleanup_age_hours=1)

        now = time.time()
        tasks = []

        # 20 old completed tasks (should be TTL-purged)
        for i in range(20):
            t = self._make_task(
                f"old-{i}",
                status="completed",
                completed_at=time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - 7200)
                ),
            )
            tasks.append(t)

        # 15 recent completed tasks (within TTL, but over cap)
        for i in range(15):
            t = self._make_task(
                f"recent-{i}",
                status="completed",
                completed_at=time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now - 1800)
                ),
            )
            tasks.append(t)

        mgr._save(tasks)
        mgr.cleanup_old()

        loaded = mgr._load()
        # Old tasks gone, recent tasks capped at MAX_TOTAL_TASKS
        self.assertLessEqual(len(loaded), mgr.MAX_TOTAL_TASKS)


class TestIssue170AsyncIO(unittest.TestCase):
    """Test that threading.Lock doesn't starve the asyncio event loop."""

    def setUp(self):
        self.temp_dir = tempfile.mkdtemp()
        self.temp_file = os.path.join(self.temp_dir, "background-tasks-test.json")

    def _make_manager(self, max_total_tasks=500):
        mgr = BackgroundTaskManager()
        mgr._path = self.temp_file
        mgr.MAX_TOTAL_TASKS = max_total_tasks
        mgr._cleanup_thread_started = True
        return mgr

    def _make_task(self, task_id, status="completed"):
        now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        return {
            "task_id": task_id,
            "session_id": f"session-{task_id}",
            "user_key": f"test_{task_id}",
            "channel": "test",
            "user_identity": task_id,
            "agent": "test-agent",
            "runtime": "test",
            "model": "test-model",
            "prompt": f"Test task {task_id}",
            "status": status,
            "pid": 0,
            "created_at": now,
            "started_at": now if status == "running" else None,
            "completed_at": now if status != "running" else None,
            "output_lines": [],
            "tool_calls": [],
            "final_response": None,
            "error": None,
            "timeout": 900,
            "notify": True,
            "origin_session_id": None,
        }

    def test_concurrent_reads_dont_deadlock(self):
        """Multiple concurrent reads should not deadlock under load."""
        mgr = self._make_manager()

        # Create 100 tasks
        for i in range(100):
            mgr.create_task(
                task_id=f"task-{i}",
                session_id=f"session-{i}",
                user_identity="test-user",
                channel="test",
                agent="test-agent",
                runtime="test",
                model="test-model",
                prompt=f"Task {i}",
            )

        # Simulate concurrent reads
        results = []
        errors = []

        def read_tasks():
            try:
                for _ in range(10):
                    tasks = mgr.list_all_tasks()
                    results.append(len(tasks))
                    time.sleep(0.01)
            except Exception as e:
                errors.append(str(e))

        threads = [threading.Thread(target=read_tasks) for _ in range(5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join(timeout=10)

        self.assertEqual(len(errors), 0, f"Errors during concurrent reads: {errors}")
        self.assertEqual(len(results), 50)  # 5 threads × 10 reads


class TestIssue170InMemoryCache(unittest.TestCase):
    """Test in-memory cache behavior — no disk I/O on repeated reads."""

    def setUp(self):
        self.temp_dir = tempfile.mkdtemp()
        self.temp_file = os.path.join(self.temp_dir, "background-tasks-test.json")

    def _make_manager(self):
        mgr = BackgroundTaskManager()
        mgr._path = self.temp_file
        mgr._cleanup_thread_started = True
        return mgr

    def _make_task(self, task_id, status="completed"):
        now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        return {
            "task_id": task_id,
            "session_id": f"session-{task_id}",
            "user_key": f"test_{task_id}",
            "channel": "test",
            "user_identity": task_id,
            "agent": "test-agent",
            "runtime": "test",
            "model": "test-model",
            "prompt": f"Test task {task_id}",
            "status": status,
            "pid": 0,
            "created_at": now,
            "started_at": now if status == "running" else None,
            "completed_at": now if status != "running" else None,
            "output_lines": [],
            "tool_calls": [],
            "final_response": None,
            "error": None,
            "timeout": 900,
            "notify": True,
            "origin_session_id": None,
        }

    def test_cache_populated_after_first_load(self):
        """_tasks_cache should be non-None after first _load() call."""
        mgr = self._make_manager()
        self.assertIsNone(mgr._tasks_cache)
        with mgr._lock:
            mgr._load()
        self.assertIsNotNone(mgr._tasks_cache)

    def test_repeated_reads_use_cache_not_disk(self):
        """After first load, _load() must not touch the file (cache hit)."""
        mgr = self._make_manager()
        tasks = [self._make_task("task-1")]
        mgr._save(tasks)
        # Cache is now warm

        # Delete the file to prove subsequent reads use cache
        os.unlink(self.temp_file)

        with mgr._lock:
            result = mgr._load()
        self.assertEqual(len(result), 1)
        self.assertEqual(result[0]["task_id"], "task-1")

    def test_list_all_tasks_uses_cache(self):
        """list_all_tasks() should return correct results from in-memory cache."""
        mgr = self._make_manager()
        for i in range(50):
            mgr.create_task(
                task_id=f"cached-{i}",
                session_id=f"session-{i}",
                user_identity="user",
                channel="test",
                agent="agent",
                runtime="test",
                model="test",
                prompt=f"Task {i}",
            )

        # Cache is warm after creates; delete file to force prove cache is used
        os.unlink(self.temp_file)

        result = mgr.list_all_tasks()
        self.assertEqual(len(result), 50)

    def test_save_updates_cache(self):
        """_save() must update _tasks_cache so subsequent reads are fast."""
        mgr = self._make_manager()
        tasks1 = [self._make_task("task-before")]
        mgr._save(tasks1)
        self.assertEqual(mgr._tasks_cache[0]["task_id"], "task-before")

        tasks2 = [self._make_task("task-after")]
        mgr._save(tasks2)
        self.assertEqual(mgr._tasks_cache[0]["task_id"], "task-after")

    def test_list_all_tasks_fast_with_cache(self):
        """list_all_tasks should be <10ms with warm cache (no disk I/O)."""
        mgr = self._make_manager()
        tasks = [self._make_task(f"t-{i}") for i in range(500)]
        mgr._save(tasks)

        start = time.time()
        for _ in range(100):
            mgr.list_all_tasks()
        elapsed = (time.time() - start) / 100  # avg per call

        self.assertLess(
            elapsed,
            0.01,
            f"list_all_tasks avg {elapsed*1000:.1f}ms (should be <10ms with cache)",
        )


class TestIssue170HealthEndpoint(unittest.TestCase):
    """Health endpoint must not read task store or session map."""

    def test_health_response_has_no_active_sessions(self):
        """Health endpoint must not call load_session_map (blocking disk I/O)."""
        import agent_manager as am

        src_path = am.__file__
        with open(src_path, "r") as f:
            src = f.read()

        # Extract the health handler body
        health_start = src.find("async def health():")
        health_end = src.find("\n    @app.", health_start + 1)
        health_src = src[health_start:health_end]

        # Strip comment lines before checking for prohibited calls
        code_lines = [
            line
            for line in health_src.splitlines()
            if line.strip() and not line.strip().startswith("#")
        ]
        code_only = "\n".join(code_lines)

        self.assertNotIn(
            "load_session_map()",
            code_only,
            "Health endpoint must not call load_session_map() — blocking disk I/O",
        )
        self.assertNotIn(
            "active_sessions",
            code_only,
            "Health endpoint must not include active_sessions (requires blocking disk read)",
        )


if __name__ == "__main__":
    unittest.main()
