diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f621e6..a2927b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **LLM call timeouts** - Phase 0 keyword extraction and Phase 4 re-ranking are wrapped in a 12s `asyncio.wait_for` with graceful fallback, so a hung Ollama call can no longer gate retrieval for the full 120s client timeout. - **`/stats` wiki page count** - The endpoint passed the bare user name as path prefix (matching nothing) and always reported 0 pages; it now counts pages under `users/{user}`. - **Wiki.js page listing** - `list_pages` applied the API-side `limit` before client-side path/tag filters, dropping matching pages that sort late; the limit now applies after filtering. `list_all_pages` replaced its fake pagination loop with a real limit-growth loop (Wiki.js 2.x `pages.list` has no offset argument) that fetches until the API returns fewer pages than requested. +- **Consolidation loop silently drained its queue on LLM failure** - Root cause of the 30-minute knowledge-consolidation loop processing 0 searches ("No unprocessed searches found" in prod): the `OLLAMA_MODEL` env collision (see below) made every consolidation `/api/generate` call fail with HTTP 400 (`"nomic-embed-text" does not support generate` — confirmed in prod logs and by direct Ollama probe), classification returned empty, and the loop STILL marked every SearchQuery `processed: true` — permanently consuming the queue with zero pages ever created (live Neo4j: 197/200 SearchQuery nodes processed with no output). LLM-infrastructure failure now raises `ConsolidationLLMUnavailableError`: the affected searches stay unprocessed (retried next run), the batch aborts after the first failure, and the response reports `searches_deferred`. Every run now logs `searches_processed` and `duration_ms` (also new response fields). The lookback boundary is now timezone-aware UTC. Regression tests added. - **Wiki.js `update_page` without tags** - Wiki.js 2.x requires `tags` on the update mutation (the server unconditionally maps over it); every `update_page(page_id, content=...)` call without tags failed with `Cannot read properties of undefined (reading 'map')` — this silently broke the consolidation service's page-update path too. The client now preserves the page's current tags when the caller does not supply any. - **Wiki.js listing completeness under pre-filter limits** - Observed live: `pages.list(limit=100)` returned 43 pages while 140 existed (`limit=500` returned all) — Wiki.js applies the limit BEFORE its own visibility filtering, so "fewer pages than requested" does not mean the listing is complete and the limit-growth loop stopped early, silently truncating listings (page counts, cleanups, integrity scans). The loop now grows the limit until the returned count stops increasing (fixed point), at the cost of one confirming fetch. diff --git a/src/models/consolidation.py b/src/models/consolidation.py index 5d2c7d9..b18a0f9 100644 --- a/src/models/consolidation.py +++ b/src/models/consolidation.py @@ -50,9 +50,14 @@ class ConsolidationResponse(BaseModel): volatile_cached: int = Field(default=0, description="Items cached to volatile storage") files_queued: int = Field(default=0, description="Files queued for Paperless") prefetch_registered: int = Field(default=0, description="Prefetch patterns registered") + searches_deferred: int = Field( + default=0, + description="Searches left unprocessed for the next run because the generation LLM was unavailable" + ) errors: List[str] = Field(default=[], description="Error messages") results: List[ConsolidationResult] = Field(description="Per-search results") dry_run: bool = Field(description="Whether this was a dry run") + duration_ms: float = Field(default=0.0, description="Run duration in milliseconds") class MemoryRouteClassification(BaseModel): diff --git a/src/services/consolidation_service.py b/src/services/consolidation_service.py index e79a27a..e6fdf7f 100644 --- a/src/services/consolidation_service.py +++ b/src/services/consolidation_service.py @@ -13,7 +13,8 @@ This service: """ import logging import json -from datetime import datetime, timedelta +import time +from datetime import datetime, timedelta, timezone from typing import List, Dict, Any, Optional from src.clients.neo4j_client import Neo4jClient @@ -32,6 +33,18 @@ from src.config import Settings logger = logging.getLogger(__name__) +class ConsolidationLLMUnavailableError(Exception): + """ + The generation LLM produced no output (infrastructure failure). + + Raised instead of silently returning an empty classification so the + caller can leave the affected SearchQuery nodes UNPROCESSED for the + next run. Marking them processed on LLM failure permanently drains + the consolidation queue with zero output (the exact failure mode that + made every run log 'No unprocessed searches found' in production). + """ + + class ConsolidationService: """ Service for consolidating knowledge from search results. @@ -77,7 +90,8 @@ class ConsolidationService: Returns: ConsolidationResponse with processing results """ - logger.info(f"Starting knowledge consolidation") + run_start = time.time() + logger.info("Starting knowledge consolidation") logger.info(f"Limits: process={process_limit}, lookback={lookback_days}d, min_web={min_web_results}") if dry_run: logger.warning("DRY RUN MODE - will not create wiki pages") @@ -86,7 +100,12 @@ class ConsolidationService: unprocessed = await self._find_unprocessed_searches(lookback_days, process_limit) if not unprocessed: - logger.info("No unprocessed searches found") + duration_ms = (time.time() - run_start) * 1000 + logger.info( + f"Consolidation run complete: searches_processed=0 " + f"searches_deferred=0 duration_ms={duration_ms:.0f} " + f"(no unprocessed searches found)" + ) return ConsolidationResponse( total_found=0, processed_count=0, @@ -95,7 +114,8 @@ class ConsolidationService: entities_added=0, errors=[], results=[], - dry_run=dry_run + dry_run=dry_run, + duration_ms=duration_ms ) logger.info(f"Found {len(unprocessed)} unprocessed searches") @@ -108,9 +128,10 @@ class ConsolidationService: total_volatile_cached = 0 total_files_queued = 0 total_prefetch_registered = 0 + searches_deferred = 0 errors: List[str] = [] - for search in unprocessed: + for index, search in enumerate(unprocessed): try: result = await self._process_search( search=search, @@ -132,6 +153,22 @@ class ConsolidationService: if not dry_run: await self._mark_search_processed(search['id']) + except ConsolidationLLMUnavailableError as e: + # Infrastructure failure: the generation LLM is unavailable. + # Do NOT consume the search - leave it (and the rest of this + # batch) unprocessed so the next run retries. Consuming + # searches here is what silently drained the queue in + # production ('No unprocessed searches found' with zero + # pages ever created). + searches_deferred = len(unprocessed) - index + error_msg = ( + f"Generation LLM unavailable ({e}); deferring " + f"{searches_deferred} search(es) to the next run" + ) + logger.error(error_msg) + errors.append(error_msg) + break + except Exception as e: error_msg = f"Search {search['id'][:8]}: {str(e)}" logger.error(f"Failed to process search: {error_msg}", exc_info=True) @@ -148,6 +185,7 @@ class ConsolidationService: # Build response processed_count = len([r for r in results if not r.error]) + duration_ms = (time.time() - run_start) * 1000 response = ConsolidationResponse( total_found=len(unprocessed), @@ -158,13 +196,16 @@ class ConsolidationService: volatile_cached=total_volatile_cached, files_queued=total_files_queued, prefetch_registered=total_prefetch_registered, + searches_deferred=searches_deferred, errors=errors, results=results, - dry_run=dry_run + dry_run=dry_run, + duration_ms=duration_ms ) logger.info( - f"Consolidation complete: {processed_count}/{len(unprocessed)} searches, " + f"Consolidation run complete: searches_processed={processed_count} " + f"searches_deferred={searches_deferred} duration_ms={duration_ms:.0f} | " f"{total_pages_created} pages created, {total_pages_updated} updated, " f"{total_entities_added} entities, {total_volatile_cached} volatile, " f"{total_files_queued} files, {total_prefetch_registered} prefetch" @@ -180,7 +221,9 @@ class ConsolidationService: """ Find unprocessed SearchQuery nodes from Neo4j. """ - lookback_date = datetime.now() - timedelta(days=lookback_days) + # UTC-aware: sq.timestamp is stored via Neo4j datetime() (UTC), and a + # naive local isoformat would be misread as UTC by datetime($param). + lookback_date = datetime.now(timezone.utc) - timedelta(days=lookback_days) query = """ MATCH (sq:SearchQuery {processed: false}) @@ -1084,9 +1127,16 @@ JSON:""" temperature=0.0 ) + # generate_text returns None on any transport/HTTP failure (e.g. + # model missing, Ollama down) and Ollama never legitimately + # returns an empty completion for this prompt: both mean the LLM + # is unavailable, NOT that there is nothing to route. Raise so + # the search is retried next run instead of being consumed. if not response: - logger.warning("Empty response from Ollama for classification") - return MemoryRoutingResult() + raise ConsolidationLLMUnavailableError( + f"no output from generation model " + f"'{self.settings.ollama_llm_model}' for classification" + ) # Extract JSON array from response response_clean = response.strip() @@ -1140,7 +1190,12 @@ JSON:""" ) return result + except ConsolidationLLMUnavailableError: + # Infrastructure failure: propagate so the search is NOT consumed + raise except json.JSONDecodeError as e: + # The model responded but with unparseable output: consume the + # search (empty routing) to avoid retrying a bad prompt forever. logger.error(f"Failed to parse classification response as JSON: {e}") return MemoryRoutingResult() except Exception as e: diff --git a/tests/test_consolidation_repair.py b/tests/test_consolidation_repair.py new file mode 100644 index 0000000..08e2fd2 --- /dev/null +++ b/tests/test_consolidation_repair.py @@ -0,0 +1,224 @@ +""" +Offline regression tests for the consolidation-loop repair (Phase C item 4). + +Production failure mode being locked in: +- The generation LLM was unavailable (OLLAMA_MODEL env collision made every + /api/generate call 400), classification returned empty, and the loop + STILL marked every SearchQuery processed - permanently draining the queue + with zero output. Every subsequent 30-minute run then logged + 'No unprocessed searches found'. + +The repaired behavior: +- LLM-infrastructure failure raises ConsolidationLLMUnavailableError, the + affected searches stay UNPROCESSED (retried next run), and the run reports + searches_deferred. +- Successful classification still consumes searches. +- Every run logs searches_processed and duration_ms. + +All clients are mocked - no shared services are contacted. +""" + +import json +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from src.services.consolidation_service import ( + ConsolidationLLMUnavailableError, + ConsolidationService, +) + +TEST_USER = "llm_tester" + + +def _search_row(search_id: str, query: str = "test query", web_count: int = 3): + return { + "id": search_id, + "query": query, + "user": TEST_USER, + "timestamp": "2026-07-14T00:00:00Z", + "total_results": web_count, + "web_count": web_count, + "keywords": ["test"], + } + + +def _web_result_row(url: str = "https://example.com/a"): + return { + "url": url, + "title": "Example", + "content": "Example content about the query.", + "rank": 1, + "rrf_score": 0.5, + } + + +def _make_service(searches, ollama_response): + """ConsolidationService with a scripted Neo4j and Ollama.""" + neo4j = AsyncMock() + executed = [] + + async def fake_query(cypher, params=None): + executed.append((cypher, params or {})) + if "processed: false" in cypher: + return searches + if "FOUND]->(wr:WebResult)" in cypher: + return [_web_result_row()] + return [] + + neo4j.execute_query = AsyncMock(side_effect=fake_query) + + ollama = AsyncMock() + ollama.generate_text = AsyncMock(return_value=ollama_response) + + wiki = AsyncMock() + wiki.get_taxonomy_structure = AsyncMock(return_value={}) + + settings = MagicMock() + settings.ollama_llm_model = "gemma4:e2b" + + service = ConsolidationService( + neo4j=neo4j, ollama=ollama, wiki=wiki, settings=settings + ) + return service, executed + + +def _mark_processed_calls(executed): + return [(c, p) for c, p in executed if "SET sq.processed = true" in c] + + +class TestLLMUnavailableDoesNotConsumeSearches: + @pytest.mark.asyncio + async def test_searches_stay_unprocessed_when_llm_returns_none(self): + """generate_text -> None (transport/HTTP failure): defer, don't consume.""" + searches = [_search_row("aaaa1111"), _search_row("bbbb2222")] + service, executed = _make_service(searches, ollama_response=None) + + response = await service.consolidate_knowledge() + + assert response.total_found == 2 + assert response.processed_count == 0 + assert response.searches_deferred == 2 + assert response.errors and "unavailable" in response.errors[0].lower() + # THE regression guard: no search was marked processed + assert _mark_processed_calls(executed) == [] + + @pytest.mark.asyncio + async def test_searches_stay_unprocessed_when_llm_returns_empty(self): + """An empty completion is an infra anomaly, not 'nothing to route'.""" + service, executed = _make_service([_search_row("cccc3333")], "") + + response = await service.consolidate_knowledge() + + assert response.searches_deferred == 1 + assert _mark_processed_calls(executed) == [] + + @pytest.mark.asyncio + async def test_classification_raises_on_no_output(self): + service, _ = _make_service([], None) + with pytest.raises(ConsolidationLLMUnavailableError): + await service._classify_web_results_unified( + query="q", web_results=[_web_result_row()], keywords=[], user=TEST_USER + ) + + @pytest.mark.asyncio + async def test_batch_aborts_after_first_llm_failure(self): + """When the LLM is down it is down for all searches: one probe, then stop.""" + searches = [_search_row(f"id{i}", web_count=5) for i in range(5)] + service, executed = _make_service(searches, ollama_response=None) + + response = await service.consolidate_knowledge() + + assert response.searches_deferred == 5 + # Only the first search's classification was attempted + assert service.ollama.generate_text.await_count == 1 + + +class TestSuccessfulRunsStillConsume: + @pytest.mark.asyncio + async def test_valid_classification_marks_processed(self): + classification = json.dumps([{ + "url": "https://example.com/a", + "title": "Example", + "route_type": "skip", + "confidence": 0.9, + "reason": "low value", + }]) + service, executed = _make_service([_search_row("dddd4444")], classification) + + response = await service.consolidate_knowledge() + + assert response.total_found == 1 + assert response.processed_count == 1 + assert response.searches_deferred == 0 + marked = _mark_processed_calls(executed) + assert len(marked) == 1 + assert marked[0][1]["search_id"] == "dddd4444" + + @pytest.mark.asyncio + async def test_unparseable_output_still_consumes_search(self): + """Model responded with junk: consume (avoid retrying a bad prompt forever).""" + service, executed = _make_service( + [_search_row("eeee5555")], "not json at all" + ) + + response = await service.consolidate_knowledge() + + assert response.searches_deferred == 0 + assert len(_mark_processed_calls(executed)) == 1 + + @pytest.mark.asyncio + async def test_skipped_low_web_search_still_consumed(self): + """Insufficient web results: intentionally consumed (existing behavior).""" + service, executed = _make_service( + [_search_row("ffff6666", web_count=0)], None + ) + + response = await service.consolidate_knowledge() + + assert response.searches_deferred == 0 + assert len(_mark_processed_calls(executed)) == 1 + # LLM never called for a skipped search + service.ollama.generate_text.assert_not_awaited() + + +class TestRunLogging: + @pytest.mark.asyncio + async def test_logs_searches_processed_and_duration(self, caplog): + service, _ = _make_service([], None) + + with caplog.at_level("INFO"): + response = await service.consolidate_knowledge() + + assert response.duration_ms >= 0 + run_logs = [r.message for r in caplog.records + if "Consolidation run complete" in r.message] + assert run_logs, "every run must emit the run-complete log line" + assert "searches_processed=0" in run_logs[0] + assert "duration_ms=" in run_logs[0] + + @pytest.mark.asyncio + async def test_logs_on_deferred_run(self, caplog): + service, _ = _make_service([_search_row("gggg7777")], None) + + with caplog.at_level("INFO"): + response = await service.consolidate_knowledge() + + assert response.duration_ms >= 0 + run_logs = [r.message for r in caplog.records + if "Consolidation run complete" in r.message] + assert run_logs + assert "searches_deferred=1" in run_logs[0] + + @pytest.mark.asyncio + async def test_lookback_parameter_is_utc_aware(self): + """The lookback boundary must be timezone-aware UTC (Neo4j datetime() + interprets naive strings as UTC, shifting the window on CET hosts).""" + service, executed = _make_service([], None) + + await service.consolidate_knowledge(lookback_days=7) + + find_calls = [(c, p) for c, p in executed if "processed: false" in c] + assert find_calls + lookback = find_calls[0][1]["lookback_date"] + assert "+00:00" in lookback or lookback.endswith("Z")