"""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
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 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 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)


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, org_node, graph_manifest, status) "
            "VALUES ('run-1', 'task-1', 'coder', '{}', '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_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, org_node, graph_manifest, status) "
            "VALUES ('run-1', 'task-1', 'planner', '{}', '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 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)


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()