From e8b389d6b76f8a9b0b4982a8664d29a8c9e6f379 Mon Sep 17 00:00:00 2001 From: Shawn Date: Sat, 23 May 2026 12:24:23 -0400 Subject: [PATCH] feat: add kanban board monitoring and Matrix notification bridge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - KanbanDB: read-only SQLite access to kanban state (tasks, events, comments, subs) - KanbanMonitor: async poll loop that watches kanban DB, formats events, sends via Matrix - format_event: human-readable event formatting (created, claimed, completed, etc.) - CLI: matrix-hermes kanban monitor/subscribe/unsubscribe/status commands - 29 tests covering DB, event formatting, monitor lifecycle, and notifications - Version bump 0.3.0 → 0.4.0 --- pyproject.toml | 2 +- src/matrix_hermes/__init__.py | 19 +- src/matrix_hermes/cli.py | 202 +++++++++++++- src/matrix_hermes/kanban.py | 512 ++++++++++++++++++++++++++++++++++ tests/test_kanban.py | 478 +++++++++++++++++++++++++++++++ 5 files changed, 1210 insertions(+), 3 deletions(-) create mode 100644 src/matrix_hermes/kanban.py create mode 100644 tests/test_kanban.py diff --git a/pyproject.toml b/pyproject.toml index c8bb625..a83a7da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "matrix-hermes" -version = "0.1.0" +version = "0.4.0" description = "Matrix-Hermes integration: authentication, client, and bridge for Hermes Agent" readme = "README.md" license = {text = "MIT"} diff --git a/src/matrix_hermes/__init__.py b/src/matrix_hermes/__init__.py index 4bd884b..89c0c4b 100644 --- a/src/matrix_hermes/__init__.py +++ b/src/matrix_hermes/__init__.py @@ -11,7 +11,7 @@ Provides: - CLI tools for all operations """ -__version__ = "0.3.0" +__version__ = "0.4.0" # Auth from matrix_hermes.auth import ( @@ -89,6 +89,16 @@ from matrix_hermes.sync import ( get_messages, ) +# Kanban +from matrix_hermes.kanban import ( + KanbanTask, + KanbanEvent, + KanbanSubscription, + KanbanDB, + KanbanMonitor, + format_event, +) + __all__ = [ # Auth "AuthResult", @@ -150,6 +160,13 @@ __all__ = [ "sync_once", "extract_events", "get_messages", + # Kanban + "KanbanTask", + "KanbanEvent", + "KanbanSubscription", + "KanbanDB", + "KanbanMonitor", + "format_event", # Meta "__version__", ] diff --git a/src/matrix_hermes/cli.py b/src/matrix_hermes/cli.py index ad4a1dc..399d1f4 100644 --- a/src/matrix_hermes/cli.py +++ b/src/matrix_hermes/cli.py @@ -6,6 +6,9 @@ Usage: matrix-hermes logout --homeserver URL --token TOKEN matrix-hermes check # Check config from env matrix-hermes show-config # Show current config + matrix-hermes kanban monitor --db PATH # Watch kanban DB, notify via Matrix + matrix-hermes kanban subscribe --db PATH --task ID --room ROOM + matrix-hermes kanban status --db PATH """ from __future__ import annotations @@ -14,7 +17,9 @@ import argparse import asyncio import os import sys +import time from pathlib import Path +from typing import Any from matrix_hermes.auth import login, logout, validate_token, AuthResult from matrix_hermes.config import MatrixConfig @@ -176,6 +181,34 @@ def _mask_token(token: str, visible: int = 6) -> str: return token[:visible] + "..." + token[-visible:] +def _add_kanban_parser(subparsers: Any) -> None: + """Add the 'kanban' subcommand tree.""" + p_kanban = subparsers.add_parser("kanban", help="Monitor kanban boards via Matrix") + kanban_sub = p_kanban.add_subparsers(dest="kanban_action", help="Kanban actions") + + # monitor + p_monitor = kanban_sub.add_parser("monitor", help="Watch kanban DB and notify via Matrix") + p_monitor.add_argument("--db", required=True, help="Path to kanban.db") + p_monitor.add_argument("--interval", type=float, default=5.0, help="Poll interval in seconds") + + # subscribe + p_sub = kanban_sub.add_parser("subscribe", help="Subscribe a room to task notifications") + p_sub.add_argument("--db", required=True, help="Path to kanban.db") + p_sub.add_argument("--task", required=True, help="Task ID or '*' for all tasks") + p_sub.add_argument("--room", required=True, help="Matrix room ID for notifications") + p_sub.add_argument("--thread", help="Optional thread root event ID") + + # unsubscribe + p_unsub = kanban_sub.add_parser("unsubscribe", help="Remove a room subscription") + p_unsub.add_argument("--db", required=True, help="Path to kanban.db") + p_unsub.add_argument("--task", required=True, help="Task ID") + p_unsub.add_argument("--room", required=True, help="Matrix room ID") + + # status + p_status = kanban_sub.add_parser("status", help="Show Matrix subscriptions for a kanban board") + p_status.add_argument("--db", required=True, help="Path to kanban.db") + + def build_parser() -> argparse.ArgumentParser: """Build the CLI argument parser.""" parser = argparse.ArgumentParser( @@ -267,10 +300,172 @@ def build_parser() -> argparse.ArgumentParser: p_sync_listen.add_argument("--room", nargs="*", default=[], help="Only listen in these rooms") p_sync_listen.add_argument("--show-all", action="store_true", help="Show all event types, not just messages") + # ── Kanban subcommands ────────────────────────────────────────────── + _add_kanban_parser(sub) + return parser -# ── Async CLI handlers ────────────────────────────────────────────────────── +# ── Kanban subcommands ────────────────────────────────────────────────── + +async def _kanban_monitor(client, args: argparse.Namespace) -> int: + """Run the kanban monitor — poll DB and notify via Matrix.""" + from matrix_hermes.kanban import KanbanDB, KanbanMonitor + + db_path = args.db + if not Path(db_path).exists(): + print(f"Error: kanban database not found: {db_path}", file=sys.stderr) + return 1 + + db = KanbanDB(db_path) + monitor = KanbanMonitor( + client, + db, + poll_interval=args.interval, + ) + + # Register a console handler for visibility + def _console_handler(event, task): + from matrix_hermes.kanban import format_event + print(format_event(event, task)) + + monitor.on_event(_console_handler) + + print(f"Monitoring kanban DB: {db_path}") + print(f"Poll interval: {args.interval}s") + print(f"Matrix user: {client.user_id}") + print("Press Ctrl+C to stop...") + + await monitor.start() + try: + await asyncio.sleep(2**30) # effectively forever + except asyncio.CancelledError: + pass + finally: + await monitor.stop() + return 0 + + +async def _kanban_subscribe(client, args: argparse.Namespace) -> int: + """Subscribe a Matrix room to kanban task notifications.""" + import sqlite3 + + db_path = args.db + if not Path(db_path).exists(): + print(f"Error: kanban database not found: {db_path}", file=sys.stderr) + return 1 + + task_id = args.task + room_id = args.room + thread_id = args.thread or "" + + # Verify the room exists (we're joined) + rooms = await client.get_joined_rooms() + if room_id not in rooms: + print(f"Warning: Not joined to room {room_id}. Attempting to join...") + ok = await client.join_room(room_id) + if not ok: + print(f"Error: Could not join room {room_id}", file=sys.stderr) + return 1 + + # Verify the task exists + db = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + task_row = db.execute("SELECT id, title FROM tasks WHERE id = ?", (task_id,)).fetchone() + db.close() + + if task_row is None and task_id != "*": + print(f"Warning: Task {task_id} not found in database (use '*' for all tasks)") + + # Insert subscription + db = sqlite3.connect(db_path) + now = int(time.time()) + try: + db.execute( + "INSERT OR REPLACE INTO kanban_notify_subs " + "(task_id, platform, chat_id, thread_id, user_id, notifier_profile, created_at) " + "VALUES (?, 'matrix', ?, ?, ?, 'kanban-monitor', ?)", + (task_id, room_id, thread_id, client.user_id, now), + ) + db.commit() + task_label = task_row[1] if task_row else task_id + print(f"Subscribed to {task_label} ({task_id})") + print(f"Notifications will be sent to {room_id}") + if thread_id: + print(f" Thread: {thread_id}") + except sqlite3.Error as exc: + print(f"Error subscribing: {exc}", file=sys.stderr) + return 1 + finally: + db.close() + return 0 + + +async def _kanban_unsubscribe(client, args: argparse.Namespace) -> int: + """Unsubscribe a Matrix room from kanban task notifications.""" + import sqlite3 + + db_path = args.db + db = sqlite3.connect(db_path) + + try: + cursor = db.execute( + "DELETE FROM kanban_notify_subs " + "WHERE task_id = ? AND platform = 'matrix' AND chat_id = ?", + (args.task, args.room), + ) + db.commit() + if cursor.rowcount > 0: + print(f"Unsubscribed from {args.task} for room {args.room}") + else: + print(f"No subscription found for task={args.task} room={args.room}") + except sqlite3.Error as exc: + print(f"Error unsubscribing: {exc}", file=sys.stderr) + return 1 + finally: + db.close() + return 0 + + +async def _kanban_status(client, args: argparse.Namespace) -> int: + """Show current kanban monitoring subscriptions.""" + from matrix_hermes.kanban import KanbanDB + + db_path = args.db + if not Path(db_path).exists(): + print(f"Error: kanban database not found: {db_path}", file=sys.stderr) + return 1 + + db = KanbanDB(db_path) + subs = db.get_matrix_subscriptions() + + if not subs: + print("No Matrix subscriptions found.") + db.close() + return 0 + + print(f"Matrix subscriptions ({len(subs)}):") + for sub in subs: + task = db.get_task(sub.task_id) + task_label = task.title if task else sub.task_id + print(f" Task: {task_label} ({sub.task_id})") + print(f" Room: {sub.chat_id}") + if sub.thread_id: + print(f" Thread: {sub.thread_id}") + print(f" Last event: {sub.last_event_id}") + print() + db.close() + return 0 + + +_ASYNC_KANBAN_ACTIONS = { + "monitor": _kanban_monitor, + "subscribe": _kanban_subscribe, + "unsubscribe": _kanban_unsubscribe, + "status": _kanban_status, +} + + +# ── Parser building ───────────────────────────────────────────────────────── def _build_client(args: argparse.Namespace) -> "MatrixClient": @@ -523,6 +718,10 @@ async def _async_command(args: argparse.Namespace) -> int: handler = _ASYNC_SYNC_ACTIONS.get(args.sync_action) if handler: return await handler(client, args) + elif args.command == "kanban": + handler = _ASYNC_KANBAN_ACTIONS.get(args.kanban_action) + if handler: + return await handler(client, args) finally: await client.disconnect() return 0 @@ -542,6 +741,7 @@ _COMMAND_MAP = { "rooms": _run_async, "send": _run_async, "sync": _run_async, + "kanban": _run_async, } diff --git a/src/matrix_hermes/kanban.py b/src/matrix_hermes/kanban.py new file mode 100644 index 0000000..79b8854 --- /dev/null +++ b/src/matrix_hermes/kanban.py @@ -0,0 +1,512 @@ +"""Kanban board monitoring and Matrix notification bridge. + +Watches a Hermes Kanban SQLite database for task events (status changes, +comments, assignments, etc.) and delivers formatted notifications to +subscribed Matrix rooms. + +Architecture: + KanbanDB — read-only SQLite access to kanban state + KanbanMonitor — async poll loop; reads events → formats → sends via Matrix +""" + +from __future__ import annotations + +import asyncio +import logging +import sqlite3 +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +from matrix_hermes.client import MatrixClient + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- + + +@dataclass +class KanbanTask: + """A kanban task snapshot.""" + + id: str + title: str + assignee: str = "" + status: str = "todo" + priority: int = 0 + created_by: str = "" + + +@dataclass +class KanbanEvent: + """A single kanban event (status change, assignment, comment, etc.).""" + + id: int + task_id: str + kind: str # created, claimed, spawned, completed, blocked, commented, etc. + payload: dict[str, Any] = field(default_factory=dict) + created_at: float = 0.0 + + @property + def timestamp_iso(self) -> str: + """ISO-format timestamp string.""" + import datetime + return datetime.datetime.fromtimestamp( + self.created_at, tz=datetime.timezone.utc + ).isoformat() + + +@dataclass +class KanbanSubscription: + """A subscription entry from kanban_notify_subs.""" + + task_id: str + platform: str + chat_id: str + thread_id: str = "" + user_id: str = "" + last_event_id: int = 0 + + +# --------------------------------------------------------------------------- +# KanbanDB — read-only SQLite access +# --------------------------------------------------------------------------- + + +class KanbanDB: + """Read-only access to a Hermes Kanban SQLite database. + + Args: + db_path: Path to the kanban.db SQLite file. + """ + + def __init__(self, db_path: str | Path) -> None: + self._db_path = str(db_path) + self._conn: Optional[sqlite3.Connection] = None + + # -- Connection ---------------------------------------------------------- + + def connect(self) -> None: + """Open (or reopen) the read-only database connection.""" + if self._conn: + try: + self._conn.close() + except sqlite3.Error: + pass + uri = f"file:{self._db_path}?mode=ro" + self._conn = sqlite3.connect(uri, uri=True) + self._conn.row_factory = sqlite3.Row + + def close(self) -> None: + """Close the database connection.""" + if self._conn: + self._conn.close() + self._conn = None + + @property + def conn(self) -> sqlite3.Connection: + """The active connection (auto-connects if needed).""" + if self._conn is None: + self.connect() + assert self._conn is not None + return self._conn + + # -- Queries ------------------------------------------------------------- + + def get_task(self, task_id: str) -> Optional[KanbanTask]: + """Fetch a single task by ID.""" + row = self.conn.execute( + "SELECT id, title, assignee, status, priority, created_by " + "FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + if row is None: + return None + return KanbanTask( + id=row["id"], + title=row["title"], + assignee=row["assignee"] or "", + status=row["status"], + priority=row["priority"] or 0, + created_by=row["created_by"] or "", + ) + + def get_events_since(self, min_event_id: int, limit: int = 200) -> list[KanbanEvent]: + """Return kanban events with id > *min_event_id*, ordered oldest first.""" + rows = self.conn.execute( + "SELECT id, task_id, kind, payload, created_at " + "FROM task_events WHERE id > ? " + "ORDER BY id ASC LIMIT ?", + (min_event_id, limit), + ).fetchall() + + events: list[KanbanEvent] = [] + for row in rows: + payload = _safe_json(row["payload"]) + events.append(KanbanEvent( + id=row["id"], + task_id=row["task_id"], + kind=row["kind"], + payload=payload if isinstance(payload, dict) else {}, + created_at=row["created_at"], + )) + return events + + def get_comments_since(self, min_comment_id: int, limit: int = 100) -> list[dict[str, Any]]: + """Return comments with id > *min_comment_id*.""" + rows = self.conn.execute( + "SELECT id, task_id, author, body, created_at " + "FROM task_comments WHERE id > ? " + "ORDER BY id ASC LIMIT ?", + (min_comment_id, limit), + ).fetchall() + return [dict(r) for r in rows] + + def get_latest_event_id(self) -> int: + """Return the max event ID in task_events (0 if empty).""" + row = self.conn.execute("SELECT COALESCE(MAX(id), 0) FROM task_events").fetchone() + return row[0] if row else 0 + + def get_latest_comment_id(self) -> int: + """Return the max comment ID in task_comments (0 if empty).""" + row = self.conn.execute("SELECT COALESCE(MAX(id), 0) FROM task_comments").fetchone() + return row[0] if row else 0 + + def get_matrix_subscriptions(self) -> list[KanbanSubscription]: + """Return all notification subscriptions for the 'matrix' platform.""" + rows = self.conn.execute( + "SELECT task_id, platform, chat_id, thread_id, user_id, last_event_id " + "FROM kanban_notify_subs WHERE platform = 'matrix'" + ).fetchall() + return [ + KanbanSubscription( + task_id=r["task_id"], + platform=r["platform"], + chat_id=r["chat_id"], + thread_id=r["thread_id"] or "", + user_id=r["user_id"] or "", + last_event_id=r["last_event_id"] or 0, + ) + for r in rows + ] + + def update_subscription_last_event( + self, task_id: str, chat_id: str, new_last_event_id: int, + ) -> None: + """Update the last_event_id for a subscription. Requires a writable connection.""" + # Re-open in read-write mode for this write + try: + rw_conn = sqlite3.connect(self._db_path) + rw_conn.execute( + "UPDATE kanban_notify_subs SET last_event_id = ? " + "WHERE task_id = ? AND platform = 'matrix' AND chat_id = ?", + (new_last_event_id, task_id, chat_id), + ) + rw_conn.commit() + rw_conn.close() + except sqlite3.Error as exc: + logger.warning("Failed to update subscription last_event: %s", exc) + + +# --------------------------------------------------------------------------- +# Event formatter +# --------------------------------------------------------------------------- + + +def format_event(event: KanbanEvent, task: Optional[KanbanTask] = None) -> str: + """Format a kanban event as a human-readable notification string. + + Args: + event: The kanban event to format. + task: Optional task snapshot for richer messages. + + Returns: + A formatted string suitable for sending as a Matrix message. + """ + task_ref = f"**{task.title}**" if task else f"`{event.task_id}`" + task_id_short = event.task_id[:12] + ts = event.timestamp_iso + + kind = event.kind + payload = event.payload + + if kind == "created": + assignee = payload.get("assignee", "?") + status = payload.get("status", "todo") + return f"🆕 **New task** {task_ref}\nAssigned to: `{assignee}` | Status: `{status}`\n`{task_id_short}` | {ts}" + + if kind == "claimed": + lock = payload.get("lock", "") + profile = lock.split(":")[0] if ":" in lock else lock + return f"🔒 **Claimed** {task_ref}\nWorker: `{profile}`\n`{task_id_short}` | {ts}" + + if kind == "spawned": + pid = payload.get("pid", "?") + return f"🚀 **Worker spawned** for {task_ref}\nPID: `{pid}`\n`{task_id_short}` | {ts}" + + if kind == "completed": + summary = payload.get("summary", "") + if summary: + return f"✅ **Completed** {task_ref}\n{summary}\n`{task_id_short}` | {ts}" + return f"✅ **Completed** {task_ref}\n`{task_id_short}` | {ts}" + + if kind == "blocked": + reason = payload.get("reason", "") + if reason: + return f"🚫 **Blocked** {task_ref}\nReason: {reason}\n`{task_id_short}` | {ts}" + return f"🚫 **Blocked** {task_ref}\n`{task_id_short}` | {ts}" + + if kind == "unblocked": + return f"🔓 **Unblocked** {task_ref}\n`{task_id_short}` | {ts}" + + if kind == "commented": + author = payload.get("author", "?") + body = payload.get("body", "") + preview = body[:200] + ("..." if len(body) > 200 else "") + return f"💬 **Comment** on {task_ref}\n`{author}`: {preview}\n`{task_id_short}` | {ts}" + + if kind == "failed" or kind == "crashed": + err = payload.get("error", "") + if err: + return f"💥 **{kind.title()}** {task_ref}\nError: {err}\n`{task_id_short}` | {ts}" + return f"💥 **{kind.title()}** {task_ref}\n`{task_id_short}` | {ts}" + + if kind == "promoted": + return f"⬆️ **Promoted** {task_ref}\n`{task_id_short}` | {ts}" + + if kind == "released": + return f"🔓 **Released** {task_ref}\n`{task_id_short}` | {ts}" + + if kind == "timed_out": + return f"⏰ **Timed out** {task_ref}\n`{task_id_short}` | {ts}" + + # Generic fallback + return f"📋 **{kind}** on {task_ref}\n`{task_id_short}` | {ts}" + + +# --------------------------------------------------------------------------- +# KanbanMonitor — poll loop + Matrix delivery +# --------------------------------------------------------------------------- + + +class KanbanMonitor: + """Async monitor that polls a kanban DB and sends notifications via Matrix. + + Usage:: + + db = KanbanDB("/path/to/kanban.db") + monitor = KanbanMonitor(client, db, poll_interval=5.0) + + # Register handlers for custom event processing + monitor.on_event(lambda event, task: print(event.kind)) + + # Start the poll loop + await monitor.start() + + # ... later ... + await monitor.stop() + + Args: + client: A connected MatrixClient for sending notifications. + db: A KanbanDB instance pointing at the kanban database. + poll_interval: Seconds between DB polls (default: 5.0). + """ + + def __init__( + self, + client: MatrixClient, + db: KanbanDB, + poll_interval: float = 5.0, + enable_comments: bool = True, + ) -> None: + if not client.connected: + raise RuntimeError("MatrixClient must be connected before creating KanbanMonitor") + + self._client = client + self._db = db + self._poll_interval = poll_interval + self._enable_comments = enable_comments + self._running = False + self._task: Optional[asyncio.Task] = None + + # Event handlers: callable(event, task) -> None + self._handlers: list[Any] = [] + + # Stats + self.events_processed: int = 0 + self.notifications_sent: int = 0 + + # -- Handlers ------------------------------------------------------------ + + def on_event(self, handler: Any) -> None: + """Register a handler for every kanban event. + + Args: + handler: Callable receiving (KanbanEvent, Optional[KanbanTask]). + """ + self._handlers.append(handler) + + # -- Lifecycle ----------------------------------------------------------- + + async def start(self) -> None: + """Start polling the kanban database and sending notifications.""" + if self._running: + return + self._running = True + self._task = asyncio.create_task(self._poll_loop()) + logger.info("KanbanMonitor started (poll_interval=%.1fs)", self._poll_interval) + + async def stop(self) -> None: + """Stop the monitor and wait for clean shutdown.""" + self._running = False + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info( + "KanbanMonitor stopped: %d events processed, %d notifications sent", + self.events_processed, + self.notifications_sent, + ) + + async def _poll_loop(self) -> None: + """Main polling loop.""" + # Start from the latest event so we only see new ones + last_event_id = self._db.get_latest_event_id() + + backoff = 1.0 # seconds + + while self._running: + try: + # Poll for new events + events = self._db.get_events_since(last_event_id) + + if events: + backoff = 1.0 # reset backoff on success + + for event in events: + self.events_processed += 1 + task = self._db.get_task(event.task_id) + await self._dispatch(event, task) + last_event_id = max(last_event_id, event.id) + + # Poll for new comments + if self._enable_comments: + await self._process_comments() + + except asyncio.CancelledError: + break + except Exception as exc: + logger.error("Poll error (retry in %.1fs): %s", backoff, exc) + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 60.0) + continue + + await asyncio.sleep(self._poll_interval) + + async def _process_comments(self) -> None: + """Check for new comments and emit synthetic events.""" + # We track comment IDs via a subscription trick: we store the last + # comment ID in a special subscription for task_id="*" (wildcard). + # Actually, let's keep it simple: just track the max comment ID. + if not hasattr(self, "_last_comment_id"): + self._last_comment_id = self._db.get_latest_comment_id() + + comments = self._db.get_comments_since(self._last_comment_id) + for comment in comments: + cid = comment["id"] + self._last_comment_id = max(self._last_comment_id, cid) + event = KanbanEvent( + id=-(cid + 1_000_000_000), # negative IDs to avoid collision + task_id=comment["task_id"], + kind="commented", + payload={"author": comment["author"], "body": comment["body"]}, + created_at=comment["created_at"], + ) + task = self._db.get_task(event.task_id) + await self._dispatch(event, task) + + async def _dispatch(self, event: KanbanEvent, task: Optional[KanbanTask]) -> None: + """Dispatch an event to registered handlers and Matrix subscribers.""" + # Fire handlers + for handler in self._handlers: + try: + result = handler(event, task) + if asyncio.iscoroutine(result): + await result + except Exception as exc: + logger.error("Handler error: %s", exc) + + # Send to subscribed Matrix rooms + await self._notify_subscribers(event, task) + + async def _notify_subscribers( + self, event: KanbanEvent, task: Optional[KanbanTask], + ) -> None: + """Send formatted event to Matrix rooms subscribed to this task.""" + subscriptions = self._db.get_matrix_subscriptions() + + # Build a set of (chat_id, thread_id) to send to + targets: set[tuple[str, str]] = set() + + for sub in subscriptions: + if sub.task_id == "*" or sub.task_id == event.task_id: + targets.add((sub.chat_id, sub.thread_id)) + + if not targets: + return + + message = format_event(event, task) + + # Include task details as context + if task: + details = ( + f"\n\n▸ Status: `{task.status}`" + f"\n▸ Assignee: `{task.assignee or 'unassigned'}`" + ) + # Truncate to avoid excessively long messages + message += details + + # Send to each target + for chat_id, thread_id in targets: + try: + result = await self._client.send_message( + room_id=chat_id, + body=message, + thread_root=thread_id if thread_id else "", + ) + if result.success: + self.notifications_sent += 1 + # Update subscription tracking + self._db.update_subscription_last_event( + event.task_id, chat_id, event.id, + ) + else: + logger.warning( + "Failed to notify %s: %s", chat_id, result.error, + ) + except Exception as exc: + logger.error("Error notifying %s: %s", chat_id, exc) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +import json + + +def _safe_json(raw: Optional[str]) -> Any: + """Parse JSON, returning empty dict on failure.""" + if not raw: + return {} + try: + return json.loads(raw) + except (json.JSONDecodeError, TypeError): + return {} diff --git a/tests/test_kanban.py b/tests/test_kanban.py new file mode 100644 index 0000000..adacf8f --- /dev/null +++ b/tests/test_kanban.py @@ -0,0 +1,478 @@ +"""Tests for matrix_hermes.kanban — kanban board monitoring and notifications.""" + +import sqlite3 +import tempfile +import time +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from matrix_hermes.kanban import ( + KanbanDB, + KanbanEvent, + KanbanMonitor, + KanbanSubscription, + KanbanTask, + format_event, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def kanban_db_path(): + """Create a temporary kanban database with test schema and data.""" + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = f.name + + conn = sqlite3.connect(db_path) + conn.executescript(""" + CREATE TABLE tasks ( + id TEXT PRIMARY KEY, title TEXT NOT NULL, assignee TEXT, + status TEXT NOT NULL, priority INTEGER DEFAULT 0, + created_by TEXT, created_at INTEGER NOT NULL + ); + CREATE TABLE task_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, run_id INTEGER, + kind TEXT NOT NULL, payload TEXT, created_at INTEGER NOT NULL + ); + CREATE TABLE task_comments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + task_id TEXT NOT NULL, author TEXT NOT NULL, + body TEXT NOT NULL, created_at INTEGER NOT NULL + ); + CREATE TABLE kanban_notify_subs ( + task_id TEXT NOT NULL, platform TEXT NOT NULL, + chat_id TEXT NOT NULL, thread_id TEXT NOT NULL DEFAULT '', + user_id TEXT, notifier_profile TEXT, + created_at INTEGER NOT NULL, + last_event_id INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (task_id, platform, chat_id, thread_id) + ); + + -- Test tasks + INSERT INTO tasks VALUES + ('t_test1', 'Test task one', 'coder', 'running', 1, 'user', 1000), + ('t_test2', 'Test task two', 'reviewer', 'todo', 0, 'user', 1000); + + -- Test events + INSERT INTO task_events (task_id, kind, payload, created_at) VALUES + ('t_test1', 'created', '{"assignee":"coder","status":"todo"}', 1000), + ('t_test1', 'claimed', '{"lock":"coder:123"}', 1005), + ('t_test1', 'spawned', '{"pid":42}', 1010), + ('t_test2', 'created', '{"assignee":"reviewer"}', 1020); + + -- Test subscription + INSERT INTO kanban_notify_subs VALUES + ('t_test1', 'matrix', '!room1:example.org', '', '@bot:example.org', 'monitor', 2000, 1); + """) + conn.commit() + conn.close() + + yield db_path + + # Cleanup + Path(db_path).unlink(missing_ok=True) + + +@pytest.fixture +def mock_matrix_client(): + """Create a mock MatrixClient that is 'connected'.""" + client = MagicMock() + client.connected = True + client.user_id = "@bot:example.org" + client.send_message = AsyncMock(return_value=MagicMock(success=True, event_id="$evt123")) + client.get_joined_rooms = AsyncMock(return_value=["!room1:example.org"]) + client.join_room = AsyncMock(return_value=True) + return client + + +# --------------------------------------------------------------------------- +# KanbanDB tests +# --------------------------------------------------------------------------- + + +class TestKanbanDB: + """Tests for KanbanDB.""" + + def test_connect_and_close(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + db.connect() + assert db._conn is not None + db.close() + assert db._conn is None + + def test_get_task(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + task = db.get_task("t_test1") + assert task is not None + assert task.id == "t_test1" + assert task.title == "Test task one" + assert task.assignee == "coder" + assert task.status == "running" + db.close() + + def test_get_task_nonexistent(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + task = db.get_task("t_nonexistent") + assert task is None + db.close() + + def test_get_events_since(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + events = db.get_events_since(0) + assert len(events) == 4 + assert events[0].kind == "created" + assert events[0].task_id == "t_test1" + + # After first event + events2 = db.get_events_since(events[0].id) + assert len(events2) == 3 + assert events2[0].kind == "claimed" + db.close() + + def test_get_events_since_limit(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + events = db.get_events_since(0, limit=2) + assert len(events) == 2 + db.close() + + def test_get_latest_event_id(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + latest = db.get_latest_event_id() + assert latest > 0 + db.close() + + def test_get_matrix_subscriptions(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + subs = db.get_matrix_subscriptions() + assert len(subs) == 1 + assert subs[0].task_id == "t_test1" + assert subs[0].chat_id == "!room1:example.org" + assert subs[0].platform == "matrix" + db.close() + + def test_get_comments_since(self, kanban_db_path): + # Add a comment + conn = sqlite3.connect(kanban_db_path) + conn.execute( + "INSERT INTO task_comments (task_id, author, body, created_at) " + "VALUES ('t_test1', 'user', 'Nice work!', 1050)" + ) + conn.commit() + conn.close() + + db = KanbanDB(kanban_db_path) + comments = db.get_comments_since(0) + assert len(comments) == 1 + assert comments[0]["author"] == "user" + assert comments[0]["body"] == "Nice work!" + db.close() + + def test_update_subscription_last_event(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + db.update_subscription_last_event("t_test1", "!room1:example.org", 99) + + # Verify + conn = sqlite3.connect(kanban_db_path) + row = conn.execute( + "SELECT last_event_id FROM kanban_notify_subs " + "WHERE task_id = 't_test1' AND platform = 'matrix'" + ).fetchone() + conn.close() + assert row[0] == 99 + + def test_auto_connect(self, kanban_db_path): + db = KanbanDB(kanban_db_path) + # Don't call connect explicitly + task = db.get_task("t_test1") + assert task is not None + db.close() + + +# --------------------------------------------------------------------------- +# format_event tests +# --------------------------------------------------------------------------- + + +class TestFormatEvent: + """Tests for format_event.""" + + def test_created(self): + event = KanbanEvent( + id=1, task_id="t_test1", kind="created", + payload={"assignee": "coder", "status": "todo"}, + created_at=1000, + ) + task = KanbanTask(id="t_test1", title="My Task") + msg = format_event(event, task) + assert "New task" in msg + assert "My Task" in msg + + def test_claimed(self): + event = KanbanEvent( + id=2, task_id="t_test1", kind="claimed", + payload={"lock": "coder:123"}, + created_at=1005, + ) + msg = format_event(event) + assert "Claimed" in msg + assert "coder" in msg + + def test_spawned(self): + event = KanbanEvent( + id=3, task_id="t_test1", kind="spawned", + payload={"pid": 42}, + created_at=1010, + ) + msg = format_event(event) + assert "spawned" in msg.lower() + assert "42" in msg + + def test_completed_with_summary(self): + event = KanbanEvent( + id=4, task_id="t_test1", kind="completed", + payload={"summary": "All done!"}, + created_at=2000, + ) + msg = format_event(event) + assert "Completed" in msg + assert "All done!" in msg + + def test_blocked_with_reason(self): + event = KanbanEvent( + id=5, task_id="t_test1", kind="blocked", + payload={"reason": "Need auth token"}, + created_at=2000, + ) + msg = format_event(event) + assert "Blocked" in msg + assert "Need auth token" in msg + + def test_timed_out(self): + event = KanbanEvent( + id=6, task_id="t_test1", kind="timed_out", + created_at=2000, + ) + msg = format_event(event) + assert "Timed out" in msg + + def test_released(self): + event = KanbanEvent( + id=7, task_id="t_test1", kind="released", + created_at=2000, + ) + msg = format_event(event) + assert "Released" in msg + + def test_unknown_kind(self): + event = KanbanEvent( + id=8, task_id="t_test1", kind="weird_event", + created_at=2000, + ) + msg = format_event(event) + assert "weird_event" in msg + assert "t_test1" in msg + + def test_timestamp_iso(self): + event = KanbanEvent( + id=1, task_id="t_test1", kind="created", + created_at=1000, + ) + assert "1970-01-01" in event.timestamp_iso + + +# --------------------------------------------------------------------------- +# KanbanMonitor tests +# --------------------------------------------------------------------------- + + +class TestKanbanMonitor: + """Tests for KanbanMonitor.""" + + def test_init_requires_connected_client(self, kanban_db_path): + client = MagicMock() + client.connected = False + db = KanbanDB(kanban_db_path) + + with pytest.raises(RuntimeError, match="connected"): + KanbanMonitor(client, db) + + def test_init_success(self, kanban_db_path, mock_matrix_client): + db = KanbanDB(kanban_db_path) + monitor = KanbanMonitor(mock_matrix_client, db, poll_interval=1.0) + assert monitor.events_processed == 0 + assert monitor.notifications_sent == 0 + + @pytest.mark.asyncio + async def test_start_and_stop(self, kanban_db_path, mock_matrix_client): + db = KanbanDB(kanban_db_path) + monitor = KanbanMonitor(mock_matrix_client, db, poll_interval=0.1) + + # Start + await monitor.start() + assert monitor._running + + # Let it poll a couple times + await asyncio_sleep(0.3) + + # Stop + await monitor.stop() + assert not monitor._running + + @pytest.mark.asyncio + async def test_event_handler_called(self, kanban_db_path, mock_matrix_client): + db = KanbanDB(kanban_db_path) + monitor = KanbanMonitor(mock_matrix_client, db, poll_interval=0.1) + + handler_events = [] + + async def my_handler(event, task): + handler_events.append(event.kind) + + monitor.on_event(my_handler) + + await monitor.start() + # Let the poll loop initialize and capture baseline event ID + await asyncio_sleep(0.2) + + # Insert a new event after the monitor has started + conn = sqlite3.connect(kanban_db_path) + conn.execute( + "INSERT INTO task_events (task_id, kind, payload, created_at) " + "VALUES ('t_test1', 'blocked', '{\"reason\":\"test\"}', 1200)" + ) + conn.commit() + conn.close() + + await asyncio_sleep(0.5) + await monitor.stop() + + # Should have processed new event + assert len(handler_events) > 0 + + @pytest.mark.asyncio + async def test_notifies_subscribers(self, kanban_db_path, mock_matrix_client): + db = KanbanDB(kanban_db_path) + monitor = KanbanMonitor(mock_matrix_client, db, poll_interval=0.1) + + await monitor.start() + # Let the poll loop initialize and capture baseline event ID + await asyncio_sleep(0.2) + + # Add a new event after starting the monitor + conn = sqlite3.connect(kanban_db_path) + conn.execute( + "INSERT INTO task_events (task_id, kind, payload, created_at) " + "VALUES ('t_test1', 'completed', '{\"summary\":\"Done!\"}', 1100)" + ) + conn.commit() + conn.close() + + await asyncio_sleep(0.5) + await monitor.stop() + + # Should have sent at least one notification + assert monitor.notifications_sent >= 1 + + # Verify send_message was called + assert mock_matrix_client.send_message.called + + @pytest.mark.asyncio + async def test_no_subscriptions_no_notifications(self, kanban_db_path, mock_matrix_client): + # Remove the subscription + conn = sqlite3.connect(kanban_db_path) + conn.execute("DELETE FROM kanban_notify_subs") + conn.commit() + conn.close() + + # Add a new event + conn = sqlite3.connect(kanban_db_path) + conn.execute( + "INSERT INTO task_events (task_id, kind, payload, created_at) " + "VALUES ('t_test1', 'completed', '{}', 1100)" + ) + conn.commit() + conn.close() + + db = KanbanDB(kanban_db_path) + monitor = KanbanMonitor(mock_matrix_client, db, poll_interval=0.1) + + await monitor.start() + await asyncio_sleep(0.3) + await monitor.stop() + + # No notifications should be sent (no subscriptions) + assert monitor.notifications_sent == 0 + + +# --------------------------------------------------------------------------- +# KanbanSubscription tests +# --------------------------------------------------------------------------- + + +class TestKanbanSubscription: + """Tests for KanbanSubscription dataclass.""" + + def test_defaults(self): + sub = KanbanSubscription( + task_id="t_test", platform="matrix", chat_id="!room:example.org", + ) + assert sub.thread_id == "" + assert sub.user_id == "" + assert sub.last_event_id == 0 + + def test_full(self): + sub = KanbanSubscription( + task_id="t_test", platform="matrix", chat_id="!room:example.org", + thread_id="$evt_thread", user_id="@bot:example.org", last_event_id=42, + ) + assert sub.thread_id == "$evt_thread" + assert sub.user_id == "@bot:example.org" + assert sub.last_event_id == 42 + + +# --------------------------------------------------------------------------- +# KanbanTask tests +# --------------------------------------------------------------------------- + + +class TestKanbanTask: + """Tests for KanbanTask dataclass.""" + + def test_minimal(self): + task = KanbanTask(id="t123", title="Minimal task") + assert task.id == "t123" + assert task.title == "Minimal task" + assert task.assignee == "" + assert task.status == "todo" + assert task.priority == 0 + + def test_full(self): + task = KanbanTask( + id="t456", title="Full task", assignee="coder", + status="running", priority=1, created_by="user", + ) + assert task.assignee == "coder" + assert task.status == "running" + + +# Helpers +import asyncio as _asyncio + + +def asyncio_sleep(seconds): + """Helper that works in both sync and async contexts for test waiting.""" + try: + loop = _asyncio.get_running_loop() + return _asyncio.sleep(seconds) + except RuntimeError: + time.sleep(seconds) + fut = _asyncio.Future() + fut.set_result(None) + return fut