initial: hermes-dashboard snapshot (post overview + git integration)

services/kanban.py

@@ -0,0 +1,139 @@
+"""Kanban database service — read-only queries against kanban.db."""
+
+import logging
+import sqlite3
+from typing import Any
+
+from extensions import get_db
+
+logger = logging.getLogger(__name__)
+
+STATUSES = ("ready", "running", "done", "blocked", "failed")
+
+
+def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
+    """Convert sqlite3.Row to plain dict."""
+    return dict(row)
+
+
+def get_tasks(status: str | None = None) -> dict[str, Any]:
+    """Return tasks list and status counts."""
+    try:
+        db = get_db()
+
+        # Get counts per status
+        counts: dict[str, int] = {}
+        for s in STATUSES:
+            row = db.execute(
+                "SELECT COUNT(*) as cnt FROM tasks WHERE status = ?", (s,)
+            ).fetchone()
+            counts[s] = row["cnt"] if row else 0
+
+        # Get filtered or all tasks
+        if status and status in STATUSES:
+            rows = db.execute(
+                "SELECT * FROM tasks WHERE status = ? ORDER BY created_at DESC",
+                (status,),
+            ).fetchall()
+        else:
+            rows = db.execute(
+                "SELECT * FROM tasks ORDER BY created_at DESC"
+            ).fetchall()
+
+        tasks = [_row_to_dict(r) for r in rows]
+        return {"tasks": tasks, "counts": counts}
+    except sqlite3.Error as exc:
+        logger.exception("get_tasks failed")
+        return {"tasks": [], "counts": {s: 0 for s in STATUSES}, "error": str(exc)}
+
+
+def get_recent_tasks(limit: int = 5) -> list[dict]:
+    """Return the most recent tasks with outcome summary."""
+    try:
+        db = get_db()
+        rows = db.execute(
+            "SELECT id, title, status, created_at, result FROM tasks "
+            "ORDER BY created_at DESC LIMIT ?",
+            (limit,),
+        ).fetchall()
+        return [
+            {
+                "id": r["id"],
+                "title": r["title"],
+                "status": r["status"],
+                "created_at": r["created_at"],
+                "outcome_summary": (r["result"] or "")[:200],
+            }
+            for r in rows
+        ]
+    except sqlite3.Error as exc:
+        logger.exception("get_recent_tasks failed")
+        return []
+
+
+def get_task_detail(task_id: str) -> dict[str, Any]:
+    """Return full task detail with events, comments, runs, attachments."""
+    try:
+        db = get_db()
+
+        task_row = db.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
+        if not task_row:
+            return {"error": "task not found"}
+
+        task = _row_to_dict(task_row)
+
+        events_rows = db.execute(
+            "SELECT * FROM task_events WHERE task_id = ? ORDER BY created_at",
+            (task_id,),
+        ).fetchall()
+        events = [_row_to_dict(r) for r in events_rows]
+
+        comments_rows = db.execute(
+            "SELECT * FROM task_comments WHERE task_id = ? ORDER BY created_at",
+            (task_id,),
+        ).fetchall()
+        comments = [_row_to_dict(r) for r in comments_rows]
+
+        runs_rows = db.execute(
+            "SELECT * FROM task_runs WHERE task_id = ? ORDER BY started_at DESC",
+            (task_id,),
+        ).fetchall()
+        runs = [_row_to_dict(r) for r in runs_rows]
+
+        attachments_rows = db.execute(
+            "SELECT * FROM task_attachments WHERE task_id = ? ORDER BY created_at",
+            (task_id,),
+        ).fetchall()
+        attachments = [_row_to_dict(r) for r in attachments_rows]
+
+        return {
+            "task": task,
+            "events": events,
+            "comments": comments,
+            "runs": runs,
+            "attachments": attachments,
+        }
+    except sqlite3.Error as exc:
+        logger.exception("get_task_detail failed")
+        return {"error": str(exc)}
+
+
+def get_kanban_board() -> dict[str, Any]:
+    """Return tasks grouped by status for kanban board view."""
+    try:
+        db = get_db()
+        board: dict[str, list[dict[str, Any]]] = {s: [] for s in STATUSES}
+
+        rows = db.execute(
+            "SELECT * FROM tasks ORDER BY created_at DESC"
+        ).fetchall()
+        for row in rows:
+            d = _row_to_dict(row)
+            status = d.get("status", "ready")
+            if status in board:
+                board[status].append(d)
+
+        return {"board": board}
+    except sqlite3.Error as exc:
+        logger.exception("get_kanban_board failed")
+        return {"board": {s: [] for s in STATUSES}, "error": str(exc)}