"""Read-only Hermes Kanban bridge for the WebUI. This module exposes a small WebUI-native API under ``/api/kanban/*`` while keeping Hermes Agent's ``hermes_cli.kanban_db`` as the only source of truth. The first integration is deliberately read-only; write/move semantics can be added in later focused PRs. """ from __future__ import annotations import json import time from dataclasses import asdict, is_dataclass from urllib.parse import parse_qs, unquote from api.helpers import bad, j BOARD_COLUMNS = ["triage", "todo", "ready", "running", "blocked", "done"] _TASK_PREFIX = "/api/kanban/tasks/" def _kb(): from hermes_cli import kanban_db as kb return kb def _conn(): kb = _kb() kb.init_db() return kb.connect() def _obj_dict(value): if value is None: return None if is_dataclass(value): return asdict(value) if isinstance(value, dict): return dict(value) return dict(getattr(value, "__dict__", {})) def _task_dict(task): data = _obj_dict(task) if not data: return data try: age = _kb().task_age(task) except Exception: age = None data["age_seconds"] = age data["age"] = age data.setdefault("progress", None) return data def _latest_event_id(conn) -> int: try: row = conn.execute("SELECT COALESCE(MAX(id), 0) AS latest FROM task_events").fetchone() return int(row["latest"] or 0) except Exception: return 0 def _bool_query(parsed, name: str, default: bool = False) -> bool: raw = (parse_qs(parsed.query or "").get(name) or [None])[0] if raw is None: return default return str(raw).strip().lower() in {"1", "true", "yes", "on"} def _str_query(parsed, name: str): raw = (parse_qs(parsed.query or "").get(name) or [None])[0] return str(raw).strip() or None if raw is not None else None def _int_query(parsed, name: str, default=None, *, minimum=None, maximum=None): raw = _str_query(parsed, name) if raw is None: return default try: value = int(raw) except (TypeError, ValueError): return default if minimum is not None: value = max(minimum, value) if maximum is not None: value = min(maximum, value) return value def _task_link_counts(conn, tasks): counts = {task.id: {"parents": 0, "children": 0} for task in tasks} try: rows = conn.execute("SELECT parent_id, child_id FROM task_links").fetchall() except Exception: return counts for row in rows: counts.setdefault(row["parent_id"], {"parents": 0, "children": 0})["children"] += 1 counts.setdefault(row["child_id"], {"parents": 0, "children": 0})["parents"] += 1 return counts def _comment_counts(conn): try: rows = conn.execute( "SELECT task_id, COUNT(*) AS n FROM task_comments GROUP BY task_id" ).fetchall() except Exception: return {} return {row["task_id"]: int(row["n"] or 0) for row in rows} def _board_payload(parsed): kb = _kb() tenant = _str_query(parsed, "tenant") assignee = _str_query(parsed, "assignee") include_archived = _bool_query(parsed, "include_archived", False) only_mine = _bool_query(parsed, "only_mine", False) since = _int_query(parsed, "since", None, minimum=0) profile = None if only_mine and not assignee: try: from api.profiles import get_active_profile_name profile = get_active_profile_name() or "default" except Exception: profile = "default" assignee = profile with _conn() as conn: latest_event_id = _latest_event_id(conn) if since is not None and since >= latest_event_id: return {"changed": False, "latest_event_id": latest_event_id, "read_only": True} tasks = kb.list_tasks( conn, tenant=tenant, assignee=assignee, include_archived=include_archived, ) link_counts = _task_link_counts(conn, tasks) comment_counts = _comment_counts(conn) def row(task): data = _task_dict(task) data["link_counts"] = link_counts.get(task.id, {"parents": 0, "children": 0}) data["comment_count"] = comment_counts.get(task.id, 0) return data columns = [ {"name": name, "tasks": [row(task) for task in tasks if task.status == name]} for name in BOARD_COLUMNS ] if include_archived: columns.append({ "name": "archived", "tasks": [row(task) for task in tasks if task.status == "archived"], }) return { "columns": columns, "tenants": sorted({task.tenant for task in tasks if getattr(task, "tenant", None)}), "assignees": sorted({task.assignee for task in tasks if getattr(task, "assignee", None)}), "latest_event_id": latest_event_id, "changed": True, "read_only": True, "filters": { "tenant": tenant, "assignee": assignee, "include_archived": include_archived, "only_mine": only_mine, "profile": profile, }, } def _validate_status(status: str) -> str: value = str(status or "").strip().lower() allowed = set(BOARD_COLUMNS) | {"archived"} if value not in allowed: raise ValueError(f"invalid status: {value}") return value def _set_status_direct(conn, task_id: str, new_status: str) -> bool: """Direct status write for drag-drop moves not covered by structured verbs. Used for ``todo <-> ready`` and ``running -> ready`` transitions. The structured verbs (``complete_task``, ``block_task``, ``unblock_task``, ``archive_task``, ``claim_task``) own their own state changes; this helper handles the remainder while preserving the dispatcher's contract: - When transitioning OFF ``running`` to anything other than the terminal verbs, claim_lock / claim_expires / worker_pid are nulled so the dispatcher doesn't see a phantom-running task. The active run (if any) is closed with ``outcome='reclaimed'`` so attempt history isn't orphaned. - When transitioning INTO ``running``, claim fields are preserved (this function is NOT used for entering 'running' — that goes through ``kb.claim_task()`` and the bridge rejects raw 'running' status writes with HTTP 400). Mirrors the agent dashboard plugin's ``_set_status_direct`` (plugins/kanban/dashboard/plugin_api.py) so first-party clients see identical behaviour from either surface. """ kb = _kb() with kb.write_txn(conn): prev = conn.execute( "SELECT status, current_run_id FROM tasks WHERE id = ?", (task_id,), ).fetchone() if prev is None: return False was_running = prev["status"] == "running" cur = conn.execute( "UPDATE tasks SET status = ?, " " claim_lock = CASE WHEN ? = 'running' THEN claim_lock ELSE NULL END, " " claim_expires = CASE WHEN ? = 'running' THEN claim_expires ELSE NULL END, " " worker_pid = CASE WHEN ? = 'running' THEN worker_pid ELSE NULL END " "WHERE id = ?", (new_status, new_status, new_status, new_status, task_id), ) if cur.rowcount != 1: return False run_id = None if was_running and new_status != "running" and prev["current_run_id"]: try: run_id = kb._end_run( conn, task_id, outcome="reclaimed", status="reclaimed", summary=f"status changed to {new_status} (webui/direct)", ) except Exception: # _end_run is best-effort here; the status flip itself is # what matters for sidebar rendering. run_id = None conn.execute( "INSERT INTO task_events (task_id, run_id, kind, payload, created_at) " "VALUES (?, ?, 'status', ?, ?)", (task_id, run_id, json.dumps({"status": new_status, "source": "webui"}), int(time.time())), ) if new_status in ("done", "ready") and hasattr(kb, "recompute_ready"): try: kb.recompute_ready(conn) except Exception: pass return True def _create_task_payload(body: dict): title = str(body.get("title") or "").strip() if not title: raise ValueError("title is required") try: priority = int(body.get("priority") or 0) except (TypeError, ValueError): raise ValueError("priority must be an integer") kb = _kb() requested_status = body.get("status") with _conn() as conn: task_id = kb.create_task( conn, title=title, body=body.get("body") or None, assignee=body.get("assignee") or None, created_by=body.get("created_by") or "webui", tenant=body.get("tenant") or None, priority=priority, parents=body.get("parents") or (), triage=bool(body.get("triage") or False), workspace_kind=body.get("workspace_kind") or "scratch", workspace_path=body.get("workspace_path") or None, idempotency_key=body.get("idempotency_key") or None, max_runtime_seconds=body.get("max_runtime_seconds") or None, skills=body.get("skills") or None, ) if requested_status: _patch_task(conn, task_id, {"status": requested_status}) return {"task": _task_dict(kb.get_task(conn, task_id)), "read_only": False} def _patch_task(conn, task_id: str, body: dict): kb = _kb() task = kb.get_task(conn, task_id) if not task: raise LookupError("task not found") updates = {} if "title" in body: title = str(body.get("title") or "").strip() if not title: raise ValueError("title is required") updates["title"] = title if "body" in body: updates["body"] = body.get("body") or None if "tenant" in body: updates["tenant"] = body.get("tenant") or None if "priority" in body: try: updates["priority"] = int(body.get("priority") or 0) except (TypeError, ValueError): raise ValueError("priority must be an integer") for field, value in updates.items(): if hasattr(task, field): try: setattr(task, field, value) except Exception: pass if updates: assignments = ", ".join(f"{field} = ?" for field in updates) conn.execute(f"UPDATE tasks SET {assignments} WHERE id = ?", [*updates.values(), task_id]) if hasattr(kb, "_append_event"): kb._append_event(conn, task_id, "updated", {"fields": list(updates), "source": "webui"}) if "assignee" in body: if not kb.assign_task(conn, task_id, body.get("assignee") or None): raise LookupError("task not found") if "status" not in body or body.get("status") in (None, ""): return status = _validate_status(body.get("status")) if status == "done": if not kb.complete_task(conn, task_id, result=body.get("result"), summary=body.get("summary")): raise LookupError("task not found") elif status == "blocked": if not kb.block_task(conn, task_id, reason=body.get("block_reason") or body.get("reason")): raise LookupError("task not found") elif status == "archived": if not kb.archive_task(conn, task_id): raise LookupError("task not found") elif status == "running": # The 'running' state is owned by the kanban dispatcher / claim # protocol — entering it via raw UPDATE bypasses claim_lock, # claim_expires, started_at, and worker_pid, which leaves the task # in a state the dispatcher treats as "phantom claimed" and may # reclaim or hide. Match the agent dashboard plugin's contract # (plugins/kanban/dashboard/plugin_api.py update_task) by rejecting # this transition with HTTP 400. Workers enter 'running' via # kb.claim_task(); UI users should use the dispatcher nudge. raise ValueError( "Cannot set status to 'running' directly; use the dispatcher/claim path" ) elif status == "ready": # If the task is currently 'blocked', use the structured unblock # verb so the unblocked event fires. Otherwise it's a legitimate # drag-drop or click move (e.g. todo → ready, running → ready when # the user yanks a stuck worker back to the queue) and we use the # claim-aware direct status write. current = kb.get_task(conn, task_id) if not current: raise LookupError("task not found") if current.status == "blocked": if not kb.unblock_task(conn, task_id): raise LookupError("task not found") else: if not _set_status_direct(conn, task_id, "ready"): raise LookupError("task not found") elif status in ("triage", "todo"): # Direct status write for drag-drop moves between non-running, # non-terminal columns. Uses the claim-aware helper that nulls out # claim_lock / claim_expires / worker_pid when leaving 'running' # and ends any active run with outcome='reclaimed'. if not _set_status_direct(conn, task_id, status): raise LookupError("task not found") else: # _validate_status guarantees we never reach here, but be defensive. raise ValueError(f"unknown status: {status}") def _patch_task_payload(task_id: str, body: dict): task_id = str(task_id or "").strip() if not task_id: raise ValueError("task_id is required") kb = _kb() with _conn() as conn: _patch_task(conn, task_id, body) return {"task": _task_dict(kb.get_task(conn, task_id)), "read_only": False} def _comment_payload(task_id: str, body: dict): task_id = str(task_id or "").strip() comment_body = str(body.get("body") or "").strip() if not task_id: raise ValueError("task_id is required") if not comment_body: raise ValueError("body is required") kb = _kb() with _conn() as conn: if not kb.get_task(conn, task_id): raise LookupError("task not found") comment_id = kb.add_comment(conn, task_id, body.get("author") or "webui", comment_body) return {"ok": True, "comment_id": comment_id, "read_only": False} def _link_tasks_payload(body: dict, *, unlink: bool = False): parent_id = str(body.get("parent_id") or "").strip() child_id = str(body.get("child_id") or "").strip() if not parent_id or not child_id: raise ValueError("parent_id and child_id are required") kb = _kb() with _conn() as conn: if not kb.get_task(conn, parent_id): raise LookupError("parent task not found") if not kb.get_task(conn, child_id): raise LookupError("child task not found") if unlink: changed = kb.unlink_tasks(conn, parent_id, child_id) return {"ok": True, "changed": bool(changed), "parent_id": parent_id, "child_id": child_id, "read_only": False} kb.link_tasks(conn, parent_id, child_id) return {"ok": True, "parent_id": parent_id, "child_id": child_id, "read_only": False} def _links_for(conn, task_id: str) -> dict: kb = _kb() return { "parents": kb.parent_ids(conn, task_id), "children": kb.child_ids(conn, task_id), } def _task_detail_payload(task_id: str): kb = _kb() with _conn() as conn: task = kb.get_task(conn, task_id) if not task: return None return { "task": _task_dict(task), "comments": [_obj_dict(c) for c in kb.list_comments(conn, task_id)], "events": [_obj_dict(e) for e in kb.list_events(conn, task_id)], "links": _links_for(conn, task_id), "runs": [_obj_dict(r) for r in kb.list_runs(conn, task_id)], "read_only": True, } def _events_payload(parsed): since = _int_query(parsed, "since", 0, minimum=0) limit = _int_query(parsed, "limit", 200, minimum=1, maximum=200) with _conn() as conn: rows = conn.execute( "SELECT id, task_id, run_id, kind, payload, created_at " "FROM task_events WHERE id > ? ORDER BY id ASC LIMIT ?", (since, limit), ).fetchall() events = [] cursor = since for row in rows: try: payload = json.loads(row["payload"]) if row["payload"] else None except Exception: payload = None events.append({ "id": row["id"], "task_id": row["task_id"], "run_id": row["run_id"], "kind": row["kind"], "payload": payload, "created_at": row["created_at"], }) cursor = int(row["id"]) latest = _latest_event_id(conn) if not events: cursor = latest if since >= latest else since return {"events": events, "cursor": cursor, "latest_event_id": cursor, "read_only": True} def _config_payload(): kb = _kb() try: with _conn() as conn: try: assignees = list(kb.known_assignees(conn)) except Exception: assignees = [] except Exception: assignees = [] try: from hermes_cli.config import load_config cfg = load_config() or {} except Exception: cfg = {} k_cfg = ((cfg.get("dashboard") or {}).get("kanban") or {}) return { "columns": BOARD_COLUMNS, "assignees": assignees, "default_tenant": k_cfg.get("default_tenant") or "", "lane_by_profile": bool(k_cfg.get("lane_by_profile", True)), "include_archived_by_default": bool(k_cfg.get("include_archived_by_default", False)), "render_markdown": bool(k_cfg.get("render_markdown", True)), "read_only": False, } def _stats_payload(): kb = _kb() with _conn() as conn: if hasattr(kb, "board_stats"): return kb.board_stats(conn) rows = conn.execute( "SELECT status, assignee, COUNT(*) AS n FROM tasks WHERE status != 'archived' GROUP BY status, assignee" ).fetchall() by_status = {} by_assignee = {} for row in rows: n = int(row["n"] or 0) by_status[row["status"]] = by_status.get(row["status"], 0) + n assignee = row["assignee"] or "unassigned" by_assignee[assignee] = by_assignee.get(assignee, 0) + n return {"by_status": by_status, "by_assignee": by_assignee} def _assignees_payload(): kb = _kb() with _conn() as conn: try: assignees = list(kb.known_assignees(conn)) except Exception: rows = conn.execute( "SELECT DISTINCT assignee FROM tasks WHERE assignee IS NOT NULL AND assignee != '' ORDER BY assignee" ).fetchall() assignees = [row["assignee"] for row in rows] return {"assignees": assignees} def _task_log_payload(parsed, task_id: str): kb = _kb() tail = _int_query(parsed, "tail", None, minimum=1, maximum=2_000_000) with _conn() as conn: if not kb.get_task(conn, task_id): return None if not hasattr(kb, "read_worker_log"): return {"task_id": task_id, "path": "", "exists": False, "size_bytes": 0, "content": "", "truncated": False} content = kb.read_worker_log(task_id, tail_bytes=tail) log_path = kb.worker_log_path(task_id) if hasattr(kb, "worker_log_path") else None try: size = log_path.stat().st_size if log_path and log_path.exists() else 0 except OSError: size = 0 return { "task_id": task_id, "path": str(log_path or ""), "exists": content is not None, "size_bytes": size, "content": content or "", "truncated": bool(tail and size > tail), } def _bulk_tasks_payload(body: dict): ids = [str(i).strip() for i in (body.get("ids") or []) if str(i).strip()] if not ids: raise ValueError("ids is required") results = [] kb = _kb() with _conn() as conn: for task_id in ids: entry = {"id": task_id, "ok": True} try: if not kb.get_task(conn, task_id): entry.update(ok=False, error="not found") results.append(entry) continue if body.get("archive"): if not kb.archive_task(conn, task_id): entry.update(ok=False, error="archive refused") elif body.get("status") is not None: _patch_task(conn, task_id, {"status": body.get("status")}) if body.get("assignee") is not None: if not kb.assign_task(conn, task_id, body.get("assignee") or None): entry.update(ok=False, error="assign refused") if body.get("priority") is not None: try: priority = int(body.get("priority")) except (TypeError, ValueError): entry.update(ok=False, error="priority must be an integer") else: conn.execute("UPDATE tasks SET priority = ? WHERE id = ?", (priority, task_id)) if hasattr(kb, "_append_event"): kb._append_event(conn, task_id, "reprioritized", {"priority": priority, "source": "webui"}) except Exception as exc: entry.update(ok=False, error=str(exc)) results.append(entry) return {"results": results, "read_only": False} def _dispatch_payload(parsed): kb = _kb() dry_run = _bool_query(parsed, "dry_run", False) max_spawn = _int_query(parsed, "max", 8, minimum=1, maximum=100) if not hasattr(kb, "dispatch_once"): raise ValueError("dispatcher is unavailable") with _conn() as conn: result = kb.dispatch_once(conn, dry_run=dry_run, max_spawn=max_spawn) if isinstance(result, dict): return result try: return asdict(result) except TypeError: return {"result": str(result)} def _task_action_payload(task_id: str, body: dict, action: str): kb = _kb() task_id = str(task_id or "").strip() if not task_id: raise ValueError("task_id is required") with _conn() as conn: if not kb.get_task(conn, task_id): raise LookupError("task not found") if action == "block": ok = kb.block_task(conn, task_id, reason=body.get("reason") or body.get("block_reason")) elif action == "unblock": if hasattr(kb, "unblock_task"): ok = kb.unblock_task(conn, task_id) else: _patch_task(conn, task_id, {"status": "ready"}) ok = True else: raise ValueError(f"invalid action: {action}") if not ok: raise RuntimeError(f"{action} refused") return {"task": _task_dict(kb.get_task(conn, task_id)), "read_only": False} def handle_kanban_get(handler, parsed) -> bool: path = parsed.path if path == "/api/kanban/board": return j(handler, _board_payload(parsed)) or True if path == "/api/kanban/config": return j(handler, _config_payload()) or True if path == "/api/kanban/stats": return j(handler, _stats_payload()) or True if path == "/api/kanban/assignees": return j(handler, _assignees_payload()) or True if path == "/api/kanban/events": return j(handler, _events_payload(parsed)) or True if path.startswith(_TASK_PREFIX) and path.endswith("/log"): task_id = unquote(path[len(_TASK_PREFIX):-len("/log")]).strip("/") if not task_id or "/" in task_id: return False payload = _task_log_payload(parsed, task_id) if payload is None: return bad(handler, "task not found", status=404) return j(handler, payload) or True if path.startswith(_TASK_PREFIX): task_id = unquote(path[len(_TASK_PREFIX):]).strip("/") if not task_id or "/" in task_id: return False payload = _task_detail_payload(task_id) if payload is None: return bad(handler, "task not found", status=404) return j(handler, payload) or True return False def handle_kanban_post(handler, parsed, body) -> bool: path = parsed.path try: if path == "/api/kanban/dispatch": return j(handler, _dispatch_payload(parsed)) or True if path == "/api/kanban/tasks/bulk": return j(handler, _bulk_tasks_payload(body)) or True if path == "/api/kanban/tasks": return j(handler, _create_task_payload(body)) or True if path == "/api/kanban/links": return j(handler, _link_tasks_payload(body)) or True if path == "/api/kanban/links/delete": return j(handler, _link_tasks_payload(body, unlink=True)) or True if path.startswith(_TASK_PREFIX) and path.endswith("/comments"): task_id = path[len(_TASK_PREFIX):-len("/comments")].strip("/") return j(handler, _comment_payload(task_id, body)) or True for suffix, action in (("/block", "block"), ("/unblock", "unblock")): if path.startswith(_TASK_PREFIX) and path.endswith(suffix): task_id = path[len(_TASK_PREFIX):-len(suffix)].strip("/") return j(handler, _task_action_payload(task_id, body, action)) or True if path.startswith(_TASK_PREFIX) and path.endswith("/patch"): task_id = path[len(_TASK_PREFIX):-len("/patch")].strip("/") return j(handler, _patch_task_payload(task_id, body)) or True except LookupError as exc: return bad(handler, str(exc), status=404) except ValueError as exc: return bad(handler, str(exc)) except RuntimeError as exc: return bad(handler, str(exc), status=409) return False def handle_kanban_patch(handler, parsed, body) -> bool: path = parsed.path try: if path.startswith(_TASK_PREFIX): task_id = unquote(path[len(_TASK_PREFIX):]).strip("/") if not task_id or "/" in task_id: return False return j(handler, _patch_task_payload(task_id, body)) or True except LookupError as exc: return bad(handler, str(exc), status=404) except ValueError as exc: return bad(handler, str(exc)) except RuntimeError as exc: return bad(handler, str(exc), status=409) return False def handle_kanban_delete(handler, parsed, body) -> bool: path = parsed.path try: if path == "/api/kanban/links": return j(handler, _link_tasks_payload(body, unlink=True)) or True except LookupError as exc: return bad(handler, str(exc), status=404) except ValueError as exc: return bad(handler, str(exc)) except RuntimeError as exc: return bad(handler, str(exc), status=409) return False