"""U-007 tests: backend process start/stop, c_pick_folder route (osascript
mocked at the subprocess boundary), and the zero-third-party-dependency
constraint (05 §5.7) checked statically via AST, not just trusted.

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

import ast
import base64
import http.client
import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
import urllib.parse
from unittest.mock import patch

from src import backend_server, execution_log, skill_update, ws_subscribe


class ServerStartStopTest(unittest.TestCase):
    def test_can_start_serve_and_stop(self):
        # deliberately hits an unrelated route, not /api/system/pick-folder --
        # that one would pop a real, unmocked osascript dialog on this machine
        # and hang waiting for a human who isn't there.
        server = backend_server.create_server(db_path=":memory:")
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            host, port = server.server_address
            conn = http.client.HTTPConnection(host, port, timeout=5)
            conn.request("POST", "/api/does-not-exist", body=b"{}")
            resp = conn.getresponse()
            resp.read()
            conn.close()
            self.assertEqual(resp.status, 404)
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=5)
            self.assertFalse(thread.is_alive())

    def test_unknown_route_returns_404(self):
        server = backend_server.create_server(db_path=":memory:")
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            host, port = server.server_address
            conn = http.client.HTTPConnection(host, port, timeout=5)
            conn.request("POST", "/api/does-not-exist")
            resp = conn.getresponse()
            body = json.loads(resp.read())
            conn.close()
            self.assertEqual(resp.status, 404)
            self.assertEqual(body["error"], "NOT_FOUND")
        finally:
            server.shutdown()
            server.server_close()
            thread.join(timeout=5)


class StaticFileServingTest(unittest.TestCase):
    """U-008: frontend shell served as static files. Actual client-side
    routing behavior was verified in a real browser this session (nav
    highlighting + content swap across all 3 routes, no page reload) --
    that's not something a stdlib-only test can exercise without a JS
    engine, so this only covers what the server is responsible for."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = resp.read()
        content_type = resp.getheader("Content-Type")
        conn.close()
        return resp.status, content_type, body

    def test_root_serves_index_html(self):
        status, content_type, body = self._get("/")
        self.assertEqual(status, 200)
        self.assertIn("text/html", content_type)
        self.assertIn(b"<main id=\"app\">", body)
        self.assertIn(b'href="#/tasks"', body)

    def test_app_js_served_with_js_content_type(self):
        status, content_type, body = self._get("/app.js")
        self.assertEqual(status, 200)
        self.assertIn("javascript", content_type)
        self.assertIn(b"hashchange", body)
        self.assertIn(b'"#/tasks"', body)
        self.assertIn(b'"#/settings/concurrency"', body)

    def test_style_css_served_with_css_content_type(self):
        status, content_type, body = self._get("/style.css")
        self.assertEqual(status, 200)
        self.assertIn("text/css", content_type)

    def test_unknown_get_path_returns_404(self):
        status, _, body = self._get("/nope")
        self.assertEqual(status, 404)
        self.assertEqual(json.loads(body)["error"], "NOT_FOUND")


class HealthRouteTest(unittest.TestCase):
    """U-038: c_deploy_supervisor_health_check's http_health probe target."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = resp.read()
        conn.close()
        return resp.status, body

    def test_returns_200_ok_when_db_connection_is_usable(self):
        status, body = self._get("/api/health")
        self.assertEqual(status, 200)
        self.assertEqual(json.loads(body), {"status": "ok"})

    def test_returns_503_when_db_connection_is_broken(self):
        self.server.conn.close()  # simulate a wedged/closed connection
        status, body = self._get("/api/health")
        self.assertEqual(status, 503)
        self.assertEqual(json.loads(body)["status"], "error")


class PickFolderTest(unittest.TestCase):
    def test_success_returns_selected_path(self):
        with patch("src.backend_server.platform.system", return_value="Darwin"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["osascript"], returncode=0, stdout="/Users/x/some-project\n", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": "/Users/x/some-project"})

    def test_user_cancel_returns_null_path_not_an_error(self):
        with patch("src.backend_server.platform.system", return_value="Darwin"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["osascript"], returncode=1, stdout="",
                stderr="execution error: User canceled. (-128)",
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": None})

    def test_osascript_failure_other_than_cancel_raises_dialog_unavailable(self):
        with patch("src.backend_server.platform.system", return_value="Darwin"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["osascript"], returncode=1, stdout="", stderr="some other AppleScript error",
            )
            with self.assertRaises(backend_server.DialogUnavailableError):
                backend_server.pick_folder()

    def test_windows_success_returns_selected_path(self):
        with patch("src.backend_server.platform.system", return_value="Windows"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["powershell"], returncode=0, stdout="C:\\Users\\x\\some-project\r\n", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": "C:\\Users\\x\\some-project"})

    def test_windows_user_cancel_returns_null_path(self):
        with patch("src.backend_server.platform.system", return_value="Windows"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["powershell"], returncode=0, stdout="", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": None})

    def test_windows_missing_powershell_raises_dialog_unavailable(self):
        with patch("src.backend_server.platform.system", return_value="Windows"), \
             patch("src.backend_server.subprocess.run", side_effect=FileNotFoundError("no powershell")):
            with self.assertRaises(backend_server.DialogUnavailableError):
                backend_server.pick_folder()

    def test_linux_zenity_success_returns_selected_path(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["zenity"], returncode=0, stdout="/home/x/some-project\n", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": "/home/x/some-project"})
        run.assert_called_once()
        self.assertEqual(run.call_args[0][0][0], "zenity")

    def test_linux_zenity_cancel_returns_null_path(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["zenity"], returncode=1, stdout="", stderr=""
            )
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": None})

    def test_linux_falls_back_to_kdialog_when_zenity_missing(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run") as run:
            run.side_effect = [
                FileNotFoundError("no zenity"),
                subprocess.CompletedProcess(args=["kdialog"], returncode=0, stdout="/home/x/proj\n", stderr=""),
            ]
            result = backend_server.pick_folder()
        self.assertEqual(result, {"selected_path": "/home/x/proj"})
        self.assertEqual(run.call_count, 2)

    def test_linux_neither_zenity_nor_kdialog_raises_dialog_unavailable(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run", side_effect=FileNotFoundError("not found")):
            with self.assertRaises(backend_server.DialogUnavailableError):
                backend_server.pick_folder()


class PickFolderRouteTest(unittest.TestCase):
    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _post(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=b"{}")
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_route_returns_selected_path_on_success(self):
        with patch("src.backend_server.platform.system", return_value="Darwin"), \
             patch("src.backend_server.subprocess.run") as run:
            run.return_value = subprocess.CompletedProcess(
                args=["osascript"], returncode=0, stdout="/tmp/chosen\n", stderr=""
            )
            status, body = self._post("/api/system/pick-folder")
        self.assertEqual(status, 200)
        self.assertEqual(body, {"selected_path": "/tmp/chosen"})

    def test_route_returns_422_dialog_unavailable_when_no_dialog_tool_present(self):
        with patch("src.backend_server.platform.system", return_value="Linux"), \
             patch("src.backend_server.subprocess.run", side_effect=FileNotFoundError("not found")):
            status, body = self._post("/api/system/pick-folder")
        self.assertEqual(status, 422)
        self.assertEqual(body["error"], "DIALOG_UNAVAILABLE")


class ProjectCreateRouteTest(unittest.TestCase):
    """c_project_create (03-C §3.0) wired to a real HTTP route for the first
    time in U-022 -- project_scaffold.create_project() itself was already
    fully implemented and tested in U-002, so this only exercises the
    request/response/error-code plumbing, not the scaffolding logic again."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _post(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_creates_project_for_existing_git_repo(self):
        status, body = self._post("/api/projects", {"target_folder_path": self.tmp.name})
        self.assertEqual(status, 200)
        self.assertTrue(body["has_git_repo"])
        self.assertTrue(body["graphlemon_subdir_initialized"])
        self.assertTrue(os.path.isdir(os.path.join(self.tmp.name, ".graphlemon")))

    def test_repeat_call_is_idempotent(self):
        status1, body1 = self._post("/api/projects", {"target_folder_path": self.tmp.name})
        status2, body2 = self._post("/api/projects", {"target_folder_path": self.tmp.name})
        self.assertEqual(status1, 200)
        self.assertEqual(status2, 200)
        self.assertEqual(body1["project_id"], body2["project_id"])

    def test_missing_target_folder_path_returns_400(self):
        status, body = self._post("/api/projects", {})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_INPUT")

    def test_nonexistent_path_returns_404(self):
        status, body = self._post("/api/projects", {"target_folder_path": "/no/such/path/at/all"})
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "PATH_NOT_FOUND")

    def test_non_git_folder_without_confirmation_returns_409(self):
        non_git = tempfile.mkdtemp()
        self.addCleanup(lambda: shutil.rmtree(non_git, ignore_errors=True))
        status, body = self._post("/api/projects", {"target_folder_path": non_git})
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "GIT_INIT_DECLINED")

    def test_non_git_folder_with_confirmation_initializes_git(self):
        non_git = tempfile.mkdtemp()
        self.addCleanup(lambda: shutil.rmtree(non_git, ignore_errors=True))
        status, body = self._post(
            "/api/projects", {"target_folder_path": non_git, "init_git_if_missing": True}
        )
        self.assertEqual(status, 200)
        self.assertTrue(body["has_git_repo"])
        self.assertTrue(os.path.isdir(os.path.join(non_git, ".git")))


class ProjectListAndCrudRouteTest(unittest.TestCase):
    """v_project_list (2026-08-03): GET /api/projects (list), GET/PUT/DELETE
    /api/projects/:id -- project_query.py's own logic is already covered by
    test_project_query.py, this only exercises the HTTP wiring."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P1', '/tmp/proj-1', 1, 'active')"
        )
        self.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def _put(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("PUT", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def _delete(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("DELETE", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_list_returns_all_projects(self):
        status, body = self._get("/api/projects")
        self.assertEqual(status, 200)
        self.assertEqual([p["project_id"] for p in body], ["proj-1"])

    def test_get_returns_project_detail(self):
        status, body = self._get("/api/projects/proj-1")
        self.assertEqual(status, 200)
        self.assertEqual(body["name"], "P1")

    def test_get_nonexistent_returns_404(self):
        status, body = self._get("/api/projects/no-such-project")
        self.assertEqual(status, 404)

    def test_put_renames_project(self):
        status, body = self._put("/api/projects/proj-1", {"name": "renamed"})
        self.assertEqual(status, 200)
        self.assertEqual(body["name"], "renamed")

    def test_delete_archives_not_removes(self):
        status, body = self._delete("/api/projects/proj-1")
        self.assertEqual(status, 200)
        self.assertEqual(body["status"], "archived")
        status, body = self._get("/api/projects/proj-1")
        self.assertEqual(status, 200)  # row still exists


class TaskCrudRouteTest(unittest.TestCase):
    """v_goal_list (2026-08-03): sort/order on GET /api/tasks, PUT/DELETE
    /api/tasks/:id -- task_query.py's own logic is already covered by
    test_task_query.py, this only exercises the HTTP wiring."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P1', '/tmp/proj-1', 1, 'active')"
        )
        self.server.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', 'proj-1', 'old goal', 'goal', 'queued', 'planner')"
        )
        self.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def _put(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("PUT", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def _delete(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("DELETE", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_list_invalid_sort_returns_400(self):
        status, body = self._get("/api/tasks?project_id=proj-1&sort=" + urllib.parse.quote("x;DROP TABLE task"))
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_list_sort_asc_by_goal_text(self):
        self.server.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-2', 'proj-1', 'aaa', 'goal', 'queued', 'planner')"
        )
        self.server.conn.commit()
        status, body = self._get("/api/tasks?project_id=proj-1&sort=goal_text&order=asc")
        self.assertEqual(status, 200)
        self.assertEqual([t["task_id"] for t in body], ["task-2", "task-1"])  # aaa, old goal

    def test_put_edits_goal_text_while_queued(self):
        status, body = self._put("/api/tasks/task-1", {"goal_text": "new goal"})
        self.assertEqual(status, 200)
        self.assertEqual(body["goal_text"], "new goal")

    def test_put_missing_goal_text_returns_400(self):
        status, body = self._put("/api/tasks/task-1", {})
        self.assertEqual(status, 400)

    def test_put_on_non_queued_task_returns_409(self):
        self.server.conn.execute("UPDATE task SET status = 'coding' WHERE task_id = 'task-1'")
        self.server.conn.commit()
        status, body = self._put("/api/tasks/task-1", {"goal_text": "new goal"})
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "TASK_NOT_EDITABLE")

    def test_delete_removes_queued_task(self):
        status, body = self._delete("/api/tasks/task-1")
        self.assertEqual(status, 200)
        status, _ = self._get("/api/tasks/task-1")
        self.assertEqual(status, 404)

    def test_delete_on_non_queued_task_returns_409(self):
        self.server.conn.execute("UPDATE task SET status = 'done' WHERE task_id = 'task-1'")
        self.server.conn.commit()
        status, body = self._delete("/api/tasks/task-1")
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "TASK_NOT_EDITABLE")


class TaskAndBugReportRouteTest(unittest.TestCase):
    """U-023: c_task_create (03-C §3.1) + c_bug_report_create (03-C §3.1.1)
    wired to real HTTP routes for v_goal_composer/v_bug_feedback."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.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.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _post(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_creates_task(self):
        status, body = self._post(
            "/api/tasks", {"project_id": "proj-1", "goal_text": "make a snake game", "created_via": "human_console"}
        )
        self.assertEqual(status, 200)
        self.assertEqual(body["status"], "queued")
        self.assertIn("task_id", body)

    def test_empty_goal_text_returns_400(self):
        status, body = self._post("/api/tasks", {"project_id": "proj-1", "goal_text": ""})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_nonexistent_project_id_returns_400(self):
        status, body = self._post("/api/tasks", {"project_id": "no-such-project", "goal_text": "x"})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_creates_bug_report(self):
        status, body = self._post(
            "/api/bug-reports", {"project_id": "proj-1", "description": "snake clips through walls"}
        )
        self.assertEqual(status, 200)
        self.assertEqual(body["status"], "open")
        self.assertIn("bug_id", body)

    def test_bug_report_with_related_task_id(self):
        _, task_body = self._post("/api/tasks", {"project_id": "proj-1", "goal_text": "x"})
        status, body = self._post(
            "/api/bug-reports",
            {"project_id": "proj-1", "description": "x", "related_task_id": task_body["task_id"]},
        )
        self.assertEqual(status, 200)

    def test_empty_description_returns_400(self):
        status, body = self._post("/api/bug-reports", {"project_id": "proj-1", "description": ""})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_bug_report_with_bad_project_id_returns_400_not_500(self):
        # a raw sqlite3.IntegrityError propagating uncaught here was a real
        # bug this smoke test caught (stale/bad project_id crashed the
        # request instead of a clean 400)
        status, body = self._post("/api/bug-reports", {"project_id": "no-such-project", "description": "x"})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")


class TaskQueryAndRouteRouteTest(unittest.TestCase):
    """U-025: c_task_list/c_task_get (03-C §3.2), c_org_loop_route (03-C
    §3.11, first HTTP wiring since U-016 built the underlying function),
    and the budget-config read route for v_dashboard's budget bar."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.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.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def _post(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_task_list_requires_project_id(self):
        status, body = self._get("/api/tasks")
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_INPUT")

    def test_task_list_returns_created_tasks(self):
        self._post("/api/tasks", {"project_id": "proj-1", "goal_text": "x"})
        status, body = self._get("/api/tasks?project_id=proj-1")
        self.assertEqual(status, 200)
        self.assertEqual(len(body), 1)

    def test_task_list_status_filter(self):
        self._post("/api/tasks", {"project_id": "proj-1", "goal_text": "x"})
        status, body = self._get("/api/tasks?project_id=proj-1&status=done")
        self.assertEqual(status, 200)
        self.assertEqual(body, [])

    def test_task_get_returns_full_detail(self):
        _, created = self._post("/api/tasks", {"project_id": "proj-1", "goal_text": "x"})
        status, body = self._get("/api/tasks/" + created["task_id"])
        self.assertEqual(status, 200)
        self.assertEqual(body["task_id"], created["task_id"])
        self.assertIn("artifacts", body)
        self.assertIn("bug_records", body)
        self.assertIn("work_graph_runs", body)

    def test_task_get_nonexistent_returns_404(self):
        status, body = self._get("/api/tasks/no-such-task")
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "NOT_FOUND")

    def test_route_manual_resume_on_blocked_task(self):
        _, created = self._post("/api/tasks", {"project_id": "proj-1", "goal_text": "x"})
        task_id = created["task_id"]
        with self.server.conn_lock:
            self.server.conn.execute(
                "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'loop_limit_reached' "
                "WHERE task_id = ?",
                (task_id,),
            )
            # _manual_resume derives the pre-blocked status to restore from
            # execution_event history (org_loop_route.py) -- a real block
            # always writes this event on the way in, so the test does too.
            self.server.conn.execute(
                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
                "VALUES ('evt-block', 'proj-1', ?, 'task_status_changed', "
                "'{\"from_status\": \"queued\", \"to_status\": \"blocked_pending_human\"}')",
                (task_id,),
            )
            self.server.conn.commit()
        status, body = self._post("/api/tasks/" + task_id + "/route", {"decided_route": "manual_resume"})
        self.assertEqual(status, 200)
        self.assertEqual(body["task_id"], task_id)

    def test_route_invalid_decided_route_returns_400(self):
        _, created = self._post("/api/tasks", {"project_id": "proj-1", "goal_text": "x"})
        status, body = self._post(
            "/api/tasks/" + created["task_id"] + "/route", {"decided_route": "not_a_real_route"}
        )
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_budget_settings_empty_when_unconfigured(self):
        status, body = self._get("/api/settings/budget")
        self.assertEqual(status, 200)
        self.assertEqual(body, [])

    def test_budget_settings_returns_configured_row(self):
        with self.server.conn_lock:
            self.server.conn.execute(
                "INSERT INTO budget_guardrail_config (config_id, scope, limit_type, limit_value, "
                "current_usage, updated_by) VALUES ('default', 'global', 'token_count', 1000, 10, 'test')"
            )
            self.server.conn.commit()
        status, body = self._get("/api/settings/budget")
        self.assertEqual(status, 200)
        self.assertEqual(len(body), 1)
        self.assertEqual(body[0]["limit_value"], 1000)
        self.assertEqual(body[0]["current_usage"], 10)


class TaskExecutionRouteTest(unittest.TestCase):
    """v_task_detail Start/Pause/Stop (2026-08-03): c_task_execute,
    c_task_pause -- task_execution.py's own logic (resume-from-status,
    pause mid-flight) is already covered by test_task_execution.py, this
    only exercises the HTTP wiring. `start_task_execution_in_background`
    is mocked throughout: it spawns a real background thread that opens
    its OWN sqlite connection via `schema.connect(db_path)`, which for a
    ':memory:' test db would silently create a second, disconnected
    database -- exactly the kind of lower-level mechanic the codebase's
    own convention (see other route test classes) says belongs in
    test_task_execution.py, not re-tested at the HTTP layer."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P1', '/tmp/proj-1', 1, 'active')"
        )
        self.server.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', 'proj-1', 'the goal', 'goal', 'queued', 'planner')"
        )
        self.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _post(self, path, payload=None):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        body = json.dumps(payload).encode("utf-8") if payload is not None else b""
        conn.request("POST", path, body=body)
        resp = conn.getresponse()
        raw = resp.read()
        body = json.loads(raw) if raw else None
        conn.close()
        return resp.status, body

    def test_execute_starts_queued_task(self):
        with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
            status, body = self._post("/api/tasks/task-1/execute")
        self.assertEqual(status, 200)
        self.assertEqual(body, {"task_id": "task-1", "started": True})
        mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")

    def test_execute_nonexistent_task_returns_404(self):
        status, body = self._post("/api/tasks/no-such-task/execute")
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "NOT_FOUND")

    def test_execute_non_queued_task_returns_409(self):
        self.server.conn.execute("UPDATE task SET status = 'coding' WHERE task_id = 'task-1'")
        self.server.conn.commit()
        status, body = self._post("/api/tasks/task-1/execute")
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "TASK_NOT_QUEUED")

    def test_pause_blocks_in_flight_task(self):
        self.server.conn.execute("UPDATE task SET status = 'coding' WHERE task_id = 'task-1'")
        self.server.conn.commit()
        status, body = self._post("/api/tasks/task-1/pause")
        self.assertEqual(status, 200)
        self.assertEqual(body, {"task_id": "task-1", "status": "blocked_pending_human", "blocked_reason": "user_paused"})
        row = self.server.conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = 'task-1'").fetchone()
        self.assertEqual(tuple(row), ("blocked_pending_human", "user_paused"))

    def test_stop_blocks_in_flight_task_with_distinct_reason(self):
        self.server.conn.execute("UPDATE task SET status = 'planning_brd_prd_sdd' WHERE task_id = 'task-1'")
        self.server.conn.commit()
        status, body = self._post("/api/tasks/task-1/stop")
        self.assertEqual(status, 200)
        self.assertEqual(body["blocked_reason"], "user_stopped")

    def test_pause_on_queued_task_returns_409(self):
        status, body = self._post("/api/tasks/task-1/pause")
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "TASK_NOT_PAUSABLE")

    def test_manual_resume_after_user_pause_auto_continues(self):
        self.server.conn.execute(
            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'user_paused' "
            "WHERE task_id = 'task-1'"
        )
        self.server.conn.execute(
            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
            "'{\"from_status\": \"coding\", \"to_status\": \"blocked_pending_human\"}')"
        )
        self.server.conn.commit()
        with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
        self.assertEqual(status, 200)
        mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")

    def test_manual_resume_after_system_block_does_not_auto_continue(self):
        self.server.conn.execute(
            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'loop_limit_reached' "
            "WHERE task_id = 'task-1'"
        )
        self.server.conn.execute(
            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
            "'{\"from_status\": \"queued\", \"to_status\": \"blocked_pending_human\"}')"
        )
        self.server.conn.commit()
        with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
        self.assertEqual(status, 200)
        mock_start.assert_not_called()

    def test_manual_resume_after_budget_exceeded_does_not_auto_continue(self):
        # 2026-08-04: broadening auto-continue to "everything except
        # loop_limit_reached/budget_exceeded" must not accidentally sweep
        # budget_exceeded back in -- it has its own dedicated resume trigger
        # (raising the limit), not a blind retry on the same budget wall.
        self.server.conn.execute(
            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'budget_exceeded' "
            "WHERE task_id = 'task-1'"
        )
        self.server.conn.execute(
            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
            "'{\"from_status\": \"coding\", \"to_status\": \"blocked_pending_human\"}')"
        )
        self.server.conn.commit()
        with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
        self.assertEqual(status, 200)
        mock_start.assert_not_called()

    def test_manual_resume_after_execution_error_auto_continues(self):
        # 2026-08-04 regression: found for real -- a transient network blip
        # ("API Error: Unable to connect to API (ENOTFOUND)") blocks a task
        # with blocked_reason='execution_error'. Resuming it used to just
        # flip the status field back with no way to actually continue
        # (/execute requires status='queued', but a resumed task lands on
        # 'planning_brd_prd_sdd'/'coding' instead) -- a dead end a second
        # click couldn't fix.
        self.server.conn.execute(
            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'execution_error' "
            "WHERE task_id = 'task-1'"
        )
        self.server.conn.execute(
            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
            "'{\"from_status\": \"planning_brd_prd_sdd\", \"to_status\": \"blocked_pending_human\", "
            "\"detail\": \"API Error: Unable to connect to API (ENOTFOUND)\"}')"
        )
        self.server.conn.commit()
        with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
        self.assertEqual(status, 200)
        mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")

    def test_manual_resume_from_validating_auto_continues(self):
        # 2026-08-07 real incident: a task stuck in 'validating' (the
        # readiness trigger already ran once -- see task_execution.py's
        # 'validating' resume section) got its status flipped back by
        # manual_resume but nothing ever actually continued it, because
        # this allowlist hadn't been updated to match task_execution.
        # run_or_resume_task_lifecycle's own 'validating' support added
        # earlier the same day.
        self.server.conn.execute(
            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'execution_error' "
            "WHERE task_id = 'task-1'"
        )
        self.server.conn.execute(
            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
            "'{\"from_status\": \"validating\", \"to_status\": \"blocked_pending_human\"}')"
        )
        self.server.conn.commit()
        with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
        self.assertEqual(status, 200)
        mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")

    def test_execute_already_in_flight_returns_409_not_500(self):
        with patch.object(
            backend_server.task_execution, "start_task_execution_in_background",
            side_effect=backend_server.task_execution.TaskExecutionError("task task-1 already has an execution in flight"),
        ):
            status, body = self._post("/api/tasks/task-1/execute")
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "ALREADY_IN_FLIGHT")

    def test_manual_resume_still_returns_200_when_already_in_flight(self):
        # The route_org_loop transition itself already succeeded; losing a
        # race to dispatch (e.g. the reconciliation loop got there a moment
        # earlier) isn't a failure of the resume request itself.
        self.server.conn.execute(
            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'execution_error' "
            "WHERE task_id = 'task-1'"
        )
        self.server.conn.execute(
            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
            "'{\"from_status\": \"coding\", \"to_status\": \"blocked_pending_human\"}')"
        )
        self.server.conn.commit()
        with patch.object(
            backend_server.task_execution, "start_task_execution_in_background",
            side_effect=backend_server.task_execution.TaskExecutionError("already in flight"),
        ):
            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
        self.assertEqual(status, 200)


def _ws_handshake_request(path, key):
    return (
        "GET {} HTTP/1.1\r\n"
        "Host: 127.0.0.1\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        "Sec-WebSocket-Key: {}\r\n"
        "Sec-WebSocket-Version: 13\r\n"
        "\r\n"
    ).format(path, key).encode("ascii")


def _read_http_response_head(sock):
    """Reads until the blank line terminating an HTTP response's headers --
    good enough for both a 101 upgrade and a plain error response, since
    neither carries a body in this handler."""
    buf = b""
    while b"\r\n\r\n" not in buf:
        chunk = sock.recv(4096)
        if not chunk:
            break
        buf += chunk
    return buf.decode("ascii", errors="replace")


def _read_http_response_with_json_body(sock):
    """Like _read_http_response_head, but also reads and parses the JSON
    body (via Content-Length) -- for pre-upgrade WS rejections, which DO
    send a body (_send_json), unlike a 101 Switching Protocols response."""
    buf = b""
    while b"\r\n\r\n" not in buf:
        chunk = sock.recv(4096)
        if not chunk:
            break
        buf += chunk
    head, _, rest = buf.partition(b"\r\n\r\n")
    head_text = head.decode("ascii", errors="replace")
    content_length = 0
    for line in head_text.splitlines():
        if line.lower().startswith("content-length:"):
            content_length = int(line.split(":", 1)[1].strip())
            break
    while len(rest) < content_length:
        chunk = sock.recv(4096)
        if not chunk:
            break
        rest += chunk
    return head_text, json.loads(rest.decode("utf-8"))


def _read_server_frame_raw(sock):
    """Server->client frames are never masked (RFC 6455 §5.1) -- simpler
    decode than ws_subscribe.decode_client_frame, which expects masking.
    Returns (opcode, payload_bytes)."""
    first_two = _recv_exact(sock, 2)
    opcode = first_two[0] & 0x0F
    length = first_two[1] & 0x7F
    if length == 126:
        length = int.from_bytes(_recv_exact(sock, 2), "big")
    elif length == 127:
        length = int.from_bytes(_recv_exact(sock, 8), "big")
    payload = _recv_exact(sock, length)
    return opcode, payload


def _read_server_frame(sock):
    _opcode, payload = _read_server_frame_raw(sock)
    return json.loads(payload.decode("utf-8"))


def _recv_exact(sock, n):
    chunks = []
    remaining = n
    while remaining > 0:
        chunk = sock.recv(remaining)
        if not chunk:
            raise ConnectionError("socket closed")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


class WsSubscribeRouteTest(unittest.TestCase):
    """U-024: c_ws_subscribe end-to-end over a real raw socket -- the actual
    RFC 6455 handshake + framing, not mocked, since that's exactly the
    hand-rolled logic (no `websockets` dependency, 05 §5.7) most likely to
    have a real bug."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.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.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _connect(self):
        host, port = self.server.server_address
        return socket.create_connection((host, port), timeout=5)

    def test_handshake_succeeds_with_correct_accept_key(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1", key))
        head = _read_http_response_head(sock)
        self.assertIn("101 Switching Protocols", head.splitlines()[0])
        expected_accept = ws_subscribe.compute_accept_key(key)
        self.assertIn("Sec-WebSocket-Accept: " + expected_accept, head)

    def test_nonexistent_project_rejected_before_upgrade(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/no-such-project", key))
        head, body = _read_http_response_with_json_body(sock)
        self.assertIn("404", head.splitlines()[0])
        self.assertNotIn("101", head.splitlines()[0])
        self.assertEqual(body["error"], "PROJECT_NOT_FOUND")

    def test_non_upgrade_request_rejected(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        sock.sendall(b"GET /api/ws/projects/proj-1 HTTP/1.1\r\nHost: x\r\n\r\n")
        head = _read_http_response_head(sock)
        self.assertIn("400", head.splitlines()[0])

    def test_new_execution_event_pushed_within_two_seconds(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1", key))
        _read_http_response_head(sock)

        start = time.monotonic()
        with self.server.conn_lock:
            self.server.conn.execute(
                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
                "VALUES ('evt-1', 'proj-1', NULL, 'budget_updated', '{\"x\": 1}')"
            )
            self.server.conn.commit()

        sock.settimeout(5)
        event = _read_server_frame(sock)
        elapsed = time.monotonic() - start
        self.assertLess(elapsed, 2.0)
        self.assertEqual(event["event_id"], "evt-1")
        self.assertEqual(event["event_type"], "budget_updated")
        self.assertEqual(event["payload"], {"x": 1})

    def test_task_id_filter_excludes_unrelated_events(self):
        with self.server.conn_lock:
            self.server.conn.execute(
                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
                "VALUES ('task-1', 'proj-1', 'g', 'goal', 'queued', 'planner')"
            )
            self.server.conn.commit()

        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/projects/proj-1?task_id=task-1", key))
        _read_http_response_head(sock)

        with self.server.conn_lock:
            self.server.conn.execute(
                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
                "VALUES ('evt-unrelated', 'proj-1', NULL, 'budget_updated', '{}')"
            )
            self.server.conn.execute(
                "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
                "VALUES ('evt-related', 'proj-1', 'task-1', 'task_status_changed', '{}')"
            )
            self.server.conn.commit()

        sock.settimeout(5)
        event = _read_server_frame(sock)
        self.assertEqual(event["event_id"], "evt-related")


class ExecutionLogRouteTest(unittest.TestCase):
    """U-027: c_execution_log_get (GET /api/work-nodes/:node_exec_id/log)
    and c_ws_log_tail (WS /api/ws/work-nodes/:node_exec_id/log)."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        conn = self.server.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.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
        )
        conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
            "VALUES ('run-1', 'task-1', '{}', 'running')"
        )
        conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')"
        )
        conn.execute(
            "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
            "input_schema_ref, output_schema_ref, input_payload, status) "
            "VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
        )
        conn.execute(
            "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
            "VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', 'active')",
            (os.path.join(self.tmp.name, "node-1.jsonl"),),
        )
        conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = resp.read()
        headers = dict(resp.getheaders())
        conn.close()
        return resp.status, body, headers

    def test_returns_static_content(self):
        with self.server.conn_lock:
            execution_log.append_log(self.server.conn, "node-1", "hello world")
        status, body, _ = self._get("/api/work-nodes/node-1/log")
        self.assertEqual(status, 200)
        parsed = json.loads(body)
        self.assertEqual(parsed["content"], "hello world")
        self.assertEqual(parsed["total_size_bytes"], len("hello world"))

    def test_pagination(self):
        with self.server.conn_lock:
            execution_log.append_log(self.server.conn, "node-1", "0123456789")
        status, body, _ = self._get("/api/work-nodes/node-1/log?offset=2&limit=3")
        self.assertEqual(status, 200)
        self.assertEqual(json.loads(body)["content"], "234")

    def test_download_returns_raw_file_with_attachment_headers(self):
        with self.server.conn_lock:
            execution_log.append_log(self.server.conn, "node-1", "raw log bytes")
        status, body, headers = self._get("/api/work-nodes/node-1/log?download=true")
        self.assertEqual(status, 200)
        self.assertEqual(body, b"raw log bytes")
        self.assertIn("attachment", headers.get("Content-Disposition", ""))

    def test_nonexistent_node_returns_404_log_not_found(self):
        status, body, _ = self._get("/api/work-nodes/no-such-node/log")
        self.assertEqual(status, 404)
        self.assertEqual(json.loads(body)["error"], "LOG_NOT_FOUND")

    def test_download_when_backing_file_missing_returns_clean_404_not_uncaught_exception(self):
        # Bug A2: execution_log_file row exists (inserted in setUp) but its
        # file_path was never actually written to disk -- append_log() is
        # what creates the file, and this test deliberately never calls it,
        # matching the exact real-world state before DU-01's fix (a node
        # that errored before its first log write, or any other row whose
        # backing file went missing). _send_file_download() must catch this
        # itself, not let a bare FileNotFoundError escape as an uncaught 500.
        self.assertFalse(os.path.exists(os.path.join(self.tmp.name, "node-1.jsonl")))
        status, body, _ = self._get("/api/work-nodes/node-1/log?download=true")
        self.assertEqual(status, 404)
        self.assertEqual(json.loads(body)["error"], "LOG_FILE_MISSING")

    def test_purged_log_returns_410(self):
        with self.server.conn_lock:
            self.server.conn.execute("UPDATE execution_log_file SET status = 'purged' WHERE log_file_id = 'log-1'")
            self.server.conn.commit()
        status, body, _ = self._get("/api/work-nodes/node-1/log")
        self.assertEqual(status, 410)
        self.assertEqual(json.loads(body)["error"], "LOG_PURGED")

    def _connect(self):
        host, port = self.server.server_address
        return socket.create_connection((host, port), timeout=5)

    def test_ws_handshake_succeeds_for_existing_node(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
        head = _read_http_response_head(sock)
        self.assertIn("101 Switching Protocols", head.splitlines()[0])

    def test_ws_nonexistent_node_rejected_before_upgrade(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/work-nodes/no-such-node/log", key))
        head, body = _read_http_response_with_json_body(sock)
        self.assertIn("404", head.splitlines()[0])
        self.assertEqual(body["error"], "NODE_EXEC_NOT_FOUND")

    def test_ws_pushes_appended_log_content(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
        _read_http_response_head(sock)

        with self.server.conn_lock:
            execution_log.append_log(self.server.conn, "node-1", "streamed chunk")

        sock.settimeout(5)
        event = _read_server_frame(sock)
        self.assertEqual(event["appended_content"], "streamed chunk")

    def test_ws_closes_when_node_reaches_terminal_status(self):
        sock = self._connect()
        self.addCleanup(sock.close)
        key = base64.b64encode(b"0123456789012345").decode("ascii")
        sock.sendall(_ws_handshake_request("/api/ws/work-nodes/node-1/log", key))
        _read_http_response_head(sock)

        with self.server.conn_lock:
            self.server.conn.execute("UPDATE work_node_execution SET status = 'done' WHERE node_exec_id = 'node-1'")
            self.server.conn.commit()

        sock.settimeout(5)
        opcode, _payload = _read_server_frame_raw(sock)
        self.assertEqual(opcode, ws_subscribe.OPCODE_CLOSE)


class ArtifactRouteTest(unittest.TestCase):
    """U-028: c_artifact_list (GET /api/artifacts) + c_artifact_get
    (GET /api/artifacts/:artifact_id) for v_artifact_browser."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        conn = self.server.conn
        conn.execute(
            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
            "VALUES ('proj-1', 'P', ?, 1, 'active')",
            (self.tmp.name,),
        )
        conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
        )
        conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
            "VALUES ('run-1', 'task-1', '{}', 'done')"
        )
        with open(os.path.join(self.tmp.name, "sdd.md"), "w") as f:
            f.write("# SDD content")
        conn.execute(
            "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, "
            "git_commit_sha) VALUES ('art-1', 'sdd', 'task-1', 'run-1', 'sdd.md', 'sha1')"
        )
        conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_list_requires_project_id(self):
        status, body = self._get("/api/artifacts")
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_INPUT")

    def test_list_returns_artifacts_for_project(self):
        status, body = self._get("/api/artifacts?project_id=proj-1")
        self.assertEqual(status, 200)
        self.assertEqual(len(body), 1)
        self.assertEqual(body[0]["artifact_id"], "art-1")

    def test_list_artifact_type_filter(self):
        status, body = self._get("/api/artifacts?project_id=proj-1&artifact_type=brd")
        self.assertEqual(status, 200)
        self.assertEqual(body, [])

    def test_get_returns_content_and_lineage(self):
        status, body = self._get("/api/artifacts/art-1")
        self.assertEqual(status, 200)
        self.assertEqual(body["content"], "# SDD content")
        self.assertEqual(body["lineage"], [{"artifact_id": "art-1", "artifact_type": "sdd"}])

    def test_get_nonexistent_returns_404(self):
        status, body = self._get("/api/artifacts/no-such-artifact")
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "NOT_FOUND")


class WorkGraphRunListRouteTest(unittest.TestCase):
    """U-GRAPH-01: c_work_graph_run_list (GET /api/work-graph-runs), the
    one net-new backend endpoint for v_graph_console's FR-5 overview panel.
    Route-dispatch-level smoke test; sort/filter/limit edge cases are
    already covered at the pure-function level in
    src/test_work_graph_run_query.py, per that Unit's own test-design note
    ("不重复造已经在纯函数层测过的排序/过滤逻辑")."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        conn = self.server.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.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
        )
        conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status, started_at) "
            "VALUES ('run-1', 'task-1', '{}', 'done', '2026-08-01T00:00:00')"
        )
        conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_list_returns_runs_for_project(self):
        status, body = self._get("/api/work-graph-runs?project_id=proj-1")
        self.assertEqual(status, 200)
        self.assertEqual(len(body), 1)
        self.assertEqual(body[0]["run_id"], "run-1")
        self.assertEqual(body[0]["node_count"], 0)

    def test_list_requires_project_id(self):
        status, body = self._get("/api/work-graph-runs")
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_list_invalid_limit_returns_400(self):
        status, body = self._get("/api/work-graph-runs?project_id=proj-1&limit=101")
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_list_org_node_filter(self):
        # 2026-08-12 (Org Graph unification Phase 2c §2.1.1): org_node now
        # filters by whether any node in the run has that subagent_type --
        # this fixture's run-1 has zero nodes, so ANY org_node filter
        # (including one that would have matched the old run-level column)
        # now excludes it.
        status, body = self._get("/api/work-graph-runs?project_id=proj-1&org_node=planner")
        self.assertEqual(status, 200)
        self.assertEqual(body, [])

    def test_get_returns_run_with_nodes(self):
        status, body = self._get("/api/work-graph-runs/run-1")
        self.assertEqual(status, 200)
        self.assertEqual(body["run_id"], "run-1")
        self.assertEqual(body["task_id"], "task-1")
        # run-1 has zero nodes seeded above -- org_nodes_involved is derived
        # from work_node_execution.subagent_type (Phase 2c §2.1.1), so it's
        # empty here, not the old run-level org_node column's fixed value.
        self.assertEqual(body["org_nodes_involved"], [])
        self.assertEqual(body["nodes"], [])

    def test_get_nonexistent_run_returns_404(self):
        status, body = self._get("/api/work-graph-runs/no-such-run")
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "NOT_FOUND")


class WorkNodeForkRouteTest(unittest.TestCase):
    """D (2026-08-06, time-travel debugging): POST
    /api/work-nodes/:node_exec_id/fork. `start_fork_in_background` is
    mocked throughout -- same reasoning as TaskExecutionRouteTest just
    above: it opens its OWN sqlite connection via schema.connect(db_path),
    which for a ':memory:' test db would silently create a second,
    disconnected database. Real fork behavior is already covered in
    test_work_graph_fork.py (real file db, real threading)."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        conn = self.server.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.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', 'proj-1', 'g', 'goal', 'validating', 'planner')"
        )
        conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
            "VALUES ('run-1', 'task-1', '{}', 'done')"
        )
        conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES ('skill-1', 'planner', 'skills/skill-1/', 'sha', 'new', 'active')"
        )
        conn.execute(
            "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
            "input_schema_ref, output_schema_ref, input_payload, output_payload, status) "
            "VALUES ('node-1', 'run-1', 'prd', 'planner', 'skill-1', 'in.json', 'out.json', '{}', '{}', 'done')"
        )
        conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _post(self, path, payload=None):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        body = json.dumps(payload).encode("utf-8") if payload is not None else b""
        conn.request("POST", path, body=body)
        resp = conn.getresponse()
        raw = resp.read()
        body = json.loads(raw) if raw else None
        conn.close()
        return resp.status, body

    def test_fork_returns_run_id_and_calls_through_with_target_folder_path(self):
        with patch.object(backend_server.work_graph_fork, "start_fork_in_background") as mock_start:
            mock_start.return_value = ("run-fork-1", None)
            status, body = self._post(
                "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "new prd"}}
            )
        self.assertEqual(status, 200)
        self.assertEqual(body, {"run_id": "run-fork-1", "started": True})
        mock_start.assert_called_once_with(
            self.server.db_path, "node-1", {"content": "new prd"}, "/tmp/proj-1",
        )

    def test_fork_nonexistent_node_returns_404(self):
        status, body = self._post(
            "/api/work-nodes/no-such-node/fork", {"new_output_payload": {"content": "x"}}
        )
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "NOT_FOUND")

    def test_fork_missing_new_output_payload_returns_400(self):
        status, body = self._post("/api/work-nodes/node-1/fork", {})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_INPUT")

    def test_fork_source_not_done_returns_409(self):
        with patch.object(
            backend_server.work_graph_fork, "start_fork_in_background",
            side_effect=backend_server.work_graph_fork.ForkSourceNotDoneError("not done"),
        ):
            status, body = self._post(
                "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "x"}}
            )
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "FORK_SOURCE_NOT_DONE")

    def test_fork_parent_run_not_terminal_returns_409(self):
        with patch.object(
            backend_server.work_graph_fork, "start_fork_in_background",
            side_effect=backend_server.work_graph_fork.ForkParentRunNotTerminalError("still running"),
        ):
            status, body = self._post(
                "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "x"}}
            )
        self.assertEqual(status, 409)
        self.assertEqual(body["error"], "FORK_PARENT_RUN_NOT_TERMINAL")

    def test_fork_unsupported_graph_shape_returns_422(self):
        with patch.object(
            backend_server.work_graph_fork, "start_fork_in_background",
            side_effect=backend_server.work_graph_fork.ForkGraphShapeUnsupportedError("nope"),
        ):
            status, body = self._post(
                "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "x"}}
            )
        self.assertEqual(status, 422)
        self.assertEqual(body["error"], "FORK_GRAPH_SHAPE_UNSUPPORTED")


class DeploymentRouteTest(unittest.TestCase):
    """U-043: c_deployment_list/c_deployment_current/c_deployment_get for
    v_deployment_status."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        conn = self.server.conn
        conn.execute(
            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
            "previous_instance_dir) VALUES ('dep-1', 'c1', 'c0', 'finalized', '/prev')"
        )
        conn.execute(
            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
            "previous_instance_dir) VALUES ('dep-2', 'c2', 'c1', 'monitoring', '/prev2')"
        )
        conn.execute(
            "INSERT INTO deploy_health_signal (signal_id, deployment_id, sampled_at, signal_type, status) "
            "VALUES ('s1', 'dep-2', CURRENT_TIMESTAMP, 'process_alive', 'ok')"
        )
        conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_list_returns_all_newest_first(self):
        status, body = self._get("/api/deployments")
        self.assertEqual(status, 200)
        self.assertEqual([d["deployment_id"] for d in body], ["dep-2", "dep-1"])

    def test_current_returns_the_active_one(self):
        status, body = self._get("/api/deployments/current")
        self.assertEqual(status, 200)
        self.assertEqual(body["deployment_id"], "dep-2")

    def test_current_is_null_when_nothing_active(self):
        with self.server.conn_lock:
            self.server.conn.execute("UPDATE deployment SET status = 'finalized' WHERE deployment_id = 'dep-2'")
            self.server.conn.commit()
        status, body = self._get("/api/deployments/current")
        self.assertEqual(status, 200)
        self.assertIsNone(body)

    def test_get_includes_health_signals(self):
        status, body = self._get("/api/deployments/dep-2")
        self.assertEqual(status, 200)
        self.assertEqual(len(body["health_signals"]), 1)
        self.assertEqual(body["health_signals"][0]["signal_id"], "s1")

    def test_get_nonexistent_returns_404(self):
        status, body = self._get("/api/deployments/no-such-deployment")
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "NOT_FOUND")


class SkillRouteTest(unittest.TestCase):
    """U-029: c_skill_diff (GET /api/skills/diff) + c_skill_rollback
    (POST /api/skills/:skill_id/rollback) + list/get for v_skill_library /
    v_skill_audit_log. skill_repo_path points at a throwaway temp repo --
    NEVER the real graphlemon checkout, since c_skill_rollback does a real
    `git commit` (see backend_server.create_server's skill_repo_path note)."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
        self.server = backend_server.create_server(db_path=":memory:", skill_repo_path=self.tmp.name)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES ('skill-1', 'coder', 'skills/skill-1', 'unknown', 'new', 'active')"
        )
        self.server.conn.execute(
            "INSERT INTO bug_record (bug_id, source, description, root_cause_category, root_cause_detail, "
            "suggested_route, status) VALUES ('bug-1', 'validator', 'x', 'skill_defect', 'd', 'fast_lane', 'routed')"
        )
        self.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def _post(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_list_returns_seeded_skill(self):
        status, body = self._get("/api/skills")
        self.assertEqual(status, 200)
        self.assertEqual(len(body), 1)
        self.assertEqual(body[0]["skill_id"], "skill-1")

    def test_audit_log_empty_for_skill_with_no_iterations(self):
        status, body = self._get("/api/skills/skill-1/audit")
        self.assertEqual(status, 200)
        self.assertEqual(body, [])

    def test_audit_log_nonexistent_skill_returns_404(self):
        status, body = self._get("/api/skills/no-such-skill/audit")
        self.assertEqual(status, 404)

    def test_diff_missing_query_params_returns_400(self):
        status, body = self._get("/api/skills/diff")
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_INPUT")

    def test_full_update_diff_rollback_cycle(self):
        skill_update._write_and_commit(self.tmp.name, "skills/skill-1/SKILL.md", "v1 content", "seed skill")
        with self.server.conn_lock:
            result = skill_update.update_skill(
                self.server.conn, "skill-1", "bug-1", "v2 content", "improve it", self.tmp.name
            )
        iteration_id = result["iteration_id"]

        status, body = self._get(
            "/api/skills/diff?before=" + self._before_sha(iteration_id) + "&after=" + result["after_git_commit_sha"]
        )
        self.assertEqual(status, 200)
        self.assertIn("v2 content", body["diff_text"])

        status, body = self._post("/api/skills/skill-1/rollback", {"iteration_id": iteration_id})
        self.assertEqual(status, 200)
        self.assertIn("new_iteration_id", body)

    def _before_sha(self, iteration_id):
        return self.server.conn.execute(
            "SELECT before_git_commit_sha FROM skill_iteration WHERE iteration_id = ?", (iteration_id,)
        ).fetchone()[0]

    def test_rollback_nonexistent_iteration_returns_404(self):
        status, body = self._post("/api/skills/skill-1/rollback", {"iteration_id": "no-such-iteration"})
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "ITERATION_NOT_FOUND")

    def test_rollback_missing_iteration_id_returns_400(self):
        status, body = self._post("/api/skills/skill-1/rollback", {})
        self.assertEqual(status, 400)

    def test_content_returns_real_file_content(self):
        skill_update._write_and_commit(self.tmp.name, "skills/skill-1/SKILL.md", "real content\n", "seed skill")
        status, body = self._get("/api/skills/skill-1/content")
        self.assertEqual(status, 200)
        self.assertEqual(body["content"], "real content\n")
        self.assertEqual(body["support_files"], [])

    def test_content_nonexistent_skill_returns_404(self):
        status, body = self._get("/api/skills/no-such-skill/content")
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "NOT_FOUND")

    def test_content_missing_file_returns_404_content_missing(self):
        status, body = self._get("/api/skills/skill-1/content")
        self.assertEqual(status, 404)
        self.assertEqual(body["error"], "CONTENT_MISSING")


class SettingsRouteTest(unittest.TestCase):
    """U-030: c_concurrency_config_update / c_budget_config_update /
    c_log_retention_config_update -- the three v_*_settings pages."""

    def setUp(self):
        self.server = backend_server.create_server(db_path=":memory:")
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.server.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.server.conn.commit()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def _put(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("PUT", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_concurrency_get_returns_null_when_unconfigured(self):
        status, body = self._get("/api/settings/concurrency")
        self.assertEqual(status, 200)
        self.assertIsNone(body)

    def test_concurrency_update_and_get(self):
        status, body = self._put("/api/settings/concurrency", {"max_concurrent_goals": 8})
        self.assertEqual(status, 200)
        self.assertEqual(body["max_concurrent_goals"], 8)
        status, body = self._get("/api/settings/concurrency")
        self.assertEqual(body["max_concurrent_goals"], 8)

    def test_concurrency_out_of_range_returns_400(self):
        status, body = self._put("/api/settings/concurrency", {"max_concurrent_goals": 100})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_concurrency_missing_field_returns_400_not_500(self):
        status, body = self._put("/api/settings/concurrency", {})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_budget_update_creates_and_returns_config(self):
        status, body = self._put(
            "/api/settings/budget", {"scope": "global", "limit_type": "token_count", "limit_value": 5000}
        )
        self.assertEqual(status, 200)
        self.assertEqual(body["config"]["limit_value"], 5000)

    def test_budget_invalid_scope_returns_400(self):
        status, body = self._put(
            "/api/settings/budget", {"scope": "bogus", "limit_type": "token_count", "limit_value": 100}
        )
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_budget_missing_limit_value_returns_400_not_500(self):
        status, body = self._put("/api/settings/budget", {"scope": "global", "limit_type": "token_count"})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_log_retention_get_returns_null_when_unconfigured(self):
        status, body = self._get("/api/settings/log-retention")
        self.assertEqual(status, 200)
        self.assertIsNone(body)

    def test_log_retention_update_and_get(self):
        status, body = self._put("/api/settings/log-retention", {"retention_days": 30, "max_total_size_mb": 200})
        self.assertEqual(status, 200)
        self.assertEqual(body["retention_days"], 30)
        status, body = self._get("/api/settings/log-retention")
        self.assertEqual(body["max_total_size_mb"], 200)

    def test_log_retention_zero_days_returns_400(self):
        status, body = self._put("/api/settings/log-retention", {"retention_days": 0})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")

    def test_log_retention_missing_field_returns_400_not_500(self):
        status, body = self._put("/api/settings/log-retention", {})
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_PARAM")


class ZeroThirdPartyDependencyTest(unittest.TestCase):
    """05 §5.7: no Flask/FastAPI/Express/Node.js -- checked statically so this
    doesn't silently rot if someone adds an import later."""

    def test_backend_server_only_imports_stdlib(self):
        path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backend_server.py")
        with open(path, encoding="utf-8") as f:
            tree = ast.parse(f.read(), filename=path)

        top_level_names = set()
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    top_level_names.add(alias.name.split(".")[0])
            elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
                top_level_names.add(node.module.split(".")[0])

        self.assertTrue(top_level_names, "expected at least one import to check")
        # "src" is graphlemon's own package (e.g. project_scaffold/schema/paths,
        # wired in U-022) -- not a third-party dependency, so it's allowed;
        # everything else must be in the standard library.
        non_stdlib = {name for name in top_level_names if name not in sys.stdlib_module_names and name != "src"}
        self.assertEqual(non_stdlib, set(), "non-stdlib imports found: {}".format(non_stdlib))


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