diff --git a/src/haiku/rag/app.py b/src/haiku/rag/app.py index 371b571f..3589c003 100644 --- a/src/haiku/rag/app.py +++ b/src/haiku/rag/app.py @@ -94,13 +94,14 @@ class HaikuRAGApp: async def serve(self, transport: str | None = None): """Start the MCP server.""" - monitor = FileWatcher(paths=Config.MONITOR_DIRECTORIES) - asyncio.create_task(monitor.observe()) - server = create_mcp_server(self.db_path) + async with HaikuRAG(self.db_path) as client: + monitor = FileWatcher(paths=Config.MONITOR_DIRECTORIES, client=client) + asyncio.create_task(monitor.observe()) + server = create_mcp_server(self.db_path) - if transport == "stdio": - await server.run_stdio_async() - elif transport == "sse": - await server.run_sse_async("sse") - else: - await server.run_http_async("streamable-http") + if transport == "stdio": + await server.run_stdio_async() + elif transport == "sse": + await server.run_sse_async("sse") + else: + await server.run_http_async("streamable-http") diff --git a/src/haiku/rag/monitor.py b/src/haiku/rag/monitor.py index be12f9d9..97809e9e 100644 --- a/src/haiku/rag/monitor.py +++ b/src/haiku/rag/monitor.py @@ -2,6 +2,7 @@ from pathlib import Path from watchfiles import Change, DefaultFilter, awatch +from haiku.rag.client import HaikuRAG from haiku.rag.logging import get_logger from haiku.rag.reader import FileReader from haiku.rag.store.models.document import Document @@ -19,8 +20,9 @@ class FileFilter(DefaultFilter): class FileWatcher: - def __init__(self, paths: list[Path]): + def __init__(self, paths: list[Path], client: HaikuRAG): self.paths = paths + self.client = client async def observe(self): logger.info(f"Watching files in {self.paths}") @@ -43,10 +45,30 @@ class FileWatcher: 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 + try: + uri = file.as_uri() + existing_doc = await self.client.get_document_by_uri(uri) + print(uri) + if existing_doc: + doc = await self.client.create_document_from_source(str(file)) + logger.info(f"Updated document {existing_doc.id} from {file}") + return doc + else: + doc = await self.client.create_document_from_source(str(file)) + logger.info(f"Created new document {doc.id} from {file}") + return doc + except Exception as e: + logger.error(f"Failed to upsert document from {file}: {e}") + return None + + async def _delete_document(self, file: Path): + try: + uri = file.as_uri() + existing_doc = await self.client.get_document_by_uri(uri) + + if existing_doc and existing_doc.id: + await self.client.delete_document(existing_doc.id) + logger.info(f"Deleted document {existing_doc.id} for {file}") + except Exception as e: + logger.error(f"Failed to delete document for {file}: {e}")