fix: backfill transport failures and snapshot check per-database folder (T-1742, T-1743)

ROADMAP.md

@@ -647,9 +647,13 @@ bounded transcript scanner. Depends on: none.
   by default with `MEMORYMASTER_BEAT_ENABLED`; read-only `check_beat_health`
   is included in operational review.
 - [x] AC4: candidates are never applied without `--apply-candidates`.
-- [ ] Activation: blocked by self-observation risk because the beat can
-  recapture `[MemoryMaster recall]` injections. Dreaming remains unchanged
-  while off.
+- [x] Activated 2026-10-22 by operator order (T-1695): `MEMORYMASTER_BEAT_ENABLED=1`
+  and the `MemoryMaster-Beat` task every 5 min. The self-observation risk is
+  now measured by `ao2-m3-lag-7d` / `ao2-m4-guard-7d` (due 2026-10-29). Daily
+  review 2026-10-23: the beat shares the Dreaming daily caps (80 extract / 24
+  consolidate) and both were exhausted by mid-day; beat-scanned captures sit
+  behind ~700 older hook captures in the FIFO, so M3 cannot be measured until
+  the caps decision above is taken.
 
 **AO-3 Per-project projection and SessionStart delta** (M, batch B;
 design T-1638; implementation not started)

docs/generated/release-truth.json

@@ -144,7 +144,7 @@
     "console_entrypoints": 9,
     "mcp_tools": 51,
     "ops_cli_commands": 5,
-    "pytest_test_functions": 5924
+    "pytest_test_functions": 5926
   },
   "feature_profile_matrix": {
     "capture_hook": [

docs/generated/release-truth.md

@@ -13,7 +13,7 @@ Do not edit this file by hand. Run `python scripts/generate_release_truth.py`.
 - Main CLI commands: **127**
 - Operations CLI commands: **5**
 - Console entrypoints: **9**
-- Pytest source test functions: **5924**
+- Pytest source test functions: **5926**
 
 ## MCP tools
 

memorymaster/dreaming/backfill.py

@@ -19,7 +19,7 @@
 from typing import Any, Callable, Mapping
 
 from memorymaster.dreaming.backfill_guard import PauseLatch, evaluate_backfill_guard
-from memorymaster.dreaming.providers import ProviderCallError
+from memorymaster.dreaming.providers import ProviderCallError, ProviderOutputError
 from memorymaster.dreaming.backfill_quality import evaluate_cohort_quality, fingerprint, receipt_is_current
 from memorymaster.dreaming.backfill_selection import (
     complete_records, frozen_identity, iter_source_turns, order_inventory,
@@ -607,15 +607,27 @@ def _extract(self, row: dict[str, Any], run_id: str) -> None:
             self._save(row)
             try:
                 result = self.extractor.extract(messages, scope=row["scope"], capture_hash=window["identity"])
-            except ProviderCallError as exc:
-                # T-1722: a malformed or failed provider reply costs this window one attempt (three
-                # strikes make it unresolved_extraction, repairs count toward the allowance); the
-                # run goes on to the next window instead of aborting the whole campaign.
+            except ProviderOutputError as exc:
+                # T-1722: a malformed provider reply costs this window one attempt (three strikes
+                # make it unresolved_extraction, repairs count toward the allowance); the run goes
+                # on to the next window instead of aborting the whole campaign.
                 window["last_error"] = type(exc).__name__
                 if window["attempts"] >= 3:
                     row["status"] = "unresolved_extraction"
                 self._save(row)
                 continue
+            except ProviderCallError:
+                # T-1742: a transport, quota or configuration failure is not the window's fault and
+                # consumed nothing. Give the attempt and the reservation back and stop the run; the
+                # scheduler retries later. Treating it like a malformed reply burned one reservation
+                # per window up to the daily cap during a 70-minute outage, and the extra attempts
+                # would have exhausted the cohort's repair allowance on the retry.
+                window["attempts"] -= 1
+                window.pop("last_error", None)
+                self._save(row)
+                with self.ledger._connect() as conn:
+                    conn.execute("UPDATE backfill_reservations SET status='released' WHERE reservation_id=?", (reservation,))
+                raise
             window.pop("last_error", None)
             self._usage(reservation, result, run_id)
             if len(result.candidates) > 5:

memorymaster/operations/operational_review.py

@@ -731,6 +731,14 @@ def check_checkpoint_delivery(config: ReviewConfig, *, now: datetime | None = No
 
 def check_snapshot_dir(config: ReviewConfig, *, now: datetime | None = None) -> ReviewResult:
     directory = config.snapshot_dir or Path(os.environ.get("MEMORYMASTER_SNAPSHOT_DIR", "~/.memorymaster/snapshots")).expanduser()
+    if config.snapshot_dir is None:
+        # T-1743: the weekly VACUUM INTO copy lives in the per-database folder below the base,
+        # so probing the base alone reported "no snapshot files found" next to a fresh snapshot.
+        from memorymaster.stores.snapshot import vacuum_dir_for
+
+        per_db = vacuum_dir_for(config.db)
+        if per_db.is_dir():
+            directory = per_db
     try:
         if not directory.is_dir():
             return ReviewResult("snapshot_dir", Verdict.FAIL, "snapshot directory missing or unreachable")

tests/test_dreaming_backfill_cli.py

@@ -671,3 +671,45 @@ def flaky_extract(messages, *, scope, capture_hash):
     assert runner._cohort_complete(rows)
     runner.run(stage="consolidate")
     assert fixture.consolidator.calls
+
+
+def test_transport_failure_gives_back_attempt_and_reservation_and_stops_the_run(tmp_path):
+    """T-1742: a transport/quota failure (plain ProviderCallError) is not the window's fault: the
+    window keeps zero attempts and no last_error, its reservation is released so the daily budget
+    only counts real calls, and the run stops so the scheduler retries later. On 2026-10-23 a
+    70-minute outage burned 459 attempts and reservations up to the daily cap through the T-1722
+    continue path, and the extra attempts would have exhausted the repair allowance on the retry."""
+    from memorymaster.dreaming.providers import ProviderCallError
+
+    fixture = _campaign(tmp_path, sessions=2)
+    runner = fixture.runner
+    runner.inventory()
+    real_extract = fixture.extractor.extract
+    failures = []
+
+    def outage(messages, *, scope, capture_hash):
+        if not failures:
+            failures.append(capture_hash)
+            raise ProviderCallError("provider request failed with HTTP 429", http_status=429)
+        return real_extract(messages, scope=scope, capture_hash=capture_hash)
+
+    fixture.extractor.extract = outage
+    with pytest.raises(ProviderCallError):
+        runner.run(stage="extract")
+    windows = [w for row in runner._rows() for w in row["metadata"]["selection"]["windows"]]
+    failed = [w for w in windows if w["identity"] == failures[0]]
+    assert len(failed) == 1 and failed[0].get("attempts", 0) == 0 and "last_error" not in failed[0]
+    assert not any(w.get("capture_id") for w in windows)
+    with runner.ledger._connect() as conn:
+        charged = conn.execute("SELECT COUNT(*) FROM backfill_reservations WHERE stage='extract' AND status!='released'").fetchone()[0]
+        released = conn.execute("SELECT COUNT(*) FROM backfill_reservations WHERE stage='extract' AND status='released'").fetchone()[0]
+    assert (charged, released) == (0, 1)
+    # The next healthy run extracts every window with one call each and nothing counts as a repair.
+    report = runner.run(stage="extract")
+    assert report["state"] != "leased"
+    windows = [w for row in runner._rows() for w in row["metadata"]["selection"]["windows"]]
+    assert all(w.get("capture_id") and w["attempts"] == 1 for w in windows)
+    assert len(fixture.extractor.calls) == len(windows)
+    with runner.ledger._connect() as conn:
+        charged = conn.execute("SELECT COUNT(*) FROM backfill_reservations WHERE stage='extract' AND status!='released'").fetchone()[0]
+    assert charged == len(windows)

tests/test_operational_review.py

@@ -482,3 +482,31 @@ def test_backfill_burndown_is_the_last_registered_check_and_silent_when_inactive
     assert result.name == "backfill_burndown" and result.verdict is review.Verdict.PASS
     assert "inactive" in result.detail
     assert not (tmp_path / "state").exists()
+
+
+def test_snapshot_dir_probes_the_per_database_folder(tmp_path: Path, monkeypatch) -> None:
+    """T-1743: the weekly VACUUM INTO copy is written to ``<base>/<db-stem>-<hash>/``; without an
+    explicit snapshot_dir the check must look there, and fall back to the base when that folder
+    does not exist. The installed review reported 'no snapshot files found' next to a one-day-old
+    snapshot because it listed only the base."""
+    import os
+
+    from memorymaster.stores.snapshot import vacuum_dir_for
+
+    now = datetime(2026, 10, 23, 12, tzinfo=timezone.utc)
+    monkeypatch.setattr(review, "_utc", lambda value=None: now if value is None else value.astimezone(timezone.utc))
+    base = tmp_path / "snapshots"
+    base.mkdir()
+    monkeypatch.setenv("MEMORYMASTER_SNAPSHOT_DIR", str(base))
+    db = tmp_path / "memorymaster.db"
+    config = review.ReviewConfig(db=db)
+    assert review.check_snapshot_dir(config, now=now).verdict is review.Verdict.WARN  # base empty, no per-db folder
+    per_db = vacuum_dir_for(db)
+    per_db.mkdir()
+    snapshot = per_db / "mm-20261022.db"
+    snapshot.write_bytes(b"snapshot")
+    os.utime(snapshot, (now.timestamp() - 86400, now.timestamp() - 86400))
+    result = review.check_snapshot_dir(config, now=now)
+    assert result.verdict is review.Verdict.PASS and result.counts == {"snapshot_files": 1, "newest_age_hours": 24}
+    # An explicit snapshot_dir is probed as given (the base), never redirected.
+    assert review.check_snapshot_dir(review.ReviewConfig(db=db, snapshot_dir=base), now=now).verdict is review.Verdict.WARN