宜博 / 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 time
import traceback
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, task_reconciliation,
    work_graph_fork, 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/"
_WORK_GRAPH_RUNS_PREFIX = "/api/work-graph-runs/"
_FORK_SUFFIX = "/fork"


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(_WORK_GRAPH_RUNS_PREFIX):
            run_id = parsed.path[len(_WORK_GRAPH_RUNS_PREFIX):]
            if run_id and "/" not in run_id:
                self._handle_work_graph_run_get(run_id)
                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.startswith(_SKILLS_PREFIX) and parsed.path.endswith("/content"):
            skill_id = parsed.path[len(_SKILLS_PREFIX):-len("/content")]
            self._handle_skill_content(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)
        elif self.path.startswith(_WORK_NODES_PREFIX) and self.path.endswith(_FORK_SUFFIX):
            node_exec_id = self.path[len(_WORK_NODES_PREFIX):-len(_FORK_SUFFIX)]
            self._handle_work_node_fork(node_exec_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
        # 'validating' added 2026-08-07 (real incident): task_execution.
        # run_or_resume_task_lifecycle has resumed from 'validating' since
        # the earlier same-day fix (skips straight to
        # validate_implementation_with_own_graph, the readiness trigger
        # already having run once) -- but THIS allowlist, a separate code
        # path, still didn't know that, so a manual Resume on a task stuck
        # in 'validating' flipped the status back with nothing ever
        # actually continuing it.
        if pre_resume_reason not in ("budget_exceeded", "loop_limit_reached") and result.get("new_status") in (
            "queued", "planning_brd_prd_sdd", "coding", "validating",
        ):
            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]
            try:
                task_execution.start_task_execution_in_background(
                    self.server.db_path, task_id, target_folder_path, task_row[1],
                )
            except task_execution.TaskExecutionError:
                # Already has an execution in flight (e.g. the
                # reconciliation loop dispatched it a moment before this
                # click landed) -- the resume itself already succeeded
                # (route_org_loop.route_org_loop above), so still report
                # 200; the in-flight run will get there on its own.
                pass
        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]
        try:
            task_execution.start_task_execution_in_background(self.server.db_path, task_id, target_folder_path, goal_text)
        except task_execution.TaskExecutionError as e:
            # 2026-08-07: already has an execution in flight (e.g. a rapid
            # double-click, or the reconciliation loop won a race against
            # this click) -- report it instead of a 500.
            self._send_json(409, {"error": "ALREADY_IN_FLIGHT", "detail": str(e)})
            return
        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_work_graph_run_get(self, run_id):
        """c_work_graph_run_get (U-046, GET /api/work-graph-runs/:run_id):
        v_run_detail's single-run lookup -- the list endpoint above has no
        per-run detail shape, and v_run_detail needs the full node list
        (same shape as task_query.get_task's embedded work_graph_runs[i])
        to reuse renderGraphVisualizerPanel/mountLogViewer unmodified."""
        try:
            with self.server.conn_lock:
                result = work_graph_run_query.get_work_graph_run(self.server.conn, run_id)
        except work_graph_run_query.WorkGraphRunNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        else:
            self._send_json(200, result)

    def _handle_work_node_fork(self, node_exec_id):
        """D (2026-08-06, time-travel debugging), POST
        /api/work-nodes/:node_exec_id/fork, body {"new_output_payload": {...}}.
        Mirrors _handle_task_execute's shape: create_fork() runs
        synchronously (fast, DB-only) so the response carries the new
        run_id immediately, real execution continues in the background --
        the console navigates straight to #/runs/:run_id and watches it
        progress live via the existing work-graph-run endpoints, no new
        polling mechanism needed."""
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        new_output_payload = payload.get("new_output_payload")
        if not isinstance(new_output_payload, dict):
            self._send_json(400, {"error": "INVALID_INPUT", "detail": "new_output_payload must be an object"})
            return

        with self.server.conn_lock:
            row = self.server.conn.execute(
                "SELECT p.target_folder_path FROM work_node_execution wne "
                "JOIN work_graph_run wgr ON wgr.run_id = wne.run_id "
                "JOIN task t ON t.task_id = wgr.task_id "
                "JOIN project p ON p.project_id = t.project_id "
                "WHERE wne.node_exec_id = ?",
                (node_exec_id,),
            ).fetchone()
        if row is None:
            self._send_json(404, {"error": "NOT_FOUND"})
            return
        target_folder_path = row[0]

        try:
            run_id, _thread = work_graph_fork.start_fork_in_background(
                self.server.db_path, node_exec_id, new_output_payload, target_folder_path,
            )
        except work_graph_fork.ForkSourceNotDoneError as e:
            self._send_json(409, {"error": "FORK_SOURCE_NOT_DONE", "detail": str(e)})
        except work_graph_fork.ForkParentRunNotTerminalError as e:
            self._send_json(409, {"error": "FORK_PARENT_RUN_NOT_TERMINAL", "detail": str(e)})
        except work_graph_fork.ForkGraphShapeUnsupportedError as e:
            self._send_json(422, {"error": "FORK_GRAPH_SHAPE_UNSUPPORTED", "detail": str(e)})
        except ValueError as e:
            self._send_json(404, {"error": "NOT_FOUND", "detail": str(e)})
        else:
            self._send_json(200, {"run_id": run_id, "started": True})

    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_content(self, skill_id):
        """c_skill_content (bug-fix Unit, no MVCP operation_id -- same
        "unowned gap" precedent as c_artifact_get before U-025). Real
        SKILL.md content off disk, alongside (not instead of) the existing
        /audit iteration history -- v_skill_audit_log's page renders both
        independently."""
        try:
            with self.server.conn_lock:
                result = skill_query.get_skill_content(self.server.conn, skill_id, repo_path=self.server.skill_repo_path)
        except skill_query.SkillNotFoundError:
            self._send_json(404, {"error": "NOT_FOUND"})
        except skill_query.SkillContentMissingError:
            self._send_json(404, {"error": "CONTENT_MISSING"})
        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):
        # execution_log_file rows are created up front (work_node_run.py) but
        # the backing file itself is only ever written by execution_log.
        # append_log() -- a node that errored before its first log write (or
        # a file removed out-of-band) leaves a row with no file on disk yet;
        # that's a clean 404, not an uncaught FileNotFoundError bubbling up
        # as a bare 500 (bug A2, same root cause as A1: see execution_log.py
        # and work_graph_orchestrate.py's DU-01 notes).
        try:
            with open(file_path, "rb") as f:
                body = f.read()
        except FileNotFoundError:
            self._send_json(404, {"error": "LOG_FILE_MISSING"})
            return
        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)
    # 2026-08-06 (real incident): create_schema() above only ever runs for a
    # BRAND NEW db file. Every schema.migrate_*() function is documented as
    # idempotent/safe-to-call-unconditionally (including against a fresh
    # create_schema()'d db, which already has whatever they'd add), but
    # nothing ever actually called them against an EXISTING, already-
    # populated database until now -- caught for real when this exact
    # process restarted against the live production db and every
    # GET /api/tasks/:id started 500ing with "no such column: parent_run_id"
    # (D's migrate_add_work_graph_fork_columns had shipped and been tested,
    # but never wired into the one code path that runs on every real
    # startup). Run all three every time, not just on first-ever creation.
    schema.migrate_expand_artifact_types(conn)
    schema.migrate_relax_org_node_constraints(conn)  # also runs migrate_add_work_graph_fork_columns internally
    schema.migrate_add_work_graph_loop_limit_config(conn)
    schema.migrate_add_loop_back_event_type(conn)
    schema.migrate_add_reviewing_planner_output_status(conn)
    schema.migrate_add_planner_review_verdict_event_type(conn)
    schema.migrate_relax_agent_profile_migrated_from_profile(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


_DEFAULT_RECONCILIATION_INTERVAL_SECONDS = 300


def _start_reconciliation_loop(db_path, interval_seconds):
    """2026-08-06 (docs/done/scheduler-task-reconciliation-requirements.md):
    background daemon thread that repeatedly calls
    `task_reconciliation.reconcile_once()`. Deliberately started here, NOT
    inside `create_server()` -- that function is instantiated fresh by
    nearly every backend_server test, and a background loop dispatching
    real `start_task_execution_in_background()` calls (real subprocess
    invocations) during unit tests would be exactly the kind of test
    pollution CLAUDE.md's Testing constraint exists to prevent. Only
    `main()` (the real CLI entry point) starts this.

    Runs one pass immediately (a restart is exactly when previously-
    abandoned tasks most need to be picked back up, not `interval_seconds`
    later), then repeats. A bad pass never kills the loop -- the next
    scheduled pass just tries again, same "one failure doesn't stop
    everything else" principle `reconcile_once()` itself already applies
    per-candidate. The traceback still goes to stderr: a silently-swallowed
    per-pass exception would otherwise look identical to "nothing to
    reconcile" from the outside -- exactly the kind of quiet failure a
    24/7-unattended loop can't afford."""
    def loop():
        while True:
            try:
                task_reconciliation.reconcile_once(db_path)
            except Exception:
                traceback.print_exc()
            time.sleep(interval_seconds)

    thread = threading.Thread(target=loop, daemon=True)
    thread.start()
    return thread


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)
    parser.add_argument("--reconciliation-interval-seconds", type=int, default=_DEFAULT_RECONCILIATION_INTERVAL_SECONDS)
    parser.add_argument("--disable-reconciliation-loop", action="store_true")
    args = parser.parse_args()

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

    if not args.disable_reconciliation_loop:
        _start_reconciliation_loop(server.db_path, args.reconciliation_interval_seconds)

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