fix(observability): address OTel review findings

Adversarial review of the merged OTel instrumentation (#352) surfaced
four issues, all on the enabled path (default-off, so no impact until
tracing is turned on):

- search LLM client was unwrapped, so hybrid/agentic token usage — the
  heaviest LLM spend — never reached Langfuse. Wrap it with
  UsageRecordingClient when observability is enabled, mirroring
  get_llm_client(); graceful keyword-only degradation is preserved.
- set_generation_usage overwrote token counts, undercounting any span
  that wraps more than one chat call (the now-wrapped agentic path).
  Accumulate instead of replacing.
- recall_hit was emitted for uncalibrated methods (unbounded BM25 /
  single-route vector), a near-constant always-hit signal that inflates
  dashboards. Gate hit on calibrated methods (HYBRID/AGENTIC); keyword
  and vector emit only the raw top_score.
- init_tracing / init_score_sink were not idempotent — a re-init without
  an intervening shutdown orphaned the export thread + OTLP socket +
  worker task. Tear down the previous instance first (init_score_sink is
  now async).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/everos/core/lifespan/tracing_lifespan.py

@@ -40,7 +40,7 @@ class TracingLifespanProvider(LifespanProvider):
         """
         settings = load_settings().observability
         enabled = init_tracing(settings)
-        scores = init_score_sink(settings)
+        scores = await init_score_sink(settings)
         logger.info("tracing_lifespan_startup", enabled=enabled, scores=scores)
         return enabled
 

src/everos/core/observability/tracing/attributes.py

@@ -181,15 +181,26 @@ def set_generation_usage(
 ) -> None:
     """Record ``gen_ai.*`` model + token attributes on the current span.
 
+    Token counts ACCUMULATE: a span that wraps more than one ``chat`` call
+    (e.g. the agentic/rank search path issues several LLM calls under one
+    ``everos.search.rank`` span) sums each call's usage rather than letting
+    the last call overwrite the rest. ``model`` is set as-is (last wins).
+
     No-op when OTel is absent or there is no active recording span, so LLM
     client wrappers can call it unconditionally.
     """
     if not _OTEL_AVAILABLE:
         return
     span = _otel_trace.get_current_span()
+    # `.attributes` exists on a recording SDK span and reflects values set
+    # earlier in this span's life; a non-recording span has none (getattr
+    # falls back to {}), and set_attribute on it is itself a no-op.
+    existing = getattr(span, "attributes", None) or {}
     if model is not None:
         span.set_attribute(GEN_AI_REQUEST_MODEL, model)
     if input_tokens is not None:
-        span.set_attribute(GEN_AI_USAGE_INPUT_TOKENS, input_tokens)
+        prior = existing.get(GEN_AI_USAGE_INPUT_TOKENS, 0)
+        span.set_attribute(GEN_AI_USAGE_INPUT_TOKENS, prior + input_tokens)
     if output_tokens is not None:
-        span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, output_tokens)
+        prior = existing.get(GEN_AI_USAGE_OUTPUT_TOKENS, 0)
+        span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, prior + output_tokens)

src/everos/core/observability/tracing/provider.py

@@ -117,6 +117,11 @@ def init_tracing(
         logger.warning("observability_enabled_but_otel_not_installed")
         return False
 
+    # Idempotent: a re-init without an intervening shutdown would otherwise
+    # orphan the previous provider's export thread + OTLP socket.
+    if _provider is not None:
+        shutdown_tracing()
+
     from everos import __version__
 
     resource = Resource.create(

src/everos/core/observability/tracing/scores.py

@@ -117,11 +117,14 @@ class RecallScoreSink:
 _sink: RecallScoreSink | None = None
 
 
-def init_score_sink(settings: ObservabilitySettings) -> bool:
+async def init_score_sink(settings: ObservabilitySettings) -> bool:
     """Build + start the sink when Langfuse creds + emit_recall_scores are set.
 
     Returns True if a sink was installed, False otherwise (disabled, scores
     off, or missing creds) — in which case ``emit_recall_scores`` is a no-op.
+
+    Idempotent: a re-init without an intervening shutdown tears down the
+    previous sink first, so its worker task + httpx client are not orphaned.
     """
     global _sink
     if not settings.enabled or not settings.emit_recall_scores:
@@ -132,6 +135,9 @@ def init_score_sink(settings: ObservabilitySettings) -> bool:
     if not (pk and sk and host):
         return False
 
+    if _sink is not None:
+        await shutdown_score_sink()
+
     import httpx
 
     endpoint = host.rstrip("/") + "/api/public/scores"
@@ -156,10 +162,15 @@ def emit_recall_scores(
     trace_id: str,
     observation_id: str | None,
     top_score: float,
-    hit: bool,
+    hit: bool | None,
     method: str,
 ) -> None:
-    """Enqueue recall_top_score + recall_hit; no-op when the sink is off."""
+    """Enqueue recall_top_score (always) + recall_hit (only when ``hit`` is
+    a verdict); no-op when the sink is off.
+
+    ``hit=None`` means the method's score is uncalibrated (KEYWORD/VECTOR),
+    so no hit verdict is pushed — only the raw top score.
+    """
     if _sink is None:
         return
     comment = f"method={method}"
@@ -168,11 +179,12 @@ def emit_recall_scores(
             trace_id, observation_id, "recall_top_score", float(top_score), comment
         )
     )
-    _sink.enqueue(
-        ScoreRecord(
-            trace_id, observation_id, "recall_hit", 1.0 if hit else 0.0, comment
+    if hit is not None:
+        _sink.enqueue(
+            ScoreRecord(
+                trace_id, observation_id, "recall_hit", 1.0 if hit else 0.0, comment
+            )
         )
-    )
 
 
 async def shutdown_score_sink() -> None:

src/everos/memory/search/manager.py

@@ -139,6 +139,14 @@ def _top_score(data: SearchData) -> float:
     return max((item.score for item in items), default=0.0)
 
 
+# Methods whose top score is calibrated to a comparable [0, 1] scale
+# (HYBRID → LR sigmoid, AGENTIC → cross-encoder), so the recall_hit
+# threshold is meaningful. KEYWORD (unbounded BM25) and single-route
+# VECTOR are excluded — a fixed threshold there yields a misleading
+# near-constant "hit" that inflates cross-method dashboards.
+_CALIBRATED_METHODS = frozenset({SearchMethod.HYBRID, SearchMethod.AGENTIC})
+
+
 class SearchManager:
     """Orchestrates per-kind recall, fusion, and shape into the public DTO."""
 
@@ -224,11 +232,16 @@ class SearchManager:
 
             # Recall-quality signal on the span (always on; the Langfuse
             # scores push is separate and gated on creds — see the score sink).
+            # `hit` is only meaningful for calibrated-score methods; leave it
+            # unset for KEYWORD/VECTOR so an unbounded score isn't forced
+            # through a fixed threshold into a misleading always-hit verdict.
             top_score = _top_score(data)
-            threshold = load_settings().observability.recall_hit_threshold
-            hit = top_score >= threshold
             span.set_attribute("everos.search.top_score", top_score)
-            span.set_attribute("everos.search.hit", hit)
+            hit: bool | None = None
+            if req.method in _CALIBRATED_METHODS:
+                threshold = load_settings().observability.recall_hit_threshold
+                hit = top_score >= threshold
+                span.set_attribute("everos.search.hit", hit)
 
             # Push recall-quality scores to Langfuse out-of-band (no-op unless
             # a score sink is configured); attach to this retriever span.

src/everos/service/search.py

@@ -105,7 +105,8 @@ def _get_llm_client() -> LLMClient | None:
     from everos.component.llm import build_llm_provider
     from everos.config import load_settings
 
-    cfg = load_settings().llm
+    settings = load_settings()
+    cfg = settings.llm
     if not cfg.api_key or not cfg.api_key.get_secret_value() or not cfg.base_url:
         logger.warning(
             "llm_not_configured",
@@ -113,7 +114,15 @@ def _get_llm_client() -> LLMClient | None:
         )
         _llm_client = None
     else:
-        _llm_client = build_llm_provider(cfg)
+        client = build_llm_provider(cfg)
+        # Record token usage for the hybrid/agentic path, mirroring
+        # get_llm_client() — otherwise the heaviest LLM spend (query
+        # decomposition + rerank judge) is invisible in Langfuse.
+        if settings.observability.enabled:
+            from everos.component.llm._usage_client import UsageRecordingClient
+
+            client = UsageRecordingClient(client)
+        _llm_client = client
         logger.info("search_llm_built", model=cfg.model)
     _llm_resolved = True
     return _llm_client

tests/unit/test_core/test_observability/test_scores.py

@@ -10,7 +10,41 @@ from __future__ import annotations
 
 import asyncio
 
-from everos.core.observability.tracing.scores import RecallScoreSink, ScoreRecord
+from pydantic import SecretStr
+
+from everos.config.settings import ObservabilitySettings
+from everos.core.observability.tracing.scores import (
+    RecallScoreSink,
+    ScoreRecord,
+    init_score_sink,
+    shutdown_score_sink,
+)
+
+
+async def test_init_score_sink_tears_down_previous() -> None:
+    """Re-init without an intervening shutdown must not leak the previous
+    sink's worker task + httpx client: the old sink is stopped first."""
+    settings = ObservabilitySettings(
+        enabled=True,
+        emit_recall_scores=True,
+        langfuse_public_key="pk",
+        langfuse_secret_key=SecretStr("sk"),
+        langfuse_host="https://us.cloud.langfuse.com",
+    )
+    from everos.core.observability.tracing import scores as scores_mod
+
+    try:
+        assert await init_score_sink(settings) is True
+        first = scores_mod._sink
+        assert first is not None and first._task is not None
+
+        assert await init_score_sink(settings) is True
+        # The previous sink was torn down (stop() nulls its task) and a new
+        # sink installed in its place.
+        assert first._task is None
+        assert scores_mod._sink is not first
+    finally:
+        await shutdown_score_sink()
 
 
 async def test_worker_sends_payload_in_langfuse_shape() -> None:

tests/unit/test_core/test_observability/test_span_attributes.py

@@ -83,6 +83,20 @@ def test_set_generation_usage_annotates_current_span(
     assert attrs["gen_ai.usage.output_tokens"] == 22
 
 
+def test_set_generation_usage_accumulates_across_calls(
+    captured: InMemorySpanExporter,
+) -> None:
+    # A multi-call operation (e.g. agentic search issuing several chats)
+    # inside one span must SUM token usage, not overwrite with the last call.
+    with memory_span("everos.search.rank", observation_type="generation"):
+        set_generation_usage(model="gpt-x", input_tokens=10, output_tokens=5)
+        set_generation_usage(model="gpt-x", input_tokens=3, output_tokens=7)
+    force_flush()
+    attrs = captured.get_finished_spans()[0].attributes
+    assert attrs["gen_ai.usage.input_tokens"] == 13
+    assert attrs["gen_ai.usage.output_tokens"] == 12
+
+
 def test_set_generation_usage_outside_span_is_noop() -> None:
     # No active span → must not raise (and nothing to record).
     set_generation_usage(model="x", input_tokens=1, output_tokens=2)

tests/unit/test_core/test_observability/test_tracing.py

@@ -125,6 +125,36 @@ async def test_child_spans_nest_across_asyncio_gather(
         assert spans[child].context.trace_id == trace_id
 
 
+def test_init_tracing_tears_down_previous_provider() -> None:
+    """Re-init without an intervening shutdown must not leak the previous
+    provider (its export thread + OTLP socket): the old provider is shut
+    down first, so its span processor receives ``shutdown()``."""
+    from opentelemetry.sdk.trace import SpanProcessor
+
+    class _SpyProcessor(SpanProcessor):
+        def __init__(self) -> None:
+            self.shutdown_called = False
+
+        def on_start(self, span: object, parent_context: object = None) -> None:
+            pass
+
+        def on_end(self, span: object) -> None:
+            pass
+
+        def shutdown(self) -> None:
+            self.shutdown_called = True
+
+        def force_flush(self, timeout_millis: int = 30000) -> bool:
+            return True
+
+    settings = ObservabilitySettings(enabled=True, endpoint="http://collector.invalid")
+    first, second = _SpyProcessor(), _SpyProcessor()
+    init_tracing(settings, span_processor=first)
+    init_tracing(settings, span_processor=second)
+    assert first.shutdown_called is True
+    assert second.shutdown_called is False
+
+
 def test_resolve_otlp_target_derives_from_langfuse_creds() -> None:
     import base64
 

tests/unit/test_memory/test_search/test_manager.py

@@ -283,6 +283,36 @@ async def test_user_keyword_returns_episodes_only() -> None:
     assert resp.data.profiles == []
 
 
+async def test_recall_hit_emitted_only_for_calibrated_methods(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """recall_hit is meaningful only for calibrated-score methods (HYBRID LR
+    / AGENTIC rerank). For KEYWORD (unbounded BM25) the code must emit
+    top_score but NOT a hit verdict — a fixed 0.6 threshold against an
+    unbounded score is an always-hit signal that inflates the dashboard."""
+    import everos.memory.search.manager as mgr_mod
+
+    calls: list[dict[str, Any]] = []
+    monkeypatch.setattr(mgr_mod, "current_trace_ids", lambda: ("t" * 32, "s" * 16))
+    monkeypatch.setattr(mgr_mod, "emit_recall_scores", lambda **kw: calls.append(kw))
+
+    # KEYWORD: raw BM25 score well above the threshold, but uncalibrated.
+    kw_mgr = _build_manager(episode_sparse=[_episode_row("ep_1", score=7.5)])
+    await kw_mgr.search(_user_req(method=SearchMethod.KEYWORD))
+    assert len(calls) == 1
+    assert calls[0]["method"] == "keyword"
+    assert calls[0]["top_score"] == 7.5
+    assert calls[0]["hit"] is None  # no hit verdict for an uncalibrated method
+
+    # HYBRID: calibrated LR score → a real boolean hit verdict is emitted.
+    calls.clear()
+    hy_mgr = _build_manager(embedding=_StubEmbedding())
+    await hy_mgr.search(_user_req(method=SearchMethod.HYBRID))
+    assert len(calls) == 1
+    assert calls[0]["method"] == "hybrid"
+    assert isinstance(calls[0]["hit"], bool)
+
+
 async def test_search_uses_propagated_request_id_when_bound() -> None:
     """When a request id is bound upstream (middleware), ``search`` reuses it
     instead of minting a fresh one, so the response id matches the trace."""
@@ -1062,41 +1092,63 @@ async def test_hybrid_agent_emits_recall_and_rank(_search_spans: Any) -> None:
     assert "everos.search.rank" in spans
 
 
-async def test_search_emits_top_score_and_hit_on_span(_search_spans: Any) -> None:
-    """search sets everos.search.top_score (max item score) + everos.search.hit
-    (>= recall_hit_threshold, default 0.6) on the retriever span — always, no
-    Langfuse needed."""
+async def test_search_emits_top_score_without_hit_for_keyword(
+    _search_spans: Any,
+) -> None:
+    """KEYWORD sets everos.search.top_score (max item score) but NOT
+    everos.search.hit — an unbounded BM25 score forced through a fixed
+    threshold would be a misleading always-hit verdict."""
     mgr = _build_manager(episode_sparse=[_episode_row("ep_1", score=0.75)])
     await mgr.search(_user_req(method=SearchMethod.KEYWORD))
-    spans = _span_index(_search_spans)
-    attrs = spans["everos.memory.search"].attributes
+    attrs = _span_index(_search_spans)["everos.memory.search"].attributes
     assert attrs["everos.search.top_score"] == pytest.approx(0.75)
+    assert "everos.search.hit" not in attrs
+
+
+async def test_search_emits_hit_when_calibrated_above_threshold(
+    _search_spans: Any, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """HYBRID is calibrated ([0, 1]) so a top score >= recall_hit_threshold
+    (default 0.6) sets everos.search.hit = True on the retriever span."""
+    import everos.memory.search.manager as mgr_mod
+
+    monkeypatch.setattr(mgr_mod, "_top_score", lambda data: 0.9)
+    mgr = _build_manager(embedding=_StubEmbedding())
+    await mgr.search(_user_req(method=SearchMethod.HYBRID))
+    attrs = _span_index(_search_spans)["everos.memory.search"].attributes
+    assert attrs["everos.search.top_score"] == pytest.approx(0.9)
     assert attrs["everos.search.hit"] is True
 
 
-async def test_search_hit_false_when_below_threshold(_search_spans: Any) -> None:
-    mgr = _build_manager(episode_sparse=[_episode_row("ep_1", score=0.3)])
-    await mgr.search(_user_req(method=SearchMethod.KEYWORD))
-    spans = _span_index(_search_spans)
-    attrs = spans["everos.memory.search"].attributes
+async def test_search_hit_false_when_calibrated_below_threshold(
+    _search_spans: Any, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    import everos.memory.search.manager as mgr_mod
+
+    monkeypatch.setattr(mgr_mod, "_top_score", lambda data: 0.3)
+    mgr = _build_manager(embedding=_StubEmbedding())
+    await mgr.search(_user_req(method=SearchMethod.HYBRID))
+    attrs = _span_index(_search_spans)["everos.memory.search"].attributes
     assert attrs["everos.search.top_score"] == pytest.approx(0.3)
     assert attrs["everos.search.hit"] is False
 
 
-async def test_search_top_score_zero_when_no_results(_search_spans: Any) -> None:
+async def test_search_top_score_zero_without_hit_for_keyword(
+    _search_spans: Any,
+) -> None:
     mgr = _build_manager()  # no candidates
     await mgr.search(_user_req(method=SearchMethod.KEYWORD))
-    spans = _span_index(_search_spans)
-    attrs = spans["everos.memory.search"].attributes
+    attrs = _span_index(_search_spans)["everos.memory.search"].attributes
     assert attrs["everos.search.top_score"] == pytest.approx(0.0)
-    assert attrs["everos.search.hit"] is False
+    assert "everos.search.hit" not in attrs
 
 
 async def test_search_enqueues_recall_scores(
     _search_spans: Any, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """When tracing is active, search hands recall_top_score/hit to the score
-    sink with the retriever span's trace_id (032x) + observation_id (016x)."""
+    """When tracing is active, search hands recall_top_score to the score
+    sink with the retriever span's trace_id (032x) + observation_id (016x).
+    KEYWORD is uncalibrated so hit is None (no recall_hit is pushed)."""
     import everos.memory.search.manager as mgr_mod
 
     captured: dict[str, Any] = {}
@@ -1109,7 +1161,7 @@ async def test_search_enqueues_recall_scores(
     await mgr.search(_user_req(method=SearchMethod.KEYWORD))
 
     assert captured["top_score"] == pytest.approx(0.75)
-    assert captured["hit"] is True
+    assert captured["hit"] is None
     assert captured["method"] == "keyword"
     assert len(captured["trace_id"]) == 32
     assert len(captured["observation_id"]) == 16

tests/unit/test_service/test_search_llm_client.py

@@ -0,0 +1,77 @@
+"""service.search._get_llm_client — token-usage wrapping parity with memorize.
+
+The hybrid/agentic search path is the most LLM-heavy flow (query
+decomposition + refine + per-query fan-out + rerank judge). Its client
+must be wrapped with ``UsageRecordingClient`` when observability is on,
+exactly like ``memorize`` / the reflection strategies — otherwise the
+biggest token spend is invisible in Langfuse.
+"""
+
+from __future__ import annotations
+
+import importlib
+
+import pytest
+from pydantic import SecretStr
+
+import everos.config as config_mod
+from everos.component.llm._usage_client import UsageRecordingClient
+from everos.config import Settings
+from everos.config.settings import LLMSettings, ObservabilitySettings
+
+# `everos.service.search` the submodule is shadowed by the re-exported
+# `search` function on the package, so resolve the module explicitly.
+search_mod = importlib.import_module("everos.service.search")
+
+
+def _reset(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(search_mod, "_llm_client", None, raising=False)
+    monkeypatch.setattr(search_mod, "_llm_resolved", False, raising=False)
+
+
+def _patch_settings(monkeypatch: pytest.MonkeyPatch, *, enabled: bool) -> None:
+    cfg = Settings(
+        llm=LLMSettings(
+            model="gpt-4.1-mini",
+            api_key=SecretStr("sk-test"),
+            base_url="https://example.test",
+        ),
+        observability=ObservabilitySettings(enabled=enabled),
+    )
+    monkeypatch.setattr(config_mod, "load_settings", lambda: cfg)
+
+
+def test_search_llm_wrapped_when_observability_enabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _reset(monkeypatch)
+    _patch_settings(monkeypatch, enabled=True)
+
+    client = search_mod._get_llm_client()
+
+    assert isinstance(client, UsageRecordingClient)
+
+
+def test_search_llm_not_wrapped_when_observability_disabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _reset(monkeypatch)
+    _patch_settings(monkeypatch, enabled=False)
+
+    client = search_mod._get_llm_client()
+
+    assert client is not None
+    assert not isinstance(client, UsageRecordingClient)
+
+
+def test_search_llm_none_when_unconfigured(monkeypatch: pytest.MonkeyPatch) -> None:
+    # No credentials → graceful None (keyword-only degradation), regardless
+    # of observability. The wrapper must not change this contract.
+    _reset(monkeypatch)
+    cfg = Settings(
+        llm=LLMSettings(model="m", api_key=None, base_url=None),
+        observability=ObservabilitySettings(enabled=True),
+    )
+    monkeypatch.setattr(config_mod, "load_settings", lambda: cfg)
+
+    assert search_mod._get_llm_client() is None