fix: g8r2h/yh044 review folds — freshness probe covers highlights, store lifecycle, message precision (critique [21092])

Critic Critical: highlights written locally in the migration-to-fix
window were stranded AND guided-upgrade's freshness probe falsely
reported the aspects slot clean (it anchored only on
document_aspects.extracted_at while migrate_highlights ships
document_highlights too). FRESHNESS_PROBES is now multi-probe per slot:
freshness confirms only when EVERY probed table answers no-newer-writes;
the stranded-window shape is regression-pinned. Recovery on an affected
box is now just re-running guided-upgrade (the probe sees the newer
ingested_at and re-ships the slot).

Reviewer High: dt highlights stores are closed per use (try/finally,
matching the _open_plan_library pattern actually cited) — the batch
loop leaked one httpx pool per DEVONthink record. Reader distinguishes
'service unavailable' from 'no highlights ingested' (ClickException,
never a raw traceback). Salience boost: first failure per process warns
with the consequence named (a sustained outage of the now-networked
read must not be silently dead); wrong no-close comment corrected
(RefreshableHttpStoreMixin.close IS load-bearing). yh044 refusal
message branches on frozen-source vs fresh-install shape; dry-run
refusal separately pinned. Service-mode salience routing + close both
test-pinned. 141 affected tests green.

Author: Hellblazer

src/nexus/commands/catalog_cmds/maintenance.py

@@ -194,19 +194,29 @@ def chash_reconcile_cmd(apply: bool) -> None:
     from nexus.db.storage_mode import StorageBackend, storage_backend_for  # noqa: PLC0415 — deferred import; rare/branch-local path
     from nexus.db.t2.chash_index import ChashIndex  # noqa: PLC0415 — deferred import; rare/branch-local path or circular-dep / startup-cost avoidance
 
-    # nexus-yh044: refuse BEFORE any work on a migrated install — the local
-    # SQLite chash_index there is the frozen migration source (RDR-176
-    # immutability, rollback target until P4b), and the "No T2 db" guard
-    # below does NOT cover migrated-in-place boxes where the file survives.
+    # nexus-yh044: refuse BEFORE any work on a service-backed install — the
+    # "No T2 db" guard below does NOT cover migrated-in-place boxes where
+    # the file survives frozen. Message branches on which shape this box is
+    # (critic Significant: SERVICE is the hard default for FRESH installs
+    # too, where "frozen migration source" would be a false premise).
     if storage_backend_for("chash_index") == StorageBackend.SERVICE:
+        db_path = default_db_path()
+        if db_path.exists():
+            raise click.ClickException(
+                "chash-reconcile is a PRE-MIGRATION repair verb and this "
+                "install is service-backed. The local SQLite chash_index "
+                "here is the FROZEN MIGRATION SOURCE (RDR-176: immutable "
+                "rollback target until RDR-155 P4b) — reconciling it "
+                "against live T3 would manufacture ghosts and --apply would "
+                "delete source rows. The PG side needs no reconciliation "
+                "(RDR-187: the router is retired; chunk tables cannot go "
+                "stale)."
+            )
         raise click.ClickException(
-            "chash-reconcile is a PRE-MIGRATION repair verb and this install "
-            "is service-backed. The local SQLite chash_index here is the "
-            "FROZEN MIGRATION SOURCE (RDR-176: immutable rollback target "
-            "until RDR-155 P4b) — reconciling it against live T3 would "
-            "manufacture ghosts and --apply would delete source rows. The "
-            "PG side needs no reconciliation (RDR-187: the router is "
-            "retired; chunk tables cannot go stale)."
+            "chash-reconcile is a PRE-MIGRATION repair verb: this install "
+            "is service-backed and has no local T2 database — there is "
+            "nothing to reconcile. The PG side needs no reconciliation "
+            "(RDR-187: the router is retired; chunk tables cannot go stale)."
         )
 
     db_path = default_db_path()

src/nexus/commands/dt.py

@@ -326,17 +326,28 @@ def _ingest_highlights_record(uuid: str) -> bool:
         # the previous direct DocumentHighlights construction wrote local
         # SQLite even on migrated boxes. Low contention either way: one
         # write per indexed record, not a long-lived worker (RDR-128 N/A).
+        # Reviewer High fold: close per use (try/finally, like every
+        # _open_plan_library caller) — this runs inside the dt-index batch
+        # loop, and an unclosed HttpDocumentHighlightsStore leaks one httpx
+        # connection pool PER RECORD. Construction itself is cheap (local
+        # env/lease resolution, no network), so per-call open/close is the
+        # minimal leak-free shape.
         store = _open_highlights_store()
         from datetime import datetime, timezone  # noqa: PLC0415 — stdlib deferred to call site (datetime)
 
-        return store.upsert(HighlightRecord(
-            doc_id=str(entry.tumbler),
-            source_uri=dt_uri,
-            collection=getattr(entry, "physical_collection", "") or "",
-            highlights_md=highlights_md,
-            mentions_md=mentions_md,
-            ingested_at=datetime.now(timezone.utc).isoformat(),
-        ))
+        try:
+            return store.upsert(HighlightRecord(
+                doc_id=str(entry.tumbler),
+                source_uri=dt_uri,
+                collection=getattr(entry, "physical_collection", "") or "",
+                highlights_md=highlights_md,
+                mentions_md=mentions_md,
+                ingested_at=datetime.now(timezone.utc).isoformat(),
+            ))
+        finally:
+            _close = getattr(store, "close", None)
+            if callable(_close):
+                _close()
     except Exception as e:  # noqa: BLE001 — DEVONthink boundary op is best-effort; failure logged via log.warning
         _log.warning("dt_highlights_failed", uuid=uuid, error=str(e))
         return False
@@ -1009,15 +1020,28 @@ def highlights_cmd(tumbler_or_uuid: str) -> None:
     ``document_highlights`` T2 table populated by ``nx dt index --highlights``.
     This is a pure T2 read — DEVONthink need not be running.
     """
-    store = _open_highlights_store()
-    if _UUID_RE.match(tumbler_or_uuid):
-        rec = store.get_by_source_uri(f"x-devonthink-item://{tumbler_or_uuid}")
-    elif _TUMBLER_RE.match(tumbler_or_uuid):
-        rec = store.get(tumbler_or_uuid)
-    else:
+    if not (_UUID_RE.match(tumbler_or_uuid) or _TUMBLER_RE.match(tumbler_or_uuid)):
         raise click.ClickException(
             "argument is neither a tumbler (e.g. 1.2.3) nor a UUID.",
         )
+    store = _open_highlights_store()
+    try:
+        if _UUID_RE.match(tumbler_or_uuid):
+            rec = store.get_by_source_uri(f"x-devonthink-item://{tumbler_or_uuid}")
+        else:
+            rec = store.get(tumbler_or_uuid)
+    except click.ClickException:
+        raise
+    except Exception as exc:  # noqa: BLE001 — reviewer Medium (nexus-g8r2h): a connection-class failure must read as "service unavailable", never a raw traceback indistinguishable from "no highlights"
+        raise click.ClickException(
+            f"highlights store unavailable ({type(exc).__name__}: {exc}) — "
+            "check `nx doctor` / service status. This is NOT 'no highlights "
+            "ingested'."
+        ) from exc
+    finally:
+        _close = getattr(store, "close", None)
+        if callable(_close):
+            _close()
     if rec is None:
         raise click.ClickException(
             f"no ingested highlights for {tumbler_or_uuid} "

src/nexus/migration/guided_upgrade.py

@@ -357,17 +357,27 @@ def detect_pending_migration_memoized() -> PreflightDetection:
 #: probe is used only to decide whether to SKIP re-shipping, never to
 #: confirm correctness, and a false skip is still safe (copy-not-move — the
 #: source is never at risk, only re-verified less eagerly).
-FRESHNESS_PROBES: dict[str, tuple[str, str]] = {
-    "memory": ("memory", "timestamp"),
-    "plans": ("plans", "created_at"),
-    "telemetry": ("search_telemetry", "ts"),
-    "taxonomy": ("topics", "created_at"),
-    "aspects": ("document_aspects", "extracted_at"),
+FRESHNESS_PROBES: dict[str, tuple[tuple[str, str], ...]] = {
+    "memory": (("memory", "timestamp"),),
+    "plans": (("plans", "created_at"),),
+    "telemetry": (("search_telemetry", "ts"),),
+    "taxonomy": (("topics", "created_at"),),
+    # nexus-g8r2h critic Critical (critique [21092]): document_highlights
+    # rides the "aspects" ETL slot (aspects_etl.migrate_highlights) but the
+    # probe never anchored on it — a highlights-only local write (the
+    # pre-g8r2h service-mode writer bug's stranded window) read as "no newer
+    # local writes": a CONFIDENTLY FALSE clean, worse than no signal. Every
+    # table an ETL slot ships must be probed; multi-probe slots confirm
+    # freshness only when EVERY probe answers "no newer writes".
+    "aspects": (
+        ("document_aspects", "extracted_at"),
+        ("document_highlights", "ingested_at"),
+    ),
     # "chash" probe removed (RDR-187/nexus-piwya.9 sweep): unreachable since
     # .10 dropped the store from LADDER_ORDER; detect_already_migrated never
     # consults it.
-    "catalog": ("documents", "indexed_at"),
-    "aspects_queue": ("aspect_extraction_queue", "enqueued_at"),
+    "catalog": (("documents", "indexed_at"),),
+    "aspects_queue": (("aspect_extraction_queue", "enqueued_at"),),
 }
 
 
@@ -537,17 +547,23 @@ def _decide_store(
             f"{verification!r} — will migrate",
         )
 
-    probe = FRESHNESS_PROBES.get(store)
+    probes = FRESHNESS_PROBES.get(store) or ()
     freshness_confirmed = False
-    if probe is not None and completed_at:
-        newer = _has_newer_local_writes(sqlite_path, probe, completed_at)
-        if newer is True:
+    if probes and completed_at:
+        answers = [
+            _has_newer_local_writes(sqlite_path, probe, completed_at)
+            for probe in probes
+        ]
+        if any(a is True for a in answers):
             return StoreMigrationStatus(
                 store, False,
                 f"{store}: local writes newer than the report ({completed_at}) "
                 "— will migrate",
             )
-        freshness_confirmed = newer is False
+        # Confirmed only when EVERY probe positively answered "no newer
+        # writes" — a missing/unreadable table (None) degrades to
+        # trust-the-report, never to a confident clean (nexus-g8r2h fold).
+        freshness_confirmed = bool(answers) and all(a is False for a in answers)
 
     if freshness_confirmed:
         line = f"{store}: already migrated {completed_at}, no newer local writes"

src/nexus/search_engine.py

@@ -876,8 +876,22 @@ def _search_one(col: str) -> dict:
                 query=query,
                 weight=float(ag_cfg.get("weight", 0.025)),
             )
-        except Exception:  # noqa: BLE001 — best-effort salience boost; failure logged at debug, results returned unboosted
-            _log.debug("salience_boost_failed", exc_info=True)
+        except Exception:  # noqa: BLE001 — best-effort salience boost; failure logged, results returned unboosted
+            # nexus-g8r2h critique fold: post-routing this guards a real HTTP
+            # round-trip on service boxes, not a rarely-failing local SQLite
+            # read. Per-call stays quiet, but the FIRST failure per process
+            # warns — a sustained outage must not be silently dead for weeks.
+            global _salience_failure_warned
+            if not _salience_failure_warned:
+                _salience_failure_warned = True
+                _log.warning(
+                    "salience_boost_failed_first",
+                    consequence="attention_guided_v1 boost inactive for this "
+                                "process (subsequent failures log at debug)",
+                    exc_info=True,
+                )
+            else:
+                _log.debug("salience_boost_failed", exc_info=True)
 
     # nexus-1qed: catalog-resolved display path attached as metadata
     # so formatters never need to import the catalog. Best-effort;
@@ -1057,6 +1071,11 @@ def _flag_contradictions(
     return out
 
 
+#: nexus-g8r2h critique fold: first-failure-per-process latch for the
+#: salience boost's swallow (see the caller's except arm).
+_salience_failure_warned: bool = False
+
+
 def _apply_salience_boost(
     results: list[SearchResult],
     *,
@@ -1117,7 +1136,10 @@ def _apply_salience_boost(
             if boost:
                 r.hybrid_score = float(r.hybrid_score) + boost
     finally:
-        # HttpDocumentAspectsStore has no close(); the SQLite store does.
+        # Both stores close(): the SQLite store closes its connection and
+        # HttpDocumentAspectsStore inherits close() from
+        # RefreshableHttpStoreMixin (closes the httpx pool — load-bearing,
+        # not a no-op; reviewer Low corrected the earlier claim here).
         close = getattr(aspects, "close", None)
         if callable(close):
             close()

tests/migration/test_guided_upgrade_already_migrated.py

@@ -94,9 +94,12 @@ def _make_sqlite(tmp_path: Path, *, rows: dict[str, str], filename: str = "t2.db
     conn = sqlite3.connect(db)
     try:
         for store, ts in rows.items():
-            table, column = FRESHNESS_PROBES[store]
-            conn.execute(f"CREATE TABLE {table} ({column} TEXT)")  # noqa: S608 — fixed internal test fixture, no external input
-            conn.execute(f"INSERT INTO {table} ({column}) VALUES (?)", (ts,))
+            # Multi-probe stores (nexus-g8r2h: "aspects" also probes
+            # document_highlights) get every probe table stamped at *ts* so
+            # all-probes-must-answer freshness semantics hold in fixtures.
+            for table, column in FRESHNESS_PROBES[store]:
+                conn.execute(f"CREATE TABLE {table} ({column} TEXT)")  # noqa: S608 — fixed internal test fixture, no external input
+                conn.execute(f"INSERT INTO {table} ({column}) VALUES (?)", (ts,))
         conn.commit()
     finally:
         conn.close()
@@ -379,3 +382,38 @@ def test_newer_clean_report_recovers_from_an_older_dirty_one(
         plan = detect_already_migrated(sqlite_path=db, reports_dir=reports_dir)
 
         assert plan.all_skipped is True
+
+
+def test_highlights_only_local_write_defeats_false_clean(tmp_path: Path) -> None:
+    """nexus-g8r2h critic Critical (critique [21092]): a document_highlights
+    row NEWER than the report — with document_aspects untouched (the exact
+    stranded-window shape the pre-fix service-mode writer produced) — must
+    read 'will migrate', never 'already migrated ... no newer local writes'.
+    Pre-fold, the aspects slot probed only document_aspects.extracted_at and
+    reported a confidently false clean."""
+    import sqlite3 as _sqlite3
+
+    from nexus.migration.guided_upgrade import _decide_store
+
+    db = tmp_path / "t2.db"
+    conn = _sqlite3.connect(db)
+    try:
+        conn.execute("CREATE TABLE document_aspects (extracted_at TEXT)")
+        conn.execute(
+            "INSERT INTO document_aspects (extracted_at) VALUES (?)",
+            (BEFORE_T0,),
+        )
+        conn.execute("CREATE TABLE document_highlights (ingested_at TEXT)")
+        conn.execute(
+            "INSERT INTO document_highlights (ingested_at) VALUES (?)",
+            (AFTER_T0,),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+    status = _decide_store(
+        "aspects", [_report(stores={"aspects": 0})], sqlite_path=db,
+    )
+    assert status.skip is False
+    assert "newer than the report" in status.line

tests/test_chash_reconcile.py

@@ -332,3 +332,24 @@ def test_service_mode_refuses_loud(monkeypatch, tmp_path) -> None:
     assert result.exit_code != 0
     assert "pre-migration" in result.output.lower()
     assert "frozen migration source" in result.output.lower()
+
+
+def test_service_mode_refuses_dry_run_too(monkeypatch, tmp_path) -> None:
+    """The gate is unconditional on --apply: even the read-only dry-run is
+    refused on a migrated install (diffing live T3 against the frozen index
+    produces garbage ghosts — a misleading report is still harm)."""
+    from click.testing import CliRunner
+
+    from nexus.commands.catalog_cmds.maintenance import chash_reconcile_cmd
+    from nexus.db.storage_mode import StorageBackend
+
+    db = tmp_path / "memory.db"
+    db.write_bytes(b"")
+    monkeypatch.setattr("nexus.commands._helpers.default_db_path", lambda: db)
+    monkeypatch.setattr(
+        "nexus.db.storage_mode.storage_backend_for",
+        lambda store: StorageBackend.SERVICE,
+    )
+    result = CliRunner().invoke(chash_reconcile_cmd, [])
+    assert result.exit_code != 0
+    assert "pre-migration" in result.output.lower()

tests/test_rdr_109_phase5_salience.py

@@ -292,3 +292,43 @@ def test_salience_boost_no_op_when_no_doc_id(tmp_path: Path, monkeypatch) -> Non
     )
     out = _apply_salience_boost([r], query="q", weight=0.5)
     assert out == [r]
+
+
+def test_salience_boost_routes_via_http_store_in_service_mode(monkeypatch) -> None:
+    """nexus-g8r2h fold (sweep [21089] item 8): on a service-backed box the
+    boost must read salient_sentences through HttpDocumentAspectsStore —
+    the old direct DocumentAspects(memory.db) read served STALE frozen
+    pre-migration rows. Also pins that the store's close() is called (it
+    closes the httpx pool — load-bearing, reviewer Low)."""
+    from nexus.db.storage_mode import StorageBackend
+
+    calls: dict = {"salient": [], "closed": 0}
+
+    class _FakeHttpAspects:
+        def get_salient_sentences(self, doc_id: str) -> list[str]:
+            calls["salient"].append(doc_id)
+            return ["hybrid retrieval cross-encoder reranking"] if doc_id == "B" else []
+
+        def close(self) -> None:
+            calls["closed"] += 1
+
+    monkeypatch.setattr(
+        "nexus.db.storage_mode.storage_backend_for",
+        lambda store: StorageBackend.SERVICE,
+    )
+    monkeypatch.setattr(
+        "nexus.db.t2.http_document_aspects_store.HttpDocumentAspectsStore",
+        lambda: _FakeHttpAspects(),
+    )
+
+    from nexus.search_engine import _apply_salience_boost
+    results = [
+        _make_result("a", "knowledge__rag", "A", score=0.50),
+        _make_result("b", "knowledge__rag", "B", score=0.45),
+    ]
+    out = _apply_salience_boost(
+        results, query="hybrid retrieval cross-encoder", weight=0.5,
+    )
+    assert [r.id for r in out] == ["b", "a"]
+    assert sorted(calls["salient"]) == ["A", "B"]
+    assert calls["closed"] == 1