宜博 / lemongraph

"""Local backend process skeleton (U-007, extended in U-022): pure stdlib
`http.server`, no third-party deps (05 §5.7) -- this process is also where
the Graph Engine scheduler will live later (05 §5.4.1), but that's a
different Unit.

MVP scope: process start/stop, `c_pick_folder` (03-C §3.-1, now all three
platforms -- U-007 shipped macOS only via `osascript`; U-022 adds Windows
via a `powershell` FolderBrowserDialog script and Linux via `zenity`/
`kdialog`, whichever is installed), `c_project_create` (03-C §3.0, wired to
the U-002 `project_scaffold.create_project()` function that had no HTTP
route until now) for `v_project_setup`'s submit flow, `c_task_create`/
`c_bug_report_create` (03-C §3.1/§3.1.1, U-023) for `v_goal_composer`/
`v_bug_feedback`'s submit flows, `c_ws_subscribe` (03-C §3.14, U-024,
hand-rolled RFC 6455 WebSocket handshake + framing in src/ws_subscribe.py --
no `websockets` dependency) for real-time `execution_event` push,
`c_execution_log_get`/`c_ws_log_tail` (03-C §3.6.1/§3.6.2, U-027, the
latter reusing the same hand-rolled WS machinery) for raw execution log
viewing, `c_artifact_list`/`c_artifact_get` (U-028, the former unnamed in
either 03-C or the MVCP JSON contract) for `v_artifact_browser`,
`c_skill_diff`/`c_skill_rollback` + skill/skill_iteration list/get (U-029,
`c_skill_diff` likewise only named in the MVCP JSON) for `v_skill_library`/
`v_skill_audit_log`, `c_concurrency_config_update`/`c_budget_config_update`/
`c_log_retention_config_update` (03-C §3.13/§3.13.1, U-030) for the three
settings pages, and serving the U-008 frontend shell as static files (no
build step, 05 §5.7: whitelisted by exact path, not a generic directory
walker, so there's no path-traversal surface to reason about).
"""

import json
import os
import platform
import socket
import sqlite3
import subprocess
import threading
import urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from src import (
    artifact_query, budget_guardrail, bug_record, deployment_query, execution_log,
    org_loop_route, paths, project_query, project_scaffold, schema, settings_config,
    skill_query, skill_update, task_create, task_execution, task_query, work_graph_run_query,
    ws_subscribe,
)

_FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend")

# U-008: exact-path whitelist, not a generic static-file server -- the
# frontend is 3 fixed files until PH3 adds real pages.
_STATIC_FILES = {
    "/": ("index.html", "text/html; charset=utf-8"),
    "/index.html": ("index.html", "text/html; charset=utf-8"),
    "/app.js": ("app.js", "text/javascript; charset=utf-8"),
    "/style.css": ("style.css", "text/css; charset=utf-8"),
}


class DialogUnavailableError(Exception):
    """DIALOG_UNAVAILABLE -- 03-C §3.-1: no native-dialog capability here."""


def pick_folder():
    """c_pick_folder (03-C §3.-1), all three platforms. Returns
    {"selected_path": str} on success, {"selected_path": None} on user
    cancel (not an error). Raises DialogUnavailableError if the platform's
    native dialog tool itself fails or isn't installed."""
    system = platform.system()
    if system == "Darwin":
        return _pick_folder_macos()
    if system == "Windows":
        return _pick_folder_windows()
    return _pick_folder_linux()


def _pick_folder_macos():
    script = 'POSIX path of (choose folder with prompt "Select a graphlemon project folder")'
    result = subprocess.run(["osascript", "-e", script], capture_output=True, text=True)
    if result.returncode != 0:
        if "User canceled" in result.stderr or "(-128)" in result.stderr:
            return {"selected_path": None}
        raise DialogUnavailableError(result.stderr.strip())
    return {"selected_path": result.stdout.strip()}


def _pick_folder_windows():
    # .NET FolderBrowserDialog via a one-shot PowerShell script -- no
    # cancel-vs-error signal other than empty stdout on a clean exit.
    script = (
        "Add-Type -AssemblyName System.Windows.Forms; "
        "$f = New-Object System.Windows.Forms.FolderBrowserDialog; "
        "if ($f.ShowDialog() -eq [System.Windows.Forms.DialogResult]::OK) { Write-Output $f.SelectedPath }"
    )
    try:
        result = subprocess.run(
            ["powershell", "-NoProfile", "-NonInteractive", "-Command", script],
            capture_output=True, text=True,
        )
    except FileNotFoundError as e:
        raise DialogUnavailableError(str(e))
    if result.returncode != 0:
        raise DialogUnavailableError(result.stderr.strip())
    return {"selected_path": result.stdout.strip() or None}


def _pick_folder_linux():
    # zenity (GNOME) and kdialog (KDE) are the two dialogs actually present
    # across mainstream Linux desktops; try whichever exists. Both exit
    # non-zero on user cancel, indistinguishable here from "no folder
    # chosen" -- both cases map to selected_path=None, not an error.
    for cmd in (
        ["zenity", "--file-selection", "--directory"],
        ["kdialog", "--getexistingdirectory", os.path.expanduser("~")],
    ):
        try:
            result = subprocess.run(cmd, capture_output=True, text=True)
        except FileNotFoundError:
            continue
        if result.returncode != 0:
            return {"selected_path": None}
        return {"selected_path": result.stdout.strip()}
    raise DialogUnavailableError("neither zenity nor kdialog is installed")


_WS_PROJECTS_PREFIX = "/api/ws/projects/"
_PROJECTS_PREFIX = "/api/projects/"
_TASKS_PREFIX = "/api/tasks/"
_WORK_NODES_PREFIX = "/api/work-nodes/"
_WS_WORK_NODES_PREFIX = "/api/ws/work-nodes/"
_LOG_SUFFIX = "/log"
_ARTIFACTS_PREFIX = "/api/artifacts/"
_SKILLS_PREFIX = "/api/skills/"
_DEPLOYMENTS_PREFIX = "/api/deployments/"


class _Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed = urllib.parse.urlsplit(self.path)
        if parsed.path == "/api/health":
            self._handle_health()
            return
        if parsed.path.startswith(_WS_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX):
            node_exec_id = parsed.path[len(_WS_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)]
            self._handle_ws_log_tail(node_exec_id)
            return
        if parsed.path.startswith(_WS_PROJECTS_PREFIX):
            self._handle_ws_subscribe(parsed.path, parsed.query)
            return
        if parsed.path == "/api/projects":
            self._handle_project_list()
            return
        if parsed.path.startswith(_PROJECTS_PREFIX):
            project_id = parsed.path[len(_PROJECTS_PREFIX):]
            if project_id and "/" not in project_id:
                self._handle_project_get(project_id)
                return
        if parsed.path == "/api/tasks":
            self._handle_task_list(parsed.query)
            return
        if parsed.path.startswith(_TASKS_PREFIX):
            task_id = parsed.path[len(_TASKS_PREFIX):]
            if task_id and "/" not in task_id:
                self._handle_task_get(task_id)
                return
        if parsed.path == "/api/settings/budget":
            self._handle_budget_list()
            return
        if parsed.path == "/api/settings/concurrency":
            self._handle_concurrency_get()
            return
        if parsed.path == "/api/settings/log-retention":
            self._handle_log_retention_get()
            return
        if parsed.path.startswith(_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX):
            node_exec_id = parsed.path[len(_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)]
            self._handle_execution_log_get(node_exec_id, parsed.query)
            return
        if parsed.path == "/api/artifacts":
            self._handle_artifact_list(parsed.query)
            return
        if parsed.path == "/api/work-graph-runs":
            self._handle_work_graph_run_list(parsed.query)
            return
        if parsed.path.startswith(_ARTIFACTS_PREFIX):
            artifact_id = parsed.path[len(_ARTIFACTS_PREFIX):]
            if artifact_id and "/" not in artifact_id:
                self._handle_artifact_get(artifact_id)
                return
        if parsed.path == "/api/skills":
            self._handle_skill_list()
            return
        if parsed.path == "/api/skills/diff":
            self._handle_skill_diff(parsed.query)
            return
        if parsed.path.startswith(_SKILLS_PREFIX) and parsed.path.endswith("/audit"):
            skill_id = parsed.path[len(_SKILLS_PREFIX):-len("/audit")]
            self._handle_skill_audit_log(skill_id)
            return
        if parsed.path == "/api/deployments":
            self._handle_deployment_list()
            return
        if parsed.path == "/api/deployments/current":
            self._handle_deployment_current()
            return
        if parsed.path.startswith(_DEPLOYMENTS_PREFIX):
            deployment_id = parsed.path[len(_DEPLOYMENTS_PREFIX):]
            if deployment_id and "/" not in deployment_id:
                self._handle_deployment_get(deployment_id)
                return

        entry = _STATIC_FILES.get(self.path)
        if entry is None:
            self._send_json(404, {"error": "NOT_FOUND"})
            return
        filename, content_type = entry
        with open(os.path.join(_FRONTEND_DIR, filename), "rb") as f:
            body = f.read()
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        if self.path == "/api/system/pick-folder":
            self._handle_pick_folder()
        elif self.path == "/api/projects":
            self._handle_project_create()
        elif self.path == "/api/tasks":
            self._handle_task_create()
        elif self.path == "/api/bug-reports":
            self._handle_bug_report_create()
        elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/route"):
            task_id = self.path[len(_TASKS_PREFIX):-len("/route")]
            self._handle_task_route(task_id)
        elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/execute"):
            task_id = self.path[len(_TASKS_PREFIX):-len("/execute")]
            self._handle_task_execute(task_id)
        elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/pause"):
            task_id = self.path[len(_TASKS_PREFIX):-len("/pause")]
            self._handle_task_pause(task_id, "user_paused")
        elif self.path.startswith(_TASKS_PREFIX) and self.path.endswith("/stop"):
            task_id = self.path[len(_TASKS_PREFIX):-len("/stop")]
            self._handle_task_pause(task_id, "user_stopped")
        elif self.path.startswith(_SKILLS_PREFIX) and self.path.endswith("/rollback"):
            skill_id = self.path[len(_SKILLS_PREFIX):-len("/rollback")]
            self._handle_skill_rollback(skill_id)
        else:
            self._send_json(404, {"error": "NOT_FOUND"})

    def do_PUT(self):
        if self.path == "/api/settings/concurrency":
            self._handle_concurrency_update()
        elif self.path == "/api/settings/budget":
            self._handle_budget_update()
        elif self.path == "/api/settings/log-retention":
            self._handle_log_retention_update()
        elif self.path.startswith(_PROJECTS_PREFIX) and "/" not in self.path[len(_PROJECTS_PREFIX):]:
            self._handle_project_update(self.path[len(_PROJECTS_PREFIX):])
        elif self.path.startswith(_TASKS_PREFIX) and "/" not in self.path[len(_TASKS_PREFIX):]:
            self._handle_task_update(self.path[len(_TASKS_PREFIX):])
        else:
            self._send_json(404, {"error": "NOT_FOUND"})

    def do_DELETE(self):
        if self.path.startswith(_PROJECTS_PREFIX) and "/" not in self.path[len(_PROJECTS_PREFIX):]:
            self._handle_project_delete(self.path[len(_PROJECTS_PREFIX):])
        elif self.path.startswith(_TASKS_PREFIX) and "/" not in self.path[len(_TASKS_PREFIX):]:
            self._handle_task_delete(self.path[len(_TASKS_PREFIX):])
        else:
            self._send_json(404, {"error": "NOT_FOUND"})

    def _handle_pick_folder(self):
        try:
            self._send_json(200, pick_folder())
        except DialogUnavailableError as e:
            self._send_json(422, {"error": "DIALOG_UNAVAILABLE", "detail": str(e)})

    def _read_json_body(self):
        """Returns (payload, None) on success, (None, error_response) if the
        body isn't valid JSON -- caller sends error_response and returns."""
        length = int(self.headers.get("Content-Length", 0))
        try:
            return json.loads(self.rfile.read(length) or b"{}"), None
        except json.JSONDecodeError:
            return None, {"error": "INVALID_INPUT", "detail": "body is not valid JSON"}

    def _handle_health(self):
        """c_deploy_supervisor_health_check's `http_health` probe (U-038,
        03-C §3.15/§6). Confirms both "the HTTP server thread answers" and
        "the sqlite connection is actually usable" -- a process that's alive
        but whose DB connection got wedged somehow would otherwise pass a
        process_alive check and fail every real request afterward; catching
        that here, not just accepting a 200 for existing, is the point of a
        health check in the first place."""
        try:
            with self.server.conn_lock:
                self.server.conn.execute("SELECT 1").fetchone()
        except sqlite3.Error as e:
            self._send_json(503, {"status": "error", "detail": str(e)})
            return
        self._send_json(200, {"status": "ok"})

    def _handle_project_create(self):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        target_folder_path = payload.get("target_folder_path")
        if not target_folder_path:
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "target_folder_path is required"})
            return
        try:
            with self.server.conn_lock:
                result = project_scaffold.create_project(
                    self.server.conn,
                    target_folder_path,
                    name=payload.get("name"),
                    init_git_if_missing=bool(payload.get("init_git_if_missing", False)),
                )
        except project_scaffold.PathNotFoundError as e:
            self._send_json(404, {"error": "PATH_NOT_FOUND", "detail": str(e)})
        except project_scaffold.PathNotWritableError as e:
            self._send_json(403, {"error": "PATH_NOT_WRITABLE", "detail": str(e)})
        except project_scaffold.GitInitDeclinedError as e:
            self._send_json(409, {"error": "GIT_INIT_DECLINED", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_project_list(self):
        with self.server.conn_lock:
            result = project_query.list_projects(self.server.conn)
        self._send_json(200, result)

    def _handle_project_get(self, project_id):
        try:
            with self.server.conn_lock:
                result = project_query.get_project(self.server.conn, project_id)
        except project_query.ProjectNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        else:
            self._send_json(200, result)

    def _handle_project_update(self, project_id):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        try:
            with self.server.conn_lock:
                result = project_query.update_project(
                    self.server.conn, project_id, name=payload.get("name"), status=payload.get("status"),
                )
        except project_query.ProjectNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        except project_query.InvalidParamError as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_project_delete(self, project_id):
        """"Delete" archives (project.status='archived') rather than
        removing the row -- see project_query.py's own docstring for why a
        real DELETE isn't safe here (FK-referenced history)."""
        try:
            with self.server.conn_lock:
                result = project_query.update_project(self.server.conn, project_id, status="archived")
        except project_query.ProjectNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        else:
            self._send_json(200, result)

    def _handle_task_create(self):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        try:
            with self.server.conn_lock:
                result = task_create.create_task(
                    self.server.conn,
                    payload.get("project_id"),
                    payload.get("goal_text"),
                    task_type=payload.get("task_type", "goal"),
                    created_via=payload.get("created_via", "api"),
                )
        except task_create.InvalidParamError as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_bug_report_create(self):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        try:
            with self.server.conn_lock:
                result = bug_record.create_bug_report(
                    self.server.conn,
                    payload.get("project_id"),
                    payload.get("description"),
                    repro_steps=payload.get("repro_steps"),
                    related_task_id=payload.get("related_task_id"),
                )
        except bug_record.InvalidParamError as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_task_list(self, query_string):
        params = urllib.parse.parse_qs(query_string)
        project_id = params.get("project_id", [None])[0]
        if not project_id:
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "project_id is required"})
            return
        try:
            with self.server.conn_lock:
                result = task_query.list_tasks(
                    self.server.conn, project_id,
                    status=params.get("status", [None])[0],
                    org_node=params.get("org_node", [None])[0],
                    keyword=params.get("keyword", [None])[0],
                    sort=params.get("sort", [None])[0],
                    order=params.get("order", [None])[0],
                )
        except task_query.InvalidParamError as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_task_get(self, task_id):
        try:
            with self.server.conn_lock:
                result = task_query.get_task(self.server.conn, task_id)
        except task_query.TaskNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        else:
            self._send_json(200, result)

    def _handle_task_update(self, task_id):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        goal_text = payload.get("goal_text")
        if not goal_text:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": "goal_text is required"})
            return
        try:
            with self.server.conn_lock:
                result = task_query.update_task_goal_text(self.server.conn, task_id, goal_text)
        except task_query.TaskNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        except task_query.TaskNotEditableError as e:
            self._send_json(409, {"error": "TASK_NOT_EDITABLE", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_task_delete(self, task_id):
        try:
            with self.server.conn_lock:
                task_query.delete_task(self.server.conn, task_id)
        except task_query.TaskNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        except task_query.TaskNotEditableError as e:
            self._send_json(409, {"error": "TASK_NOT_EDITABLE", "detail": str(e)})
        else:
            self._send_json(200, {"task_id": task_id, "deleted": True})

    def _handle_task_route(self, task_id):
        """c_org_loop_route (03-C §3.11) -- the underlying route_org_loop()
        was built in U-016 with no HTTP route until now; v_task_detail's
        click_retry (manual_resume) is what actually needs it wired up.

        **2026-08-03**: if the task was blocked via a user-initiated
        pause/stop (task_execution.request_pause, blocked_reason in
        ('user_paused', 'user_stopped')), a successful manual_resume also
        spawns a background thread to actually continue real execution --
        matching what the ORIGINAL blocked_pending_human/manual_resume
        design already implies (04-P: "调度器继续执行队列中其他任务" --
        resuming should let the scheduler keep going, not just flip a
        status flag and stop).

        **2026-08-04 broadened**: found for real -- a transient network
        blip (`claude` CLI: "API Error: Unable to connect to API
        (ENOTFOUND)") blocks a task with blocked_reason='execution_error',
        same as any other unroutable exception (task_execution._mark_blocked's
        catch-all). The original 'user_paused'/'user_stopped' allowlist left
        this stuck: manual_resume flipped the status back but nothing ever
        re-triggered real execution, and /execute's own guard requires
        status='queued' (a resumed task lands on 'planning_brd_prd_sdd' or
        'coding' instead) -- clicking "resume" did nothing a second click
        could fix. The user has ALREADY made the judgment call to retry by
        clicking resume at all; withholding continuation only made sense for
        the two reasons that have their OWN dedicated resume path requiring
        a different human action first: `budget_exceeded` (must raise the
        limit -- see budget_guardrail.update_budget_limit_and_resume/
        upsert_budget_config, which are the actual intended resume trigger)
        and `loop_limit_reached` (excluded explicitly by name, not just by
        relying on its real from_status usually being 'validating' -- a
        wrong-status edge case, real or a test seeding one, shouldn't be
        able to accidentally slip a 3-strikes task back into auto-retry).
        Every other blocked_reason (execution_error, classifier_blocked,
        corrupt_persisted_state, usage_limit_wait_timeout,
        prd_consistency_check_failed, user_paused, user_stopped, ...) now
        auto-continues on a successful resume to a resumable status."""
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        decided_route = payload.get("decided_route")
        pre_resume_reason = None
        if decided_route == "manual_resume":
            row = self.server.conn.execute(
                "SELECT blocked_reason FROM task WHERE task_id = ?", (task_id,)
            ).fetchone()
            pre_resume_reason = row[0] if row else None
        try:
            with self.server.conn_lock:
                result = org_loop_route.route_org_loop(self.server.conn, task_id, decided_route)
        except org_loop_route.LoopLimitReachedError as e:
            self._send_json(409, {"error": "LOOP_LIMIT_REACHED", "detail": str(e)})
            return
        except (org_loop_route.TaskNotBlockedError, org_loop_route.TaskNotValidatingError, ValueError) as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
            return
        if pre_resume_reason not in ("budget_exceeded", "loop_limit_reached") and result.get("new_status") in (
            "queued", "planning_brd_prd_sdd", "coding",
        ):
            task_row = self.server.conn.execute(
                "SELECT project_id, goal_text FROM task WHERE task_id = ?", (task_id,)
            ).fetchone()
            target_folder_path = self.server.conn.execute(
                "SELECT target_folder_path FROM project WHERE project_id = ?", (task_row[0],)
            ).fetchone()[0]
            task_execution.start_task_execution_in_background(
                self.server.db_path, task_id, target_folder_path, task_row[1],
            )
        self._send_json(200, result)

    def _handle_task_execute(self, task_id):
        """c_task_execute (2026-08-03): the manual "Start" trigger --
        confirmed with the user this is genuinely missing (03-C's own
        c_task_create contract already assumed task creation would
        auto-trigger scheduling via an always-on scheduler that was never
        actually wired up as a running process, see task_execution.py's
        module docstring)."""
        row = self.server.conn.execute("SELECT project_id, goal_text, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
        if row is None:
            self._send_json(404, {"error": "NOT_FOUND"})
            return
        project_id, goal_text, status = row
        if status != "queued":
            self._send_json(409, {"error": "TASK_NOT_QUEUED", "detail": "task is status={}, not queued".format(status)})
            return
        target_folder_path = self.server.conn.execute(
            "SELECT target_folder_path FROM project WHERE project_id = ?", (project_id,)
        ).fetchone()[0]
        task_execution.start_task_execution_in_background(self.server.db_path, task_id, target_folder_path, goal_text)
        self._send_json(200, {"task_id": task_id, "started": True})

    def _handle_task_pause(self, task_id, reason):
        try:
            with self.server.conn_lock:
                result = task_execution.request_pause(self.server.conn, task_id, reason)
        except task_execution.TaskExecutionError as e:
            self._send_json(409, {"error": "TASK_NOT_PAUSABLE", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_budget_list(self):
        with self.server.conn_lock:
            result = budget_guardrail.list_budget_configs(self.server.conn)
        self._send_json(200, result)

    def _handle_budget_update(self):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        try:
            with self.server.conn_lock:
                result = budget_guardrail.upsert_budget_config(
                    self.server.conn, payload.get("scope"), payload.get("limit_type"),
                    payload.get("limit_value"), payload.get("updated_by", "console"),
                )
        except (budget_guardrail.InvalidParamError, TypeError) as e:
            # TypeError here means a required numeric field (limit_value) was
            # missing/null -- comparing None <= 0 raises TypeError, not
            # InvalidParamError, before upsert_budget_config's own checks run
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_concurrency_get(self):
        with self.server.conn_lock:
            result = settings_config.get_concurrency_config(self.server.conn)
        self._send_json(200, result)

    def _handle_concurrency_update(self):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        try:
            with self.server.conn_lock:
                result = settings_config.upsert_concurrency_config(
                    self.server.conn, payload.get("max_concurrent_goals"), payload.get("updated_by", "console"),
                )
        except (settings_config.InvalidParamError, TypeError) as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_log_retention_get(self):
        with self.server.conn_lock:
            result = settings_config.get_log_retention_config(self.server.conn)
        self._send_json(200, result)

    def _handle_log_retention_update(self):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        try:
            with self.server.conn_lock:
                result = settings_config.upsert_log_retention_config(
                    self.server.conn, payload.get("retention_days"), payload.get("updated_by", "console"),
                    max_total_size_mb=payload.get("max_total_size_mb"),
                )
        except (settings_config.InvalidParamError, TypeError) as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_artifact_list(self, query_string):
        params = urllib.parse.parse_qs(query_string)
        project_id = params.get("project_id", [None])[0]
        if not project_id:
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "project_id is required"})
            return
        with self.server.conn_lock:
            result = artifact_query.list_artifacts(
                self.server.conn, project_id,
                task_id=params.get("task_id", [None])[0],
                artifact_type=params.get("artifact_type", [None])[0],
            )
        self._send_json(200, result)

    def _handle_work_graph_run_list(self, query_string):
        params = urllib.parse.parse_qs(query_string)
        try:
            with self.server.conn_lock:
                result = work_graph_run_query.list_work_graph_runs(
                    self.server.conn,
                    params.get("project_id", [None])[0],
                    org_node=params.get("org_node", [None])[0],
                    limit=params.get("limit", [None])[0],
                )
        except work_graph_run_query.InvalidParamError as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_artifact_get(self, artifact_id):
        try:
            with self.server.conn_lock:
                result = artifact_query.get_artifact(self.server.conn, artifact_id)
        except artifact_query.ArtifactNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        else:
            self._send_json(200, result)

    def _handle_deployment_list(self):
        with self.server.conn_lock:
            result = deployment_query.list_deployments(self.server.conn)
        self._send_json(200, result)

    def _handle_deployment_current(self):
        with self.server.conn_lock:
            result = deployment_query.get_current_deployment(self.server.conn)
        self._send_json(200, result)

    def _handle_deployment_get(self, deployment_id):
        try:
            with self.server.conn_lock:
                result = deployment_query.get_deployment(self.server.conn, deployment_id)
        except deployment_query.DeploymentNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        else:
            self._send_json(200, result)

    def _handle_skill_list(self):
        with self.server.conn_lock:
            result = skill_query.list_skills(self.server.conn)
        self._send_json(200, result)

    def _handle_skill_audit_log(self, skill_id):
        try:
            with self.server.conn_lock:
                result = skill_query.get_skill_audit_log(self.server.conn, skill_id)
        except skill_query.SkillNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        else:
            self._send_json(200, result)

    def _handle_skill_diff(self, query_string):
        params = urllib.parse.parse_qs(query_string)
        before = params.get("before", [None])[0]
        after = params.get("after", [None])[0]
        if not before or not after:
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "before and after are both required"})
            return
        try:
            diff_text = skill_query.get_diff(before, after, repo_path=self.server.skill_repo_path)
        except subprocess.CalledProcessError as e:
            self._send_json(400, {"error": "INVALID_PARAM", "detail": e.stderr})
        else:
            self._send_json(200, {"diff_text": diff_text})

    def _handle_skill_rollback(self, skill_id):
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        iteration_id = payload.get("iteration_id")
        if not iteration_id:
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "iteration_id is required"})
            return
        try:
            with self.server.conn_lock:
                result = skill_update.rollback_skill(self.server.conn, iteration_id, self.server.skill_repo_path)
        except skill_update.IterationNotFoundError as e:
            self._send_json(404, {"error": "ITERATION_NOT_FOUND", "detail": str(e)})
        else:
            self._send_json(200, result)

    def _handle_ws_subscribe(self, path, query_string):
        """c_ws_subscribe (03-C §3.14): WS /api/ws/projects/:project_id.
        Rejects with a normal HTTP response (no 101) before ever touching
        the socket-hijack path if the request isn't a real WS upgrade or
        project_id doesn't exist -- 03-C: "直接拒绝连接", not a connection
        that opens and then immediately closes."""
        project_id = path[len(_WS_PROJECTS_PREFIX):]
        if not project_id or "/" in project_id:
            self._send_json(404, {"error": "NOT_FOUND"})
            return
        task_id = urllib.parse.parse_qs(query_string).get("task_id", [None])[0]

        ws_key = self.headers.get("Sec-WebSocket-Key")
        if self.headers.get("Upgrade", "").lower() != "websocket" or not ws_key:
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "expected a WebSocket upgrade request"})
            return

        with self.server.conn_lock:
            project_exists = self.server.conn.execute(
                "SELECT 1 FROM project WHERE project_id = ?", (project_id,)
            ).fetchone() is not None
        if not project_exists:
            self._send_json(404, {"error": "PROJECT_NOT_FOUND"})
            return

        self._send_ws_handshake_response(ws_key)

        with self.server.conn_lock:
            after_rowid = ws_subscribe.latest_rowid(self.server.conn, project_id)
        self.connection.settimeout(ws_subscribe.POLL_INTERVAL_SECONDS)
        try:
            while True:
                try:
                    opcode, _payload = ws_subscribe.decode_client_frame(self.connection)
                    if opcode == ws_subscribe.OPCODE_CLOSE:
                        break
                    if opcode == ws_subscribe.OPCODE_PING:
                        self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_PONG))
                except socket.timeout:
                    pass  # normal: nothing from the client this poll tick

                with self.server.conn_lock:
                    events, after_rowid = ws_subscribe.poll_new_events(
                        self.server.conn, project_id, after_rowid, task_id=task_id
                    )
                for event in events:
                    self.connection.sendall(ws_subscribe.encode_text_frame(json.dumps(event)))
        except (ConnectionError, OSError):
            pass  # client disconnected -- nothing more to send
        except sqlite3.ProgrammingError:
            pass  # server shutting down closed the shared conn out from under this poll loop

    def _send_ws_handshake_response(self, ws_key):
        accept_key = ws_subscribe.compute_accept_key(ws_key)
        self.connection.sendall(
            (
                "HTTP/1.1 101 Switching Protocols\r\n"
                "Upgrade: websocket\r\n"
                "Connection: Upgrade\r\n"
                "Sec-WebSocket-Accept: " + accept_key + "\r\n\r\n"
            ).encode("ascii")
        )
        self.close_connection = True

    def _handle_execution_log_get(self, node_exec_id, query_string):
        params = urllib.parse.parse_qs(query_string)
        download = params.get("download", ["false"])[0].lower() in ("true", "1")
        try:
            if download:
                with self.server.conn_lock:
                    file_path = execution_log.get_log_file_path(self.server.conn, node_exec_id)
                self._send_file_download(file_path, node_exec_id)
                return
            offset = int(params.get("offset", ["0"])[0])
            limit_param = params.get("limit", [None])[0]
            limit = int(limit_param) if limit_param is not None else None
            with self.server.conn_lock:
                result = execution_log.get_log(self.server.conn, node_exec_id, offset=offset, limit=limit)
        except execution_log.LogNotFoundError:
            self._send_json(404, {"error": "LOG_NOT_FOUND"})
        except execution_log.LogPurgedError:
            self._send_json(410, {"error": "LOG_PURGED"})
        else:
            self._send_json(200, result)

    def _send_file_download(self, file_path, node_exec_id):
        with open(file_path, "rb") as f:
            body = f.read()
        self.send_response(200)
        self.send_header("Content-Type", "application/octet-stream")
        self.send_header("Content-Disposition", 'attachment; filename="' + node_exec_id + '.log"')
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _handle_ws_log_tail(self, node_exec_id):
        """c_ws_log_tail (03-C §3.6.2): WS /api/ws/work-nodes/:node_exec_id/log.
        The server actively closes the connection once the node reaches a
        terminal status (done/blocked/failed) -- 03-C's own wording, not
        left for the client to keep polling forever after there's nothing
        more to send; the frontend switches to c_execution_log_get's static
        mode on close."""
        ws_key = self.headers.get("Sec-WebSocket-Key")
        if self.headers.get("Upgrade", "").lower() != "websocket" or not ws_key:
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "expected a WebSocket upgrade request"})
            return

        try:
            with self.server.conn_lock:
                execution_log.node_exec_status(self.server.conn, node_exec_id)
        except execution_log.NodeExecNotFoundError:
            self._send_json(404, {"error": "NODE_EXEC_NOT_FOUND"})
            return

        self._send_ws_handshake_response(ws_key)

        with self.server.conn_lock:
            after_rowid = execution_log.latest_log_event_rowid(self.server.conn, node_exec_id)
        self.connection.settimeout(ws_subscribe.POLL_INTERVAL_SECONDS)
        try:
            while True:
                try:
                    opcode, _payload = ws_subscribe.decode_client_frame(self.connection)
                    if opcode == ws_subscribe.OPCODE_CLOSE:
                        break
                    if opcode == ws_subscribe.OPCODE_PING:
                        self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_PONG))
                except socket.timeout:
                    pass  # normal: nothing from the client this poll tick

                with self.server.conn_lock:
                    events, after_rowid = execution_log.poll_log_events(self.server.conn, node_exec_id, after_rowid)
                    status = execution_log.node_exec_status(self.server.conn, node_exec_id)
                for event in events:
                    self.connection.sendall(ws_subscribe.encode_text_frame(json.dumps(event)))
                if execution_log.is_terminal_status(status):
                    self.connection.sendall(ws_subscribe.encode_frame(b"", opcode=ws_subscribe.OPCODE_CLOSE))
                    break
        except (ConnectionError, OSError):
            pass  # client disconnected -- nothing more to send
        except sqlite3.ProgrammingError:
            pass  # server shutting down closed the shared conn out from under this poll loop

    def _send_json(self, status, obj):
        body = json.dumps(obj).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass  # quiets test output; real request logging is a later Unit's concern


def _ensure_schema(conn):
    exists = conn.execute(
        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'project'"
    ).fetchone()
    if exists is None:
        schema.create_schema(conn)


class _Server(ThreadingHTTPServer):
    # long-lived WS connections (c_ws_subscribe) shouldn't be able to block
    # process shutdown -- ThreadingMixIn.daemon_threads defaults to False.
    daemon_threads = True

    def server_close(self):
        super().server_close()
        # a WS handler thread can still be mid-poll on another thread at
        # shutdown (its next tick would just get a clean ProgrammingError,
        # already caught) -- closing conn without this lock races the C
        # sqlite3 connection object across threads with no synchronization
        # at all, which segfaults rather than raising cleanly.
        with self.conn_lock:
            self.conn.close()


def create_server(host="127.0.0.1", port=0, db_path=None, skill_repo_path=None):
    """port=0 lets the OS assign a free port -- avoids collisions in tests
    and lets multiple graphlemon instances run side by side if ever needed.

    db_path defaults to paths.db_path() (U-003's platform_home, so it never
    lives inside a target project's own repo); tests pass ":memory:" or a
    tempfile so runs don't touch the real developer's ~/.graphlemon/.

    skill_repo_path defaults to skill_query._REPO_ROOT (graphlemon's own
    checkout -- Skills genuinely live there, U-018/U-029) but is injectable
    for the same reason db_path is: c_skill_rollback does a real `git
    commit`, and a test hitting that route with the real default would
    write an actual commit into whatever repo this process happens to be
    running from. Never exposed to HTTP clients -- only settable here, at
    process/test-construction time."""
    resolved_db_path = db_path if db_path is not None else paths.db_path()
    conn = schema.connect(resolved_db_path, check_same_thread=False)
    _ensure_schema(conn)
    server = _Server((host, port), _Handler)
    server.conn = conn
    server.db_path = resolved_db_path
    server.conn_lock = threading.Lock()
    server.skill_repo_path = skill_repo_path if skill_repo_path is not None else skill_query._REPO_ROOT
    return server


def main():
    """U-037: CLI entry point (`python3 -m src.backend_server ...`) -- until
    this Unit, `create_server()` was only ever instantiated in-process by
    tests; there was no way to actually launch graphlemon's own backend as a
    standalone OS process. c_deploy_supervisor_stage needs exactly that (a
    real candidate process, `subprocess.Popen`, listening on a real port) --
    found while building it, added here since it's the natural, minimal home
    for "run this server for real" regardless of who launches it.
    """
    import argparse
    import signal

    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=0)
    parser.add_argument("--db-path", default=None)
    parser.add_argument("--skill-repo-path", default=None)
    args = parser.parse_args()

    server = create_server(args.host, args.port, args.db_path, args.skill_repo_path)

    def _handle_sigterm(signum, frame):
        # server.shutdown() blocks until serve_forever()'s loop notices and
        # returns -- calling it from the SAME thread that's running
        # serve_forever() (the signal handler runs on the main thread, same
        # as serve_forever() below) deadlocks, per http.server's own docs.
        threading.Thread(target=server.shutdown, daemon=True).start()

    signal.signal(signal.SIGTERM, _handle_sigterm)
    print("graphlemon backend listening on {}:{}".format(*server.server_address), flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == "__main__":
    main()