"""Local backend process skeleton (U-007, extended in U-022): pure stdlib `http.server`, no third-party deps (05 §5.7) -- this process is also where the Graph Engine scheduler will live later (05 §5.4.1), but that's a different Unit. MVP scope: process start/stop, `c_pick_folder` (03-C §3.-1, now all three platforms -- U-007 shipped macOS only via `osascript`; U-022 adds Windows via a `powershell` FolderBrowserDialog script and Linux via `zenity`/ `kdialog`, whichever is installed), `c_project_create` (03-C §3.0, wired to the U-002 `project_scaffold.create_project()` function that had no HTTP route until now) for `v_project_setup`'s submit flow, `c_task_create`/ `c_bug_report_create` (03-C §3.1/§3.1.1, U-023) for `v_goal_composer`/ `v_bug_feedback`'s submit flows, `c_ws_subscribe` (03-C §3.14, U-024, hand-rolled RFC 6455 WebSocket handshake + framing in src/ws_subscribe.py -- no `websockets` dependency) for real-time `execution_event` push, `c_execution_log_get`/`c_ws_log_tail` (03-C §3.6.1/§3.6.2, U-027, the latter reusing the same hand-rolled WS machinery) for raw execution log viewing, `c_artifact_list`/`c_artifact_get` (U-028, the former unnamed in either 03-C or the MVCP JSON contract) for `v_artifact_browser`, `c_skill_diff`/`c_skill_rollback` + skill/skill_iteration list/get (U-029, `c_skill_diff` likewise only named in the MVCP JSON) for `v_skill_library`/ `v_skill_audit_log`, `c_concurrency_config_update`/`c_budget_config_update`/ `c_log_retention_config_update` (03-C §3.13/§3.13.1, U-030) for the three settings pages, and serving the U-008 frontend shell as static files (no build step, 05 §5.7: whitelisted by exact path, not a generic directory walker, so there's no path-traversal surface to reason about). """ import json import os import platform import socket import sqlite3 import subprocess import threading import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from src import ( artifact_query, budget_guardrail, bug_record, deployment_query, execution_log, org_loop_route, paths, project_query, project_scaffold, schema, settings_config, skill_query, skill_update, task_create, task_execution, task_query, work_graph_run_query, ws_subscribe, ) _FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend") # U-008: exact-path whitelist, not a generic static-file server -- the # frontend is 3 fixed files until PH3 adds real pages. _STATIC_FILES = { "/": ("index.html", "text/html; charset=utf-8"), "/index.html": ("index.html", "text/html; charset=utf-8"), "/app.js": ("app.js", "text/javascript; charset=utf-8"), "/style.css": ("style.css", "text/css; charset=utf-8"), } class DialogUnavailableError(Exception): """DIALOG_UNAVAILABLE -- 03-C §3.-1: no native-dialog capability here.""" def pick_folder(): """c_pick_folder (03-C §3.-1), all three platforms. Returns {"selected_path": str} on success, {"selected_path": None} on user cancel (not an error). Raises DialogUnavailableError if the platform's native dialog tool itself fails or isn't installed.""" system = platform.system() if system == "Darwin": return _pick_folder_macos() if system == "Windows": return _pick_folder_windows() return _pick_folder_linux() def _pick_folder_macos(): script = 'POSIX path of (choose folder with prompt "Select a graphlemon project folder")' result = subprocess.run(["osascript", "-e", script], capture_output=True, text=True) if result.returncode != 0: if "User canceled" in result.stderr or "(-128)" in result.stderr: return {"selected_path": None} raise DialogUnavailableError(result.stderr.strip()) return {"selected_path": result.stdout.strip()} def _pick_folder_windows(): # .NET FolderBrowserDialog via a one-shot PowerShell script -- no # cancel-vs-error signal other than empty stdout on a clean exit. script = ( "Add-Type -AssemblyName System.Windows.Forms; " "$f = New-Object System.Windows.Forms.FolderBrowserDialog; " "if ($f.ShowDialog() -eq [System.Windows.Forms.DialogResult]::OK) { Write-Output $f.SelectedPath }" ) try: result = subprocess.run( ["powershell", "-NoProfile", "-NonInteractive", "-Command", script], capture_output=True, text=True, ) except FileNotFoundError as e: raise DialogUnavailableError(str(e)) if result.returncode != 0: raise DialogUnavailableError(result.stderr.strip()) return {"selected_path": result.stdout.strip() or None} def _pick_folder_linux(): # zenity (GNOME) and kdialog (KDE) are the two dialogs actually present # across mainstream Linux desktops; try whichever exists. Both exit # non-zero on user cancel, indistinguishable here from "no folder # chosen" -- both cases map to selected_path=None, not an error. for cmd in ( ["zenity", "--file-selection", "--directory"], ["kdialog", "--getexistingdirectory", os.path.expanduser("~")], ): try: result = subprocess.run(cmd, capture_output=True, text=True) except FileNotFoundError: continue if result.returncode != 0: return {"selected_path": None} return {"selected_path": result.stdout.strip()} raise DialogUnavailableError("neither zenity nor kdialog is installed") _WS_PROJECTS_PREFIX = "/api/ws/projects/" _PROJECTS_PREFIX = "/api/projects/" _TASKS_PREFIX = "/api/tasks/" _WORK_NODES_PREFIX = "/api/work-nodes/" _WS_WORK_NODES_PREFIX = "/api/ws/work-nodes/" _LOG_SUFFIX = "/log" _ARTIFACTS_PREFIX = "/api/artifacts/" _SKILLS_PREFIX = "/api/skills/" _DEPLOYMENTS_PREFIX = "/api/deployments/" class _Handler(BaseHTTPRequestHandler): def do_GET(self): parsed = urllib.parse.urlsplit(self.path) if parsed.path == "/api/health": self._handle_health() return if parsed.path.startswith(_WS_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX): node_exec_id = parsed.path[len(_WS_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)] self._handle_ws_log_tail(node_exec_id) return if parsed.path.startswith(_WS_PROJECTS_PREFIX): self._handle_ws_subscribe(parsed.path, parsed.query) return if parsed.path == "/api/projects": self._handle_project_list() return if parsed.path.startswith(_PROJECTS_PREFIX): project_id = parsed.path[len(_PROJECTS_PREFIX):] if project_id and "/" not in project_id: self._handle_project_get(project_id) return if parsed.path == "/api/tasks": self._handle_task_list(parsed.query) return if parsed.path.startswith(_TASKS_PREFIX): task_id = parsed.path[len(_TASKS_PREFIX):] if task_id and "/" not in task_id: self._handle_task_get(task_id) return if parsed.path == "/api/settings/budget": self._handle_budget_list() return if parsed.path == "/api/settings/concurrency": self._handle_concurrency_get() return if parsed.path == "/api/settings/log-retention": self._handle_log_retention_get() return if parsed.path.startswith(_WORK_NODES_PREFIX) and parsed.path.endswith(_LOG_SUFFIX): node_exec_id = parsed.path[len(_WORK_NODES_PREFIX):-len(_LOG_SUFFIX)] self._handle_execution_log_get(node_exec_id, parsed.query) return if parsed.path == "/api/artifacts": self._handle_artifact_list(parsed.query) return if parsed.path == "/api/work-graph-runs": self._handle_work_graph_run_list(parsed.query) return if parsed.path.startswith(_ARTIFACTS_PREFIX): artifact_id = parsed.path[len(_ARTIFACTS_PREFIX):] if artifact_id and "/" not in artifact_id: self._handle_artifact_get(artifact_id) return if parsed.path == "/api/skills": self._handle_skill_list() return if parsed.path == "/api/skills/diff": self._handle_skill_diff(parsed.query) return if parsed.path.startswith(_SKILLS_PREFIX) and parsed.path.endswith("/audit"): skill_id = parsed.path[len(_SKILLS_PREFIX):-len("/audit")] self._handle_skill_audit_log(skill_id) return if parsed.path.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) else: self._send_json(404, {"error": "NOT_FOUND"}) def do_PUT(self): if self.path == "/api/settings/concurrency": self._handle_concurrency_update() elif self.path == "/api/settings/budget": self._handle_budget_update() elif self.path == "/api/settings/log-retention": self._handle_log_retention_update() elif self.path.startswith(_PROJECTS_PREFIX) and "/" not in self.path[len(_PROJECTS_PREFIX):]: self._handle_project_update(self.path[len(_PROJECTS_PREFIX):]) elif self.path.startswith(_TASKS_PREFIX) and "/" not in self.path[len(_TASKS_PREFIX):]: self._handle_task_update(self.path[len(_TASKS_PREFIX):]) else: self._send_json(404, {"error": "NOT_FOUND"}) def do_DELETE(self): if self.path.startswith(_PROJECTS_PREFIX) and "/" not in self.path[len(_PROJECTS_PREFIX):]: self._handle_project_delete(self.path[len(_PROJECTS_PREFIX):]) elif self.path.startswith(_TASKS_PREFIX) and "/" not in self.path[len(_TASKS_PREFIX):]: self._handle_task_delete(self.path[len(_TASKS_PREFIX):]) else: self._send_json(404, {"error": "NOT_FOUND"}) def _handle_pick_folder(self): try: self._send_json(200, pick_folder()) except DialogUnavailableError as e: self._send_json(422, {"error": "DIALOG_UNAVAILABLE", "detail": str(e)}) def _read_json_body(self): """Returns (payload, None) on success, (None, error_response) if the body isn't valid JSON -- caller sends error_response and returns.""" length = int(self.headers.get("Content-Length", 0)) try: return json.loads(self.rfile.read(length) or b"{}"), None except json.JSONDecodeError: return None, {"error": "INVALID_INPUT", "detail": "body is not valid JSON"} def _handle_health(self): """c_deploy_supervisor_health_check's `http_health` probe (U-038, 03-C §3.15/§6). Confirms both "the HTTP server thread answers" and "the sqlite connection is actually usable" -- a process that's alive but whose DB connection got wedged somehow would otherwise pass a process_alive check and fail every real request afterward; catching that here, not just accepting a 200 for existing, is the point of a health check in the first place.""" try: with self.server.conn_lock: self.server.conn.execute("SELECT 1").fetchone() except sqlite3.Error as e: self._send_json(503, {"status": "error", "detail": str(e)}) return self._send_json(200, {"status": "ok"}) def _handle_project_create(self): payload, error = self._read_json_body() if error: self._send_json(400, error) return target_folder_path = payload.get("target_folder_path") if not target_folder_path: self._send_json(400, {"error": "INVALID_INPUT", "detail": "target_folder_path is required"}) return try: with self.server.conn_lock: result = project_scaffold.create_project( self.server.conn, target_folder_path, name=payload.get("name"), init_git_if_missing=bool(payload.get("init_git_if_missing", False)), ) except project_scaffold.PathNotFoundError as e: self._send_json(404, {"error": "PATH_NOT_FOUND", "detail": str(e)}) except project_scaffold.PathNotWritableError as e: self._send_json(403, {"error": "PATH_NOT_WRITABLE", "detail": str(e)}) except project_scaffold.GitInitDeclinedError as e: self._send_json(409, {"error": "GIT_INIT_DECLINED", "detail": str(e)}) else: self._send_json(200, result) def _handle_project_list(self): with self.server.conn_lock: result = project_query.list_projects(self.server.conn) self._send_json(200, result) def _handle_project_get(self, project_id): try: with self.server.conn_lock: result = project_query.get_project(self.server.conn, project_id) except project_query.ProjectNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) else: self._send_json(200, result) def _handle_project_update(self, project_id): payload, error = self._read_json_body() if error: self._send_json(400, error) return try: with self.server.conn_lock: result = project_query.update_project( self.server.conn, project_id, name=payload.get("name"), status=payload.get("status"), ) except project_query.ProjectNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) except project_query.InvalidParamError as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_project_delete(self, project_id): """"Delete" archives (project.status='archived') rather than removing the row -- see project_query.py's own docstring for why a real DELETE isn't safe here (FK-referenced history).""" try: with self.server.conn_lock: result = project_query.update_project(self.server.conn, project_id, status="archived") except project_query.ProjectNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) else: self._send_json(200, result) def _handle_task_create(self): payload, error = self._read_json_body() if error: self._send_json(400, error) return try: with self.server.conn_lock: result = task_create.create_task( self.server.conn, payload.get("project_id"), payload.get("goal_text"), task_type=payload.get("task_type", "goal"), created_via=payload.get("created_via", "api"), ) except task_create.InvalidParamError as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_bug_report_create(self): payload, error = self._read_json_body() if error: self._send_json(400, error) return try: with self.server.conn_lock: result = bug_record.create_bug_report( self.server.conn, payload.get("project_id"), payload.get("description"), repro_steps=payload.get("repro_steps"), related_task_id=payload.get("related_task_id"), ) except bug_record.InvalidParamError as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_task_list(self, query_string): params = urllib.parse.parse_qs(query_string) project_id = params.get("project_id", [None])[0] if not project_id: self._send_json(400, {"error": "INVALID_INPUT", "detail": "project_id is required"}) return try: with self.server.conn_lock: result = task_query.list_tasks( self.server.conn, project_id, status=params.get("status", [None])[0], org_node=params.get("org_node", [None])[0], keyword=params.get("keyword", [None])[0], sort=params.get("sort", [None])[0], order=params.get("order", [None])[0], ) except task_query.InvalidParamError as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_task_get(self, task_id): try: with self.server.conn_lock: result = task_query.get_task(self.server.conn, task_id) except task_query.TaskNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) else: self._send_json(200, result) def _handle_task_update(self, task_id): payload, error = self._read_json_body() if error: self._send_json(400, error) return goal_text = payload.get("goal_text") if not goal_text: self._send_json(400, {"error": "INVALID_PARAM", "detail": "goal_text is required"}) return try: with self.server.conn_lock: result = task_query.update_task_goal_text(self.server.conn, task_id, goal_text) except task_query.TaskNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) except task_query.TaskNotEditableError as e: self._send_json(409, {"error": "TASK_NOT_EDITABLE", "detail": str(e)}) else: self._send_json(200, result) def _handle_task_delete(self, task_id): try: with self.server.conn_lock: task_query.delete_task(self.server.conn, task_id) except task_query.TaskNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) except task_query.TaskNotEditableError as e: self._send_json(409, {"error": "TASK_NOT_EDITABLE", "detail": str(e)}) else: self._send_json(200, {"task_id": task_id, "deleted": True}) def _handle_task_route(self, task_id): """c_org_loop_route (03-C §3.11) -- the underlying route_org_loop() was built in U-016 with no HTTP route until now; v_task_detail's click_retry (manual_resume) is what actually needs it wired up. **2026-08-03**: if the task was blocked via a user-initiated pause/stop (task_execution.request_pause, blocked_reason in ('user_paused', 'user_stopped')), a successful manual_resume also spawns a background thread to actually continue real execution -- matching what the ORIGINAL blocked_pending_human/manual_resume design already implies (04-P: "调度器继续执行队列中其他任务" -- resuming should let the scheduler keep going, not just flip a status flag and stop). **2026-08-04 broadened**: found for real -- a transient network blip (`claude` CLI: "API Error: Unable to connect to API (ENOTFOUND)") blocks a task with blocked_reason='execution_error', same as any other unroutable exception (task_execution._mark_blocked's catch-all). The original 'user_paused'/'user_stopped' allowlist left this stuck: manual_resume flipped the status back but nothing ever re-triggered real execution, and /execute's own guard requires status='queued' (a resumed task lands on 'planning_brd_prd_sdd' or 'coding' instead) -- clicking "resume" did nothing a second click could fix. The user has ALREADY made the judgment call to retry by clicking resume at all; withholding continuation only made sense for the two reasons that have their OWN dedicated resume path requiring a different human action first: `budget_exceeded` (must raise the limit -- see budget_guardrail.update_budget_limit_and_resume/ upsert_budget_config, which are the actual intended resume trigger) and `loop_limit_reached` (excluded explicitly by name, not just by relying on its real from_status usually being 'validating' -- a wrong-status edge case, real or a test seeding one, shouldn't be able to accidentally slip a 3-strikes task back into auto-retry). Every other blocked_reason (execution_error, classifier_blocked, corrupt_persisted_state, usage_limit_wait_timeout, prd_consistency_check_failed, user_paused, user_stopped, ...) now auto-continues on a successful resume to a resumable status.""" payload, error = self._read_json_body() if error: self._send_json(400, error) return decided_route = payload.get("decided_route") pre_resume_reason = None if decided_route == "manual_resume": row = self.server.conn.execute( "SELECT blocked_reason FROM task WHERE task_id = ?", (task_id,) ).fetchone() pre_resume_reason = row[0] if row else None try: with self.server.conn_lock: result = org_loop_route.route_org_loop(self.server.conn, task_id, decided_route) except org_loop_route.LoopLimitReachedError as e: self._send_json(409, {"error": "LOOP_LIMIT_REACHED", "detail": str(e)}) return except (org_loop_route.TaskNotBlockedError, org_loop_route.TaskNotValidatingError, ValueError) as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) return if pre_resume_reason not in ("budget_exceeded", "loop_limit_reached") and result.get("new_status") in ( "queued", "planning_brd_prd_sdd", "coding", ): task_row = self.server.conn.execute( "SELECT project_id, goal_text FROM task WHERE task_id = ?", (task_id,) ).fetchone() target_folder_path = self.server.conn.execute( "SELECT target_folder_path FROM project WHERE project_id = ?", (task_row[0],) ).fetchone()[0] task_execution.start_task_execution_in_background( self.server.db_path, task_id, target_folder_path, task_row[1], ) self._send_json(200, result) def _handle_task_execute(self, task_id): """c_task_execute (2026-08-03): the manual "Start" trigger -- confirmed with the user this is genuinely missing (03-C's own c_task_create contract already assumed task creation would auto-trigger scheduling via an always-on scheduler that was never actually wired up as a running process, see task_execution.py's module docstring).""" row = self.server.conn.execute("SELECT project_id, goal_text, status FROM task WHERE task_id = ?", (task_id,)).fetchone() if row is None: self._send_json(404, {"error": "NOT_FOUND"}) return project_id, goal_text, status = row if status != "queued": self._send_json(409, {"error": "TASK_NOT_QUEUED", "detail": "task is status={}, not queued".format(status)}) return target_folder_path = self.server.conn.execute( "SELECT target_folder_path FROM project WHERE project_id = ?", (project_id,) ).fetchone()[0] task_execution.start_task_execution_in_background(self.server.db_path, task_id, target_folder_path, goal_text) self._send_json(200, {"task_id": task_id, "started": True}) def _handle_task_pause(self, task_id, reason): try: with self.server.conn_lock: result = task_execution.request_pause(self.server.conn, task_id, reason) except task_execution.TaskExecutionError as e: self._send_json(409, {"error": "TASK_NOT_PAUSABLE", "detail": str(e)}) else: self._send_json(200, result) def _handle_budget_list(self): with self.server.conn_lock: result = budget_guardrail.list_budget_configs(self.server.conn) self._send_json(200, result) def _handle_budget_update(self): payload, error = self._read_json_body() if error: self._send_json(400, error) return try: with self.server.conn_lock: result = budget_guardrail.upsert_budget_config( self.server.conn, payload.get("scope"), payload.get("limit_type"), payload.get("limit_value"), payload.get("updated_by", "console"), ) except (budget_guardrail.InvalidParamError, TypeError) as e: # TypeError here means a required numeric field (limit_value) was # missing/null -- comparing None <= 0 raises TypeError, not # InvalidParamError, before upsert_budget_config's own checks run self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_concurrency_get(self): with self.server.conn_lock: result = settings_config.get_concurrency_config(self.server.conn) self._send_json(200, result) def _handle_concurrency_update(self): payload, error = self._read_json_body() if error: self._send_json(400, error) return try: with self.server.conn_lock: result = settings_config.upsert_concurrency_config( self.server.conn, payload.get("max_concurrent_goals"), payload.get("updated_by", "console"), ) except (settings_config.InvalidParamError, TypeError) as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_log_retention_get(self): with self.server.conn_lock: result = settings_config.get_log_retention_config(self.server.conn) self._send_json(200, result) def _handle_log_retention_update(self): payload, error = self._read_json_body() if error: self._send_json(400, error) return try: with self.server.conn_lock: result = settings_config.upsert_log_retention_config( self.server.conn, payload.get("retention_days"), payload.get("updated_by", "console"), max_total_size_mb=payload.get("max_total_size_mb"), ) except (settings_config.InvalidParamError, TypeError) as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_artifact_list(self, query_string): params = urllib.parse.parse_qs(query_string) project_id = params.get("project_id", [None])[0] if not project_id: self._send_json(400, {"error": "INVALID_INPUT", "detail": "project_id is required"}) return with self.server.conn_lock: result = artifact_query.list_artifacts( self.server.conn, project_id, task_id=params.get("task_id", [None])[0], artifact_type=params.get("artifact_type", [None])[0], ) self._send_json(200, result) def _handle_work_graph_run_list(self, query_string): params = urllib.parse.parse_qs(query_string) try: with self.server.conn_lock: result = work_graph_run_query.list_work_graph_runs( self.server.conn, params.get("project_id", [None])[0], org_node=params.get("org_node", [None])[0], limit=params.get("limit", [None])[0], ) except work_graph_run_query.InvalidParamError as e: self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)}) else: self._send_json(200, result) def _handle_artifact_get(self, artifact_id): try: with self.server.conn_lock: result = artifact_query.get_artifact(self.server.conn, artifact_id) except artifact_query.ArtifactNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) else: self._send_json(200, result) def _handle_deployment_list(self): with self.server.conn_lock: result = deployment_query.list_deployments(self.server.conn) self._send_json(200, result) def _handle_deployment_current(self): with self.server.conn_lock: result = deployment_query.get_current_deployment(self.server.conn) self._send_json(200, result) def _handle_deployment_get(self, deployment_id): try: with self.server.conn_lock: result = deployment_query.get_deployment(self.server.conn, deployment_id) except deployment_query.DeploymentNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) else: self._send_json(200, result) def _handle_skill_list(self): with self.server.conn_lock: result = skill_query.list_skills(self.server.conn) self._send_json(200, result) def _handle_skill_audit_log(self, skill_id): try: with self.server.conn_lock: result = skill_query.get_skill_audit_log(self.server.conn, skill_id) except skill_query.SkillNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) else: self._send_json(200, result) def _handle_skill_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) class _Server(ThreadingHTTPServer): # long-lived WS connections (c_ws_subscribe) shouldn't be able to block # process shutdown -- ThreadingMixIn.daemon_threads defaults to False. daemon_threads = True def server_close(self): super().server_close() # a WS handler thread can still be mid-poll on another thread at # shutdown (its next tick would just get a clean ProgrammingError, # already caught) -- closing conn without this lock races the C # sqlite3 connection object across threads with no synchronization # at all, which segfaults rather than raising cleanly. with self.conn_lock: self.conn.close() def create_server(host="127.0.0.1", port=0, db_path=None, skill_repo_path=None): """port=0 lets the OS assign a free port -- avoids collisions in tests and lets multiple graphlemon instances run side by side if ever needed. db_path defaults to paths.db_path() (U-003's platform_home, so it never lives inside a target project's own repo); tests pass ":memory:" or a tempfile so runs don't touch the real developer's ~/.graphlemon/. skill_repo_path defaults to skill_query._REPO_ROOT (graphlemon's own checkout -- Skills genuinely live there, U-018/U-029) but is injectable for the same reason db_path is: c_skill_rollback does a real `git commit`, and a test hitting that route with the real default would write an actual commit into whatever repo this process happens to be running from. Never exposed to HTTP clients -- only settable here, at process/test-construction time.""" resolved_db_path = db_path if db_path is not None else paths.db_path() conn = schema.connect(resolved_db_path, check_same_thread=False) _ensure_schema(conn) server = _Server((host, port), _Handler) server.conn = conn server.db_path = resolved_db_path server.conn_lock = threading.Lock() server.skill_repo_path = skill_repo_path if skill_repo_path is not None else skill_query._REPO_ROOT return server def main(): """U-037: CLI entry point (`python3 -m src.backend_server ...`) -- until this Unit, `create_server()` was only ever instantiated in-process by tests; there was no way to actually launch graphlemon's own backend as a standalone OS process. c_deploy_supervisor_stage needs exactly that (a real candidate process, `subprocess.Popen`, listening on a real port) -- found while building it, added here since it's the natural, minimal home for "run this server for real" regardless of who launches it. """ import argparse import signal parser = argparse.ArgumentParser() parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=0) parser.add_argument("--db-path", default=None) parser.add_argument("--skill-repo-path", default=None) args = parser.parse_args() server = create_server(args.host, args.port, args.db_path, args.skill_repo_path) def _handle_sigterm(signum, frame): # server.shutdown() blocks until serve_forever()'s loop notices and # returns -- calling it from the SAME thread that's running # serve_forever() (the signal handler runs on the main thread, same # as serve_forever() below) deadlocks, per http.server's own docs. threading.Thread(target=server.shutdown, daemon=True).start() signal.signal(signal.SIGTERM, _handle_sigterm) print("graphlemon backend listening on {}:{}".format(*server.server_address), flush=True) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()