feat(mail-body-ingest): circuit breaker na martwy backend embed przed Etapem B

Tolerancja pojedynczego nieudanego batcha jest słuszna, przeżycie martwej Ollamy
już nie. Parse archiwum jest jednowątkowy i wyprzedza GPU, więc przy Etapie B
(~212k kopert bez --since) padnięta Ollama przemieliłaby resztę korpusu z
prędkością parse'u, oznaczając każdy chunk jako chunks_errors — bez ani jednego
zapisu, ale kosztem ~2 h przebiegu do powtórzenia. Znany tryb awarii
Ollama@SOLARIA jest totalny (zniknięcie kontenera / network-detach, 4 incydenty,
§1.4/§7), nie częściowy, więc próg z kolejnych porażek trafia w niego od razu.

--max-embed-failures N (domyślnie 5, 0 wyłącza) → EmbedBackendUnavailableError
i exit 2, odrębny od exit 1 (który pełny korpus osiąga legalnie na pojedynczych
parse_errors — §1.5). Licznik zeruje się po udanym batchu, więc kryterium jest
"kolejnych", nie "łącznie". Przy abortcie dopychane są zaległe wpisy
entities[type=threading]: nie zależą od Ollamy, są idempotentne, a ich odtworzenie
oznaczałoby ponowny odczyt tych samych 27 GB. Nowy licznik embed_batch_failures
jest wyłącznie diagnostyczny — równania bilansu bez zmian.

Plan §9: dopisane decyzje operatora do Etapu B (plastry po 50k, breaker, pominięty
dry-run całości, hybrid default poza zakresem) + nota jak czytać exit 1 vs exit 2.

Testy: 4 nowe (trip po N kolejnych, reset po sukcesie, 0 wyłącza, flush threadingu
przy abortcie); 55 passed mail-body-ingest, 25 passed kb-retrieval. Smoke:
--limit 5 dry-run na żywym kb-postgres@PIHA — bilans domknięty, zero zapisów,
zero wywołań Ollamy.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/kb/modules/05-faza-mailowa-plan.md

@@ -589,6 +589,33 @@ mała zmiana w serwisie, nie w `packages/kb-retrieval`).
 
 **Szacunek: 1 sesja (run w tle).**
 
+### Decyzje operatora do Etapu B (2026-08-04) — przed runem
+
+Recon przed Etapem B (mirror archiwum na SOLARII żyje: 225 057 plików / 27 GB; RTT
+SOLARIA→PIHA 0,83 ms; PIHA 140 GB wolne, baza 397 MB; M1 — `NODE_TYPE=lte_node`
+na node-agencie SOLARII — zdeployowane, więc kontener Ollamy nie zniknie po
+zatrzymaniu) wykazał dwie rzeczy do rozstrzygnięcia. Decyzje:
+
+1. **Run w plastrach po 50k** (`--limit 50000 --offset 0/50k/100k/150k/200k`),
+   log per plaster, `nice`/`ionice`. Powód: brak checkpointu (restart = ponowny
+   parse od początku listy, ~1 h) + nocne wyłączanie SOLARII. Plaster ≈ 25–40 min.
+   Tempo kolejnych plastrów po obserwacji PIHA po pierwszym.
+2. **Circuit breaker w jobie: TAK** — `--max-embed-failures` (domyślnie 5),
+   abort z kodem wyjścia 2 po N kolejnych nieudanych batchach embed. Powód:
+   parse jest jednowątkowy i wyprzedza GPU, więc martwa Ollama (4 incydenty)
+   zamieniłaby 2-godzinny przebieg w 200k+ `chunks_errors` bez ani jednego
+   zapisu. Licznik zeruje się po udanym batchu.
+3. **Dry-run całości pomijamy** — idempotencja i odwracalność flag newsletterowych
+   wystarczają; ewentualna kalibracja heurystyki na dekadzie 2010–2015 po fakcie,
+   na już zapisanych flagach.
+4. Przełączenie domyślnego `mode` kb-query na `hybrid` (DoD (d)) — **poza zakresem
+   Etapu B**, osobny task po PASS regresji.
+
+Uwaga do czytania wyników: na pełnym korpusie `exit 1` jest spodziewany
+(pojedyncze `parse_errors` — §1.5 dokumentuje ~9 maili na fallbacku compat32).
+Werdyktem jest bilans i liczniki w linii `summary`, nie kod wyjścia. `exit 2`
+oznacza co innego: backend embed padł, trzeba wznowić plaster po naprawie Ollamy.
+
 ## 10. Krok 7 — IMAP/JMAP przyrostówka (zarys; szczegóły = osobny recon)
 
 Zakotwiczone w kb-00 jako etapy 3–4 (`jobs/fastmail-poller`,
@@ -621,16 +648,16 @@ Zakotwiczone w kb-00 jako etapy 3–4 (`jobs/fastmail-poller`,
 
 ## 12. Plan implementacji (kolejność = zależności)
 
-| # | Krok | Zależy od | Szacunek |
+| # | Krok | Zależy od | Szacunek | Stan | Dowód (2026-08-04) |
-|---|---|---|---|
+|---|---|---|---|---|---|
-| 0 | Chunker → `packages/kb-mail` | — | 0,5 sesji |
+| 0 | Chunker → `packages/kb-mail` | — | 0,5 sesji | **WYKONANE** | `348ce10`; `packages/kb-mail/src/kb_mail/chunking.py` + `tests/test_chunking.py` |
-| 1 | `embed_batch` w kb-retrieval | — | 0,5 sesji |
+| 1 | `embed_batch` w kb-retrieval | — | 0,5 sesji | **WYKONANE** | `51998fd`; `kb_retrieval/embed.py:61` (`embed_batch`) + `tests/test_embed.py` |
-| 2 | Job `mail-body-ingest` | 0, 1 | 2 sesje |
+| 2 | Job `mail-body-ingest` | 0, 1 | 2 sesje | **WYKONANE** | `ad0ef40` (job), `a95524c` (README), `fc5c698` (fix html_to_text); `jobs/mail-body-ingest/` + `tests/test_ingest.py` |
-| 3 | Tryb hybrid (kb-retrieval + kb-query) | — (równolegle z 2) | 1 sesja |
+| 3 | Tryb hybrid (kb-retrieval + kb-query) | — (równolegle z 2) | 1 sesja | **WYKONANE** | `a640cf1`; `kb_retrieval/retrieval.py:112` (`hybrid_retrieve`), `:195` (`hybrid_query`), `kb-query/app/main.py:119` (`mode` pattern). Uwaga: domyślny `mode` to nadal `cascade` — przełączenie to follow-up z §8, nie część Kroku 3 |
-| 4 | rsync + Etap A (12 mies.) + kalibracja | 2 | 1 sesja |
+| 4 | rsync + Etap A (12 mies.) + kalibracja | 2 | 1 sesja | **WYKONANE** | §7 „Wynik Etapu A" (run na żywo 2026-07-23); potwierdzone na żywej bazie 2026-08-04: `document_chunk` gmail = 33 871 (6 398 z embeddingiem + 27 473 `newsletter`) — zgodne co do sztuki z tabelą §7 |
-| 5 | Bramka jakościowa (eval mailowy + regresja) | 3, 4 + zapytania od operatora | 1 sesja |
+| 5 | Bramka jakościowa (eval mailowy + regresja) | 3, 4 + zapytania od operatora | 1 sesja | **WYKONANE** (PASS) | §8 „Wynik bramki"; `56f64e9` (eval + queries.yaml dla hybrid), `bce635c` (`mail_hit@3`, próg N2, werdykt PASS), `71eb264` (`--transport http`) |
-| 6 | Etap B (pełne archiwum) + regresja + obserwacja PIHA | 5 = PASS | 1 sesja |
+| 6 | Etap B (pełne archiwum) + regresja + obserwacja PIHA | 5 = PASS | 1 sesja | **OTWARTE** | Brak commitu, brak sekcji z wynikiem w tym dokumencie; żywa baza pokazuje wyłącznie wolumen Etapu A (33 871 chunków gmail vs oczekiwane ~496k), więc run bez `--since` nie był wykonany |
-| 7 | Recon przyrostówki IMAP/JMAP | — (po 6) | 1 sesja (poza DoD fazy) |
+| 7 | Recon przyrostówki IMAP/JMAP | — (po 6) | 1 sesja (poza DoD fazy) | **OTWARTE** | Brak `jobs/fastmail-poller` / `jobs/gmail-imap-poller`, brak dokumentu reconu; IMAP/JMAP występuje wyłącznie jako zarys w §10 i w `kb-00-overview.md` |
 
 **Kryterium ukończenia fazy mailowej:** (a) pełny korpus gmail zchunkowany
 (bilans domknięty, `parse_errors` na poziomie pojedynczych sztuk jak

hosts/vps/runtime/node-agent/docker-compose.override.yml

@@ -3,6 +3,18 @@ services:
     environment:
       - NODE_NAME=vps
       - CHECK_INTERVAL=60
+      # TEMPORARY mitigation (M1) for the unfiltered-prune incident
+      # (docs/incidents/2026-07-30-ollama-solaria-vanish.md §7). node-agent runs
+      # `docker container prune()` with NO filters every CHECK_INTERVAL, and the
+      # Docker API removes EVERY non-running container regardless of restart
+      # policy or compose labels — this already destroyed ollama@solaria. On VPS
+      # the loss is worse: humanai-mailer and humanai-landing have no compose
+      # definition in this repo, so a pruned container cannot be recreated.
+      # node_type is read ONLY by run_safe_cleanup() (plus two log lines), so
+      # lte_node disables cleanup and nothing else — monitoring, event shipping
+      # and action dispatch keep working.
+      # REMOVE once R1 (explicit-enumeration prune) is deployed to VPS.
+      - NODE_TYPE=lte_node
     # host network mode: node-agent on VPS shares the host's network namespace
     # so that localhost:18180 resolves to the control-plane's exposed port.
     # Without this, localhost inside the container is the container's own loopback

jobs/mail-body-ingest/README.md

@@ -38,6 +38,11 @@ mail-body-ingest --dsn postgresql://kb:<pw>@piha:5433/kb --archive-root /home/os
 # Etap A pilot — last 12 months only (plan Decyzja 9):
 mail-body-ingest --dsn ... --since 2025-07-01 --apply > mail-ingest-etapA.log 2>&1
 
+# Etap B — full archive in 50k slices (plan §9; ORDER BY id is stable, so slices are
+# reproducible, and idempotency covers their boundaries):
+nice -n 10 ionice -c2 -n7 mail-body-ingest --dsn ... --apply \
+    --limit 50000 --offset 0 > mail-ingest-etapB-0.log 2>&1
+
 # Smoke-test slice:
 mail-body-ingest --dsn ... --apply --limit 10
 ```
@@ -89,14 +94,39 @@ Any non-zero `read_errors`/`parse_errors`/`missing_file`/`chunks_errors`/
 `chunks_conflict_skipped`, or an unbalanced sum, makes the CLI exit 1 — same convention as
 `gmail-header-backfill`/`documents-ingest`'s `chunk_embed`.
 
-## Ollama-offline tolerance
+**Reading exit 1 on a full-corpus run**: it is a "look at this", not "the run failed". Across
+225k mails a handful of `parse_errors` is expected (plan §1.5 documents ~9 mails that need the
+compat32 fallback), and any one of them alone trips exit 1. The verdict is the balance and the
+counters in the `summary` line, not the exit code. Exit 2 is different — see below.
+
+## Exit codes
+
+| Code | Meaning |
+|---|---|
+| 0 | Balanced, zero errors |
+| 1 | Balanced-but-imperfect (any `parse_errors`/`missing_file`/`read_errors`/`chunks_errors`/`chunks_conflict_skipped`), an unbalanced sum, or an embedding-dimension abort |
+| 2 | `--max-embed-failures` consecutive embed batches failed — the embed backend is down; re-run once it is back |
+
+## Ollama-offline tolerance and the circuit breaker
 
 A failed `embed_batch()` call is caught per-batch (`aiohttp.ClientError` -> the whole batch,
 up to `--batch-size` chunks, counts as `chunks_errors`; the run logs a warning and continues).
 Those chunks never enter the idempotency set, so a later re-run retries them automatically —
-no separate checkpointing needed. Only a wrong embedding dimension
+no separate checkpointing needed.
-(`EmbeddingDimensionError`) aborts the entire run, since that would otherwise silently index
+
-a vector that doesn't match `document_chunk.embedding VECTOR(1024)`.
+Tolerating a *flaky* backend is right; surviving a *dead* one is not. The archive is parsed
+single-threaded ahead of the GPU, so on a full-corpus run (Etap B) a dead Ollama would let the
+job chew through 200k+ mails at parse speed, mark every chunk `chunks_errors`, and throw away a
+multi-hour pass. `--max-embed-failures` (default 5, `0` disables) therefore stops the run after
+that many *consecutive* failed batches, with exit code 2; a single successful batch resets the
+counter. Ollama@SOLARIA's known failure mode is total (container vanishes, network-detached —
+4 incidents, plan §1.4/§7), so the breaker trips within seconds of it. On abort, pending
+`entities[type=threading]` appends are flushed first: they don't depend on Ollama, they're
+idempotent, and re-deriving them would mean re-reading the same 27 GB.
+
+Only a wrong embedding dimension is more severe (`EmbeddingDimensionError`, exit 1) — it aborts
+immediately, since that would otherwise silently index a vector that doesn't match
+`document_chunk.embedding VECTOR(1024)`.
 
 ## Idempotency
 
@@ -114,13 +144,15 @@ pip install -e "jobs/mail-body-ingest[dev]"
 cd jobs/mail-body-ingest && pytest
 ```
 
-Pure unit tests (48), no DB/Ollama — `run()` is tested by monkeypatching `asyncpg.connect`
+Pure unit tests (55), no DB/Ollama — `run()` is tested by monkeypatching `asyncpg.connect`
 and `aiohttp.ClientSession` with in-memory fakes, `.eml` bytes written to `tmp_path`. Covers:
 quote-strip (EN/PL/Outlook markers, bare `>` lines), HTML->text (style/script/blockquote/
 gmail_quote skipping), newsletter classification, threading extraction, prefix building,
 body extraction (plain-preferred, HTML fallback, attachment-only), the typed/compat32 parse
 fallback, stats balance, idempotency (second run inserts nothing new), newsletter chunks
-never reaching Ollama, Ollama-offline batch isolation, and dimension-mismatch abort.
+never reaching Ollama, Ollama-offline batch isolation, dimension-mismatch abort, and the
+circuit breaker (trips on N consecutive failures, resets on a success, disabled by `0`,
+flushes pending threading on abort).
 
 ## Definition of Done
 

jobs/mail-body-ingest/src/mail_body_ingest/ingest.py

@@ -43,9 +43,16 @@ stays in the DB (reversible: `UPDATE ... SET excluded_reason=NULL WHERE excluded
 'newsletter'` + a re-embed run un-flags them later, per plan Decyzja 4).
 
 Ollama-offline tolerance: a failed `embed_batch()` call is caught per-batch (`chunks_errors +=
-len(batch)`), never aborting the run — those chunks never enter the idempotency set, so a later
+len(batch)`) — those chunks never enter the idempotency set, so a later re-run naturally retries
-re-run naturally retries them. Only a wrong embedding dimension aborts the whole run
+them. Transient single-batch failures therefore cost nothing. What they must NOT do is let a run
-(`EmbeddingDimensionError`) — never silently indexes a mismatched vector.
+survive a *dead* backend: with the archive parsed single-threaded ahead of the GPU, a full-corpus
+run (Etap B, plan §9) would otherwise keep chewing through 200k+ mails at parse speed, marking
+every chunk `chunks_errors`, and the whole multi-hour pass would have to be repeated. Hence the
+circuit breaker: `--max-embed-failures` consecutive failed batches (default 5, `0` disables)
+raise `EmbedBackendUnavailableError` and stop the run early with exit code 2, after flushing the
+threading updates already earned. A single successful batch resets the counter. Only a wrong
+embedding dimension is more severe (`EmbeddingDimensionError`, exit 1) — never silently indexes
+a mismatched vector.
 """
 from __future__ import annotations
 
@@ -81,6 +88,17 @@ _log = structlog.get_logger(__name__)
 DEFAULT_ARCHIVE_ROOT = Path("/home/oskar/kb/mail/archive")
 DEFAULT_BATCH_SIZE = 64
 THREADING_UPDATE_BATCH_SIZE = 500
+# Circuit breaker: consecutive failed embed batches that mean "the backend is down, not flaky".
+# 5 x --batch-size chunks written off before stopping; Ollama@SOLARIA's known failure mode is
+# total (container vanishes / network-detached), not partial, so this trips within seconds of it.
+DEFAULT_MAX_EMBED_FAILURES = 5
+EXIT_EMBED_BACKEND_UNAVAILABLE = 2
+
+
+class EmbedBackendUnavailableError(RuntimeError):
+    """Raised when `--max-embed-failures` consecutive embed batches failed — the run stops instead
+    of parsing the rest of the archive into `chunks_errors`. Nothing is corrupted: failed chunks
+    were never inserted, so a re-run picks them up through the ordinary idempotency path."""
 
 _CHUNK_INSERT_SQL = """
 INSERT INTO document_chunk (envelope_id, chunk_index, text, embedding, model, excluded_reason)
@@ -374,6 +392,7 @@ def _new_stats() -> dict:
         "chunks_errors": 0,
         "quoted_chars_stripped_total": 0,
         "embed_calls": 0,
+        "embed_batch_failures": 0,  # diagnostic only — not part of the balance equations
         "embed_seconds_total": 0.0,
         "threading_updated": 0,
         "threading_already_present": 0,
@@ -390,6 +409,7 @@ async def run(
     offset: Optional[int] = None,
     apply: bool = False,
     batch_size: int = DEFAULT_BATCH_SIZE,
+    max_embed_failures: int = DEFAULT_MAX_EMBED_FAILURES,
 ) -> dict:
     """Process one --limit/--offset (optionally --since-filtered) slice of `source='gmail'`
     envelopes. dry-run (apply=False): parse + quote-strip + classify + chunk + count, zero
@@ -413,8 +433,10 @@ async def run(
 
         embed_buffer: list[tuple[str, int, str]] = []
         threading_pending: list[tuple[str, str]] = []
+        consecutive_embed_failures = 0
 
         async def flush_embed_buffer() -> None:
+            nonlocal consecutive_embed_failures
             if not embed_buffer:
                 return
             texts = [t for (_eid, _idx, t) in embed_buffer]
@@ -425,9 +447,18 @@ async def run(
             except aiohttp.ClientError:
                 _log.warning("skip.embed_batch_error", count=len(embed_buffer), exc_info=True)
                 stats["chunks_errors"] += len(embed_buffer)
+                stats["embed_batch_failures"] += 1
+                consecutive_embed_failures += 1
                 embed_buffer.clear()
+                if max_embed_failures and consecutive_embed_failures >= max_embed_failures:
+                    raise EmbedBackendUnavailableError(
+                        f"{consecutive_embed_failures} consecutive embed batches failed "
+                        f"({ollama_url}) — stopping instead of parsing the rest of the archive "
+                        f"into chunks_errors; re-run to retry the missed chunks"
+                    )
                 return
 
+            consecutive_embed_failures = 0
             stats["embed_calls"] += 1
             stats["embed_seconds_total"] += elapsed
             for (eid, idx, chunk), embedding in zip(embed_buffer, embeddings):
@@ -540,6 +571,12 @@ async def run(
             if apply:
                 await flush_embed_buffer()
                 await flush_threading()
+        except EmbedBackendUnavailableError:
+            # The threading appends already earned by the mails parsed so far are independent of
+            # Ollama and idempotent — flush them rather than making the next run re-derive them.
+            await flush_threading()
+            _log.error("embed_backend_unavailable_abort", ollama_url=ollama_url, **stats)
+            raise
         finally:
             if session is not None:
                 await session.close()
@@ -586,6 +623,12 @@ def main() -> None:
                         help="Slice offset, ordered by envelope id (default: 0)")
     parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE,
                         help=f"Ollama /api/embed batch size (default: {DEFAULT_BATCH_SIZE})")
+    parser.add_argument("--max-embed-failures", type=int, default=DEFAULT_MAX_EMBED_FAILURES,
+                        metavar="N",
+                        help=f"Abort (exit {EXIT_EMBED_BACKEND_UNAVAILABLE}) after N consecutive "
+                        f"failed embed batches — a dead Ollama must not turn a multi-hour run into "
+                        f"200k chunks_errors. 0 disables the breaker. "
+                        f"Default: {DEFAULT_MAX_EMBED_FAILURES}.")
     parser.add_argument("--apply", action="store_true",
                         help="Actually call Ollama, insert chunks, and update threading entities. "
                         "Default is dry-run (parse + classify + chunk + count only).")
@@ -610,11 +653,17 @@ def main() -> None:
                 offset=args.offset,
                 apply=args.apply,
                 batch_size=args.batch_size,
+                max_embed_failures=args.max_embed_failures,
             )
         )
     except EmbeddingDimensionError as exc:
         _log.error("dim_mismatch_abort", error=str(exc))
         sys.exit(1)
+    except EmbedBackendUnavailableError as exc:
+        # Distinct exit code: unlike exit 1 (which a full-corpus run can legitimately reach on a
+        # handful of parse_errors), this one means "nothing more will succeed until Ollama is back".
+        _log.error("embed_backend_unavailable", error=str(exc))
+        sys.exit(EXIT_EMBED_BACKEND_UNAVAILABLE)
 
     mode = "APPLY" if args.apply else "DRY-RUN"
     avg_embed_ms = (

jobs/mail-body-ingest/tests/test_ingest.py

@@ -17,6 +17,7 @@ import pytest
 from kb_retrieval.embed import EmbeddingDimensionError
 from mail_body_ingest.ingest import (
     _CHUNK_INSERT_SQL,
+    EmbedBackendUnavailableError,
     _decode_jsonb,
     _has_threading,
     build_prefix,
@@ -388,17 +389,26 @@ class _FakeTagsResponse:
 
 
 class _FakeOllamaSession:
-    def __init__(self, dim=1024, fail_batches=False, health_up=True):
+    def __init__(self, dim=1024, fail_batches=False, health_up=True, fail_pattern=None):
         self._dim = dim
         self._fail_batches = fail_batches
         self._health_up = health_up
+        # Per-request failure sequence (True = this batch fails); exhausting it falls back to
+        # `fail_batches`. Lets a test interleave failures and successes to exercise the breaker's
+        # "consecutive" semantics rather than a plain total.
+        self._fail_pattern = list(fail_pattern) if fail_pattern is not None else None
         self.requests: list[dict] = []
         self.closed = False
 
+    def _should_fail(self) -> bool:
+        if self._fail_pattern:
+            return self._fail_pattern.pop(0)
+        return self._fail_batches
+
     def post(self, url, json):
         assert url.endswith("/api/embed")
         self.requests.append({"url": url, "json": json})
-        if self._fail_batches:
+        if self._should_fail():
             return _FakeEmbedResponse({}, status=500)
         n = len(json["input"])
         return _FakeEmbedResponse({"embeddings": [[0.01] * self._dim for _ in range(n)]})
@@ -558,6 +568,76 @@ class TestRun:
             + stats["chunks_conflict_skipped"] + stats["chunks_errors"]
         )
 
+    def _write_n_mails(self, tmp_path, n: int) -> list:
+        rows = []
+        for i in range(n):
+            eid = f"m{i}@x"
+            _write_eml(tmp_path, f"gmail/2025/08/{eid}.eml", _plain_eml(
+                {"From": "a@b.com", "Subject": f"hi {i}", "Date": "Fri, 01 Aug 2025 10:00:00 +0000"},
+                f"hello there number {i}",
+            ))
+            rows.append(_env_row(eid, f"hi {i}"))
+        return rows
+
+    async def test_breaker_aborts_after_consecutive_embed_failures(self, tmp_path, monkeypatch):
+        """A dead backend must stop the run, not let it parse the rest of the archive into
+        chunks_errors (plan §9 Etap B: 200k+ mails behind a single-threaded parse)."""
+        conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 8))
+        ollama = _FakeOllamaSession(dim=1024, fail_batches=True)
+        self._patch(monkeypatch, conn, ollama)
+
+        with pytest.raises(EmbedBackendUnavailableError):
+            await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True,
+                      batch_size=1, max_embed_failures=5)
+
+        # Stopped at the 5th failed batch — mails 6-8 were never parsed, let alone embedded.
+        assert len(ollama.requests) == 5
+        assert conn.execute_calls == []  # nothing inserted: failed chunks stay retryable
+
+    async def test_breaker_counter_resets_on_successful_batch(self, tmp_path, monkeypatch):
+        """"Consecutive", not "total" — flaky batches interleaved with successes must not trip it."""
+        conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 5))
+        ollama = _FakeOllamaSession(dim=1024, fail_pattern=[True, True, False, True, True])
+        self._patch(monkeypatch, conn, ollama)
+
+        stats = await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True,
+                          batch_size=1, max_embed_failures=3)
+
+        assert len(ollama.requests) == 5  # ran to completion
+        assert stats["chunks_errors"] == 4
+        assert stats["chunks_inserted"] == 1
+        assert stats["embed_batch_failures"] == 4
+        assert stats["chunks_total"] == (
+            stats["chunks_inserted"] + stats["chunks_newsletter_flagged"] + stats["chunks_already_embedded"]
+            + stats["chunks_conflict_skipped"] + stats["chunks_errors"]
+        )
+
+    async def test_breaker_disabled_with_zero(self, tmp_path, monkeypatch):
+        conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 6))
+        ollama = _FakeOllamaSession(dim=1024, fail_batches=True)
+        self._patch(monkeypatch, conn, ollama)
+
+        stats = await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True,
+                          batch_size=1, max_embed_failures=0)
+
+        assert len(ollama.requests) == 6
+        assert stats["chunks_errors"] == 6
+
+    async def test_breaker_abort_flushes_pending_threading(self, tmp_path, monkeypatch):
+        """Threading appends are Ollama-independent and idempotent — the abort keeps them rather
+        than making the next run re-derive them from the same 27 GB read."""
+        conn = _FakeConn(envelopes=self._write_n_mails(tmp_path, 8))
+        ollama = _FakeOllamaSession(dim=1024, fail_batches=True)
+        self._patch(monkeypatch, conn, ollama)
+
+        with pytest.raises(EmbedBackendUnavailableError):
+            await run(dsn="postgresql://fake", archive_root=tmp_path, apply=True,
+                      batch_size=1, max_embed_failures=5)
+
+        assert len(conn.executemany_calls) == 1
+        # The 5 mails processed before the breaker tripped, none of the 3 after it.
+        assert len(conn.executemany_calls[0][1]) == 5
+
     async def test_dimension_mismatch_aborts(self, tmp_path, monkeypatch):
         _write_eml(tmp_path, "gmail/2025/08/m1@x.eml", _plain_eml(
             {"From": "a@b.com", "Subject": "hi", "Date": "Fri, 01 Aug 2025 10:00:00 +0000"}, "hello there"

services/node-agent/src/node_agent.py

@@ -13,10 +13,10 @@ Runs as a Docker container on every managed node. Each cycle it:
   6. Optionally rsyncs events (including action_result) to VPS so the
      control-plane observer/executor can process them.
 
-Cleanup policy (matches health-monitor.sh):
+Cleanup policy (matches health-monitor.sh). Every type below is additionally
+rate-limited to one cleanup run per CLEANUP_INTERVAL_SECS (24 h):
   lte_node  (chelsty-infra, chelsty-ha) : NO cleanup, NO image operations
-  sd_card   (piha, saturn)              : dangling images + stopped containers,
+  sd_card   (piha, saturn)              : dangling images + stopped containers
-                                          max once per 24 h
   ai_node   (solaria)                   : dangling + containers + build cache,
                                           NEVER docker image prune -a
   standard  (vps)                       : dangling + containers + build cache +
@@ -27,6 +27,9 @@ NEVER TOUCHED on any node:
   /opt/homelab/config/   All hand-crafted and repo-seeded configuration
   /opt/homelab/state/    Heartbeat files, observer checkpoint
   actions/pending|approved|running   Live work queue
+  Stopped containers with a restart policy or a compose project label — a
+  stopped service is recorded operator intent, not garbage. Container cleanup
+  enumerates explicitly and never calls the unfiltered containers.prune().
 """
 
 import json
@@ -131,7 +134,8 @@ MEM_CRIT_PCT  = 95
 # that is actually stuck flapping. Configurable via env for tuning per fleet.
 CRASH_LOOP_RESTART_THRESHOLD = int(os.getenv("CRASH_LOOP_RESTART_THRESHOLD", "3"))
 
-# SD-card nodes: enforce 24-hour gap between Docker cleanup runs
+# Minimum gap between Docker cleanup runs, on EVERY cleanup-eligible node type
+# (was sd_card-only until the 2026-07-30 unfiltered-prune incident)
 CLEANUP_INTERVAL_SECS = 86_400
 LAST_CLEANUP_FILE = STATE_DIR / "last-docker-cleanup"
 
@@ -591,8 +595,14 @@ class NodeAgent:
     # Safe Docker cleanup
     # ------------------------------------------------------------------
 
-    def _sd_card_rate_ok(self) -> bool:
+    def _cleanup_rate_ok(self) -> bool:
-        """Return True only if 24 hours have elapsed since last cleanup."""
+        """Return True only if CLEANUP_INTERVAL_SECS has elapsed since last cleanup.
+
+        Applies to EVERY cleanup-eligible node type, not just sd_card. Before
+        R2 this guard covered sd_card alone, so ai_node and standard pruned
+        every 60 s — 1440 chances a day to delete a deliberately stopped
+        container, for `0 MB reclaimed` in practically every cycle.
+        """
         if LAST_CLEANUP_FILE.exists():
             try:
                 last_ts  = int(LAST_CLEANUP_FILE.read_text().strip())
@@ -623,14 +633,58 @@ class NodeAgent:
             logger.error(f"Image prune failed: {exc}")
 
     def _prune_stopped_containers(self):
+        """Remove exited containers that are unambiguously disposable.
+
+        NEVER uses containers.prune(): the Docker API removes *every*
+        non-running container, ignoring RestartPolicy and compose labels. That
+        is what deleted `ollama` on SOLARIA 19 s after an operator stopped it
+        (docs/incidents/2026-07-30-ollama-solaria-vanish.md). No filter can fix
+        it either — `until` filters on creation time, not stop time, so it
+        never protects a long-lived service.
+
+        A container carrying `restart: unless-stopped|always|on-failure` or a
+        compose project label is recorded operator intent, exactly like the
+        paths in this module's NEVER TOUCHED list. Only one-off leftovers
+        (restart policy `no`, no compose project) are removed.
+        """
         if not self.docker_client:
             return
         try:
-            result   = self.docker_client.containers.prune()
+            containers = self.docker_client.containers.list(
-            reclaimed = result.get("SpaceReclaimed", 0) // (1024 * 1024)
+                all=True, filters={"status": "exited"}
-            logger.info(f"Pruned stopped containers ({reclaimed} MB reclaimed)")
+            )
         except Exception as exc:
-            logger.error(f"Container prune failed: {exc}")
+            logger.error(f"Container list for prune failed: {exc}")
+            return
+
+        removed, kept = [], 0
+        for c in containers:
+            try:
+                policy = (
+                    (c.attrs.get("HostConfig") or {})
+                    .get("RestartPolicy", {})
+                    .get("Name", "")
+                )
+                if policy in ("unless-stopped", "always", "on-failure"):
+                    kept += 1  # operator intent — do not touch
+                    continue
+                if (c.labels or {}).get("com.docker.compose.project"):
+                    kept += 1  # compose-managed — do not touch
+                    continue
+                c.remove()
+                removed.append(c.name)
+            except Exception as exc:
+                logger.error(f"Failed to remove stopped container {c.name}: {exc}")
+
+        # R3: name what was deleted. The old log line reported megabytes only,
+        # which is why ollama's removal looked identical to every no-op cycle.
+        if removed:
+            logger.warning(
+                f"Removed {len(removed)} disposable stopped container(s): "
+                f"{', '.join(removed)} (kept {kept} managed)"
+            )
+        else:
+            logger.info(f"No disposable stopped containers (kept {kept} managed)")
 
     def _prune_build_cache(self):
         if not self.docker_client:
@@ -651,9 +705,14 @@ class NodeAgent:
             logger.debug("Skipping Docker cleanup: LTE node")
             return
 
+        # Rate limit applies to every cleanup-eligible node type (R2). Cleanup
+        # is housekeeping, not a health function: once per CLEANUP_INTERVAL_SECS
+        # reclaims the same space as once per minute, at a fraction of the I/O
+        # and of the exposure to accidental deletion.
+        if not self._cleanup_rate_ok():
+            return
+
         if self.node_type == "sd_card":
-            if not self._sd_card_rate_ok():
-                return
             self._prune_dangling_images()
             self._prune_stopped_containers()
             # No builder prune: minimise write cycles on SD card
@@ -665,6 +724,7 @@ class NodeAgent:
         self._prune_dangling_images()
         self._prune_stopped_containers()
         self._prune_build_cache()
+        self._mark_cleanup_done()
 
     # ------------------------------------------------------------------
     # VPS-specific: control-plane filesystem rotation

services/node-agent/tests/test_safe_cleanup.py

@@ -0,0 +1,201 @@
+"""Tests for NodeAgent Docker cleanup — regression cover for the 2026-07-30
+unfiltered-prune incident (docs/incidents/2026-07-30-ollama-solaria-vanish.md).
+
+`containers.prune()` removes EVERY non-running container, ignoring restart
+policy and compose labels; that is what deleted `ollama` on SOLARIA 19 s after
+an operator stopped it. These tests pin down the replacement:
+
+  R1 — explicit enumeration; a stopped container with `restart: unless-stopped`
+       or a compose project label is never removed, and prune() is never called.
+  R2 — the cleanup rate limit applies to ai_node / standard, not sd_card only.
+  R3 — removals are named in a WARNING log line.
+"""
+from __future__ import annotations
+
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+import node_agent
+
+
+# ---------------------------------------------------------------------------
+# Fake Docker container helper
+# ---------------------------------------------------------------------------
+
+def make_stopped(name, *, restart_policy="no", compose_project=None):
+    c = MagicMock()
+    c.name = name
+    c.status = "exited"
+    c.attrs = {"HostConfig": {"RestartPolicy": {"Name": restart_policy}}}
+    c.labels = {"com.docker.compose.project": compose_project} if compose_project else {}
+    return c
+
+
+def with_containers(agent, containers):
+    client = MagicMock()
+    client.containers.list.return_value = containers
+    agent.docker_client = client
+    return client
+
+
+# ---------------------------------------------------------------------------
+# R1 — the incident itself
+# ---------------------------------------------------------------------------
+
+def test_unless_stopped_container_survives(agent):
+    """THE regression: ollama@solaria — exited, restart=unless-stopped, must live."""
+    ollama = make_stopped("ollama", restart_policy="unless-stopped")
+    client = with_containers(agent, [ollama])
+
+    agent._prune_stopped_containers()
+
+    ollama.remove.assert_not_called()
+    client.containers.prune.assert_not_called()
+
+
+@pytest.mark.parametrize("policy", ["unless-stopped", "always", "on-failure"])
+def test_every_restart_policy_is_operator_intent(agent, policy):
+    c = make_stopped("svc", restart_policy=policy)
+    with_containers(agent, [c])
+    agent._prune_stopped_containers()
+    c.remove.assert_not_called()
+
+
+def test_compose_managed_container_survives(agent):
+    """No restart policy, but compose owns it → still off limits."""
+    c = make_stopped("outline-redis-1", restart_policy="no", compose_project="outline")
+    with_containers(agent, [c])
+    agent._prune_stopped_containers()
+    c.remove.assert_not_called()
+
+
+def test_disposable_leftover_is_removed(agent):
+    """restart=no and no compose project → a one-off leftover, safe to remove."""
+    c = make_stopped("nervous-shell-42", restart_policy="no")
+    with_containers(agent, [c])
+    agent._prune_stopped_containers()
+    c.remove.assert_called_once()
+
+
+def test_only_exited_containers_are_enumerated(agent):
+    """Running containers must never even enter the candidate list."""
+    client = with_containers(agent, [])
+    agent._prune_stopped_containers()
+    _args, kwargs = client.containers.list.call_args
+    assert kwargs["all"] is True
+    assert kwargs["filters"] == {"status": "exited"}
+
+
+def test_mixed_set_removes_only_the_disposable_one(agent):
+    keep_policy = make_stopped("ollama", restart_policy="unless-stopped")
+    keep_compose = make_stopped("umami-db", compose_project="umami")
+    drop = make_stopped("tmp-build", restart_policy="no")
+    with_containers(agent, [keep_policy, keep_compose, drop])
+
+    agent._prune_stopped_containers()
+
+    keep_policy.remove.assert_not_called()
+    keep_compose.remove.assert_not_called()
+    drop.remove.assert_called_once()
+
+
+def test_remove_failure_does_not_abort_the_sweep(agent):
+    boom = make_stopped("boom", restart_policy="no")
+    boom.remove.side_effect = RuntimeError("device or resource busy")
+    later = make_stopped("later", restart_policy="no")
+    with_containers(agent, [boom, later])
+
+    agent._prune_stopped_containers()
+
+    later.remove.assert_called_once()
+
+
+def test_no_docker_client_is_noop(agent):
+    agent.docker_client = None
+    agent._prune_stopped_containers()  # must not raise
+
+
+# ---------------------------------------------------------------------------
+# R3 — say what was deleted
+# ---------------------------------------------------------------------------
+
+def test_removal_is_logged_at_warning_with_names(agent, caplog):
+    with_containers(agent, [make_stopped("tmp-build", restart_policy="no")])
+
+    with caplog.at_level("INFO", logger="node-agent"):
+        agent._prune_stopped_containers()
+
+    warnings = [r for r in caplog.records if r.levelname == "WARNING"]
+    assert len(warnings) == 1
+    assert "tmp-build" in warnings[0].message
+
+
+def test_nothing_removed_stays_at_info(agent, caplog):
+    with_containers(agent, [make_stopped("ollama", restart_policy="unless-stopped")])
+
+    with caplog.at_level("INFO", logger="node-agent"):
+        agent._prune_stopped_containers()
+
+    assert [r for r in caplog.records if r.levelname == "WARNING"] == []
+
+
+# ---------------------------------------------------------------------------
+# R2 — rate limit covers every node type
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def cleanup_calls(agent, monkeypatch):
+    """Record which prune helpers run, without touching Docker."""
+    calls = []
+    for helper in ("_prune_dangling_images", "_prune_stopped_containers",
+                   "_prune_build_cache"):
+        monkeypatch.setattr(agent, helper, lambda h=helper: calls.append(h))
+    return calls
+
+
+@pytest.fixture
+def fresh_cleanup_marker(tmp_path, monkeypatch):
+    monkeypatch.setattr(node_agent, "LAST_CLEANUP_FILE", tmp_path / "last-docker-cleanup")
+    return node_agent.LAST_CLEANUP_FILE
+
+
+@pytest.mark.parametrize("node_type", ["ai_node", "standard", "sd_card"])
+def test_recent_cleanup_blocks_next_run(agent, cleanup_calls, fresh_cleanup_marker,
+                                        node_type):
+    """Was true for sd_card only; after R2 it holds for ai_node and standard too."""
+    agent.node_type = node_type
+    fresh_cleanup_marker.write_text(str(int(time.time())))
+
+    agent.run_safe_cleanup()
+
+    assert cleanup_calls == []
+
+
+@pytest.mark.parametrize("node_type,expected", [
+    ("ai_node", ["_prune_dangling_images", "_prune_stopped_containers",
+                 "_prune_build_cache"]),
+    ("standard", ["_prune_dangling_images", "_prune_stopped_containers",
+                  "_prune_build_cache"]),
+    ("sd_card", ["_prune_dangling_images", "_prune_stopped_containers"]),
+])
+def test_stale_marker_allows_cleanup_and_is_refreshed(agent, cleanup_calls,
+                                                      fresh_cleanup_marker,
+                                                      node_type, expected):
+    agent.node_type = node_type
+    stale = int(time.time()) - node_agent.CLEANUP_INTERVAL_SECS - 1
+    fresh_cleanup_marker.write_text(str(stale))
+
+    agent.run_safe_cleanup()
+
+    assert cleanup_calls == expected
+    # Marker refreshed for every type, otherwise the guard never engages.
+    assert int(fresh_cleanup_marker.read_text()) > stale
+
+
+def test_lte_node_still_does_nothing(agent, cleanup_calls, fresh_cleanup_marker):
+    agent.node_type = "lte_node"
+    agent.run_safe_cleanup()
+    assert cleanup_calls == []
+    assert not fresh_cleanup_marker.exists()