fix(metering): alarm when a billable meter emit is swallowed

reflexio/server/routes/_metering.py

@@ -12,6 +12,7 @@
 from fastapi import Request
 
 from reflexio.server.cache import reflexio_cache
+from reflexio.server.error_reporting import capture_anomaly
 
 logger = logging.getLogger(__name__)
 
@@ -62,9 +63,22 @@ def _meter_applied_learnings(
         )
         return True
     except Exception:
+        # ALARM, not just a log line. This is a billable event being dropped, and
+        # the drop is otherwise invisible: the caller ignores the return value,
+        # and `_worker_loop`'s own `search.metering.job_failed` capture can never
+        # fire because this handler swallows first. Note the asymmetry that makes
+        # it dangerous -- `_meter_search_request` below performs NO config lookup,
+        # so a config-store blip drops `learning_applied` while `search_request`
+        # keeps flowing, and the two meters silently diverge.
         logger.warning(
             "applied-learnings metering failed for org %s", org_id, exc_info=True
         )
+        capture_anomaly(
+            "search.metering.emit_failed",
+            level="error",
+            meter="applied_learnings",
+            org_id=org_id,
+        )
         return False
 
 
@@ -96,7 +110,15 @@ def _meter_search_request(
         )
         return True
     except Exception:
+        # Same reasoning as `_meter_applied_learnings`: a swallowed emit is lost
+        # billable usage, so it must reach the error reporter and not only a log.
         logger.warning(
             "search-request metering failed for org %s", org_id, exc_info=True
         )
+        capture_anomaly(
+            "search.metering.emit_failed",
+            level="error",
+            meter="search_requests",
+            org_id=org_id,
+        )
         return False

tests/server/api_endpoints/test_applied_learnings_metering.py

@@ -261,9 +261,20 @@ def _gate() -> None:
 
 
 def test_metering_failure_does_not_break_search_response() -> None:
-    """A get_reflexio error inside the metering helper must not turn a 200 into a 500."""
+    """A get_reflexio error inside the metering helper must not turn a 200 into a 500.
+
+    It must ALSO raise an alarm. A dropped emit is lost billable usage, and it is
+    otherwise undetectable: the caller ignores the helper's return value, and
+    ``_worker_loop``'s own ``search.metering.job_failed`` capture can never fire
+    because this handler swallows the exception first. A ``logger.warning`` does
+    not close that gap -- the enterprise Sentry integration is wired at
+    ``event_level=logging.ERROR`` with Sentry Logs off, so a warning produces no
+    event at all, and searching for one returns a false clean whether or not the
+    drop is happening.
+    """
     events = _capture()
     profiles = [_make_profile_view("u1")]
+    anomalies: list[tuple[str, dict[str, object]]] = []
     try:
         # The unified_search mock is wired via _patch_unified_search (first get_reflexio call).
         # Inside the helper, get_reflexio is called a second time; we make *that* call's
@@ -285,9 +296,15 @@ def test_metering_failure_does_not_break_search_response() -> None:
             RuntimeError("boom"),
         ]
 
-        with patch(
-            "reflexio.server.cache.reflexio_cache.get_reflexio",
-            return_value=mock_reflexio_search,
+        with (
+            patch(
+                "reflexio.server.cache.reflexio_cache.get_reflexio",
+                return_value=mock_reflexio_search,
+            ),
+            patch(
+                "reflexio.server.routes._metering.capture_anomaly",
+                side_effect=lambda name, **kw: anomalies.append((name, kw)),
+            ),
         ):
             resp = _client("production_agent").post(
                 "/api/search", json={"query": "x", "user_id": "u1"}
@@ -297,8 +314,13 @@ def test_metering_failure_does_not_break_search_response() -> None:
     finally:
         configure_usage_event_recorder(None)
 
-    # Metering failed silently — no learning_applied event should have been emitted.
+    # The emit really was dropped...
     assert [e for e in events if e.event_name == "learning_applied"] == []
+    # ...but NOT silently. This is the half that used to be missing, and its
+    # absence is why a revenue-affecting drop could run indefinitely unnoticed.
+    assert [name for name, _ in anomalies] == ["search.metering.emit_failed"]
+    assert anomalies[0][1]["meter"] == "applied_learnings"
+    assert anomalies[0][1]["level"] == "error"
 
 
 def test_unified_search_response_does_not_wait_for_metering_database_work(