From 5f604d31dd545da5e2fb0c3ef36e7071e88d8cb9 Mon Sep 17 00:00:00 2001
From: Yiorgis Gozadinos <ggozadinos@gmail.com>
Date: Fri, 20 Jun 2025 07:44:03 +0200
Subject: [PATCH] Add monitor

---
 src/haiku/rag/app.py     | 20 +++++++--------
 src/haiku/rag/cli.py     |  2 +-
 src/haiku/rag/logging.py | 24 ++++++++++++++++++
 src/haiku/rag/monitor.py | 54 ++++++++++++++++++++++++++++++++++++++++
 4 files changed, 88 insertions(+), 12 deletions(-)
 create mode 100644 src/haiku/rag/logging.py
 create mode 100644 src/haiku/rag/monitor.py

diff --git a/src/haiku/rag/app.py b/src/haiku/rag/app.py
index 1972afae..b8e3c6d0 100644
--- a/src/haiku/rag/app.py
+++ b/src/haiku/rag/app.py
@@ -1,9 +1,12 @@
+import asyncio
 from pathlib import Path
 
 from rich.console import Console
 from rich.markdown import Markdown
 
 from haiku.rag.client import HaikuRAG
+from haiku.rag.mcp import create_mcp_server
+from haiku.rag.monitor import FileWatcher
 from haiku.rag.store.models.chunk import Chunk
 from haiku.rag.store.models.document import Document
 
@@ -88,20 +91,15 @@ class HaikuRAGApp:
         self.console.print(content)
         self.console.rule()
 
-    def serve(self, transport: str | None = None):
+    async def serve(self, transport: str | None = None):
         """Start the MCP server."""
-        from haiku.rag.mcp import create_mcp_server
-
+        monitor = FileWatcher(paths=[])
+        asyncio.create_task(monitor.observe())
         server = create_mcp_server(self.db_path)
 
         if transport == "stdio":
-            self.console.print("[green]Starting MCP server on stdio...[/green]")
-            server.run("stdio")
+            await server.run_stdio_async()
         elif transport == "sse":
-            self.console.print(
-                "[green]Starting MCP server with streamable HTTP...[/green]"
-            )
-            server.run("sse")
+            await server.run_sse_async("sse")
         else:
-            self.console.print("[green]Starting MCP server with HTTP...[/green]")
-            server.run("streamable-http")
+            await server.run_http_async("streamable-http")
diff --git a/src/haiku/rag/cli.py b/src/haiku/rag/cli.py
index 21b5d4d1..71e2c8b9 100644
--- a/src/haiku/rag/cli.py
+++ b/src/haiku/rag/cli.py
@@ -146,7 +146,7 @@ def serve(
     elif sse:
         transport = "sse"
 
-    app.serve(transport=transport)
+    event_loop.run_until_complete(app.serve(transport=transport))
 
 
 if __name__ == "__main__":
diff --git a/src/haiku/rag/logging.py b/src/haiku/rag/logging.py
new file mode 100644
index 00000000..33197b8d
--- /dev/null
+++ b/src/haiku/rag/logging.py
@@ -0,0 +1,24 @@
+import logging
+
+from rich.console import Console
+from rich.logging import RichHandler
+
+
+def get_logger() -> logging.Logger:
+    logger = logging.getLogger("haiku.rag")
+
+    handler = RichHandler(
+        console=Console(stderr=True),
+        rich_tracebacks=True,
+    )
+    formatter = logging.Formatter("%(message)s")
+    handler.setFormatter(formatter)
+
+    logger.setLevel("INFO")
+
+    # Remove any existing handlers to avoid duplicates on reconfiguration
+    for hdlr in logger.handlers[:]:
+        logger.removeHandler(hdlr)
+
+    logger.addHandler(handler)
+    return logger
diff --git a/src/haiku/rag/monitor.py b/src/haiku/rag/monitor.py
new file mode 100644
index 00000000..dccdb764
--- /dev/null
+++ b/src/haiku/rag/monitor.py
@@ -0,0 +1,54 @@
+from pathlib import Path
+
+from watchfiles import Change, DefaultFilter, awatch
+
+from haiku.rag.logging import get_logger
+from haiku.rag.reader import FileReader
+from haiku.rag.store.models.document import Document
+
+logger = get_logger()
+
+
+class FileFilter(DefaultFilter):
+    def __init__(self, *, ignore_paths: list[str | Path] | None = None) -> None:
+        self.extensions = tuple(FileReader.extensions)
+        super().__init__(ignore_paths=ignore_paths)
+
+    def __call__(self, change: "Change", path: str) -> bool:
+        return path.endswith(self.extensions) and super().__call__(change, path)
+
+
+class FileWatcher:
+    def __init__(self, paths: list[str | Path]):
+        self.paths = paths
+
+    async def observe(self):
+        logger.info(f"Watching files in {self.paths}")
+        filter = FileFilter()
+        await self.refresh()
+
+        async for changes in awatch(*self.paths, watch_filter=filter):
+            await self.handler(changes)
+
+    async def handler(self, changes: set[tuple[Change, str]]):
+        for change, path in changes:
+            if change == Change.added or change == Change.modified:
+                await self._upsert_document(Path(path))
+            elif change == Change.deleted:
+                await self._delete_document(Path(path))
+
+    async def refresh(self, paths: list[str | Path] | None = None):
+        if paths is None:
+            paths = self.paths
+        for path in paths:
+            for f in Path(path).rglob("**/*"):
+                if f.is_file() and f.suffix in FileReader.extensions:
+                    await self._upsert_document(f)
+
+    async def _delete_document(self, file: Path):
+        logger.info(f"Deleting document from {file}")
+        pass
+
+    async def _upsert_document(self, file: Path) -> Document | None:
+        logger.info(f"Updating document from {file}")
+        pass