"""Cyclic ingest wrapper -- module 5 phase 3, plan step 5 (kb/phases/kb-m5-faza3.md, §7). Orchestrates one run of the recurring ingest pipeline for `kb-ingest.timer` on PIHA: paperless_adapter.run() -- new source='paperless' envelopes -> chunk_embed.run() -- new document_chunk rows (bge-m3 via Ollama@SOLARIA) -> mail_body_ingest.run(only_unchunked=True) -- chunks+embeds mail the incremental IMAP sync fetched (faza mailowa Krok 7) -> summarize.run_summarize(backend='anthropic') -- new document_summary rows -> summarize.run_embed_summaries() -- embeds those summaries (bge-m3) The mail stage is the consumer end of `jobs/mail-imap-sync`, which runs on its own hourly timer and deliberately does no GPU work (recon kb/audits/mail-sync-2026-08-06.md §3.2). The handoff between them needs no queue of its own: an envelope with no `document_chunk` rows IS the queue, so an interrupted run, a mail fetched while SOLARIA slept, or one inserted with a backdated header all stay pending until they have chunks, with no state to reconcile. The stage is soft-imported -- a venv without `mail-body-ingest` installed skips it and says so, rather than failing to start the whole wrapper. The two embed stages are gated on a pre-flight Ollama reachability probe (`GET /api/tags`) because SOLARIA has `availability_target: medium` (planned power-off, plan §1.3): "Ollama unreachable" is skipped, not failed -- new chunks/summaries are picked up by the next tick since both embed passes are idempotent. Without the fourth stage, new summaries would sit with `embedding IS NULL` and stay invisible to `cascade_query` (module 5 phase 3 step 4, `retrieval.py`), which filters on `embedding IS NOT NULL` -- so the write pass and the embed pass both run every tick, gated by the same probe. A real failure anywhere else -- Paperless unreachable, a DB error, a non-zero job error counter, a broken stats-balance invariant, the Anthropic API failing -- is NOT tolerated: each stage's pass/fail predicate mirrors that job's own `main()` exit-code check 1:1 (see `_adapter_failed` / `_chunk_embed_failed` / `_summarize_failed` / `_embed_summaries_failed` below -- duplicated deliberately, since each job's real predicate lives inline in argparse wiring, not as an importable function; a test pins each one against its source job). Stages are isolated, not fail-fast: an earlier stage failing doesn't skip later ones, mirroring the per-row isolation the underlying jobs already use (a stale Paperless token this run shouldn't also block chunk-embedding of documents already adapted in prior runs). The wrapper's own exit code is 1 if ANY stage failed, 0 otherwise (Ollama-skips don't count). Writes a Prometheus textfile-collector `.prom` file (plan §7.2) alongside a structured log line; `kb_ingest_last_success_timestamp` is carried forward from any previous file when this run fails, so a transient failure doesn't reset the `KbIngestStale` alert's clock to zero. This module only ever calls the existing job functions -- no changes to paperless_adapter.py / chunk_embed.py / summarize.py, and no new CLI flags on them. """ from __future__ import annotations import argparse import asyncio import os import re import sys import time from pathlib import Path from typing import Optional import aiohttp import asyncpg import structlog from documents_ingest import chunk_embed, paperless_adapter, summarize try: # optional: only installed in venvs that also run the mail pipeline from mail_body_ingest import ingest as mail_body_ingest except ImportError: # pragma: no cover - exercised by the "not installed" test via monkeypatch mail_body_ingest = None _log = structlog.get_logger(__name__) # Wrapper always runs on PIHA -- Ollama lives on SOLARIA, reached over Tailscale (plan §7.1). DEFAULT_OLLAMA_URL = "http://solaria:11434" # plan §2 decision 3, resolved 2026-07-17: claude-haiku-4-5 is the compilation-track model; # cascade_query's default summary_model matches this. DEFAULT_SUMMARY_MODEL = "claude-haiku-4-5" DEFAULT_EMBED_MODEL = "bge-m3" DEFAULT_OLLAMA_PROBE_TIMEOUT = 5.0 DEFAULT_PROM_PATH = Path("/opt/homelab/state/node-exporter/kb-ingest.prom") # Mail sources whose envelopes this wrapper chunks+embeds. The archive root is PIHA's # canonical one -- this wrapper only ever runs there. DEFAULT_MAIL_SOURCES = ("gmail", "fastmail") DEFAULT_MAIL_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive") _BACKLOG_QUERY = ( "SELECT count(*) AS n FROM document_chunk WHERE excluded_reason IS NULL AND embedding IS NULL" ) PROM_METRIC_HELP: dict[str, tuple[str, str]] = { "kb_ingest_last_run_timestamp": ( "gauge", "Unix timestamp of the last kb-ingest wrapper run (success or failure).", ), "kb_ingest_last_success_timestamp": ( "gauge", "Unix timestamp of the last kb-ingest run with no hard failure " "(Ollama-offline skips do not count as failure).", ), "kb_ingest_last_exit_code": ("gauge", "Exit code of the last kb-ingest wrapper run."), "kb_ingest_documents_inserted": ( "gauge", "New envelope rows inserted by the Paperless adapter stage in the last run.", ), "kb_ingest_chunks_inserted": ( "gauge", "New document_chunk rows inserted by the chunk/embed stage in the last run " "(0 if skipped -- Ollama unreachable).", ), "kb_ingest_summaries_inserted": ( "gauge", "New document_summary rows inserted by the summarize stage in the last run.", ), "kb_ingest_embed_skipped": ( "gauge", "1 if the embed stages (chunk + summary) were skipped this run because " "Ollama@SOLARIA was unreachable, 0 otherwise.", ), "kb_ingest_embed_backlog": ( "gauge", "Active document_chunk rows (excluded_reason IS NULL) still missing an embedding.", ), "kb_ingest_mail_chunks_inserted": ( "gauge", "New document_chunk rows inserted by the mail body stage in the last run.", ), "kb_ingest_mail_parse_errors": ( "gauge", "Mails the body stage could not read or parse in the last run. Tolerated (they do " "not fail the wrapper) but tracked -- a rising number is a real signal.", ), "kb_ingest_mail_skipped": ( "gauge", "1 if the mail body stage was skipped this run (Ollama unreachable, the " "mail-body-ingest package not installed, or no .eml archive on this host).", ), } _PROM_LINE_RE = re.compile(r"^(kb_ingest_\w+)\s+([0-9.eE+-]+)\s*$") def _stage_result(stats: Optional[dict] = None, failed: bool = False, skipped: bool = False, error: Optional[str] = None) -> dict: return {"stats": stats, "failed": failed, "skipped": skipped, "error": error} def _adapter_failed(stats: dict) -> bool: """Mirrors paperless_adapter.main()'s exit predicate.""" balanced = stats["fetched"] == stats["already_in_db"] + stats["inserted"] + stats["errors"] return stats["errors"] > 0 or not balanced def _chunk_embed_failed(stats: dict) -> bool: """Mirrors chunk_embed.main()'s exit predicate.""" balanced = ( stats["documents_fetched"] == stats["empty_content"] + stats["documents_chunked"] and stats["chunks_total"] == ( stats["chunks_already_embedded"] + stats["chunks_inserted"] + stats["chunks_junk_flagged"] + stats["chunks_conflict_skipped"] + stats["chunks_errors"] ) ) return stats["chunks_errors"] > 0 or stats["chunks_conflict_skipped"] > 0 or not balanced def _summarize_failed(stats: dict) -> bool: """Mirrors summarize.main()'s write-mode exit predicate.""" balanced = stats["documents_fetched"] == ( stats["duplicates_skipped"] + stats["no_active_chunks"] + stats["already_summarized"] + stats["summarized"] + stats["llm_errors"] ) return stats["llm_errors"] > 0 or not balanced def _embed_summaries_failed(stats: dict) -> bool: """Mirrors summarize.main()'s --embed-summaries exit predicate.""" balanced = stats["embedded"] + stats["errors"] == stats["summaries_fetched"] return stats["errors"] > 0 or not balanced def _mail_body_failed(stats: dict) -> bool: """Deliberately NOT a 1:1 mirror of mail_body_ingest.main() -- the one predicate in this wrapper that is looser than its job's own. That CLI exits 1 on any `parse_errors`/`missing_file`/`read_errors`, which is right for a supervised slice run: across 225k mails a handful is expected and worth a look. On a recurring unattended tick it would be corrosive. A single permanently-unparseable mail would mark every run failed, `kb_ingest_last_success_timestamp` would stop advancing, and KbIngestStale would fire after 48h and stay lit -- reporting nothing but one bad mail nobody can fix. That is the same "an alert that always fires is not an alert" failure the schedule change in kb-ingest.timer exists to avoid. So per-mail failures are tolerated and published as `kb_ingest_mail_parse_errors` instead. What still fails the wrapper is what means the stage itself is broken: unbalanced stats, or `chunks_conflict_skipped` (an insert silently no-op'd -- a real defect, not bad input). """ balanced = ( stats["mails_scanned"] == ( stats["missing_file"] + stats["read_errors"] + stats["parse_errors"] + stats["body_empty"] + stats["mails_chunked"] ) and stats["chunks_total"] == ( stats["chunks_inserted"] + stats["chunks_newsletter_flagged"] + stats["chunks_already_embedded"] + stats["chunks_conflict_skipped"] + stats["chunks_errors"] ) ) return stats["chunks_conflict_skipped"] > 0 or not balanced def _mail_parse_errors(stats: dict) -> int: return stats["missing_file"] + stats["read_errors"] + stats["parse_errors"] async def probe_ollama(ollama_url: str, timeout: float = DEFAULT_OLLAMA_PROBE_TIMEOUT) -> bool: """`GET /api/tags` reachability check (plan §7.1). Any exception or non-2xx means "treat as offline, skip the embed stages this tick" -- never raises.""" try: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session: async with session.get(f"{ollama_url}/api/tags") as resp: return resp.status == 200 except Exception: return False async def fetch_embed_backlog(dsn: str) -> Optional[int]: """Read-only count for `kb_ingest_embed_backlog` (plan §7.2). Returns None (metric omitted, not zeroed) on a DB error -- a real DB outage is already reflected by every other stage's failure, this is purely observability.""" try: conn = await asyncpg.connect(dsn) except Exception: return None try: row = await conn.fetchrow(_BACKLOG_QUERY) return int(row["n"]) except Exception: return None finally: await conn.close() async def run_cyclic( dsn: str, paperless_url: str, paperless_token: str, ollama_url: str = DEFAULT_OLLAMA_URL, anthropic_api_key: Optional[str] = None, summary_model: str = DEFAULT_SUMMARY_MODEL, embed_model: str = DEFAULT_EMBED_MODEL, tags_vocab_path: Path = summarize.DEFAULT_TAGS_VOCAB_PATH, apply: bool = False, ollama_probe_timeout: float = DEFAULT_OLLAMA_PROBE_TIMEOUT, mail_sources: tuple[str, ...] = DEFAULT_MAIL_SOURCES, mail_archive_root: Path = DEFAULT_MAIL_ARCHIVE_ROOT, ) -> dict: """Runs all five stages, isolating failures per stage (an earlier failure never skips a later stage). Returns a dict with one entry per stage (`adapter`, `chunk_embed`, `mail_body`, `summarize`, `embed_summaries`, each `_stage_result()`-shaped), plus `ollama_up`, `embed_backlog`, and the aggregate `failed` bool the caller should exit non-zero on.""" ollama_up = await probe_ollama(ollama_url, timeout=ollama_probe_timeout) adapter_stage = _stage_result() try: stats = await paperless_adapter.run( dsn=dsn, paperless_url=paperless_url, paperless_token=paperless_token, apply=apply, ) adapter_stage = _stage_result(stats=stats, failed=_adapter_failed(stats)) except Exception as exc: _log.error("adapter_exception", error=str(exc)) adapter_stage = _stage_result(failed=True, error=str(exc)) chunk_stage = _stage_result(skipped=not ollama_up) if ollama_up: try: stats = await chunk_embed.run( dsn=dsn, ollama_url=ollama_url, model=embed_model, apply=apply, ) chunk_stage = _stage_result(stats=stats, failed=_chunk_embed_failed(stats)) except Exception as exc: _log.error("chunk_embed_exception", error=str(exc)) chunk_stage = _stage_result(failed=True, error=str(exc)) else: _log.warning("skip.chunk_embed", reason="ollama_unreachable") # Mail body stage -- the consumer end of the incremental sync (recon §3.2). It drains the # queue that IS the data: envelopes with no document_chunk rows. Nothing to reconcile and # nothing to reset, so a tick that skips it (GPU asleep) costs only latency. mail_stage = _stage_result(skipped=True) mail_skip_reason = None if not mail_sources: mail_skip_reason = "no_mail_sources_configured" elif not ollama_up: mail_skip_reason = "ollama_unreachable" elif mail_body_ingest is None: mail_skip_reason = "mail_body_ingest_not_installed" elif not Path(mail_archive_root).is_dir(): mail_skip_reason = "archive_root_missing" if mail_skip_reason is None: try: stats = await mail_body_ingest.run( dsn=dsn, archive_root=Path(mail_archive_root), ollama_url=ollama_url, model=embed_model, sources=tuple(mail_sources), only_unchunked=True, apply=apply, ) mail_stage = _stage_result(stats=stats, failed=_mail_body_failed(stats)) except Exception as exc: _log.error("mail_body_exception", error=str(exc)) mail_stage = _stage_result(failed=True, error=str(exc)) else: _log.warning("skip.mail_body", reason=mail_skip_reason) mail_stage = _stage_result(skipped=True, error=None) summarize_stage = _stage_result() try: stats = await summarize.run_summarize( dsn=dsn, backend_name="anthropic", model=summary_model, anthropic_api_key=anthropic_api_key, tags_vocab_path=tags_vocab_path, apply=apply, ) summarize_stage = _stage_result(stats=stats, failed=_summarize_failed(stats)) except Exception as exc: _log.error("summarize_exception", error=str(exc)) summarize_stage = _stage_result(failed=True, error=str(exc)) embed_summaries_stage = _stage_result(skipped=not ollama_up) if ollama_up: try: stats = await summarize.run_embed_summaries( dsn=dsn, ollama_url=ollama_url, embed_model=embed_model, model_filter=summary_model, apply=apply, ) embed_summaries_stage = _stage_result(stats=stats, failed=_embed_summaries_failed(stats)) except Exception as exc: _log.error("embed_summaries_exception", error=str(exc)) embed_summaries_stage = _stage_result(failed=True, error=str(exc)) else: _log.warning("skip.embed_summaries", reason="ollama_unreachable") embed_backlog = await fetch_embed_backlog(dsn) failed = any( stage["failed"] for stage in (adapter_stage, chunk_stage, mail_stage, summarize_stage, embed_summaries_stage) ) return { "ollama_up": ollama_up, "adapter": adapter_stage, "chunk_embed": chunk_stage, "mail_body": mail_stage, "mail_skip_reason": mail_skip_reason, "summarize": summarize_stage, "embed_summaries": embed_summaries_stage, "embed_backlog": embed_backlog, "failed": failed, } def _read_prev_metric(path: Path, name: str) -> Optional[float]: if not path.exists(): return None try: text = path.read_text() except OSError: return None for line in text.splitlines(): m = _PROM_LINE_RE.match(line) if m and m.group(1) == name: return float(m.group(2)) return None def build_metrics(result: dict, now_ts: float, prom_path: Path) -> dict: """Turns a `run_cyclic()` result into the flat name->value map `render_prom()` expects. `kb_ingest_last_success_timestamp` carries forward the previous file's value on failure -- never resets to 0/now, or `KbIngestStale` would flap on every transient error.""" failed = result["failed"] prev_success = _read_prev_metric(prom_path, "kb_ingest_last_success_timestamp") last_success = now_ts if not failed else (prev_success if prev_success is not None else 0.0) metrics = { "kb_ingest_last_run_timestamp": now_ts, "kb_ingest_last_success_timestamp": last_success, "kb_ingest_last_exit_code": 1 if failed else 0, "kb_ingest_documents_inserted": (result["adapter"]["stats"] or {}).get("inserted", 0), "kb_ingest_chunks_inserted": (result["chunk_embed"]["stats"] or {}).get("chunks_inserted", 0), "kb_ingest_summaries_inserted": (result["summarize"]["stats"] or {}).get("summarized", 0), "kb_ingest_embed_skipped": 0 if result["ollama_up"] else 1, "kb_ingest_mail_chunks_inserted": (result["mail_body"]["stats"] or {}).get("chunks_inserted", 0), "kb_ingest_mail_parse_errors": _mail_parse_errors(result["mail_body"]["stats"]) if result["mail_body"]["stats"] else 0, "kb_ingest_mail_skipped": 1 if result["mail_body"]["skipped"] else 0, } if result["embed_backlog"] is not None: metrics["kb_ingest_embed_backlog"] = result["embed_backlog"] return metrics def render_prom(metrics: dict) -> str: lines = [] for name, (mtype, help_text) in PROM_METRIC_HELP.items(): if name not in metrics: continue lines.append(f"# HELP {name} {help_text}") lines.append(f"# TYPE {name} {mtype}") lines.append(f"{name} {metrics[name]}") return "\n".join(lines) + "\n" def write_prom_atomic(path: Path, content: str) -> None: """tmp-write + rename (plan §7.2) -- node_exporter's textfile collector never observes a half-written file.""" path.parent.mkdir(parents=True, exist_ok=True) tmp_path = path.with_name(f"{path.name}.tmp{os.getpid()}") tmp_path.write_text(content) tmp_path.replace(path) def main() -> None: parser = argparse.ArgumentParser( description="Cyclic KB ingest wrapper: paperless_adapter -> chunk_embed -> " "summarize -> embed-summaries (module 5 phase 3, plan §7). Tolerates " "Ollama@SOLARIA being offline; does not tolerate anything else failing." ) parser.add_argument("--dsn", default=os.environ.get("KB_DSN"), help="asyncpg DSN for kb-postgres (or set KB_DSN env var)") parser.add_argument("--paperless-url", default=os.environ.get("PAPERLESS_URL", paperless_adapter.DEFAULT_PAPERLESS_URL), help="Paperless base URL (or set PAPERLESS_URL env var)") parser.add_argument("--paperless-token", default=os.environ.get("PAPERLESS_API_TOKEN"), help="Paperless API token (or set PAPERLESS_API_TOKEN env var) -- never logged") parser.add_argument("--ollama-url", default=os.environ.get("OLLAMA_URL", DEFAULT_OLLAMA_URL), help=f"Ollama base URL (default: {DEFAULT_OLLAMA_URL}, or set OLLAMA_URL)") parser.add_argument("--anthropic-api-key", default=os.environ.get("ANTHROPIC_API_KEY"), help="Anthropic API key (or set ANTHROPIC_API_KEY env var) -- never logged") parser.add_argument("--summary-model", default=os.environ.get("KB_SUMMARY_MODEL", DEFAULT_SUMMARY_MODEL), help=f"Summarizing model, both write+embed filter (default: {DEFAULT_SUMMARY_MODEL})") parser.add_argument("--embed-model", default=os.environ.get("KB_EMBED_MODEL", DEFAULT_EMBED_MODEL), help=f"Ollama embedding model, chunks+summaries (default: {DEFAULT_EMBED_MODEL})") parser.add_argument("--tags-vocab", type=Path, default=summarize.DEFAULT_TAGS_VOCAB_PATH, help="Path to tags-vocab.yaml") parser.add_argument("--mail-sources", default=os.environ.get("KB_MAIL_SOURCES", ",".join(DEFAULT_MAIL_SOURCES)), metavar="A,B", help=f"Envelope sources the mail body stage drains " f"(default: {','.join(DEFAULT_MAIL_SOURCES)}, or set KB_MAIL_SOURCES). " f"Empty disables the stage.") parser.add_argument("--mail-archive-root", type=Path, default=Path(os.environ.get("MAIL_ARCHIVE_ROOT", str(DEFAULT_MAIL_ARCHIVE_ROOT))), help=f"Mail .eml archive root (default: {DEFAULT_MAIL_ARCHIVE_ROOT}). " f"The stage is skipped if this directory does not exist.") parser.add_argument("--prom-path", type=Path, default=Path(os.environ.get("KB_INGEST_PROM_PATH", str(DEFAULT_PROM_PATH))), help=f"Textfile-collector output path (default: {DEFAULT_PROM_PATH})") parser.add_argument("--ollama-probe-timeout", type=float, default=DEFAULT_OLLAMA_PROBE_TIMEOUT, help=f"Seconds before the Ollama reachability probe gives up (default: {DEFAULT_OLLAMA_PROBE_TIMEOUT})") parser.add_argument("--apply", action="store_true", help="Actually write to Paperless-read/DB/call the LLM+Ollama. Default is dry-run.") args = parser.parse_args() if not args.dsn: _log.error("missing_dsn", hint="pass --dsn or set KB_DSN") sys.exit(1) if not args.paperless_token: _log.error("missing_paperless_token", hint="pass --paperless-token or set PAPERLESS_API_TOKEN") sys.exit(1) if args.apply and not args.anthropic_api_key: _log.error("missing_anthropic_api_key", hint="pass --anthropic-api-key or set ANTHROPIC_API_KEY") sys.exit(1) result = asyncio.run(run_cyclic( dsn=args.dsn, paperless_url=args.paperless_url, paperless_token=args.paperless_token, ollama_url=args.ollama_url, anthropic_api_key=args.anthropic_api_key, summary_model=args.summary_model, embed_model=args.embed_model, tags_vocab_path=args.tags_vocab, apply=args.apply, ollama_probe_timeout=args.ollama_probe_timeout, mail_sources=tuple(s.strip() for s in args.mail_sources.split(",") if s.strip()), mail_archive_root=args.mail_archive_root, )) now_ts = time.time() metrics = build_metrics(result, now_ts, args.prom_path) write_prom_atomic(args.prom_path, render_prom(metrics)) mode = "APPLY" if args.apply else "DRY-RUN" _log.info( "cyclic_summary", mode=mode, ollama_up=result["ollama_up"], failed=result["failed"], adapter=result["adapter"]["stats"], adapter_error=result["adapter"]["error"], chunk_embed=result["chunk_embed"]["stats"], chunk_embed_skipped=result["chunk_embed"]["skipped"], chunk_embed_error=result["chunk_embed"]["error"], mail_body=result["mail_body"]["stats"], mail_body_skipped=result["mail_body"]["skipped"], mail_skip_reason=result["mail_skip_reason"], mail_body_error=result["mail_body"]["error"], summarize=result["summarize"]["stats"], summarize_error=result["summarize"]["error"], embed_summaries=result["embed_summaries"]["stats"], embed_summaries_skipped=result["embed_summaries"]["skipped"], embed_summaries_error=result["embed_summaries"]["error"], embed_backlog=result["embed_backlog"], ) sys.exit(1 if result["failed"] else 0) if __name__ == "__main__": main()