From cb8b60fee6deda9ed7592d74f316732e16fc2337 Mon Sep 17 00:00:00 2001 From: oskar Date: Fri, 17 Jul 2026 15:05:18 +0200 Subject: [PATCH] =?UTF-8?q?feat(kb):=20faza=203=20krok=205=20=E2=80=94=20c?= =?UTF-8?q?ykliczny=20ingest=20(systemd=20timer)=20+=20alerting?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit documents-ingest-cyclic (jobs/documents-ingest/src/documents_ingest/cyclic_ingest.py): orkiestruje paperless_adapter -> chunk_embed -> summarize(--backend anthropic, claude-haiku-4-5) -> summarize(--embed-summaries) bez zmian w samych jobach. Ollama@SOLARIA (availability_target: medium) jest tolerowana offline: probe GET /api/tags przed obu etapami embed, brak -> pominięcie, nie fail (oba embed passy idempotentne, nadrobią się na kolejnym ticku). Czwarty etap (embed-summaries) dopisany ponad plan §7.1 (który wymieniał tylko 3 kroki) — bez niego nowe streszczenia miałyby embedding=NULL i byłyby niewidoczne dla cascade_query (bramka kroku 4, WHERE embedding IS NOT NULL); potwierdzone z Oskarem. Predykaty pass/fail każdego etapu 1:1 z exit-checkiem danego joba (chunks_errors, llm_errors, stats-balance itd.) — etapy izolowane, nie fail-fast (wcześniejszy fail nie blokuje kolejnych, tak jak joby izolują błędy per wiersz). Metryki .prom (atomowy zapis, last_success_timestamp trzymany z poprzedniego pliku przy failu) do /opt/homelab/state/node-exporter/kb-ingest.prom. 36 nowych testów (202/202 pakietu). systemd (jobs/documents-ingest/systemd/): pierwszy systemd-timer w repo — kb-ingest.timer (OnCalendar=*-*-* 03:30, Persistent=true, plan §7.1) + kb-ingest.service (host-level, User oskar, EnvironmentFile /opt/homelab/kb/.env) + kb-ingest-run.sh (log per-run do /opt/homelab/logs/kb-ingest/, konwencja repo). Instalacja i sekrety udokumentowane w README (Faza 3 krok 5) — instalacja na PIHA dopiero po merge. fleet-prometheus (rules/kb-ingest.yml): KbIngestStale (>172800s od last_success, critical) + KbEmbedBacklogGrowing (backlog>0 przez 72h, warning) — dostawa istniejącym torem brain-watchdog->Telegram, bez Alertmanagera (konwencja liveness.yml). node_exporter: owner_node vps -> per-host (service.yaml) + wpis + override (--collector.textfile.directory, bez nowego mountu — czyta przez istniejący /:/host:ro) + topology.yaml dla PIHA. Domyka pozycję z docs/backlog.md "stability-agent / node_exporter owner_node single, biegaja wielomiejscowo -> per-host" (połowę — node_exporter; stability-agent zostaje osobnym follow-upem) w ramach paczki B inwentaryzacji monitoringu dla PIHA. Test end-to-end na żywo na PIHA (2× --apply, po potwierdzeniu z Oskarem): pierwszy run złapał 5 dokumentów faktycznie nowych w Paperless (nieoczekiwane, niezwiązane z tym taskiem) -> 82 nowe chunki (2 ocr_junk), 5 nowych streszczeń, 5 embeddingów streszczeń, 0 błędów, metryki zapisane. Drugi run: pełna idempotencja, wszystko 0. ANTHROPIC_API_KEY dodany przez Oskara ręcznie do /opt/homelab/kb/.env (nigdy nie logowany/generowany). Co dalej: prawdziwa instalacja systemd (services.yaml już przygotowany, po merge), zdecydowanie czy stability-agent też idzie na per-host przy okazji. Co-Authored-By: Claude Sonnet 5 --- .../node_exporter/docker-compose.override.yml | 15 + hosts/piha/services.yaml | 14 + inventory/topology.yaml | 1 + jobs/documents-ingest/README.md | 108 +++++ jobs/documents-ingest/pyproject.toml | 1 + .../src/documents_ingest/cyclic_ingest.py | 396 ++++++++++++++++ .../documents-ingest/systemd/kb-ingest-run.sh | 20 + .../systemd/kb-ingest.service | 19 + jobs/documents-ingest/systemd/kb-ingest.timer | 14 + .../tests/test_cyclic_ingest.py | 428 ++++++++++++++++++ services/fleet-prometheus/rules/kb-ingest.yml | 38 ++ services/node_exporter/service.yaml | 7 +- 12 files changed, 1060 insertions(+), 1 deletion(-) create mode 100644 hosts/piha/runtime/node_exporter/docker-compose.override.yml create mode 100644 jobs/documents-ingest/src/documents_ingest/cyclic_ingest.py create mode 100755 jobs/documents-ingest/systemd/kb-ingest-run.sh create mode 100644 jobs/documents-ingest/systemd/kb-ingest.service create mode 100644 jobs/documents-ingest/systemd/kb-ingest.timer create mode 100644 jobs/documents-ingest/tests/test_cyclic_ingest.py create mode 100644 services/fleet-prometheus/rules/kb-ingest.yml diff --git a/hosts/piha/runtime/node_exporter/docker-compose.override.yml b/hosts/piha/runtime/node_exporter/docker-compose.override.yml new file mode 100644 index 0000000..b261a10 --- /dev/null +++ b/hosts/piha/runtime/node_exporter/docker-compose.override.yml @@ -0,0 +1,15 @@ +# PIHA-specific override for node_exporter. +# +# WHY: KB module 5 phase 3 step 5 (docs/kb/modules/05-faza3-plan.md §7.2) needs the +# textfile collector so kb-ingest's systemd timer can publish +# kb_ingest_last_success_timestamp / kb_ingest_last_exit_code / kb_ingest_embed_backlog +# etc. for fleet-prometheus to scrape and alert on. +# +# NO NEW MOUNT: the base compose already binds `/:/host:ro,rslave` — the textfile +# directory is read through that existing mount, not a new volume. Command lists are +# replaced wholesale by Compose overrides (not merged), so both flags are restated here. +services: + node_exporter: + command: + - '--path.rootfs=/host' + - '--collector.textfile.directory=/host/opt/homelab/state/node-exporter' diff --git a/hosts/piha/services.yaml b/hosts/piha/services.yaml index 6cdafc4..d9d3987 100644 --- a/hosts/piha/services.yaml +++ b/hosts/piha/services.yaml @@ -74,6 +74,20 @@ services: # .env (TAILSCALE_BIND_IP) lives alongside the compose file; stateless, no data path config_path: services/llm-gateway + node_exporter: + role: metrics-exporter + deployment_model: docker-compose + exposure: local-only + offline_required: true + depends_on: + local: [] + external: [] + runtime: + # textfile collector reads /opt/homelab/state/node-exporter (module 5 phase 3 step 5, + # docs/kb/modules/05-faza3-plan.md §7.2 — kb-ingest.prom) via the existing /:/host:ro + # mount, see hosts/piha/runtime/node_exporter/docker-compose.override.yml. + data_path: /opt/homelab/state/node-exporter + kb-postgres: role: kb-database # KB spine: Postgres 16 + pgvector (always-on) deployment_model: docker-compose diff --git a/inventory/topology.yaml b/inventory/topology.yaml index 83dffe0..9de5a4a 100644 --- a/inventory/topology.yaml +++ b/inventory/topology.yaml @@ -49,6 +49,7 @@ nodes: - vikunja # Task management (vikunja + postgres), public via npm - kb-postgres # KB spine: Postgres 16 + pgvector, port 5433 (always-on) - llm-gateway # FastAPI router -> Ollama @ SOLARIA (Tailscale-internal :8080) + - node_exporter # per-host (also vps) — textfile collector for kb-ingest metrics solaria: roles: diff --git a/jobs/documents-ingest/README.md b/jobs/documents-ingest/README.md index 689fcc8..bf3fce1 100644 --- a/jobs/documents-ingest/README.md +++ b/jobs/documents-ingest/README.md @@ -535,3 +535,111 @@ cd jobs/documents-ingest && pytest all envelopes, cascade stage-1-narrows-stage-2, an envelope whose summary exists but has no active chunks, N larger than the number of summarized envelopes, the no-summaries short-circuit (stage 2 never queried), and both query entry points embedding exactly once. + +--- + +## Phase 3 step 5 — cyclic ingest (`documents-ingest-cyclic`) + systemd timer + +Module 5, phase 3, plan step 5 (`docs/kb/modules/05-faza3-plan.md`, §7). Orchestrates one +run of the recurring ingest pipeline: `paperless_adapter.run()` (new `source='paperless'` +envelopes) → `chunk_embed.run()` (new `document_chunk` rows) → +`summarize.run_summarize(backend='anthropic')` (new `document_summary` rows, +`model='claude-haiku-4-5'` — plan §2 decision 3) → `summarize.run_embed_summaries()` +(embeds those summaries). All four are the same job functions used elsewhere in this +package, called directly — no changes to `paperless_adapter.py` / `chunk_embed.py` / +`summarize.py`, no new CLI flags on them. + +### Ollama-offline tolerance + +SOLARIA has `availability_target: medium` (planned power-off, plan §1.3). The wrapper +probes `GET {OLLAMA_URL}/api/tags` before the two embed stages (chunk embedding, summary +embedding); unreachable means **skip, not fail** — both embed passes are idempotent, so +new chunks/summaries left unembedded this tick are picked up whole on the next one. A +growing backlog is what `kb_ingest_embed_backlog` + the `KbEmbedBacklogGrowing` alert are +for, not this wrapper's exit code. + +Anything else failing **is** a hard failure: Paperless unreachable, a DB error, a non-zero +job error counter, a broken stats-balance invariant, the Anthropic API failing. Each +stage's pass/fail predicate mirrors that job's own `main()` exit check 1:1 (see +`cyclic_ingest.py`'s module docstring). Stages are isolated, not fail-fast — an earlier +stage failing never skips a later one, mirroring the per-row isolation the underlying jobs +already use. + +### Usage + +```bash +# Dry run (default) — same idempotent counting as every other job in this family, no writes: +documents-ingest-cyclic --dsn postgresql://kb:@localhost:5433/kb \ + --paperless-token --anthropic-api-key + +# Real run (what the timer invokes): +documents-ingest-cyclic --dsn ... --paperless-token ... --anthropic-api-key ... --apply +``` + +`--dsn`/`--paperless-token`/`--anthropic-api-key` also read from `KB_DSN` / +`PAPERLESS_API_TOKEN` / `ANTHROPIC_API_KEY` env vars — never logged. `--ollama-url` +defaults to `http://solaria:11434` (this wrapper always runs on PIHA, unlike +`chunk_embed`/`summarize`'s own CLI defaults which assume co-location with Ollama). + +### Metrics (Prometheus textfile collector) + +Every run — success or failure — writes `--prom-path` +(default `/opt/homelab/state/node-exporter/kb-ingest.prom`) atomically (tmp + rename): + +| Metric | Meaning | +|---|---| +| `kb_ingest_last_run_timestamp` | Unix ts of the last run, success or failure | +| `kb_ingest_last_success_timestamp` | Unix ts of the last run with no hard failure — carried forward from the previous file on a failing run, never reset to 0/now | +| `kb_ingest_last_exit_code` | 0 or 1 | +| `kb_ingest_documents_inserted` | New envelope rows this run | +| `kb_ingest_chunks_inserted` | New document_chunk rows this run (0 if the embed stage was skipped) | +| `kb_ingest_summaries_inserted` | New document_summary rows this run | +| `kb_ingest_embed_skipped` | 1 if Ollama was unreachable this run (both embed stages skipped), 0 otherwise | +| `kb_ingest_embed_backlog` | Active chunks (`excluded_reason IS NULL`) still missing an embedding | + +Scraped by fleet-prometheus via node_exporter's textfile collector on PIHA +(`hosts/piha/runtime/node_exporter/docker-compose.override.yml`); alert rules in +`services/fleet-prometheus/rules/kb-ingest.yml`. + +### Install (PIHA) + +1. Dedicated venv (per plan §7.1 — not the ad hoc rsync-to-`/tmp` pattern used for the + one-shot jobs elsewhere in this README; this is a permanent, recurring installation): + ```bash + python3 -m venv /opt/homelab/kb/venv + /opt/homelab/kb/venv/bin/pip install -e packages/kb-mail -e jobs/documents-ingest + ``` + (run from a checkout of this repo on PIHA — the checkout is used as an install source + only, per CLAUDE.md's "deploy-only" rule; no development happens there). +2. Secrets in `/opt/homelab/kb/.env` (already holds `PAPERLESS_API_TOKEN`; add + `KB_DSN=postgresql://kb:@localhost:5433/kb` and `ANTHROPIC_API_KEY=`), + `chmod 600`, never in Git. +3. Copy `jobs/documents-ingest/systemd/kb-ingest-run.sh` to `/opt/homelab/kb/` and + `chmod +x` it. +4. Copy (or symlink) `kb-ingest.service` and `kb-ingest.timer` to `/etc/systemd/system/`, + then: + ```bash + systemctl daemon-reload + systemctl enable --now kb-ingest.timer + ``` +5. Verify: `systemctl list-timers kb-ingest.timer`, `journalctl -u kb-ingest.service`, + `/opt/homelab/logs/kb-ingest/run-YYYYMMDD.log`, and + `/opt/homelab/state/node-exporter/kb-ingest.prom` after the first run (manual + `systemctl start kb-ingest.service` to trigger one immediately without waiting for + 03:30). + +### Tests + +```bash +pip install -e packages/kb-mail/ +pip install -e jobs/documents-ingest/ +cd jobs/documents-ingest && pytest +``` + +`tests/test_cyclic_ingest.py` — pure unit tests, no DB or real HTTP/Ollama/Anthropic; every +stage function and the Ollama probe are monkeypatched. Covers: each stage's failure +predicate (pinned against its source job's own exit check), the Ollama-down skip path +(chunk_embed/embed_summaries never even called), stage isolation (an earlier stage failing +never skips a later one, whether via a failed predicate or a raised exception), `.prom` +rendering, atomic write, `last_success_timestamp` carry-forward across a failing run, and +`main()`'s CLI guardrails + exit-code propagation. diff --git a/jobs/documents-ingest/pyproject.toml b/jobs/documents-ingest/pyproject.toml index b930283..379ecd7 100644 --- a/jobs/documents-ingest/pyproject.toml +++ b/jobs/documents-ingest/pyproject.toml @@ -20,6 +20,7 @@ documents-ingest = "documents_ingest.extractor:main" documents-ingest-paperless = "documents_ingest.paperless_adapter:main" documents-ingest-embed = "documents_ingest.chunk_embed:main" documents-ingest-summarize = "documents_ingest.summarize:main" +documents-ingest-cyclic = "documents_ingest.cyclic_ingest:main" [tool.setuptools.packages.find] where = ["src"] diff --git a/jobs/documents-ingest/src/documents_ingest/cyclic_ingest.py b/jobs/documents-ingest/src/documents_ingest/cyclic_ingest.py new file mode 100644 index 0000000..d0f086f --- /dev/null +++ b/jobs/documents-ingest/src/documents_ingest/cyclic_ingest.py @@ -0,0 +1,396 @@ +"""Cyclic ingest wrapper -- module 5 phase 3, plan step 5 (docs/kb/modules/05-faza3-plan.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) + -> summarize.run_summarize(backend='anthropic') -- new document_summary rows + -> summarize.run_embed_summaries() -- embeds those summaries (bge-m3) + +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 + +_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") + +_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.", + ), +} + +_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 + + +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, +) -> dict: + """Runs all four stages, isolating failures per stage (an earlier failure never skips + a later stage). Returns a dict with one entry per stage (`adapter`, `chunk_embed`, + `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") + + 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, summarize_stage, embed_summaries_stage) + ) + + return { + "ollama_up": ollama_up, + "adapter": adapter_stage, + "chunk_embed": chunk_stage, + "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, + } + 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("--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, + )) + + 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"], + 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() diff --git a/jobs/documents-ingest/systemd/kb-ingest-run.sh b/jobs/documents-ingest/systemd/kb-ingest-run.sh new file mode 100755 index 0000000..c2f0bb3 --- /dev/null +++ b/jobs/documents-ingest/systemd/kb-ingest-run.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# kb-ingest-run.sh — thin launcher for kb-ingest.service (module 5 phase 3 step 5, +# docs/kb/modules/05-faza3-plan.md §7.1). +# +# All sequencing/tolerance/exit-code logic lives in documents-ingest-cyclic (Python, +# jobs/documents-ingest/src/documents_ingest/cyclic_ingest.py) — this script only computes +# a date-stamped log path and redirects, per the repo convention that a long run must never +# log only to a terminal/journal (lesson from the 2026-07 backfill that lost 4999 rows to a +# closed tmux — see the kb-piha-job-execution memory note). +# +# Installed at /opt/homelab/kb/kb-ingest-run.sh (copied from repo at deploy time), invoked +# by kb-ingest.service. `exec` at the end preserves documents-ingest-cyclic's exit code as +# this script's own, so systemd sees the real success/failure. +set -euo pipefail + +LOG_DIR="/opt/homelab/logs/kb-ingest" +mkdir -p "$LOG_DIR" +LOG_FILE="${LOG_DIR}/run-$(date +%Y%m%d).log" + +exec /opt/homelab/kb/venv/bin/documents-ingest-cyclic --apply >> "$LOG_FILE" 2>&1 diff --git a/jobs/documents-ingest/systemd/kb-ingest.service b/jobs/documents-ingest/systemd/kb-ingest.service new file mode 100644 index 0000000..236baff --- /dev/null +++ b/jobs/documents-ingest/systemd/kb-ingest.service @@ -0,0 +1,19 @@ +# kb-ingest.service — module 5 phase 3 step 5 (docs/kb/modules/05-faza3-plan.md §7.1). +# First systemd unit in this repo: deliberately host-level, not a container — the job +# needs simultaneous LAN (Paperless), local DB (kb-postgres@PIHA), and Tailscale (Ollama@ +# SOLARIA) access; containerizing it buys nothing here. +# +# Install: copy or symlink this file (and kb-ingest.timer) to /etc/systemd/system/, then +# `systemctl daemon-reload`. See ../README.md "Cyclic ingest (systemd timer)" for the full +# install sequence (venv, .env, kb-ingest-run.sh). +[Unit] +Description=KB cyclic ingest (Paperless adapter + chunk embed + summarize) +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +User=oskar +EnvironmentFile=/opt/homelab/kb/.env +WorkingDirectory=/opt/homelab/kb +ExecStart=/opt/homelab/kb/kb-ingest-run.sh diff --git a/jobs/documents-ingest/systemd/kb-ingest.timer b/jobs/documents-ingest/systemd/kb-ingest.timer new file mode 100644 index 0000000..cf45c5b --- /dev/null +++ b/jobs/documents-ingest/systemd/kb-ingest.timer @@ -0,0 +1,14 @@ +# kb-ingest.timer — module 5 phase 3 step 5 (docs/kb/modules/05-faza3-plan.md §7.1). +# Daily at 03:30, not hourly: the plan fixes this schedule explicitly (low-traffic window, +# and the underlying jobs are full-corpus re-scans each tick — cheap at 186 documents +# today, but daily keeps headroom as the corpus grows). Persistent=true catches up after a +# PIHA reboot instead of silently skipping the day's run. +[Unit] +Description=Run kb-ingest.service daily (module 5 phase 3 step 5) + +[Timer] +OnCalendar=*-*-* 03:30:00 +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/jobs/documents-ingest/tests/test_cyclic_ingest.py b/jobs/documents-ingest/tests/test_cyclic_ingest.py new file mode 100644 index 0000000..496eb3c --- /dev/null +++ b/jobs/documents-ingest/tests/test_cyclic_ingest.py @@ -0,0 +1,428 @@ +"""Unit tests for the cyclic ingest wrapper — no DB, no real HTTP, no real +Ollama/Anthropic/Paperless. Every stage function (paperless_adapter.run, chunk_embed.run, +summarize.run_summarize, summarize.run_embed_summaries) and probe_ollama/fetch_embed_backlog +are monkeypatched with fakes; only cyclic_ingest's own orchestration, predicates, and +.prom rendering are under test.""" +from __future__ import annotations + +import sys + +import pytest + +from documents_ingest import cyclic_ingest +from documents_ingest.cyclic_ingest import ( + PROM_METRIC_HELP, + _adapter_failed, + _chunk_embed_failed, + _embed_summaries_failed, + _read_prev_metric, + _summarize_failed, + build_metrics, + main, + probe_ollama, + render_prom, + run_cyclic, + write_prom_atomic, +) + + +# --------------------------------------------------------------------------- +# Per-stage failure predicates — pinned against each job's own main() exit check +# --------------------------------------------------------------------------- + +class TestAdapterFailed: + def test_balanced_no_errors_passes(self): + stats = {"fetched": 3, "already_in_db": 1, "inserted": 2, "errors": 0} + assert _adapter_failed(stats) is False + + def test_any_error_fails(self): + stats = {"fetched": 3, "already_in_db": 1, "inserted": 1, "errors": 1} + assert _adapter_failed(stats) is True + + def test_imbalance_fails_even_with_zero_errors(self): + stats = {"fetched": 3, "already_in_db": 1, "inserted": 1, "errors": 0} + assert _adapter_failed(stats) is True + + +class TestChunkEmbedFailed: + def _stats(self, **overrides): + stats = { + "documents_fetched": 2, "empty_content": 0, "documents_chunked": 2, + "chunks_total": 4, "chunks_already_embedded": 0, "chunks_inserted": 4, + "chunks_junk_flagged": 0, "chunks_conflict_skipped": 0, "chunks_errors": 0, + } + stats.update(overrides) + return stats + + def test_balanced_passes(self): + assert _chunk_embed_failed(self._stats()) is False + + def test_chunks_errors_fails(self): + assert _chunk_embed_failed(self._stats(chunks_errors=1, chunks_inserted=3)) is True + + def test_conflict_skipped_fails(self): + assert _chunk_embed_failed(self._stats(chunks_conflict_skipped=1, chunks_inserted=3)) is True + + def test_doc_balance_break_fails(self): + assert _chunk_embed_failed(self._stats(documents_fetched=99)) is True + + +class TestSummarizeFailed: + def _stats(self, **overrides): + stats = { + "documents_fetched": 2, "duplicates_skipped": 0, "no_active_chunks": 0, + "already_summarized": 0, "summarized": 2, "llm_errors": 0, + } + stats.update(overrides) + return stats + + def test_balanced_passes(self): + assert _summarize_failed(self._stats()) is False + + def test_llm_errors_fails(self): + assert _summarize_failed(self._stats(llm_errors=1, summarized=1)) is True + + def test_imbalance_fails(self): + assert _summarize_failed(self._stats(documents_fetched=5)) is True + + +class TestEmbedSummariesFailed: + def test_balanced_passes(self): + assert _embed_summaries_failed({"summaries_fetched": 2, "embedded": 2, "errors": 0}) is False + + def test_errors_fails(self): + assert _embed_summaries_failed({"summaries_fetched": 2, "embedded": 1, "errors": 1}) is True + + def test_imbalance_fails(self): + assert _embed_summaries_failed({"summaries_fetched": 5, "embedded": 2, "errors": 0}) is True + + +# --------------------------------------------------------------------------- +# probe_ollama — reachability probe, must never raise +# --------------------------------------------------------------------------- + +class _FakeProbeResponse: + def __init__(self, status): + self.status = status + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class _FakeProbeSession: + def __init__(self, status=200, raise_on_get=None): + self._status = status + self._raise_on_get = raise_on_get + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + def get(self, url): + if self._raise_on_get is not None: + raise self._raise_on_get + return _FakeProbeResponse(self._status) + + +class TestProbeOllama: + async def test_200_is_up(self, monkeypatch): + monkeypatch.setattr( + "documents_ingest.cyclic_ingest.aiohttp.ClientSession", + lambda *a, **kw: _FakeProbeSession(status=200), + ) + assert await probe_ollama("http://solaria:11434") is True + + async def test_non_200_is_down(self, monkeypatch): + monkeypatch.setattr( + "documents_ingest.cyclic_ingest.aiohttp.ClientSession", + lambda *a, **kw: _FakeProbeSession(status=500), + ) + assert await probe_ollama("http://solaria:11434") is False + + async def test_connection_error_is_down_not_raised(self, monkeypatch): + monkeypatch.setattr( + "documents_ingest.cyclic_ingest.aiohttp.ClientSession", + lambda *a, **kw: _FakeProbeSession(raise_on_get=ConnectionError("refused")), + ) + assert await probe_ollama("http://solaria:11434") is False + + +# --------------------------------------------------------------------------- +# run_cyclic — sequencing, Ollama-offline tolerance, stage isolation +# --------------------------------------------------------------------------- + +ADAPTER_OK = {"fetched": 1, "already_in_db": 0, "inserted": 1, "source_mail_linked": 0, + "empty_content": 0, "errors": 0} +CHUNK_OK = {"documents_fetched": 1, "empty_content": 0, "documents_chunked": 1, + "chunks_total": 1, "chunks_already_embedded": 0, "chunks_inserted": 1, + "chunks_junk_flagged": 0, "chunks_conflict_skipped": 0, "chunks_errors": 0, + "embed_calls": 1, "embed_seconds_total": 0.1} +SUMMARIZE_OK = {"documents_fetched": 1, "duplicates_skipped": 0, "no_active_chunks": 0, + "already_summarized": 0, "summarized": 1, "llm_errors": 0, + "documents_mapreduce": 0, "tags_truncated": 0} +EMBED_SUMMARIES_OK = {"summaries_fetched": 1, "embedded": 1, "errors": 0} + + +def _patch_stages(monkeypatch, *, ollama_up=True, adapter=None, chunk=None, summarize=None, + embed_summaries=None, backlog=0): + calls = {"adapter": 0, "chunk_embed": 0, "summarize": 0, "embed_summaries": 0} + + async def _fake_probe(url, timeout=5.0): + return ollama_up + + async def _fake_adapter_run(**kwargs): + calls["adapter"] += 1 + return adapter if adapter is not None else dict(ADAPTER_OK) + + async def _fake_chunk_run(**kwargs): + calls["chunk_embed"] += 1 + if chunk is not None and isinstance(chunk, Exception): + raise chunk + return chunk if chunk is not None else dict(CHUNK_OK) + + async def _fake_run_summarize(**kwargs): + calls["summarize"] += 1 + if summarize is not None and isinstance(summarize, Exception): + raise summarize + return summarize if summarize is not None else dict(SUMMARIZE_OK) + + async def _fake_run_embed_summaries(**kwargs): + calls["embed_summaries"] += 1 + return embed_summaries if embed_summaries is not None else dict(EMBED_SUMMARIES_OK) + + async def _fake_backlog(dsn): + return backlog + + monkeypatch.setattr(cyclic_ingest, "probe_ollama", _fake_probe) + monkeypatch.setattr(cyclic_ingest.paperless_adapter, "run", _fake_adapter_run) + monkeypatch.setattr(cyclic_ingest.chunk_embed, "run", _fake_chunk_run) + monkeypatch.setattr(cyclic_ingest.summarize, "run_summarize", _fake_run_summarize) + monkeypatch.setattr(cyclic_ingest.summarize, "run_embed_summaries", _fake_run_embed_summaries) + monkeypatch.setattr(cyclic_ingest, "fetch_embed_backlog", _fake_backlog) + return calls + + +class TestRunCyclic: + async def test_all_stages_succeed_ollama_up(self, monkeypatch): + calls = _patch_stages(monkeypatch, ollama_up=True) + result = await run_cyclic( + dsn="dsn", paperless_url="http://paperless", paperless_token="tok", + anthropic_api_key="key", apply=True, + ) + assert result["failed"] is False + assert result["ollama_up"] is True + assert calls == {"adapter": 1, "chunk_embed": 1, "summarize": 1, "embed_summaries": 1} + assert result["chunk_embed"]["skipped"] is False + assert result["embed_summaries"]["skipped"] is False + assert result["embed_backlog"] == 0 + + async def test_ollama_down_skips_embed_stages_not_a_failure(self, monkeypatch): + calls = _patch_stages(monkeypatch, ollama_up=False) + result = await run_cyclic( + dsn="dsn", paperless_url="http://paperless", paperless_token="tok", + anthropic_api_key="key", apply=True, + ) + assert result["failed"] is False + assert result["ollama_up"] is False + # chunk_embed and embed_summaries must never even be invoked when Ollama is down. + assert calls["chunk_embed"] == 0 + assert calls["embed_summaries"] == 0 + assert calls["adapter"] == 1 + assert calls["summarize"] == 1 + assert result["chunk_embed"]["skipped"] is True + assert result["chunk_embed"]["stats"] is None + assert result["embed_summaries"]["skipped"] is True + + async def test_adapter_failure_does_not_skip_later_stages(self, monkeypatch): + broken_adapter = dict(ADAPTER_OK, errors=1) + calls = _patch_stages(monkeypatch, ollama_up=True, adapter=broken_adapter) + result = await run_cyclic( + dsn="dsn", paperless_url="http://paperless", paperless_token="tok", + anthropic_api_key="key", apply=True, + ) + assert result["failed"] is True + assert result["adapter"]["failed"] is True + # every other stage still ran despite the adapter failing this tick. + assert calls == {"adapter": 1, "chunk_embed": 1, "summarize": 1, "embed_summaries": 1} + assert result["chunk_embed"]["failed"] is False + assert result["summarize"]["failed"] is False + assert result["embed_summaries"]["failed"] is False + + async def test_chunk_embed_exception_is_isolated(self, monkeypatch): + calls = _patch_stages(monkeypatch, ollama_up=True, chunk=RuntimeError("ollama 500")) + result = await run_cyclic( + dsn="dsn", paperless_url="http://paperless", paperless_token="tok", + anthropic_api_key="key", apply=True, + ) + assert result["failed"] is True + assert result["chunk_embed"]["failed"] is True + assert "ollama 500" in result["chunk_embed"]["error"] + # summarize + embed_summaries still ran despite chunk_embed raising. + assert calls["summarize"] == 1 + assert calls["embed_summaries"] == 1 + assert result["summarize"]["failed"] is False + + async def test_stats_mismatch_marks_stage_failed_without_exception(self, monkeypatch): + imbalanced_summarize = dict(SUMMARIZE_OK, documents_fetched=99) + _patch_stages(monkeypatch, ollama_up=True, summarize=imbalanced_summarize) + result = await run_cyclic( + dsn="dsn", paperless_url="http://paperless", paperless_token="tok", + anthropic_api_key="key", apply=True, + ) + assert result["failed"] is True + assert result["summarize"]["failed"] is True + assert result["summarize"]["error"] is None # no exception, just a failed predicate + + +# --------------------------------------------------------------------------- +# .prom rendering + carry-forward +# --------------------------------------------------------------------------- + +class TestRenderProm: + def test_includes_help_and_type_for_every_metric(self): + metrics = {name: 1 for name in PROM_METRIC_HELP} + content = render_prom(metrics) + for name in PROM_METRIC_HELP: + assert f"# HELP {name} " in content + assert f"# TYPE {name} gauge" in content + assert f"\n{name} 1\n" in content + + def test_omits_metrics_not_present(self): + content = render_prom({"kb_ingest_last_run_timestamp": 123.0}) + assert "kb_ingest_last_run_timestamp 123.0" in content + assert "kb_ingest_embed_backlog" not in content + + +class TestWriteAndReadPromFile: + def test_write_then_read_prev_metric(self, tmp_path): + path = tmp_path / "kb-ingest.prom" + content = render_prom({"kb_ingest_last_success_timestamp": 1700000000.0}) + write_prom_atomic(path, content) + + assert path.read_text() == content + assert list(tmp_path.iterdir()) == [path] # no leftover .tmp file + + assert _read_prev_metric(path, "kb_ingest_last_success_timestamp") == 1700000000.0 + assert _read_prev_metric(path, "kb_ingest_missing_metric") is None + + def test_read_prev_metric_missing_file_returns_none(self, tmp_path): + assert _read_prev_metric(tmp_path / "nope.prom", "kb_ingest_last_success_timestamp") is None + + +class TestBuildMetrics: + def _result(self, failed, ollama_up=True, embed_backlog=0): + return { + "ollama_up": ollama_up, + "adapter": {"stats": dict(ADAPTER_OK), "failed": False, "skipped": False, "error": None}, + "chunk_embed": {"stats": dict(CHUNK_OK), "failed": False, "skipped": False, "error": None}, + "summarize": {"stats": dict(SUMMARIZE_OK), "failed": False, "skipped": False, "error": None}, + "embed_summaries": {"stats": dict(EMBED_SUMMARIES_OK), "failed": False, "skipped": False, "error": None}, + "embed_backlog": embed_backlog, + "failed": failed, + } + + def test_success_sets_last_success_to_now(self, tmp_path): + prom_path = tmp_path / "kb-ingest.prom" + metrics = build_metrics(self._result(failed=False), now_ts=1000.0, prom_path=prom_path) + assert metrics["kb_ingest_last_success_timestamp"] == 1000.0 + assert metrics["kb_ingest_last_exit_code"] == 0 + assert metrics["kb_ingest_documents_inserted"] == 1 + assert metrics["kb_ingest_embed_skipped"] == 0 + + def test_failure_carries_forward_previous_success(self, tmp_path): + prom_path = tmp_path / "kb-ingest.prom" + write_prom_atomic(prom_path, render_prom({"kb_ingest_last_success_timestamp": 500.0})) + + metrics = build_metrics(self._result(failed=True), now_ts=1000.0, prom_path=prom_path) + assert metrics["kb_ingest_last_success_timestamp"] == 500.0 + assert metrics["kb_ingest_last_run_timestamp"] == 1000.0 + assert metrics["kb_ingest_last_exit_code"] == 1 + + def test_failure_with_no_previous_file_defaults_to_zero(self, tmp_path): + prom_path = tmp_path / "kb-ingest.prom" + metrics = build_metrics(self._result(failed=True), now_ts=1000.0, prom_path=prom_path) + assert metrics["kb_ingest_last_success_timestamp"] == 0.0 + + def test_ollama_down_sets_embed_skipped(self, tmp_path): + result = self._result(failed=False, ollama_up=False) + result["chunk_embed"] = {"stats": None, "failed": False, "skipped": True, "error": None} + metrics = build_metrics(result, now_ts=1000.0, prom_path=tmp_path / "x.prom") + assert metrics["kb_ingest_embed_skipped"] == 1 + assert metrics["kb_ingest_chunks_inserted"] == 0 + + def test_embed_backlog_none_is_omitted(self, tmp_path): + metrics = build_metrics( + self._result(failed=False, embed_backlog=None), now_ts=1000.0, prom_path=tmp_path / "x.prom", + ) + assert "kb_ingest_embed_backlog" not in metrics + + +# --------------------------------------------------------------------------- +# main() — CLI-level guardrails and exit-code propagation +# --------------------------------------------------------------------------- + +class TestMainGuards: + def test_missing_dsn_exits_1(self, monkeypatch): + monkeypatch.setattr(sys, "argv", ["documents-ingest-cyclic", + "--paperless-token", "tok"]) + monkeypatch.delenv("KB_DSN", raising=False) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 1 + + def test_missing_paperless_token_exits_1(self, monkeypatch): + monkeypatch.setattr(sys, "argv", ["documents-ingest-cyclic", "--dsn", "dsn"]) + monkeypatch.delenv("PAPERLESS_API_TOKEN", raising=False) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 1 + + def test_apply_without_anthropic_key_exits_1(self, monkeypatch): + monkeypatch.setattr(sys, "argv", [ + "documents-ingest-cyclic", "--dsn", "dsn", "--paperless-token", "tok", "--apply", + ]) + monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 1 + + +class TestMainExitCodePropagation: + def test_exit_0_on_success(self, monkeypatch, tmp_path): + _patch_stages(monkeypatch, ollama_up=True) + monkeypatch.setattr(sys, "argv", [ + "documents-ingest-cyclic", "--dsn", "dsn", "--paperless-token", "tok", + "--anthropic-api-key", "key", "--prom-path", str(tmp_path / "kb-ingest.prom"), + ]) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 0 + assert (tmp_path / "kb-ingest.prom").exists() + + def test_exit_1_on_stage_failure(self, monkeypatch, tmp_path): + _patch_stages(monkeypatch, ollama_up=True, adapter=dict(ADAPTER_OK, errors=1)) + monkeypatch.setattr(sys, "argv", [ + "documents-ingest-cyclic", "--dsn", "dsn", "--paperless-token", "tok", + "--anthropic-api-key", "key", "--prom-path", str(tmp_path / "kb-ingest.prom"), + ]) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 1 + + def test_exit_0_when_ollama_down_apply_still_true(self, monkeypatch, tmp_path): + _patch_stages(monkeypatch, ollama_up=False) + monkeypatch.setattr(sys, "argv", [ + "documents-ingest-cyclic", "--dsn", "dsn", "--paperless-token", "tok", + "--anthropic-api-key", "key", "--apply", + "--prom-path", str(tmp_path / "kb-ingest.prom"), + ]) + with pytest.raises(SystemExit) as exc: + main() + assert exc.value.code == 0 + prom_content = (tmp_path / "kb-ingest.prom").read_text() + assert "kb_ingest_embed_skipped 1" in prom_content diff --git a/services/fleet-prometheus/rules/kb-ingest.yml b/services/fleet-prometheus/rules/kb-ingest.yml new file mode 100644 index 0000000..5ef6325 --- /dev/null +++ b/services/fleet-prometheus/rules/kb-ingest.yml @@ -0,0 +1,38 @@ +# fleet-prometheus kb-ingest rules — module 5 phase 3 step 5 +# (docs/kb/modules/05-faza3-plan.md §7.2). +# +# Same delivery convention as liveness.yml: no Alertmanager, these rules only make alerts +# FIRING (visible at GET /api/v1/alerts on this Prometheus instance); brain-watchdog@PIHA +# polls that API and forwards to Telegram. Do NOT add Alertmanager config here. +# +# Metrics come from documents-ingest-cyclic's Prometheus textfile-collector output +# (jobs/documents-ingest/systemd, `/opt/homelab/state/node-exporter/kb-ingest.prom`), +# scraped via node_exporter's textfile collector on PIHA — every series from that target +# already carries `node="piha"` (fleet-node job's static_configs label), same as liveness.yml. + +groups: + - name: kb-ingest + rules: + - alert: KbIngestStale + # 2 missed daily runs (OnCalendar=*-*-* 03:30) — the timer itself down, or every + # run since has hard-failed (Ollama-down alone never blocks this: the wrapper only + # skips its two embed stages then, still updating last_success_timestamp). + expr: time() - kb_ingest_last_success_timestamp{node="piha"} > 172800 + for: 5m + labels: + severity: critical + annotations: + summary: "kb-ingest stale on {{ $labels.node }}" + description: "kb-ingest.timer has not completed a successful run in over 48h on {{ $labels.node }} — check `journalctl -u kb-ingest.service` and the latest /opt/homelab/logs/kb-ingest/run-*.log." + + - alert: KbEmbedBacklogGrowing + # SOLARIA/Ollama has been unreachable (or embedding has been failing) for 3 straight + # days of ticks — the backlog itself isn't an incident (Ollama sleeps by design, + # plan §1.3), sustained non-zero for this long is. + expr: kb_ingest_embed_backlog{node="piha"} > 0 + for: 72h + labels: + severity: warning + annotations: + summary: "kb-ingest embed backlog growing on {{ $labels.node }}" + description: "kb_ingest_embed_backlog on {{ $labels.node }} has been > 0 for 72h straight — Ollama@SOLARIA may not have come back up in that window, or chunk/summary embedding has been failing on live Ollama." diff --git a/services/node_exporter/service.yaml b/services/node_exporter/service.yaml index f2743a6..f0e1f1d 100644 --- a/services/node_exporter/service.yaml +++ b/services/node_exporter/service.yaml @@ -1,5 +1,10 @@ name: node_exporter -owner_node: vps +# Deployed per-host: vps and piha (each with its own hosts//services.yaml entry + +# hosts//runtime/node_exporter/docker-compose.override.yml). Was owner_node: vps only +# until KB module 5 phase 3 step 5 (docs/kb/modules/05-faza3-plan.md §7.2) needed the +# textfile collector on PIHA — closes the docs/backlog.md "stability-agent / node_exporter +# owner_node single, biegaja wielomiejscowo -> per-host" item for node_exporter's half. +owner_node: per-host role: metrics-exporter description: > Prometheus Node Exporter — exposes host-level metrics (CPU, memory, disk,