"""U-032 tests: the scheduler's own bookkeeping (loop-closing wiring + real
thread-level concurrency bookkeeping), mocked at the org_level_loop/
org_loop_route/run_goal_to_completion boundary -- this file is not about
re-proving e2e_pipeline/org_level_loop's own mechanics (covered in their own
test files), it's about proving finish_task_lifecycle applies the right
loop-closing step and run_scheduler_once actually runs goals concurrently
(not accidentally serially) while still collecting every result. Real
multi-Goal coverage (RUN_REAL_AGENT_E2E=1) is in src/test_scheduler_e2e.py.

Run: python3 -m unittest src.test_scheduler -v
"""

import os
import tempfile
import threading
import time
import unittest
from unittest.mock import patch

from src import org_level_loop, org_loop_route, schema, scheduler, settings_config


class FinishTaskLifecycleTestBase(unittest.TestCase):
    def setUp(self):
        self.conn = schema.connect(":memory:")
        schema.create_schema(self.conn)
        self.conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
        )
        self.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, loop_round, assigned_org_node) "
            "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 0, 'validator')"
        )
        self.conn.commit()

    def tearDown(self):
        self.conn.close()


class FinishTaskLifecyclePassTest(FinishTaskLifecycleTestBase):
    def test_pass_verdict_closes_task_done(self):
        run_result = {
            "task_id": "task-1", "doc_artifact_ids": {}, "code_artifact_ids": {},
            "verdict": {"verdict": "PASS", "raw_response": "VERDICT: PASS"},
        }
        result = scheduler.finish_task_lifecycle(self.conn, run_result, "/tmp/proj-1", "goal text")
        self.assertEqual(result["final_status"], "done")
        status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
        self.assertEqual(status, "done")


class FinishTaskLifecycleFailTest(FinishTaskLifecycleTestBase):
    def test_fail_verdict_runs_fix_loop(self):
        run_result = {
            "task_id": "task-1",
            "doc_artifact_ids": {"benchmark": "artifact-b"},
            "code_artifact_ids": {},
            "verdict": {"verdict": "FAIL", "raw_response": "VERDICT: FAIL\nwall collision broken"},
        }
        with patch.object(scheduler.e2e_pipeline, "_read_artifact_content", return_value="benchmark content"), \
             patch.object(org_level_loop, "run_fix_loop", return_value={"final_status": "done"}) as mock_fix:
            result = scheduler.finish_task_lifecycle(self.conn, run_result, "/tmp/proj-1", "goal text")

        self.assertEqual(result["final_status"], "done")
        mock_fix.assert_called_once()
        args, kwargs = mock_fix.call_args
        self.assertEqual(args[0], self.conn)
        self.assertEqual(args[1], "task-1")
        self.assertEqual(args[2], "/tmp/proj-1")


class RunSchedulerOnceTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
        self.tmp.close()
        self.db_path = self.tmp.name
        conn = schema.connect(self.db_path)
        schema.create_schema(conn)
        conn.close()
        self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))

    def test_runs_all_goals_and_collects_results_in_order(self):
        def fake_run(db_path, target_folder_path, goal_text):
            return {"goal_text": goal_text, "final_status": "done"}

        goals = [("/tmp/p", "goal-{}".format(i)) for i in range(4)]
        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=2)

        self.assertEqual([r["goal_text"] for r in results], [g[1] for g in goals])
        self.assertTrue(all("result" in r for r in results))

    def test_one_goal_raising_does_not_stop_the_others(self):
        def fake_run(db_path, target_folder_path, goal_text):
            if goal_text == "goal-bad":
                raise RuntimeError("boom")
            return {"goal_text": goal_text, "final_status": "done"}

        goals = [("/tmp/p", "goal-ok-1"), ("/tmp/p", "goal-bad"), ("/tmp/p", "goal-ok-2")]
        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=3)

        self.assertIn("result", results[0])
        self.assertIn("error", results[1])
        self.assertEqual(results[1]["error"], "boom")
        self.assertIn("result", results[2])

    def test_max_workers_bounds_real_concurrency(self):
        current = [0]
        peak = [0]
        lock = threading.Lock()

        def fake_run(db_path, target_folder_path, goal_text):
            with lock:
                current[0] += 1
                peak[0] = max(peak[0], current[0])
            time.sleep(0.05)
            with lock:
                current[0] -= 1
            return {"goal_text": goal_text, "final_status": "done"}

        goals = [("/tmp/p", "goal-{}".format(i)) for i in range(6)]
        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=2)

        self.assertEqual(len(results), 6)
        self.assertLessEqual(peak[0], 2)
        self.assertEqual(peak[0], 2, "expected real parallelism up to the max_workers bound, not accidental serial execution")

    def test_defaults_max_workers_to_org_concurrency_config(self):
        conn = schema.connect(self.db_path)
        settings_config.upsert_concurrency_config(conn, max_concurrent_goals=3, updated_by="test")
        conn.close()

        current = [0]
        peak = [0]
        lock = threading.Lock()

        def fake_run(db_path, target_folder_path, goal_text):
            with lock:
                current[0] += 1
                peak[0] = max(peak[0], current[0])
            time.sleep(0.05)
            with lock:
                current[0] -= 1
            return {"goal_text": goal_text, "final_status": "done"}

        goals = [("/tmp/p", "goal-{}".format(i)) for i in range(6)]
        with patch.object(scheduler, "run_goal_to_completion", side_effect=fake_run):
            scheduler.run_scheduler_once(self.db_path, goals, max_workers=None)

        self.assertEqual(peak[0], 3)


class ConcurrentSqliteWritesStressTest(unittest.TestCase):
    """Not gated behind RUN_REAL_AGENT_E2E -- doesn't call the real Claude
    API, but DOES exercise the actual novel risk in this module's design
    (see module docstring's "Threading model" note): multiple threads, each
    holding its OWN sqlite connection, writing concurrently to the SAME db
    file. e2e_pipeline.run_end_to_end is faked (no real API calls) but
    finish_task_lifecycle's own close_task_done still runs for real against
    a real connection, so this is a genuine concurrent-write stress test,
    not a fully mocked one."""

    def setUp(self):
        self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
        self.tmp.close()
        self.db_path = self.tmp.name
        conn = schema.connect(self.db_path)
        schema.create_schema(conn)
        conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
        )
        conn.commit()
        conn.close()
        self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))

    def test_many_concurrent_goals_write_to_the_same_db_file_without_locking_errors(self):
        def fake_run_end_to_end(conn, target_folder_path, goal_text):
            task_id = goal_text  # unique per goal, stands in for a real uuid
            conn.execute(
                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
                "VALUES (?, 'proj-1', ?, 'goal', 'validating', 'validator')",
                (task_id, goal_text),
            )
            conn.commit()
            time.sleep(0.02)  # widen the contention window
            return {
                "task_id": task_id, "doc_artifact_ids": {}, "code_artifact_ids": {},
                "verdict": {"verdict": "PASS", "raw_response": "VERDICT: PASS"},
            }

        goals = [("/tmp/proj-1", "goal-{}".format(i)) for i in range(10)]
        with patch.object(scheduler.e2e_pipeline, "run_end_to_end", side_effect=fake_run_end_to_end):
            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=5)

        self.assertTrue(all("result" in r for r in results), results)

        conn = schema.connect(self.db_path)
        done_count = conn.execute("SELECT COUNT(*) FROM task WHERE status = 'done'").fetchone()[0]
        conn.close()
        self.assertEqual(done_count, 10)


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