homelab-codex-ws/jobs/documents-ingest/tests/test_cyclic_ingest.py

"""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}
MAIL_BODY_OK = {"mails_scanned": 1, "missing_file": 0, "read_errors": 0, "parse_errors": 0,
                "body_empty": 0, "mails_chunked": 1, "chunks_total": 2, "chunks_inserted": 2,
                "chunks_newsletter_flagged": 0, "chunks_already_embedded": 0,
                "chunks_conflict_skipped": 0, "chunks_errors": 0}


def _patch_stages(monkeypatch, *, ollama_up=True, adapter=None, chunk=None, summarize=None,
                   embed_summaries=None, mail_body=None, backlog=0, mail_archive=True):
    calls = {"adapter": 0, "chunk_embed": 0, "summarize": 0, "embed_summaries": 0,
             "mail_body": 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_mail_body_run(**kwargs):
        calls["mail_body"] += 1
        if mail_body is not None and isinstance(mail_body, Exception):
            raise mail_body
        return mail_body if mail_body is not None else dict(MAIL_BODY_OK)

    async def _fake_backlog(dsn):
        return backlog

    class _FakeMailBodyModule:
        run = staticmethod(_fake_mail_body_run)

    monkeypatch.setattr(cyclic_ingest, "mail_body_ingest", _FakeMailBodyModule)
    # The archive root check is a real filesystem probe; pin it so the suite does not depend
    # on whether the host running the tests happens to have a mail archive.
    monkeypatch.setattr(cyclic_ingest.Path, "is_dir", lambda self: mail_archive)
    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,
                         "mail_body": 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_mail_stage_drains_only_unchunked_envelopes(self, monkeypatch):
        # The handoff from mail-imap-sync is the unchunked-envelope queue, not a timestamp.
        captured = {}

        async def _capture(**kwargs):
            captured.update(kwargs)
            return dict(MAIL_BODY_OK)

        _patch_stages(monkeypatch, ollama_up=True)
        monkeypatch.setattr(cyclic_ingest.mail_body_ingest, "run", staticmethod(_capture))
        await run_cyclic(
            dsn="dsn", paperless_url="http://paperless", paperless_token="tok",
            anthropic_api_key="key", apply=True,
        )
        assert captured["only_unchunked"] is True
        assert captured["sources"] == ("gmail", "fastmail")
        assert captured["apply"] is True

    async def test_mail_stage_is_skipped_when_ollama_is_down(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 calls["mail_body"] == 0
        assert result["mail_body"]["skipped"] is True
        assert result["mail_skip_reason"] == "ollama_unreachable"
        assert result["failed"] is False

    async def test_mail_stage_is_skipped_when_the_package_is_not_installed(self, monkeypatch):
        # PIHA's venv may predate mail-body-ingest; that must not stop the whole wrapper.
        calls = _patch_stages(monkeypatch, ollama_up=True)
        monkeypatch.setattr(cyclic_ingest, "mail_body_ingest", None)
        result = await run_cyclic(
            dsn="dsn", paperless_url="http://paperless", paperless_token="tok",
            anthropic_api_key="key", apply=True,
        )
        assert result["mail_skip_reason"] == "mail_body_ingest_not_installed"
        assert result["failed"] is False
        assert calls["adapter"] == 1

    async def test_mail_stage_is_skipped_without_an_archive(self, monkeypatch):
        _patch_stages(monkeypatch, ollama_up=True, mail_archive=False)
        result = await run_cyclic(
            dsn="dsn", paperless_url="http://paperless", paperless_token="tok",
            anthropic_api_key="key", apply=True,
        )
        assert result["mail_skip_reason"] == "archive_root_missing"
        assert result["failed"] is False

    async def test_mail_stage_is_skipped_when_no_sources_are_configured(self, monkeypatch):
        _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, mail_sources=(),
        )
        assert result["mail_skip_reason"] == "no_mail_sources_configured"

    async def test_per_mail_parse_errors_do_not_fail_the_wrapper(self, monkeypatch):
        # Looser than mail-body-ingest's own CLI predicate, on purpose: one permanently
        # unparseable mail must not keep last_success frozen and light KbIngestStale forever.
        flaky = dict(MAIL_BODY_OK, mails_scanned=2, parse_errors=1)
        _patch_stages(monkeypatch, ollama_up=True, mail_body=flaky)
        result = await run_cyclic(
            dsn="dsn", paperless_url="http://paperless", paperless_token="tok",
            anthropic_api_key="key", apply=True,
        )
        assert result["mail_body"]["failed"] is False
        assert result["failed"] is False

    async def test_unbalanced_mail_stats_do_fail_the_wrapper(self, monkeypatch):
        _patch_stages(monkeypatch, ollama_up=True,
                      mail_body=dict(MAIL_BODY_OK, mails_scanned=9))
        result = await run_cyclic(
            dsn="dsn", paperless_url="http://paperless", paperless_token="tok",
            anthropic_api_key="key", apply=True,
        )
        assert result["mail_body"]["failed"] is True
        assert result["failed"] is True

    async def test_silent_chunk_conflict_fails_the_wrapper(self, monkeypatch):
        # An insert that no-op'd is a defect in the code, not bad input.
        broken = dict(MAIL_BODY_OK, chunks_inserted=1, chunks_conflict_skipped=1)
        _patch_stages(monkeypatch, ollama_up=True, mail_body=broken)
        result = await run_cyclic(
            dsn="dsn", paperless_url="http://paperless", paperless_token="tok",
            anthropic_api_key="key", apply=True,
        )
        assert result["mail_body"]["failed"] is True

    async def test_mail_stage_exception_is_isolated(self, monkeypatch):
        calls = _patch_stages(monkeypatch, ollama_up=True,
                              mail_body=RuntimeError("archive vanished"))
        result = await run_cyclic(
            dsn="dsn", paperless_url="http://paperless", paperless_token="tok",
            anthropic_api_key="key", apply=True,
        )
        assert result["mail_body"]["failed"] is True
        assert result["failed"] is True
        assert calls["summarize"] == 1      # later stages still ran

    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,
                         "mail_body": 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},
            "mail_body": {"stats": dict(MAIL_BODY_OK), "failed": False, "skipped": False, "error": None},
            "mail_skip_reason": None,
            "embed_backlog": embed_backlog,
            "failed": failed,
        }

    def test_mail_metrics_are_published(self, tmp_path):
        metrics = build_metrics(self._result(failed=False), now_ts=1000.0,
                                prom_path=tmp_path / "kb-ingest.prom")
        assert metrics["kb_ingest_mail_chunks_inserted"] == 2
        assert metrics["kb_ingest_mail_parse_errors"] == 0
        assert metrics["kb_ingest_mail_skipped"] == 0

    def test_tolerated_mail_parse_errors_are_still_published(self, tmp_path):
        # Tolerated by the exit code, but not invisible — a rising count is a real signal.
        result = self._result(failed=False)
        result["mail_body"]["stats"] = dict(MAIL_BODY_OK, parse_errors=2, missing_file=1)
        metrics = build_metrics(result, now_ts=1000.0, prom_path=tmp_path / "kb-ingest.prom")
        assert metrics["kb_ingest_mail_parse_errors"] == 3

    def test_skipped_mail_stage_reports_zero_chunks_and_a_skip_flag(self, tmp_path):
        result = self._result(failed=False)
        result["mail_body"] = {"stats": None, "failed": False, "skipped": True, "error": None}
        metrics = build_metrics(result, now_ts=1000.0, prom_path=tmp_path / "kb-ingest.prom")
        assert metrics["kb_ingest_mail_skipped"] == 1
        assert metrics["kb_ingest_mail_chunks_inserted"] == 0

    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