jpmschweitzer/library-desk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

fix: stop consolidation from consuming searches when the LLM is down

Browse Source 
ROOT CAUSE (investigated read-only against prod): the production
container sets OLLAMA_MODEL=nomic-embed-text (the embedding model), which
the pre-rename generation setting also read, so every consolidation
/api/generate call failed with HTTP 400 ('"nomic-embed-text" does not
support generate' - confirmed in prod logs and by a direct Ollama probe).
_classify_web_results_unified swallowed that as an empty classification,
and consolidate_knowledge marked EVERY SearchQuery processed anyway -
permanently draining the queue with zero pages ever created. Live Neo4j
shows 197/200 SearchQuery nodes processed=true with no output; every
subsequent 30-minute run then logged 'No unprocessed searches found'.
The label/tenant scoping was NOT at fault: persistence writes both the
tenant label and the plain :SearchQuery label the loop matches on.

The model resolution itself was already fixed in Phase A (94482bc,
ollama_llm_model / OLLAMA_LLM_MODEL). This commit repairs the pipeline
defect that masked it:

- LLM infrastructure failure (no output from generate) now raises
  ConsolidationLLMUnavailableError instead of returning an empty routing
- consolidate_knowledge leaves those searches UNPROCESSED for the next
  run, aborts the rest of the batch (the LLM is down for all of them),
  and reports searches_deferred
- unparseable-but-present model output is still consumed (avoids
  retrying a bad prompt forever); low-web skips unchanged
- every run logs 'Consolidation run complete: searches_processed=N
  searches_deferred=M duration_ms=X'; both fields added to the response
- lookback boundary is now timezone-aware UTC (Neo4j datetime() reads
  naive strings as UTC, shifting the window on CET hosts)

10 new offline regression tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L
This commit is contained in:
jpmschweitzer
2026-07-14 12:26:14 +02:00
co-authored by Claude Fable 5
parent 191a8be6c5
commit 51f9ce08ec
4 changed files with 295 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/models/consolidation.py
+5
View File
@@ -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):
src/services/consolidation_service.py
+65 -10
View File
@@ -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: