homelab-codex-ws/jobs/documents-ingest/src/documents_ingest/cyclic_ingest.py

"""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()