"""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, new_goal_dispatch, 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") # 2026-08-13: execution_priority is an independent, optional field on # this same PUT -- unlike goal_text it's never gated to status= # 'queued' (see update_task_execution_priority's own docstring), so # it's handled as its own branch rather than folded into the # goal_text-only contract this endpoint originally had. has_priority = "execution_priority" in payload if not goal_text and not has_priority: self._send_json(400, {"error": "INVALID_PARAM", "detail": "goal_text or execution_priority is required"}) return try: with self.server.conn_lock: result = None if goal_text: result = task_query.update_task_goal_text(self.server.conn, task_id, goal_text) if has_priority: result = task_query.update_task_execution_priority( self.server.conn, task_id, payload.get("execution_priority") ) except task_query.TaskNotFoundError: self._send_json(404, {"error": "NOT_FOUND"}) except (task_query.TaskNotEditableError, task_query.InvalidParamError) as e: status = 409 if isinstance(e, task_query.TaskNotEditableError) else 400 self._send_json(status, {"error": "TASK_NOT_EDITABLE" if status == 409 else "INVALID_PARAM", "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. # # 'reviewing_planner_output' added 2026-08-13 (real incident, task # b243dfe8): same exact bug, one status later -- Org Graph # unification Phase 1 (2026-08-08) added this status to # task_execution._RESUMABLE_STATUSES, but nobody updated THIS # separate allowlist to match, so a manual_resume off # blocked_pending_human onto 'reviewing_planner_output' flipped the # DB status and spawned nothing -- the task sat completely idle # (zero new execution_event rows, no claude subprocess) looking # exactly like it was "stuck again", when actually nothing was # running at all. Any status added to _RESUMABLE_STATUSES going # forward needs to land here too -- the two lists have now drifted # out of sync twice for the identical reason. if pre_resume_reason not in ("budget_exceeded", "loop_limit_reached") and result.get("new_status") in ( "queued", "planning_brd_prd_sdd", "reviewing_planner_output", "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) schema.migrate_drop_work_graph_run_org_node(conn) schema.migrate_add_concurrency_wait_event_type(conn) schema.migrate_add_task_execution_priority(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 # 45s (2026-08-13, docs/done/scheduler-new-goal-dispatch-requirements.md FR-1): # deliberately far below _DEFAULT_RECONCILIATION_INTERVAL_SECONDS -- the # acceptance bar is a brand-new Goal starting without a human clicking Start, # and the requirements doc's own framing ("接近实时") rules out reusing the # 300s figure, which was tuned for "eventually recover a stuck task", not # "don't make someone wait a Pomodoro to see their new Goal start". Not # sub-10s either: this is a full-table scan (same "no index" simplification # task_reconciliation.py already documents) plus a DB connect/close per pass. _DEFAULT_NEW_GOAL_DISPATCH_INTERVAL_SECONDS = 45 def _start_new_goal_dispatch_loop(db_path, interval_seconds): """Sibling of `_start_reconciliation_loop` for `new_goal_dispatch.dispatch_once()` -- same "own daemon thread, only started from `main()`, run immediately then repeat, never let one bad pass kill the loop, traceback to stderr instead of a silent `except: pass`" shape, kept as a genuinely separate loop/thread rather than folded into the reconciliation one: `docs/done/ scheduler-task-reconciliation-design.md` §7 records that decision -- new-Goal dispatch and stuck-task recovery are independent concerns with different urgency, and merging them into one "万能调度循环" was deliberately rejected.""" def loop(): while True: try: new_goal_dispatch.dispatch_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") parser.add_argument( "--new-goal-dispatch-interval-seconds", type=int, default=_DEFAULT_NEW_GOAL_DISPATCH_INTERVAL_SECONDS ) parser.add_argument("--disable-new-goal-dispatch-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) if not args.disable_new_goal_dispatch_loop: _start_new_goal_dispatch_loop(server.db_path, args.new_goal_dispatch_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()