"""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, 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 ( 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" 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 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_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