宜博 / 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`, 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, execution_log,
    org_loop_route, paths, project_scaffold, schema, task_create,
    task_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/"
_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/"


class _Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed = urllib.parse.urlsplit(self.path)
        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/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.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.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

        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)
        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_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_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
        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],
            )
        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_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."""
        payload, error = self._read_json_body()
        if error:
            self._send_json(400, error)
            return
        try:
            with self.server.conn_lock:
                result = org_loop_route.route_org_loop(self.server.conn, task_id, payload.get("decided_route"))
        except org_loop_route.LoopLimitReachedError as e:
            self._send_json(409, {"error": "LOOP_LIMIT_REACHED", "detail": str(e)})
        except (org_loop_route.TaskNotBlockedError, org_loop_route.TaskNotValidatingError, ValueError) as e:
            self._send_json(400, {"error": "INVALID_PARAM", "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_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_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_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):
    """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/."""
    conn = schema.connect(db_path if db_path is not None else paths.db_path(), check_same_thread=False)
    _ensure_schema(conn)
    server = _Server((host, port), _Handler)
    server.conn = conn
    server.conn_lock = threading.Lock()
    return server