From ca6e0290af457ad00bbab46b791b622ae3a3fc08 Mon Sep 17 00:00:00 2001 From: Arch Date: Tue, 30 Jun 2026 04:13:59 +0100 Subject: [PATCH] Scan resilience: model preflight, shared per-post path, crash-safe memory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three robustness/maintainability improvements to scan mode: 1. Crash-safe incremental memory (was: saved once at end of run). A local LLM scan can run for minutes; a kill / reboot / OOM mid-run discarded every analysis since the last run, so the next run re-burned the model over all of them. cmd_scan now checkpoints memory every SCAN_SAVE_EVERY (10) newly-analyzed posts AND in a finally, so an interrupted scan keeps its finished work. Writes are atomic (existing _atomic_json_write). 2. Model-availability preflight. The startup check only confirmed the Ollama daemon was reachable; if the requested model wasn't pulled, every call_ollama returned None and the whole run silently no-op'd with per-post errors. ensure_model_available() now checks /api/tags and fails fast with "ollama pull " — wired into both scan and webhook entry points. 3. Extracted _process_post(). The single-post (--post-id) and bulk-scan branches duplicated the analyze -> stamp -> record -> act pipeline; one shared helper removes the drift risk and makes the per-post path unit- testable. Pinned by tests/test_scan_resilience.py (12 tests): preflight match/miss/ daemon-down, _process_post success/dry-run/model-failure, and the checkpoint cadence + that a mid-scan crash still persists completed work. Full suite: 130 passing. Co-Authored-By: Claude Opus 4.8 (1M context) --- sentinel.py | 197 ++++++++++++++++++++++------------ tests/test_scan_resilience.py | 171 +++++++++++++++++++++++++++++ 2 files changed, 300 insertions(+), 68 deletions(-) create mode 100644 tests/test_scan_resilience.py diff --git a/sentinel.py b/sentinel.py index 467fcbf..46cf6fc 100644 --- a/sentinel.py +++ b/sentinel.py @@ -78,6 +78,11 @@ # Prune memory every N processed posts in webhook mode (scan mode prunes on # every run). WEBHOOK_PRUNE_EVERY = 50 +# Checkpoint scan-mode memory to disk every N newly-analyzed posts (and again +# in a finally). A local LLM scan can run for minutes; without checkpoints a +# kill / reboot / OOM mid-run discards every analysis since the last run, so +# the next run re-burns the model over all of them. The write is atomic. +SCAN_SAVE_EVERY = 10 DEFAULT_USERNAME = "qwen-jorwhol-analyzer" DEFAULT_DISPLAY_NAME = "Qwen 3.5 Jorwhol Analyzer" @@ -887,12 +892,83 @@ def log_results(results: list[dict]) -> None: # ─── Scan mode (one-shot) ─────────────────────────────────────────────── +def ensure_model_available(model: str) -> None: + """Abort early if the requested model isn't pulled into Ollama. + + The startup preflight only confirms the daemon is reachable. If the + model itself is missing, every ``call_ollama`` would just return None + and the whole run would silently do nothing but log per-post errors — + so fail loudly here with the exact fix instead. + """ + try: + resp = requests.get( + f"{OLLAMA_HOST}/api/tags", timeout=OLLAMA_CONNECT_TIMEOUT) + resp.raise_for_status() + models = resp.json().get("models", []) or [] + except Exception as e: + logger.error("Could not query Ollama models at %s: %s", OLLAMA_HOST, e) + sys.exit(1) + names = {m.get("name", "") for m in models} + # Ollama tags carry an explicit ":tag" (e.g. "qwen3.5:9b-q4_K_M"); a bare + # name defaults to ":latest". Accept an exact match, the implicit :latest + # form, or a base-name match. + bases = {n.split(":", 1)[0] for n in names} + if model in names or f"{model}:latest" in names or model in bases: + return + available = ", ".join(sorted(n for n in names if n)) or "(none installed)" + logger.error( + "Ollama model %r is not available. Pull it with: ollama pull %s " + "(installed: %s)", model, model, available, + ) + sys.exit(1) + + +def _process_post( + client: ColonyClient, + post_data: dict, + args: argparse.Namespace, + memory: dict, + results: list[dict], +) -> bool: + """Analyze one already-fetched post, record it in ``memory`` + + ``results``, and apply its moderation actions. + + Shared by the single-post (``--post-id``) and bulk-scan paths so the two + can't drift. Returns True when the post was analyzed; False when the + model call failed (caller leaves it unrecorded to retry next run). + """ + post_id = post_data["post"].get("id") + judgement = analyze_post(post_data, args.model) + if judgement is None: + return False + + judgement["analyzed_at"] = datetime.now().isoformat() + results.append(judgement) + memory[post_id] = judgement + + if not args.dry_run: + failed = act_on_judgement( + client, + post_id, + judgement, + allow_vote=not args.no_vote, + allow_pii=not args.no_pii, + confirm=args.confirm, + ) + if failed: + judgement["_pending_actions"] = failed + return True + + def cmd_scan(args: argparse.Namespace) -> None: logger.info("Sentinel — scan mode") if args.dry_run: args.no_vote = True args.no_pii = True + # Fail fast if the model isn't pulled rather than silently no-op'ing. + ensure_model_available(args.model) + username = (args.username or DEFAULT_USERNAME).lower().replace(" ", "-") client, config = get_or_register_client(username) @@ -909,80 +985,61 @@ def cmd_scan(args: argparse.Namespace) -> None: results: list[dict] = [] new_analyses = 0 - if args.post_id: - data = fetch_post_with_comments(client, args.post_id) - if data is None: - logger.error("Could not fetch post — aborting") - sys.exit(1) - judgement = analyze_post(data, args.model) - if judgement: - judgement["analyzed_at"] = datetime.now().isoformat() - results.append(judgement) - memory[args.post_id] = judgement - new_analyses += 1 - if not args.dry_run: - failed = act_on_judgement( - client, - args.post_id, - judgement, - allow_vote=not args.no_vote, - allow_pii=not args.no_pii, - confirm=args.confirm, - ) - if failed: - judgement["_pending_actions"] = failed - else: - try: - posts = list(client.iter_posts(sort=args.sort, max_results=args.limit)) - except ColonyAPIError as e: - logger.error("Failed to fetch posts: %s", e) - sys.exit(1) + def _checkpoint() -> None: + if not args.dry_run: + save_memory(memory) - for post in posts: - post_id = post.get("id") - title = (post.get("title") or "")[:70] - created_at = post.get("created_at") + try: + if args.post_id: + data = fetch_post_with_comments(client, args.post_id) + if data is None: + logger.error("Could not fetch post — aborting") + sys.exit(1) + if _process_post(client, data, args, memory, results): + new_analyses += 1 + else: + try: + posts = list(client.iter_posts(sort=args.sort, max_results=args.limit)) + except ColonyAPIError as e: + logger.error("Failed to fetch posts: %s", e) + sys.exit(1) - if not post_id: - continue - if post_id in processed and not args.force: - logger.debug("Skipping (already analyzed): %s %s", post_id[:8], title) - continue - if not is_within_days(created_at or "", args.days): - logger.debug("Skipping (older than %d days): %s %s", args.days, post_id[:8], title) - continue - if (post.get("author") or {}).get("username", "") == config.get("username"): - logger.debug("Skipping (own post): %s %s", post_id[:8], title) - continue + for post in posts: + post_id = post.get("id") + title = (post.get("title") or "")[:70] + created_at = post.get("created_at") - logger.info("Analyzing %s: %s", post_id[:8], title) - data = fetch_post_with_comments(client, post_id) - if data is None: - continue - judgement = analyze_post(data, args.model) - if judgement is None: - logger.warning("Analysis failed — will retry next run") - continue + if not post_id: + continue + if post_id in processed and not args.force: + logger.debug("Skipping (already analyzed): %s %s", post_id[:8], title) + continue + if not is_within_days(created_at or "", args.days): + logger.debug("Skipping (older than %d days): %s %s", args.days, post_id[:8], title) + continue + if (post.get("author") or {}).get("username", "") == config.get("username"): + logger.debug("Skipping (own post): %s %s", post_id[:8], title) + continue - judgement["analyzed_at"] = datetime.now().isoformat() - results.append(judgement) - memory[post_id] = judgement - new_analyses += 1 - - if not args.dry_run: - failed = act_on_judgement( - client, - post_id, - judgement, - allow_vote=not args.no_vote, - allow_pii=not args.no_pii, - confirm=args.confirm, - ) - if failed: - judgement["_pending_actions"] = failed + logger.info("Analyzing %s: %s", post_id[:8], title) + data = fetch_post_with_comments(client, post_id) + if data is None: + continue + if not _process_post(client, data, args, memory, results): + logger.warning("Analysis failed — will retry next run") + continue + + new_analyses += 1 + # Checkpoint periodically so an interrupted scan keeps its + # finished work instead of re-analyzing it on the next run. + if new_analyses % SCAN_SAVE_EVERY == 0: + _checkpoint() + logger.debug("Checkpointed memory after %d new analyses", new_analyses) + finally: + # Always persist what we finished — even on Ctrl-C / crash / sys.exit. + _checkpoint() if not args.dry_run: - save_memory(memory) logger.info("Memory updated: %d posts (added %d new)", len(memory), new_analyses) else: logger.info("Dry run — memory not saved (%d posts analyzed)", new_analyses) @@ -1237,6 +1294,10 @@ def log_message(self, format: str, *args: Any) -> None: # noqa: A002 def cmd_webhook(args: argparse.Namespace) -> None: logger.info("Sentinel — webhook mode") + # Fail fast if the model isn't pulled — otherwise the server would accept + # webhooks and silently fail every analysis. + ensure_model_available(args.model) + username = (args.username or DEFAULT_USERNAME).lower().replace(" ", "-") client, config = get_or_register_client(username) diff --git a/tests/test_scan_resilience.py b/tests/test_scan_resilience.py new file mode 100644 index 0000000..115da23 --- /dev/null +++ b/tests/test_scan_resilience.py @@ -0,0 +1,171 @@ +"""Tests for scan-mode resilience: model preflight, the shared per-post +pipeline, and crash-safe incremental memory checkpointing. +""" +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +import requests + +import sentinel as s + + +def _args(**over): + base = dict( + dry_run=False, no_vote=False, no_pii=False, model="qwen3.5:9b-q4_K_M", + username=None, force=False, post_id=None, sort="new", limit=50, + days=7, confirm=False, + ) + base.update(over) + return SimpleNamespace(**base) + + +def _tags_response(names): + r = MagicMock() + r.status_code = 200 + r.raise_for_status = MagicMock() + r.json.return_value = {"models": [{"name": n} for n in names]} + return r + + +# ─── #2 ensure_model_available ─────────────────────────────────────────── + +class TestEnsureModelAvailable: + def test_exact_match_ok(self, monkeypatch): + monkeypatch.setattr(s.requests, "get", + lambda *a, **k: _tags_response(["qwen3.5:9b-q4_K_M"])) + s.ensure_model_available("qwen3.5:9b-q4_K_M") # no raise + + def test_base_name_match_ok(self, monkeypatch): + monkeypatch.setattr(s.requests, "get", + lambda *a, **k: _tags_response(["llama3:latest"])) + s.ensure_model_available("llama3") # base name matches + + def test_implicit_latest_ok(self, monkeypatch): + monkeypatch.setattr(s.requests, "get", + lambda *a, **k: _tags_response(["llama3:latest"])) + s.ensure_model_available("llama3") + + def test_missing_model_exits(self, monkeypatch): + monkeypatch.setattr(s.requests, "get", + lambda *a, **k: _tags_response(["other:7b"])) + with pytest.raises(SystemExit) as e: + s.ensure_model_available("qwen3.5:9b-q4_K_M") + assert e.value.code == 1 + + def test_daemon_error_exits(self, monkeypatch): + def boom(*a, **k): + raise requests.exceptions.ConnectionError("down") + monkeypatch.setattr(s.requests, "get", boom) + with pytest.raises(SystemExit) as e: + s.ensure_model_available("m") + assert e.value.code == 1 + + +# ─── #3 _process_post ──────────────────────────────────────────────────── + +class TestProcessPost: + def _post(self): + return {"post": {"id": "p1", "title": "t", "colony_id": "c"}, "comments": []} + + def test_success_records_and_acts(self, monkeypatch, mock_client): + monkeypatch.setattr(s, "analyze_post", + lambda data, model: {"category": "OKAY", "score": 5}) + act = MagicMock(return_value=[]) + monkeypatch.setattr(s, "act_on_judgement", act) + memory, results = {}, [] + ok = s._process_post(mock_client, self._post(), _args(), memory, results) + assert ok is True + assert memory["p1"]["category"] == "OKAY" + assert "analyzed_at" in memory["p1"] + assert len(results) == 1 + act.assert_called_once() + + def test_dry_run_skips_actions(self, monkeypatch, mock_client): + monkeypatch.setattr(s, "analyze_post", lambda d, m: {"category": "OKAY"}) + act = MagicMock(return_value=[]) + monkeypatch.setattr(s, "act_on_judgement", act) + memory, results = {}, [] + ok = s._process_post(mock_client, self._post(), _args(dry_run=True), memory, results) + assert ok is True + assert "p1" in memory + act.assert_not_called() + + def test_model_failure_records_nothing(self, monkeypatch, mock_client): + monkeypatch.setattr(s, "analyze_post", lambda d, m: None) + act = MagicMock() + monkeypatch.setattr(s, "act_on_judgement", act) + memory, results = {}, [] + ok = s._process_post(mock_client, self._post(), _args(), memory, results) + assert ok is False + assert memory == {} and results == [] + act.assert_not_called() + + +# ─── #1 incremental + crash-safe checkpointing in cmd_scan ─────────────── + +def _wire_scan(monkeypatch, mock_client, posts, *, analyze): + """Stub cmd_scan's collaborators; return the list that records each + save_memory() call's snapshot of memory keys.""" + monkeypatch.setattr(s, "ensure_model_available", lambda model: None) + monkeypatch.setattr(s, "get_or_register_client", + lambda u: (mock_client, {"username": "sentinel-bot"})) + monkeypatch.setattr(s, "load_memory", lambda: {}) + monkeypatch.setattr(s, "prune_memory", lambda m, **k: m) + monkeypatch.setattr(s, "retry_pending_actions", lambda c, m: 0) + monkeypatch.setattr(s, "is_within_days", lambda *a, **k: True) + monkeypatch.setattr(s, "fetch_post_with_comments", + lambda c, pid: {"post": {"id": pid, "title": "t"}, "comments": []}) + monkeypatch.setattr(s, "act_on_judgement", lambda *a, **k: []) + monkeypatch.setattr(s, "log_results", lambda r: None) + monkeypatch.setattr(s, "analyze_post", analyze) + mock_client.iter_posts = MagicMock(return_value=posts) + saves = [] + monkeypatch.setattr(s, "save_memory", lambda m: saves.append(set(m.keys()))) + return saves + + +def test_scan_checkpoints_periodically(monkeypatch, mock_client): + posts = [{"id": f"p{i}", "title": "t", "created_at": "2026-06-30", + "author": {"username": "someone"}} for i in range(25)] + saves = _wire_scan(monkeypatch, mock_client, posts, + analyze=lambda d, m: {"category": "OKAY"}) + s.cmd_scan(_args()) + # 25 posts, SCAN_SAVE_EVERY=10 → checkpoints at 10, 20, + final finally = 3. + assert len(saves) >= 3 + assert len(saves[-1]) == 25 # final save has everything + + +def test_scan_persists_on_crash(monkeypatch, mock_client): + posts = [{"id": f"p{i}", "title": "t", "created_at": "2026-06-30", + "author": {"username": "someone"}} for i in range(5)] + calls = {"n": 0} + + def exploding_analyze(data, model): + calls["n"] += 1 + if calls["n"] == 3: + raise RuntimeError("simulated crash mid-scan") + return {"category": "OKAY"} + + saves = _wire_scan(monkeypatch, mock_client, posts, analyze=exploding_analyze) + with pytest.raises(RuntimeError): + s.cmd_scan(_args()) + # The finally block must have saved the two posts finished before the crash. + assert saves, "memory was never checkpointed on crash" + assert saves[-1] == {"p0", "p1"} + + +def test_dry_run_never_saves(monkeypatch, mock_client): + posts = [{"id": "p0", "title": "t", "created_at": "2026-06-30", + "author": {"username": "someone"}}] + saves = _wire_scan(monkeypatch, mock_client, posts, + analyze=lambda d, m: {"category": "OKAY"}) + s.cmd_scan(_args(dry_run=True)) + assert saves == [] + + +def test_scan_save_every_constant_sane(): + assert isinstance(s.SCAN_SAVE_EVERY, int) + assert 1 <= s.SCAN_SAVE_EVERY <= 100