from __future__ import annotations import hashlib import json import re import secrets import sqlite3 from datetime import datetime, timezone from pathlib import Path from typing import Any def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def hash_token(token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() def row_to_dict(row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return {key: row[key] for key in row.keys()} def json_loads(value: str | None, default: Any) -> Any: if not value: return default try: return json.loads(value) except json.JSONDecodeError: return default def json_dumps(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True) def as_int(value: Any, default: int = 0) -> int: try: return int(value) except (TypeError, ValueError): return default def as_float(value: Any, default: float = 0.0) -> float: try: return float(value) except (TypeError, ValueError): return default class AlreadyRegisteredError(RuntimeError): pass class ManagedConnection(sqlite3.Connection): def __exit__(self, exc_type, exc_value, exc_traceback): try: return super().__exit__(exc_type, exc_value, exc_traceback) finally: self.close() DEFAULT_MODEL_CONFIG = { "target_system_url": "", "vllm_base_url": "", "vllm_model": "", "api_key": "", "grounding_url": "", "grounding_model": "", "ground_coordinate_mode": "absolute", "resolution": "1920x1080", "screenshot_max_width": 1920, "screenshot_max_height": 1200, "max_steps": 40, "request_timeout_seconds": 30, "temperature": 0, } DEFAULT_TOOLS = [ { "id": "browser_control", "name": "Browser Control", "type": "interaction", "version": "0.1", "source": "controller", "capability": "Open pages, click, type, navigate, and inspect Web UI state.", "scenario": "Web / ERP / management system operation", "available": True, }, { "id": "screen_capture", "name": "Screen Capture", "type": "observation", "version": "0.1", "source": "agent", "capability": "Capture controlled-device screenshots for model reasoning and evidence.", "scenario": "Live monitoring and task evidence", "available": True, }, { "id": "keyboard_mouse", "name": "Keyboard Mouse", "type": "interaction", "version": "0.1", "source": "agent", "capability": "Send keyboard and pointer actions to the controlled desktop.", "scenario": "Desktop and browser automation", "available": True, }, { "id": "log_reader", "name": "Log Reader", "type": "evidence", "version": "0.1", "source": "agent", "capability": "Read task logs and expose trace evidence for troubleshooting.", "scenario": "Failure analysis and report evidence", "available": True, }, { "id": "report_generator", "name": "Report Generator", "type": "report", "version": "0.1", "source": "controller", "capability": "Build run summaries from task status, duration, logs, and screenshots.", "scenario": "Acceptance report and evidence package", "available": True, }, ] DEFAULT_TOOL_CONFIG = { "global_enabled": {tool["id"]: True for tool in DEFAULT_TOOLS}, "agent_overrides": {}, "plan_overrides": {}, } class Database: def __init__(self, path: Path, agent_stale_after_seconds: int = 5): self.path = path self.agent_stale_after_seconds = agent_stale_after_seconds def connect(self) -> sqlite3.Connection: self.path.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(self.path, factory=ManagedConnection) connection.row_factory = sqlite3.Row connection.execute("PRAGMA journal_mode=WAL") connection.execute("PRAGMA foreign_keys=ON") return connection def init(self) -> None: with self.connect() as db: db.executescript( """ CREATE TABLE IF NOT EXISTS agents ( id INTEGER PRIMARY KEY AUTOINCREMENT, public_id TEXT NOT NULL UNIQUE, display_name TEXT NOT NULL, deployment_type TEXT NOT NULL, machine_fingerprint TEXT NOT NULL, token_hash TEXT NOT NULL, agent_version TEXT NOT NULL, registered_at TEXT NOT NULL, last_seen_at TEXT, status TEXT NOT NULL DEFAULT 'registered', reported_host TEXT, reported_ip TEXT, reported_log_web_port INTEGER, metadata_json TEXT NOT NULL DEFAULT '{}' ); CREATE INDEX IF NOT EXISTS idx_agents_last_seen ON agents(last_seen_at); CREATE TABLE IF NOT EXISTS heartbeats ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id INTEGER NOT NULL, received_at TEXT NOT NULL, payload_json TEXT NOT NULL, FOREIGN KEY(agent_id) REFERENCES agents(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, external_id TEXT UNIQUE, title TEXT, plan_name TEXT, plan_version INTEGER NOT NULL DEFAULT 1, module TEXT, parent_external_id TEXT, root_external_id TEXT, depth INTEGER NOT NULL DEFAULT 0, task_level TEXT, risk_level TEXT, status TEXT NOT NULL DEFAULT 'ready', priority INTEGER NOT NULL DEFAULT 100, assigned_agent_id INTEGER, payload_json TEXT NOT NULL DEFAULT '{}', failure_policy TEXT NOT NULL DEFAULT 'skip', max_retries INTEGER NOT NULL DEFAULT 0, timeout_seconds INTEGER, dispatch_mode TEXT NOT NULL DEFAULT 'auto', dependencies_json TEXT NOT NULL DEFAULT '[]', tags_json TEXT NOT NULL DEFAULT '[]', attempt INTEGER NOT NULL DEFAULT 0, result_json TEXT, last_error TEXT, duration_seconds REAL, model_duration_seconds REAL, action_duration_seconds REAL, operator_note TEXT, created_at TEXT NOT NULL, claimed_at TEXT, started_at TEXT, finished_at TEXT, updated_at TEXT NOT NULL, FOREIGN KEY(assigned_agent_id) REFERENCES agents(id) ON DELETE SET NULL ); CREATE INDEX IF NOT EXISTS idx_tasks_status_priority ON tasks(status, priority, created_at); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value_json TEXT NOT NULL, version INTEGER NOT NULL DEFAULT 1, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS audit_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL, entity_type TEXT NOT NULL, entity_id TEXT, actor TEXT NOT NULL DEFAULT 'system', reason TEXT, payload_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL ); """ ) self._migrate(db) self._ensure_setting(db, "model_config", DEFAULT_MODEL_CONFIG) self._ensure_setting( db, "tool_config", { "tools": DEFAULT_TOOLS, "config": DEFAULT_TOOL_CONFIG, }, ) def _migrate(self, db: sqlite3.Connection) -> None: self._ensure_column(db, "agents", "reported_log_web_port", "INTEGER") self._ensure_column(db, "agents", "config_revision", "INTEGER NOT NULL DEFAULT 0") self._ensure_column(db, "agents", "config_applied_at", "TEXT") self._ensure_column(db, "agents", "config_restart_requested_at", "TEXT") self._ensure_column(db, "agents", "config_json", "TEXT NOT NULL DEFAULT '{}'") task_columns = { "title": "TEXT", "plan_name": "TEXT", "plan_version": "INTEGER NOT NULL DEFAULT 1", "module": "TEXT", "parent_external_id": "TEXT", "root_external_id": "TEXT", "depth": "INTEGER NOT NULL DEFAULT 0", "task_level": "TEXT", "risk_level": "TEXT", "failure_policy": "TEXT NOT NULL DEFAULT 'skip'", "max_retries": "INTEGER NOT NULL DEFAULT 0", "timeout_seconds": "INTEGER", "dispatch_mode": "TEXT NOT NULL DEFAULT 'auto'", "dependencies_json": "TEXT NOT NULL DEFAULT '[]'", "tags_json": "TEXT NOT NULL DEFAULT '[]'", "attempt": "INTEGER NOT NULL DEFAULT 0", "result_json": "TEXT", "last_error": "TEXT", "duration_seconds": "REAL", "model_duration_seconds": "REAL", "action_duration_seconds": "REAL", "operator_note": "TEXT", "started_at": "TEXT", "finished_at": "TEXT", } for name, definition in task_columns.items(): self._ensure_column(db, "tasks", name, definition) def _ensure_column( self, db: sqlite3.Connection, table_name: str, column_name: str, column_sql: str, ) -> None: columns = { row["name"] for row in db.execute(f"PRAGMA table_info({table_name})").fetchall() } if column_name not in columns: db.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_sql}") def _ensure_setting(self, db: sqlite3.Connection, key: str, value: Any) -> None: existing = db.execute("SELECT key FROM settings WHERE key = ?", (key,)).fetchone() if existing is None: db.execute( """ INSERT INTO settings (key, value_json, version, updated_at) VALUES (?, ?, 1, ?) """, (key, json_dumps(value), utc_now()), ) def _audit( self, db: sqlite3.Connection, event_type: str, entity_type: str, entity_id: str | None, payload: dict[str, Any] | None = None, reason: str | None = None, actor: str = "system", ) -> None: db.execute( """ INSERT INTO audit_events ( event_type, entity_type, entity_id, actor, reason, payload_json, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( event_type, entity_type, entity_id, actor, reason, json_dumps(payload or {}), utc_now(), ), ) def health(self) -> dict[str, Any]: try: with self.connect() as db: db.execute("SELECT 1").fetchone() agent_count = db.execute("SELECT COUNT(*) FROM agents").fetchone()[0] task_count = db.execute("SELECT COUNT(*) FROM tasks").fetchone()[0] return { "ok": True, "agent_count": agent_count, "task_count": task_count, "path": str(self.path), } except sqlite3.Error as exc: return {"ok": False, "error": str(exc), "path": str(self.path)} def register_agent(self, payload: dict[str, Any]) -> dict[str, Any]: public_id = str(payload["public_id"]) now = utc_now() token = secrets.token_urlsafe(32) metadata = payload.get("metadata") or {} with self.connect() as db: existing = db.execute( "SELECT id, machine_fingerprint, registered_at FROM agents WHERE public_id = ?", (public_id,), ).fetchone() if existing is not None: incoming_fingerprint = str(payload.get("machine_fingerprint") or "unknown") existing_fingerprint = str(existing["machine_fingerprint"] or "unknown") if existing_fingerprint != incoming_fingerprint: raise AlreadyRegisteredError( "Agent public_id is already registered by another machine fingerprint." ) agent_id = int(existing["id"]) db.execute( """ UPDATE agents SET display_name = ?, deployment_type = ?, machine_fingerprint = ?, token_hash = ?, agent_version = ?, status = 'registered', reported_host = ?, reported_ip = ?, reported_log_web_port = ?, metadata_json = ? WHERE id = ? """, ( str(payload.get("display_name") or public_id), str(payload.get("deployment_type") or "unknown"), incoming_fingerprint, hash_token(token), str(payload.get("agent_version") or "unknown"), str(payload.get("reported_host") or ""), str(payload.get("reported_ip") or ""), payload.get("reported_log_web_port"), json_dumps(metadata), agent_id, ), ) self._audit(db, "token_rotated", "agent", str(agent_id), {"public_id": public_id}) return { "agent_id": agent_id, "public_id": public_id, "token": token, "registered": False, "token_rotated": True, "registered_at": str(existing["registered_at"]), } cursor = db.execute( """ INSERT INTO agents ( public_id, display_name, deployment_type, machine_fingerprint, token_hash, agent_version, registered_at, status, reported_host, reported_ip, reported_log_web_port, metadata_json ) VALUES (?, ?, ?, ?, ?, ?, ?, 'registered', ?, ?, ?, ?) """, ( public_id, str(payload.get("display_name") or public_id), str(payload.get("deployment_type") or "unknown"), str(payload.get("machine_fingerprint") or "unknown"), hash_token(token), str(payload.get("agent_version") or "unknown"), now, str(payload.get("reported_host") or ""), str(payload.get("reported_ip") or ""), payload.get("reported_log_web_port"), json_dumps(metadata), ), ) agent_id = cursor.lastrowid self._audit(db, "registered", "agent", str(agent_id), {"public_id": public_id}) return { "agent_id": agent_id, "public_id": public_id, "token": token, "registered": True, "registered_at": now, } def validate_agent(self, agent_id: int, token: str) -> dict[str, Any] | None: with self.connect() as db: row = db.execute( "SELECT * FROM agents WHERE id = ? AND token_hash = ?", (agent_id, hash_token(token)), ).fetchone() return row_to_dict(row) def record_heartbeat(self, agent_id: int, payload: dict[str, Any]) -> dict[str, Any]: now = utc_now() deployment_type = str(payload.get("deployment_type") or "unknown") display_name = str(payload.get("display_name") or "") metadata = payload.get("metadata") or {} status = str(payload.get("status") or "online") reported_revision = as_int( payload.get("config_revision", payload.get("model_config_version")), 0, ) model_setting = self.get_model_config(public=False) model_revision = as_int(model_setting.get("version"), 1) update_available = reported_revision < model_revision config_applied_at = now if reported_revision >= model_revision else None config_restart_requested_at = now if update_available else None with self.connect() as db: db.execute( """ UPDATE agents SET last_seen_at = ?, status = ?, deployment_type = ?, display_name = COALESCE(NULLIF(?, ''), display_name), reported_host = COALESCE(NULLIF(?, ''), reported_host), reported_ip = COALESCE(NULLIF(?, ''), reported_ip), reported_log_web_port = ?, metadata_json = ?, config_json = ?, config_revision = ?, config_applied_at = COALESCE(?, config_applied_at), config_restart_requested_at = COALESCE(?, config_restart_requested_at) WHERE id = ? """, ( now, status, deployment_type, display_name, str(payload.get("reported_host") or ""), str(payload.get("reported_ip") or ""), payload.get("reported_log_web_port"), json_dumps(metadata), json_dumps( { "model_config_version": reported_revision, "runtime_generation": payload.get("runtime_generation"), } ), reported_revision, config_applied_at, config_restart_requested_at, agent_id, ), ) db.execute( """ INSERT INTO heartbeats (agent_id, received_at, payload_json) VALUES (?, ?, ?) """, (agent_id, now, json_dumps(payload)), ) receipt: dict[str, Any] = { "ok": True, "agent_id": agent_id, "received_at": now, "model_config_revision": model_revision, "agent_config_revision": reported_revision, "config_update_available": update_available, } if update_available: receipt["model_config"] = { "version": model_revision, "updated_at": model_setting.get("updated_at"), "requires_agent_restart": True, "value": self._public_model_config_value(model_setting.get("value") or DEFAULT_MODEL_CONFIG), } return receipt def _agent_status_with_stale_check(self, status: str, last_seen_at: str | None) -> str: if not last_seen_at: return status try: last_seen = datetime.fromisoformat(last_seen_at) except ValueError: return status if last_seen.tzinfo is None: last_seen = last_seen.replace(tzinfo=timezone.utc) elapsed = (datetime.now(timezone.utc) - last_seen).total_seconds() if elapsed > self.agent_stale_after_seconds: return "offline" return status def list_agents(self) -> list[dict[str, Any]]: with self.connect() as db: rows = db.execute( """ SELECT id, public_id, display_name, deployment_type, machine_fingerprint, agent_version, registered_at, last_seen_at, status, reported_host, reported_ip, reported_log_web_port, config_revision, config_applied_at, config_restart_requested_at, config_json, metadata_json FROM agents ORDER BY last_seen_at DESC NULLS LAST, registered_at DESC """ ).fetchall() agents: list[dict[str, Any]] = [] for row in rows: item = row_to_dict(row) or {} item["config"] = json_loads(item.pop("config_json", "{}"), {}) item["metadata"] = json_loads(item.pop("metadata_json", "{}"), {}) item["raw_status"] = item.get("status") item["status"] = self._agent_status_with_stale_check( str(item.get("status") or "unknown"), item.get("last_seen_at"), ) agents.append(item) return agents def get_setting(self, key: str, default: Any) -> dict[str, Any]: with self.connect() as db: row = db.execute( "SELECT value_json, version, updated_at FROM settings WHERE key = ?", (key,), ).fetchone() if row is None: return {"value": default, "version": 1, "updated_at": None} return { "value": json_loads(row["value_json"], default), "version": row["version"], "updated_at": row["updated_at"], } def set_setting(self, key: str, value: Any, actor: str = "web") -> dict[str, Any]: now = utc_now() with self.connect() as db: existing = db.execute( "SELECT version FROM settings WHERE key = ?", (key,), ).fetchone() version = int(existing["version"]) + 1 if existing else 1 db.execute( """ INSERT INTO settings (key, value_json, version, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value_json = excluded.value_json, version = excluded.version, updated_at = excluded.updated_at """, (key, json_dumps(value), version, now), ) self._audit(db, "updated", "setting", key, {"version": version}, actor=actor) return {"value": value, "version": version, "updated_at": now} def _public_model_config_value(self, value: dict[str, Any]) -> dict[str, Any]: public_value = {**DEFAULT_MODEL_CONFIG, **value} public_value["api_key_set"] = bool(public_value.get("api_key")) public_value["api_key"] = "" return public_value def _normalize_model_config( self, payload: dict[str, Any], current: dict[str, Any] | None = None, ) -> dict[str, Any]: current_value = {**DEFAULT_MODEL_CONFIG, **(current or {})} merged = {**current_value} ignored = {"api_key_set", "version", "updated_at"} for key, value in payload.items(): if key in ignored or key == "api_key": continue if key in DEFAULT_MODEL_CONFIG: merged[key] = value raw_api_key = str(payload.get("api_key") or "") if payload.get("clear_api_key"): merged["api_key"] = "" elif raw_api_key: merged["api_key"] = raw_api_key else: merged["api_key"] = str(current_value.get("api_key") or "") text_fields = [ "target_system_url", "vllm_base_url", "vllm_model", "grounding_url", "grounding_model", "ground_coordinate_mode", "resolution", "api_key", ] for field in text_fields: merged[field] = str(merged.get(field) or "").strip() for field in ["target_system_url", "vllm_base_url", "grounding_url"]: value = merged.get(field) if value and not str(value).startswith(("http://", "https://")): raise ValueError(f"{field} must start with http:// or https://") if not re.fullmatch(r"\d{3,5}x\d{3,5}", str(merged.get("resolution") or "")): raise ValueError("resolution must use WIDTHxHEIGHT format, for example 1920x1080") numeric_ranges = { "screenshot_max_width": (1, 7680), "screenshot_max_height": (1, 7680), "max_steps": (1, 500), "request_timeout_seconds": (1, 300), } for field, (minimum, maximum) in numeric_ranges.items(): value = as_int(merged.get(field), as_int(DEFAULT_MODEL_CONFIG[field])) if not minimum <= value <= maximum: raise ValueError(f"{field} must be between {minimum} and {maximum}") merged[field] = value temperature = as_float(merged.get("temperature"), as_float(DEFAULT_MODEL_CONFIG["temperature"])) if not 0 <= temperature <= 2: raise ValueError("temperature must be between 0 and 2") merged["temperature"] = temperature if merged["ground_coordinate_mode"] not in {"absolute", "normalized"}: raise ValueError("ground_coordinate_mode must be absolute or normalized") return merged def get_model_config(self, public: bool = True) -> dict[str, Any]: setting = self.get_setting("model_config", DEFAULT_MODEL_CONFIG) value = {**DEFAULT_MODEL_CONFIG, **(setting.get("value") or {})} if public: value = self._public_model_config_value(value) return { "value": value, "version": setting["version"], "updated_at": setting["updated_at"], } def save_model_config(self, payload: dict[str, Any]) -> dict[str, Any]: current = self.get_model_config(public=False)["value"] merged = self._normalize_model_config(payload, current) saved = self.set_setting("model_config", merged, actor="web") return { "value": self._public_model_config_value(saved["value"]), "version": saved["version"], "updated_at": saved["updated_at"], } def get_tool_config(self) -> dict[str, Any]: value = self.get_setting( "tool_config", {"tools": DEFAULT_TOOLS, "config": DEFAULT_TOOL_CONFIG}, ) tool_payload = value["value"] tools = tool_payload.get("tools") or DEFAULT_TOOLS config = {**DEFAULT_TOOL_CONFIG, **(tool_payload.get("config") or {})} return { "tools": tools, "config": config, "version": value["version"], "updated_at": value["updated_at"], } def save_tool_config(self, payload: dict[str, Any]) -> dict[str, Any]: current = self.get_tool_config() tools = payload.get("tools") or current["tools"] config = {**DEFAULT_TOOL_CONFIG, **current["config"], **(payload.get("config") or {})} saved = self.set_setting( "tool_config", {"tools": tools, "config": config}, actor="web", ) return { "tools": tools, "config": config, "version": saved["version"], "updated_at": saved["updated_at"], } def _dedupe_external_id(self, db: sqlite3.Connection, external_id: str | None) -> str | None: if not external_id: return None candidate = external_id suffix = 2 while db.execute("SELECT 1 FROM tasks WHERE external_id = ?", (candidate,)).fetchone(): candidate = f"{external_id}-{suffix}" suffix += 1 return candidate def _normalize_task_payload( self, payload: dict[str, Any], defaults: dict[str, Any] | None = None, ) -> dict[str, Any]: defaults = defaults or {} task_payload = payload.get("payload") if isinstance(payload.get("payload"), dict) else payload external_id = ( payload.get("external_id") or payload.get("task_id") or payload.get("id") or task_payload.get("task_id") or task_payload.get("id") ) instruction = task_payload.get("instruction") or payload.get("instruction") or "" title = ( payload.get("title") or task_payload.get("title") or task_payload.get("name") or (str(instruction)[:64] if instruction else external_id) or "Untitled task" ) return { "external_id": str(external_id) if external_id else None, "title": str(title), "plan_name": str(payload.get("plan_name") or task_payload.get("plan_name") or defaults.get("plan_name") or "Default Plan"), "plan_version": as_int(payload.get("plan_version") or defaults.get("plan_version"), 1), "module": str(payload.get("module") or task_payload.get("module") or task_payload.get("category") or ""), "parent_external_id": ( payload.get("parent_external_id") or payload.get("parent_id") or task_payload.get("parent_external_id") or task_payload.get("parent_id") ), "root_external_id": ( payload.get("root_external_id") or payload.get("root_id") or task_payload.get("root_external_id") or task_payload.get("root_id") ), "depth": as_int(payload.get("depth") or task_payload.get("depth"), 0), "task_level": str(payload.get("task_level") or payload.get("level") or task_payload.get("task_level") or task_payload.get("level") or ""), "risk_level": str(payload.get("risk_level") or task_payload.get("risk_level") or "low"), "status": str(payload.get("status") or defaults.get("status") or "ready"), "priority": as_int( payload.get("priority") or task_payload.get("priority") or task_payload.get("task_priority") or defaults.get("priority"), 100, ), "assigned_agent_id": payload.get("assigned_agent_id") or defaults.get("assigned_agent_id"), "payload": task_payload, "failure_policy": str( payload.get("failure_policy") or task_payload.get("failure_policy") or task_payload.get("task_failure_policy") or defaults.get("failure_policy") or "skip" ), "max_retries": as_int(payload.get("max_retries") or task_payload.get("retries") or defaults.get("max_retries"), 0), "timeout_seconds": payload.get("timeout_seconds") or task_payload.get("timeout") or defaults.get("timeout_seconds"), "dispatch_mode": str(payload.get("dispatch_mode") or defaults.get("dispatch_mode") or "auto"), "dependencies": payload.get("dependencies") or task_payload.get("depends_on") or [], "tags": payload.get("tags") or task_payload.get("tags") or [], } def create_task(self, payload: dict[str, Any]) -> dict[str, Any]: task = self._normalize_task_payload(payload) now = utc_now() with self.connect() as db: task["external_id"] = self._dedupe_external_id(db, task["external_id"]) if not task["root_external_id"]: task["root_external_id"] = task["external_id"] cursor = db.execute( """ INSERT INTO tasks ( external_id, title, plan_name, plan_version, module, parent_external_id, root_external_id, depth, task_level, risk_level, status, priority, assigned_agent_id, payload_json, failure_policy, max_retries, timeout_seconds, dispatch_mode, dependencies_json, tags_json, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( task["external_id"], task["title"], task["plan_name"], task["plan_version"], task["module"], task["parent_external_id"], task["root_external_id"], task["depth"], task["task_level"], task["risk_level"], task["status"], task["priority"], task["assigned_agent_id"], json_dumps(task["payload"]), task["failure_policy"], task["max_retries"], task["timeout_seconds"], task["dispatch_mode"], json_dumps(task["dependencies"]), json_dumps(task["tags"]), now, now, ), ) task_id = cursor.lastrowid self._audit(db, "created", "task", str(task_id), {"external_id": task["external_id"]}, actor="web") return self.get_task(int(task_id)) or {} def _row_to_task(self, row: sqlite3.Row) -> dict[str, Any]: task = row_to_dict(row) or {} task["payload"] = json_loads(task.pop("payload_json", "{}"), {}) task["dependencies"] = json_loads(task.pop("dependencies_json", "[]"), []) task["tags"] = json_loads(task.pop("tags_json", "[]"), []) task["result"] = json_loads(task.pop("result_json", None), None) return task def get_task(self, task_id: int) -> dict[str, Any] | None: with self.connect() as db: row = db.execute( """ SELECT t.*, a.display_name AS assigned_agent_name FROM tasks t LEFT JOIN agents a ON a.id = t.assigned_agent_id WHERE t.id = ? """, (task_id,), ).fetchone() return self._row_to_task(row) if row else None def list_tasks(self, status: str | None = None, plan_name: str | None = None) -> list[dict[str, Any]]: where: list[str] = [] params: list[Any] = [] if status: where.append("t.status = ?") params.append(status) if plan_name: where.append("t.plan_name = ?") params.append(plan_name) where_sql = "WHERE " + " AND ".join(where) if where else "" with self.connect() as db: rows = db.execute( f""" SELECT t.*, a.display_name AS assigned_agent_name FROM tasks t LEFT JOIN agents a ON a.id = t.assigned_agent_id {where_sql} ORDER BY t.priority ASC, t.created_at DESC """, params, ).fetchall() return [self._row_to_task(row) for row in rows] def import_tasks(self, payload: dict[str, Any]) -> dict[str, Any]: raw = payload.get("content", payload.get("tasks", payload)) if isinstance(raw, str): raw = json.loads(raw) plan_name = payload.get("plan_name") plan_version = as_int(payload.get("plan_version"), 1) entries: list[dict[str, Any]] = [] if isinstance(raw, list): for item in raw: if isinstance(item, dict) and isinstance(item.get("tasks"), list): name = item.get("plan_name") or item.get("name") or plan_name for child in item["tasks"]: if isinstance(child, dict): entries.append({**child, "plan_name": child.get("plan_name") or name}) elif isinstance(item, dict): entries.append(item) elif isinstance(raw, dict): plan_name = raw.get("plan_name") or raw.get("name") or plan_name if isinstance(raw.get("plans"), list): for plan in raw["plans"]: if not isinstance(plan, dict): continue name = plan.get("plan_name") or plan.get("name") or plan_name for child in plan.get("tasks") or []: if isinstance(child, dict): entries.append({**child, "plan_name": child.get("plan_name") or name}) elif isinstance(raw.get("tasks"), list): for child in raw["tasks"]: if isinstance(child, dict): entries.append({**child, "plan_name": child.get("plan_name") or plan_name}) elif isinstance(raw.get("tasks"), dict): for key, child in raw["tasks"].items(): if isinstance(child, dict): entries.append({**child, "task_id": child.get("task_id") or key, "plan_name": child.get("plan_name") or plan_name}) else: entries.append(raw) else: raise ValueError("Unsupported task JSON shape") defaults = { "plan_name": plan_name or "Imported Plan", "plan_version": plan_version, "priority": payload.get("priority") or 100, } created = [self.create_task({**entry, **{"plan_name": entry.get("plan_name") or defaults["plan_name"]}, "plan_version": plan_version}) for entry in entries] return {"count": len(created), "tasks": created} def claim_task(self, agent_id: int) -> dict[str, Any] | None: now = utc_now() with self.connect() as db: row = db.execute( """ SELECT * FROM tasks WHERE status IN ('ready', 'routed') AND (assigned_agent_id IS NULL OR assigned_agent_id = ?) ORDER BY priority ASC, created_at ASC LIMIT 1 """, (agent_id,), ).fetchone() if row is None: return None db.execute( """ UPDATE tasks SET status = 'running', assigned_agent_id = ?, claimed_at = ?, started_at = COALESCE(started_at, ?), updated_at = ? WHERE id = ? """, (agent_id, now, now, now, row["id"]), ) self._audit(db, "claimed", "task", str(row["id"]), {"agent_id": agent_id}) return self.get_task(int(row["id"])) def assign_task(self, task_id: int, agent_id: int | None, reason: str | None = None) -> dict[str, Any]: now = utc_now() with self.connect() as db: db.execute( """ UPDATE tasks SET assigned_agent_id = ?, dispatch_mode = CASE WHEN ? IS NULL THEN 'auto' ELSE 'manual' END, status = CASE WHEN ? IS NULL THEN 'ready' ELSE 'routed' END, updated_at = ? WHERE id = ? """, (agent_id, agent_id, agent_id, now, task_id), ) self._audit(db, "assigned", "task", str(task_id), {"agent_id": agent_id}, reason=reason, actor="web") return self.get_task(task_id) or {} def update_task_status( self, task_id: int, status: str, reason: str | None = None, payload: dict[str, Any] | None = None, ) -> dict[str, Any]: now = utc_now() finish_statuses = {"succeeded", "failed", "cancelled"} with self.connect() as db: db.execute( """ UPDATE tasks SET status = ?, last_error = COALESCE(?, last_error), result_json = COALESCE(?, result_json), finished_at = CASE WHEN ? THEN COALESCE(finished_at, ?) ELSE finished_at END, updated_at = ? WHERE id = ? """, ( status, reason, json_dumps(payload) if payload is not None else None, 1 if status in finish_statuses else 0, now, now, task_id, ), ) self._audit(db, "status_changed", "task", str(task_id), {"status": status}, reason=reason, actor="web") return self.get_task(task_id) or {} def retry_task(self, task_id: int, reason: str | None = None) -> dict[str, Any]: now = utc_now() with self.connect() as db: db.execute( """ UPDATE tasks SET status = 'ready', attempt = attempt + 1, assigned_agent_id = NULL, claimed_at = NULL, started_at = NULL, finished_at = NULL, updated_at = ? WHERE id = ? """, (now, task_id), ) self._audit(db, "retried", "task", str(task_id), {}, reason=reason, actor="web") return self.get_task(task_id) or {} def record_task_result( self, task_id: int, result: dict[str, Any], status: str | None = None, ) -> dict[str, Any]: now = utc_now() final_status = status or ("succeeded" if result.get("success") else "failed") with self.connect() as db: db.execute( """ UPDATE tasks SET status = ?, result_json = ?, last_error = ?, duration_seconds = ?, model_duration_seconds = ?, action_duration_seconds = ?, finished_at = ?, updated_at = ? WHERE id = ? """, ( final_status, json_dumps(result), result.get("message") if not result.get("success") else None, result.get("duration_seconds") or result.get("duration"), result.get("model_duration_seconds"), result.get("action_duration_seconds"), now, now, task_id, ), ) self._audit(db, "result_reported", "task", str(task_id), {"status": final_status}) return self.get_task(task_id) or {} def dashboard(self) -> dict[str, Any]: agents = self.list_agents() tasks = self.list_tasks() metrics = self.metrics() return { "agents": agents, "tasks": tasks[:50], "metrics": metrics, "model_config": self.get_model_config(), "tool_config": self.get_tool_config(), } def metrics(self) -> dict[str, Any]: agents = self.list_agents() tasks = self.list_tasks() total = len(tasks) status_counts: dict[str, int] = {} for task in tasks: status_counts[task["status"]] = status_counts.get(task["status"], 0) + 1 succeeded = status_counts.get("succeeded", 0) failed = status_counts.get("failed", 0) cancelled = status_counts.get("cancelled", 0) completed = succeeded + failed + cancelled running = status_counts.get("running", 0) ready = status_counts.get("ready", 0) + status_counts.get("routed", 0) active_agents = len([a for a in agents if a.get("status") in {"online", "idle", "running"}]) durations = [as_float(t.get("duration_seconds")) for t in tasks if t.get("duration_seconds")] avg_duration = round(sum(durations) / len(durations), 2) if durations else 0 interventions = len([t for t in tasks if t.get("dispatch_mode") == "manual" or t.get("operator_note")]) retried = len([t for t in tasks if as_int(t.get("attempt")) > 0]) plan_names = sorted({t.get("plan_name") or "Default Plan" for t in tasks}) return { "total_tasks": total, "ready_tasks": ready, "running_tasks": running, "completed_tasks": completed, "succeeded_tasks": succeeded, "failed_tasks": failed, "cancelled_tasks": cancelled, "agent_total": len(agents), "agent_active": active_agents, "pass_rate": round((succeeded / (succeeded + failed)) * 100, 1) if succeeded + failed else 0, "fail_rate": round((failed / (succeeded + failed)) * 100, 1) if succeeded + failed else 0, "coverage_rate": round((completed / total) * 100, 1) if total else 0, "avg_duration_seconds": avg_duration, "concurrency_utilization": round((running / active_agents) * 100, 1) if active_agents else 0, "manual_intervention_rate": round((interventions / total) * 100, 1) if total else 0, "recovery_rate": round((retried / max(failed + retried, 1)) * 100, 1) if (failed or retried) else 0, "tool_consistency_rate": 100.0 if self.get_tool_config()["tools"] else 0, "status_counts": status_counts, "plans": plan_names, } def report_summary(self) -> dict[str, Any]: tasks = self.list_tasks() metrics = self.metrics() failed = [t for t in tasks if t.get("status") == "failed"] plans: dict[str, dict[str, Any]] = {} for task in tasks: name = task.get("plan_name") or "Default Plan" item = plans.setdefault( name, {"plan_name": name, "total": 0, "succeeded": 0, "failed": 0, "cancelled": 0}, ) item["total"] += 1 if task["status"] in item: item[task["status"]] += 1 return { "generated_at": utc_now(), "metrics": metrics, "plans": list(plans.values()), "failed_tasks": [ { "id": task["id"], "external_id": task.get("external_id"), "title": task.get("title"), "plan_name": task.get("plan_name"), "reason": task.get("last_error"), "evidence": (task.get("result") or {}).get("screenshots") or (task.get("result") or {}).get("logs"), } for task in failed[:50] ], }