danvics/haiku.rag
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #135 from ggozad/feat/monitor-improvements

Browse source 
Optional orphan deletion when monitoring. Fix skipping updates of non-changed files.
This commit is contained in:
Yiorgis Gozadinos 2025-11-07 14:40:49 +02:00 committed by GitHub
commit 51d64259fd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 302 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
CHANGELOG.md
View file

@ -1,6 +1,10 @@
# Changelog
## [Unreleased]
### Added
- **File Monitor**: Orphan deletion feature - automatically removes documents from database when source files are deleted (enabled via `monitor.delete_orphans` config option, default: false)
### Changed
- **Configuration**: All CLI commands now properly support `--config` parameter for specifying custom configuration files
@ -9,6 +13,10 @@
- Updated CLI documentation to clarify global vs per-command options
- **BREAKING**: Standardized configuration filename to `haiku.rag.yaml` in user directories (was incorrectly using `config.yaml`). Users with existing `config.yaml` in their user directory will need to rename it to `haiku.rag.yaml`
### Fixed
- **File Monitor**: Fixed incorrect "Updated document" logging for unchanged files - monitor now properly skips files when MD5 hash hasn't changed
### Removed
- **BREAKING**: A2A (Agent-to-Agent) protocol support has been moved to a separate self-contained package in `examples/a2a-server/`. The A2A server is no longer part of the main haiku.rag package. Users who need A2A functionality can install and run it from the examples directory with `cd examples/a2a-server && uv sync`.

19
haiku_rag_slim/haiku/rag/client.py
View file

@ -219,9 +219,13 @@ class HaikuRAG:
if title is not None and title != existing_doc.title:
existing_doc.title = title
updated = True
if metadata:
existing_doc.metadata = {**(existing_doc.metadata or {}), **metadata}
# Check if metadata actually changed (beyond contentType and md5)
merged_metadata = {**(existing_doc.metadata or {}), **metadata}
if merged_metadata != existing_doc.metadata:
existing_doc.metadata = merged_metadata
updated = True
if updated:
return await self.document_repository.update(existing_doc)
return existing_doc
@ -290,13 +294,14 @@ class HaikuRAG:
if title is not None and title != existing_doc.title:
existing_doc.title = title
updated = True
metadata.update({"contentType": content_type, "md5": md5_hash})
if metadata:
existing_doc.metadata = {
**(existing_doc.metadata or {}),
**metadata,
}
# Check if metadata actually changed (beyond contentType and md5)
merged_metadata = {**(existing_doc.metadata or {}), **metadata}
if merged_metadata != existing_doc.metadata:
existing_doc.metadata = merged_metadata
updated = True
if updated:
return await self.document_repository.update(existing_doc)
return existing_doc

1
haiku_rag_slim/haiku/rag/config/models.py
View file

@ -15,6 +15,7 @@ class MonitorConfig(BaseModel):
directories: list[Path] = []
ignore_patterns: list[str] = []
include_patterns: list[str] = []
delete_orphans: bool = False
class LanceDBConfig(BaseModel):

71
haiku_rag_slim/haiku/rag/monitor.py
View file

@ -1,3 +1,4 @@
import asyncio
import logging
from pathlib import Path
from typing import TYPE_CHECKING
@ -75,6 +76,7 @@ class FileWatcher:
self.client = client
self.ignore_patterns = config.monitor.ignore_patterns or None
self.include_patterns = config.monitor.include_patterns or None
self.delete_orphans = config.monitor.delete_orphans
async def observe(self):
logger.info(f"Watching files in {self.paths}")
@ -97,6 +99,11 @@ class FileWatcher:
# Lazy import to avoid loading docling
from haiku.rag.reader import FileReader
# Delete orphaned documents in background if enabled
if self.delete_orphans:
logger.info("Starting orphan cleanup in background")
asyncio.create_task(self._delete_orphans())
# Create filter to apply same logic as observe()
filter = FileFilter(
ignore_patterns=self.ignore_patterns, include_patterns=self.include_patterns
@ -113,22 +120,68 @@ class FileWatcher:
try:
uri = file.as_uri()
existing_doc = await self.client.get_document_by_uri(uri)
result = await self.client.create_document_from_source(str(file))
doc = result if isinstance(result, Document) else result[0]
if existing_doc:
result = await self.client.create_document_from_source(str(file))
# Since we're passing a file (not directory), result should be a single Document
doc = result if isinstance(result, Document) else result[0]
logger.info(f"Updated document {existing_doc.id} from {file}")
return doc
# Check if document was actually updated by comparing updated_at timestamps
if doc.updated_at > existing_doc.updated_at:
logger.info(f"Updated document {existing_doc.id} from {file}")
else:
logger.info(
f"Skipped unchanged document {existing_doc.id} from {file}"
)
else:
result = await self.client.create_document_from_source(str(file))
# Since we're passing a file (not directory), result should be a single Document
doc = result if isinstance(result, Document) else result[0]
logger.info(f"Created new document {doc.id} from {file}")
return doc
return doc
except Exception as e:
logger.error(f"Failed to upsert document from {file}: {e}")
return None
async def _delete_orphans(self):
"""Delete documents whose source files no longer exist."""
try:
from urllib.parse import unquote, urlparse
# Create filter to apply same include/exclude logic
filter = FileFilter(
ignore_patterns=self.ignore_patterns,
include_patterns=self.include_patterns,
)
all_docs = await self.client.list_documents()
for doc in all_docs:
if not doc.uri or not doc.id:
continue
# Only check file:// URIs
parsed = urlparse(doc.uri)
if parsed.scheme != "file":
continue
# Convert URI to Path, decoding URL-encoded characters (like %20 for spaces)
file_path = Path(unquote(parsed.path))
# Check if file exists
if not file_path.exists():
# Check if file is within monitored directories
is_monitored = any(
file_path.is_relative_to(monitored_path)
for monitored_path in self.paths
)
# Check if file would have been included by filters
if is_monitored and filter.include_file(str(file_path)):
await self.client.delete_document(doc.id)
logger.info(
f"Deleted orphaned document {doc.id} for {file_path}"
)
except Exception as e:
logger.error(f"Failed to delete orphaned documents: {e}")
async def _delete_document(self, file: Path):
try:
uri = file.as_uri()

22
tests/test_client.py
View file

@ -485,6 +485,28 @@ async def test_client_create_update_no_op_behavior(temp_db_path):
assert retrieved_doc.content == updated_content
@pytest.mark.asyncio
async def test_client_unchanged_file_keeps_timestamp(temp_db_path):
"""Test that unchanged files don't update the updated_at timestamp."""
async with HaikuRAG(temp_db_path) as client:
# Create a temporary file
test_content = "Test content for timestamp check."
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir) / "test.txt"
temp_path.write_text(test_content)
# First call - create document
doc1 = await client.create_document_from_source(temp_path)
assert isinstance(doc1, Document)
original_updated_at = doc1.updated_at
# Second call with same content - should not update timestamp
doc2 = await client.create_document_from_source(temp_path)
assert isinstance(doc2, Document)
assert doc2.id == doc1.id
assert doc2.updated_at == original_updated_at # Timestamp should not change
@pytest.mark.asyncio
async def test_client_url_create_update_no_op_behavior(temp_db_path):
"""Test create/update/no-op behavior for URLs based on MD5 changes."""

197
tests/test_monitor.py
View file

@ -243,3 +243,200 @@ async def test_file_watcher_with_include_patterns():
# Should have only called for the .md file
assert mock_client.create_document_from_source.call_count == 1
mock_client.create_document_from_source.assert_called_with(str(md_file))
@pytest.mark.asyncio
async def test_file_watcher_skips_unchanged_document():
"""Test FileWatcher returns existing document when content hasn't changed."""
from datetime import datetime
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
test_file = temp_path / "test.txt"
test_content = "Test content"
test_file.write_text(test_content)
mock_client = AsyncMock(spec=HaikuRAG)
# Existing document with a timestamp
now = datetime.now()
existing_doc = Document(
id="1",
content=test_content,
uri=test_file.as_uri(),
created_at=now,
updated_at=now,
)
mock_client.get_document_by_uri.return_value = existing_doc
# Client returns same document with same timestamp (unchanged)
mock_client.create_document_from_source.return_value = existing_doc
test_config = AppConfig(monitor=MonitorConfig(directories=[temp_path]))
watcher = FileWatcher(client=mock_client, config=test_config)
result = await watcher._upsert_document(test_file)
assert result is not None
assert result.id == "1"
# Verify timestamp hasn't changed (document wasn't updated)
assert result.updated_at == now
@pytest.mark.asyncio
async def test_file_watcher_deletes_orphans():
"""Test FileWatcher deletes documents whose files no longer exist."""
import asyncio
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
existing_file = temp_path / "exists.txt"
existing_file.write_text("Existing file")
# Create a document for a file that doesn't exist
orphan_uri = (temp_path / "deleted.txt").as_uri()
mock_client = AsyncMock(spec=HaikuRAG)
orphan_doc = Document(id="orphan-1", content="Orphaned content", uri=orphan_uri)
existing_doc = Document(
id="existing-1", content="Existing content", uri=existing_file.as_uri()
)
# Mock list_documents to return both documents
mock_client.list_documents.return_value = [orphan_doc, existing_doc]
mock_client.get_document_by_uri.return_value = None
mock_client.create_document_from_source.return_value = existing_doc
test_config = AppConfig(
monitor=MonitorConfig(directories=[temp_path], delete_orphans=True)
)
watcher = FileWatcher(client=mock_client, config=test_config)
# Run refresh which should delete orphan and process existing file
await watcher.refresh()
# Give background task time to complete
await asyncio.sleep(0.1)
# Should have deleted the orphan document
mock_client.delete_document.assert_called_once_with("orphan-1")
# Should have processed the existing file
mock_client.create_document_from_source.assert_called_once()
@pytest.mark.asyncio
async def test_file_watcher_skips_orphan_deletion_when_disabled():
"""Test FileWatcher does not delete orphans when delete_orphans is False."""
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Create a document for a file that doesn't exist
orphan_uri = (temp_path / "deleted.txt").as_uri()
mock_client = AsyncMock(spec=HaikuRAG)
orphan_doc = Document(id="orphan-1", content="Orphaned content", uri=orphan_uri)
# Mock list_documents to return orphan document
mock_client.list_documents.return_value = [orphan_doc]
test_config = AppConfig(
monitor=MonitorConfig(directories=[temp_path], delete_orphans=False)
)
watcher = FileWatcher(client=mock_client, config=test_config)
# Run refresh
await watcher.refresh()
# Should NOT have deleted the orphan document
mock_client.delete_document.assert_not_called()
@pytest.mark.asyncio
async def test_file_watcher_orphan_deletion_respects_patterns():
"""Test orphan deletion respects include/ignore patterns."""
import asyncio
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Create documents for files that don't exist
ignored_orphan_uri = (temp_path / "draft.md").as_uri()
excluded_orphan_uri = (temp_path / "file.pdf").as_uri()
included_orphan_uri = (temp_path / "readme.md").as_uri()
mock_client = AsyncMock(spec=HaikuRAG)
ignored_doc = Document(
id="ignored-1", content="Ignored", uri=ignored_orphan_uri
)
excluded_doc = Document(
id="excluded-1", content="Excluded", uri=excluded_orphan_uri
)
included_doc = Document(
id="included-1", content="Included", uri=included_orphan_uri
)
# Mock list_documents to return all orphan documents
mock_client.list_documents.return_value = [
ignored_doc,
excluded_doc,
included_doc,
]
# Config with patterns: only .md files, but exclude draft*
test_config = AppConfig(
monitor=MonitorConfig(
directories=[temp_path],
delete_orphans=True,
include_patterns=["*.md"],
ignore_patterns=["draft*"],
)
)
watcher = FileWatcher(client=mock_client, config=test_config)
# Run refresh
await watcher.refresh()
# Give background task time to complete
await asyncio.sleep(0.1)
# Should only delete the included orphan (readme.md)
# - draft.md matches ignore pattern -> NOT deleted
# - file.pdf doesn't match include pattern -> NOT deleted
# - readme.md matches include and not ignored -> DELETED
mock_client.delete_document.assert_called_once_with("included-1")
@pytest.mark.asyncio
async def test_file_watcher_orphan_handles_spaces_in_filenames():
"""Test orphan deletion correctly handles files with spaces in names."""
import asyncio
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Create a file with spaces that exists
existing_file = temp_path / "my file with spaces.txt"
existing_file.write_text("Existing file")
mock_client = AsyncMock(spec=HaikuRAG)
# Document with URI that has URL-encoded spaces (%20)
existing_doc = Document(
id="existing-1", content="Existing", uri=existing_file.as_uri()
)
# Mock list_documents to return document with encoded spaces
mock_client.list_documents.return_value = [existing_doc]
mock_client.get_document_by_uri.return_value = None
mock_client.create_document_from_source.return_value = existing_doc
test_config = AppConfig(
monitor=MonitorConfig(directories=[temp_path], delete_orphans=True)
)
watcher = FileWatcher(client=mock_client, config=test_config)
# Run refresh
await watcher.refresh()
# Give background task time to complete
await asyncio.sleep(0.1)
# Should NOT delete the document since file exists
mock_client.delete_document.assert_not_called()