Encapsulate WG21Index.papers behind snapshot and count accessors

CONTRIBUTING.md

@@ -15,7 +15,7 @@ Thank you for your interest in improving paperscout. This document describes how
 When adding blocking I/O or new data sources:
 
 1. Use **`run_blocking_io()`** (see `src/paperscout/concurrency.py`) only for pure blocking I/O with **no shared in-process mutable state** (e.g. psycopg2 via the connection pool).
-2. **Never access `ISOProber._stats`, `WG21Index.papers`, or other source internals from a thread** — keep probes and index refresh on the asyncio event loop.
+2. **Never access `ISOProber._stats`, `WG21Index._papers`, or other source internals from a thread** — keep probes and index refresh on the asyncio event loop. Use `paper_count()` or `get_papers_snapshot()` for cross-thread reads.
 3. See **[docs/architecture.md](docs/architecture.md)** for the full model and **[docs/probe-operations.md](docs/probe-operations.md)** for probe tuning.
 
 ## Workflow

docs/architecture.md

@@ -17,7 +17,7 @@ These components share one thread and the main event loop. They may await I/O bu
 
 | Thread | Role |
 |--------|------|
-| **Health server** (`health.py`) | Serves `GET /health`; reads `len(index.papers)` via a callback and scheduler fields from `Scheduler.health_snapshot()` (immutable snapshot, lock-protected publish). |
+| **Health server** (`health.py`) | Serves `GET /health`; reads paper count via `index.paper_count()` callback and scheduler fields from `Scheduler.health_snapshot()` (immutable snapshot, lock-protected publish). |
 | **MessageQueue sender** (`scout.py`) | Drains Slack post queue with rate limiting. |
 | **`run_blocking_io` / `asyncio.to_thread`** | Runs blocking psycopg2 calls (e.g. `UserWatchlist.matches_for_users`) off the loop. |
 
@@ -26,8 +26,8 @@ These components share one thread and the main event loop. They may await I/O bu
 When adding or changing code:
 
 1. **Use `run_blocking_io()` (or `asyncio.to_thread`) only for pure blocking I/O** with no shared in-process mutable state. The function should use its own DB connection from the pool.
-2. **Never access `ISOProber._stats`, `WG21Index.papers`, `WG21Index._max_rev`, or other source internals from a thread.** Read them only on the event loop, or use lock-protected snapshots (`snapshot_stats()`).
-3. **`WG21Index.papers` is replaced wholesale on every `refresh()`** — do not mutate the dict in place. Assign a new dict from `_parse_and_index()`.
+2. **Never access `ISOProber._stats`, `WG21Index._papers`, `WG21Index._max_rev`, or other source internals from a thread.** Read them only on the event loop, or use snapshot accessors (`paper_count()`, `get_papers_snapshot()`, `snapshot_stats()`).
+3. **`WG21Index._papers` is replaced wholesale on every `refresh()`** — do not mutate the dict in place. Publication is an atomic reference swap in `_commit_index()`.
 4. **New HTTP data sources** should follow the async pattern (`httpx` + coroutines on the loop), like `WG21Index` and `ISOProber`. The optional open-std.org scraper in `sources.py` is a future extension point: if integrated, either keep it async on the loop or isolate it in a thread with no shared mutable state.
 
 ## Related docs

src/paperscout/__main__.py

@@ -250,7 +250,7 @@ async def _async_main() -> None:
         mq.start()
 
         def paper_count_fn() -> int:
-            return len(index.papers)
+            return index.paper_count()
 
         def _on_poll_result(result: PollResult) -> None:
             notify_channel(app, result, mq)

src/paperscout/concurrency.py

@@ -14,7 +14,7 @@ async def run_blocking_io(fn: Callable[..., _T], /, *args, **kwargs) -> _T:
 
     Use only for pure blocking I/O (e.g. psycopg2 queries) that does **not**
     touch shared in-process mutable state such as ``ISOProber._stats``,
-    ``WG21Index.papers``, or other source internals. Each call should use its
+    ``WG21Index._papers``, or other source internals. Each call should use its
     own database connection from the pool.
 
     See ``docs/architecture.md`` for the full concurrency model.

src/paperscout/monitor.py

@@ -371,12 +371,12 @@ async def seed(self) -> SeedResult:
 
         wg21 = self._wg21_index()
         paper_count = (
-            len(wg21.papers)
+            wg21.paper_count()
             if wg21 is not None
             else len(self._snapshots.get(SOURCE_WG21_INDEX, {}))
         )
         if self.cfg.enable_bulk_wg21 and wg21 is not None:
-            log.info("SEED  wg21.link loaded  papers=%d", len(wg21.papers))
+            log.info("SEED  wg21.link loaded  papers=%d", wg21.paper_count())
         if self.cfg.enable_iso_probe:
             log.info("SEED  isocpp.org probe  existing=%d", len(hits))
 
@@ -496,7 +496,7 @@ async def poll_once(self) -> PollResult:
 
         # Safe to run off the event loop: matches_for_users only performs blocking
         # PostgreSQL I/O via psycopg2 on its own pool connection — it does not
-        # touch ISOProber._stats, WG21Index.papers, or other shared source state.
+        # touch ISOProber._stats, WG21Index._papers, or other shared source state.
         per_user_matches = await run_blocking_io(
             self.user_watchlist.matches_for_users,
             diff.new_papers,

src/paperscout/sources.py

@@ -52,7 +52,7 @@ class WG21Index:
 
     Likewise, do not access from the MessageQueue sender thread or any
     other sync worker — ``refresh()`` mutates internal state and would
-    race with cross-thread reads on ``papers`` / ``_max_rev``.
+    race with cross-thread reads on ``_papers`` / ``_max_rev``.
     """
 
     source_id: str = SOURCE_WG21_INDEX
@@ -61,20 +61,20 @@ def __init__(self, pool, cfg: Settings | None = None):
         self._cfg = cfg or settings
         self._cache = PaperCache(pool, ttl_hours=self._cfg.cache_ttl_hours)
         # Replaced wholesale on every refresh(); never mutate in place.
-        self.papers: dict[str, Paper] = {}
+        self._papers: dict[str, Paper] = {}
         self._max_rev: dict[int, int] = {}  # P-number -> highest revision
         self._max_p: int = 0  # absolute highest P-number
         self._sorted_p_nums: list[int] = []  # sorted unique P-numbers, for gap analysis
 
     async def refresh(self) -> IndexRefreshResult:
-        """Load index from cache or network; populate ``self.papers``."""
+        """Load index from cache or network; populate ``_papers``."""
         cached = self._cache.read_if_fresh()
         if cached:
             papers, max_rev, max_p = self._parse_raw(cached)
             if papers:
                 self._commit_index(papers, max_rev, max_p)
                 log.info("Loaded %d entries from cache", len(papers))
-                return IndexRefreshResult(self.papers, stale=False)
+                return IndexRefreshResult(dict(self._papers), stale=False)
 
         raw, category = await self._download()
         if raw:
@@ -83,7 +83,7 @@ async def refresh(self) -> IndexRefreshResult:
                 self._cache.write(raw)
                 self._commit_index(papers, max_rev, max_p)
                 log.info("Downloaded and cached %d entries", len(papers))
-                return IndexRefreshResult(self.papers, stale=False)
+                return IndexRefreshResult(dict(self._papers), stale=False)
             log.warning(
                 "INDEX-FETCH  failure_category=%s  url=%s  payload produced zero parseable papers",
                 FailureCategory.CONFIGURATION.value,
@@ -100,7 +100,7 @@ async def refresh(self) -> IndexRefreshResult:
                     "INDEX-STALE-FALLBACK  entries=%d  (download failed; using persisted cache)",
                     len(papers),
                 )
-                return IndexRefreshResult(self.papers, stale=True)
+                return IndexRefreshResult(dict(self._papers), stale=True)
 
         log.error("No index data available")
         self._raise_on_download_failure(category or FailureCategory.CONFIGURATION)
@@ -192,10 +192,17 @@ def _parse_raw(self, raw: dict) -> tuple[dict[str, Paper], dict[int, int], int]:
         return papers, max_rev, max_p
 
     def _commit_index(self, papers: dict[str, Paper], max_rev: dict[int, int], max_p: int) -> None:
-        self.papers = papers
-        self._max_rev = max_rev
+        """Atomically publish a new index snapshot.
+
+        Assigns a new dict reference so readers see either the old or new map,
+        never a partially-built one. Callers must read via ``paper_count()``,
+        ``get_papers_snapshot()``, or ``get_known_paper_ids()`` — not the live
+        ``_papers`` dict, especially from other threads.
+        """
+        self._papers = dict(papers)
+        self._max_rev = dict(max_rev)
         self._max_p = max_p
-        self._sorted_p_nums = sorted(max_rev.keys())
+        self._sorted_p_nums = sorted(self._max_rev.keys())
 
     def _parse_and_index(self, raw: dict) -> dict[str, Paper]:
         papers, max_rev, max_p = self._parse_raw(raw)
@@ -238,11 +245,19 @@ def known_p_numbers(self) -> set[int]:
 
     def get_known_paper_ids(self) -> frozenset[str]:
         """Immutable snapshot of all paper ids currently in the index."""
-        return frozenset(self.papers)
+        return frozenset(self._papers)
 
     def get_papers_snapshot(self) -> Mapping[str, Paper]:
-        """Read-only mapping copy of ``papers`` (not the live dict object)."""
-        return MappingProxyType(dict(self.papers))
+        """Read-only mapping copy of the index (not the live dict object)."""
+        return MappingProxyType(dict(self._papers))
+
+    def paper_count(self) -> int:
+        """Number of papers in the current index snapshot."""
+        return len(self._papers)
+
+    def _replace_papers_for_test(self, papers: Mapping[str, Paper]) -> None:
+        """Test-only atomic swap of the papers map (does not recompute ``_max_rev``)."""
+        self._papers = dict(papers)
 
     async def fetch(self) -> IndexRefreshResult:
         """DataSource: load index and return papers plus stale signal."""

tests/test_callback_protocols.py

@@ -65,7 +65,7 @@ def on_ops_alert(message: str) -> None:
     assert isinstance(scheduler.notify_callback, NotifyCallback)
     assert isinstance(scheduler.ops_alert_fn, OpsAlertFn)
 
-    wg21.papers = {}
+    wg21._papers = {}
     await scheduler.poll_once()
     assert notified == []
 

tests/test_monitor.py

@@ -234,10 +234,11 @@ def test_rejects_invalid_per_user_matches_value(self):
 def _make_mock_wg21() -> MagicMock:
     mock = MagicMock(spec=WG21Index)
     mock.source_id = SOURCE_WG21_INDEX
-    mock.papers = {}
+    mock._papers = {}
+    mock.paper_count = lambda: len(mock._papers)
 
     async def _fetch():
-        return IndexRefreshResult(dict(mock.papers), stale=False)
+        return IndexRefreshResult(dict(mock._papers), stale=False)
 
     mock.fetch = AsyncMock(side_effect=_fetch)
     mock.diff = lambda previous, current: diff_snapshots(previous or {}, current)
@@ -320,7 +321,7 @@ async def test_poll_once_detects_new_papers(self, fake_pool):
         await scheduler.poll_once()
 
         new_paper = Paper(id="P9999R0", title="New", author="Author", date="2024-01-01")
-        index.papers = {"P9999R0": new_paper}
+        index._papers = {"P9999R0": new_paper}
         _set_iso_cycle(prober, _empty_cycle())
         result = await scheduler.poll_once()
         assert len(result.diff.new_papers) == 1
@@ -331,7 +332,7 @@ async def test_poll_once_surfaces_only_recent_probe_hits(self, fake_pool):
 
         recent = _recent_hit()
         old = _old_hit()
-        index.papers = {}
+        index._papers = {}
         _set_iso_cycle(prober, _success_cycle([recent, old]))
         result = await scheduler.poll_once()
         assert len(result.probe_hits) == 1
@@ -350,7 +351,7 @@ async def test_poll_once_detects_dp_transition(self, fake_pool):
             author="Author",
             date="2025-01-01",
         )
-        index.papers = {"P9999R0": new_paper}
+        index._papers = {"P9999R0": new_paper}
         _set_iso_cycle(prober, _empty_cycle())
 
         result = await scheduler.poll_once()
@@ -365,7 +366,7 @@ async def test_poll_once_dp_skip_non_p_papers(self, fake_pool):
         await scheduler.poll_once()
 
         n_paper = Paper(id="N4950", title="Working Draft", author="Ed", date="2025-01-01")
-        index.papers = {"N4950": n_paper}
+        index._papers = {"N4950": n_paper}
         _set_iso_cycle(prober, _empty_cycle())
 
         result = await scheduler.poll_once()
@@ -376,7 +377,7 @@ async def test_poll_once_no_dp_transition_when_no_draft(self, fake_pool):
         await scheduler.poll_once()
 
         new_paper = Paper(id="P8888R0", title="Entirely New", author="X", date="2025-01-01")
-        index.papers = {"P8888R0": new_paper}
+        index._papers = {"P8888R0": new_paper}
         _set_iso_cycle(prober, _empty_cycle())
 
         result = await scheduler.poll_once()
@@ -391,7 +392,7 @@ async def test_poll_once_dp_transition_logged(self, fake_pool, caplog):
         draft_url = "https://isocpp.org/files/papers/D7777R0.pdf"
         state.mark_discovered(draft_url)
         new_paper = Paper(id="P7777R0", title="X", author="Y", date="2025-01-01")
-        index.papers = {"P7777R0": new_paper}
+        index._papers = {"P7777R0": new_paper}
         _set_iso_cycle(prober, _empty_cycle())
 
         with caplog.at_level(logging.INFO):
@@ -416,7 +417,7 @@ async def test_poll_once_logs_updated_papers(self, fake_pool, caplog):
         old_paper = Paper(id="P9999R0", title="Old Title", author="A", date="2024-01-01")
         scheduler._snapshots[SOURCE_WG21_INDEX] = {"P9999R0": old_paper}
         updated_paper = Paper(id="P9999R0", title="New Title", author="A", date="2024-01-01")
-        index.papers = {"P9999R0": updated_paper}
+        index._papers = {"P9999R0": updated_paper}
         _set_iso_cycle(prober, _empty_cycle())
         with caplog.at_level(logging.DEBUG):
             await scheduler.poll_once()
@@ -428,7 +429,7 @@ async def test_poll_old_hits_logged(self, fake_pool, caplog):
         scheduler, index, prober, _, _ = _make_scheduler(fake_pool)
         await scheduler.poll_once()
         old = _old_hit()
-        index.papers = {}
+        index._papers = {}
         _set_iso_cycle(prober, _success_cycle([old]))
         with caplog.at_level(logging.INFO):
             result = await scheduler.poll_once()
@@ -440,7 +441,7 @@ async def test_poll_once_populates_per_user_matches(self, fake_pool):
         await scheduler.poll_once()
 
         new_paper = Paper(id="P9999R0", title="Senders", author="Eric Niebler", date="2024-01-01")
-        index.papers = {"P9999R0": new_paper}
+        index._papers = {"P9999R0": new_paper}
         _set_iso_cycle(prober, _empty_cycle())
 
         user_watchlist.matches_for_users.return_value = {
@@ -456,7 +457,7 @@ async def test_poll_once_per_user_probe_hit(self, fake_pool):
 
         hit = _recent_hit(front_text="written by eric niebler")
         _set_iso_cycle(prober, _success_cycle([hit]))
-        index.papers = {}
+        index._papers = {}
 
         user_watchlist.matches_for_users.return_value = {
             "U123": PerUserMatches(papers=[], probe_hits=[(hit, MatchReason.AUTHOR)])
@@ -543,7 +544,7 @@ async def test_seed_does_not_log_index_new_on_baseline(self, fake_pool, caplog):
 
         scheduler, wg21, _, _, _ = _make_scheduler(fake_pool)
         paper = Paper(id="P9999R0", title="Baseline", author="Author", date="2024-01-01")
-        wg21.papers = {"P9999R0": paper}
+        wg21._papers = {"P9999R0": paper}
 
         with caplog.at_level(logging.INFO):
             await scheduler.seed()
@@ -702,7 +703,7 @@ async def test_failed_probe_cycle_does_not_advance_last_successful_poll_normal_p
         scheduler, index, prober, _, _ = _make_scheduler(fake_pool)
         await scheduler.poll_once()
         before = scheduler._last_successful_poll
-        index.papers = {}
+        index._papers = {}
         _set_iso_cycle(prober, _failed_cycle("network down"))
         await scheduler.poll_once()
         assert scheduler._last_successful_poll == before
@@ -832,7 +833,7 @@ async def test_poll_once_propagates_index_refresh_error(self, fake_pool):
             with pytest.raises(IndexRefreshError) as exc_info:
                 await scheduler.poll_once()
         assert exc_info.value.category is FailureCategory.TIMEOUT
-        assert wg21.papers == {}
+        assert wg21.paper_count() == 0
 
     async def test_poll_once_uses_stale_index(self, fake_pool, caplog):
         scheduler, wg21, prober, _, _ = _make_scheduler_with_real_wg21(fake_pool)
@@ -846,7 +847,7 @@ async def test_poll_once_uses_stale_index(self, fake_pool, caplog):
             mock_cls.return_value.__aexit__ = AsyncMock(return_value=False)
             with caplog.at_level(logging.WARNING, logger="paperscout.monitor"):
                 await scheduler.poll_once()
-            assert "P2300R10" in wg21.papers
+            assert "P2300R10" in wg21.get_known_paper_ids()
             assert "INDEX-STALE" in caplog.text
             assert scheduler._last_successful_poll is None
             prober.fetch.assert_called_once()