Merge pull request #353 from EverMind-AI/fix/otel-review-followups

fix(observability): address OTel review findings

CHANGELOG.md

@@ -15,6 +15,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   backward-compatible alias: both prefixes resolve to the same handlers with
   identical request/response contracts, so existing integrations keep working
   unchanged. Infrastructure endpoints (`/health`, `/metrics`) stay unversioned.
+- **Native OpenTelemetry tracing** — memory operations (add / flush, memcell
+  boundary, episode extraction, search, and OME reflection) export to any
+  OTLP backend (e.g. Langfuse) as nested traces carrying LLM/embedding token
+  usage, per-request correlation, and recall-quality scores. Off by default;
+  enabled via the `[observability]` config with the optional `otel` extra.
+  Content capture (query / extracted memory) is opt-in and redaction-aware.
 
 ## [1.1.4] - 2026-07-20
 

src/everos/component/embedding/openai_provider.py

@@ -23,7 +23,7 @@ from collections.abc import Sequence
 
 import openai
 
-from everos.core.observability.tracing import set_generation_usage
+from everos.core.observability.tracing import memory_span, set_generation_usage
 
 from .protocol import EmbeddingServiceError
 
@@ -88,20 +88,26 @@ class OpenAIEmbeddingProvider:
 
     async def _embed_chunk(self, chunk: list[str]) -> list[list[float]]:
         """One ``/embeddings`` call, semaphore-guarded."""
-        async with self._semaphore:
-            try:
-                response = await self._client.embeddings.create(
-                    model=self._model,
-                    input=chunk,
-                )
-            except openai.OpenAIError as exc:
-                raise EmbeddingServiceError(str(exc)) from exc
-        # Surface token usage onto the active span (e.g. everos.search.embed_query).
-        # No-op when tracing is off; embeddings report only input (prompt) tokens.
-        usage = getattr(response, "usage", None)
-        set_generation_usage(
-            model=self._model,
-            input_tokens=usage.prompt_tokens if usage else None,
-        )
-        # OpenAI returns ``data`` indexed by request order; truncate to ``dim``.
-        return [list(item.embedding[: self.dim]) for item in response.data]
+        # Wrap in an EMBEDDING-typed span so token usage lands on an embedding
+        # observation (which Langfuse can price) rather than the enclosing
+        # retriever span. nested_only: skip when there is no active trace (e.g.
+        # cascade-time indexing) so we don't spawn one root trace per chunk.
+        with memory_span(
+            "everos.embedding", observation_type="embedding", nested_only=True
+        ):
+            async with self._semaphore:
+                try:
+                    response = await self._client.embeddings.create(
+                        model=self._model,
+                        input=chunk,
+                    )
+                except openai.OpenAIError as exc:
+                    raise EmbeddingServiceError(str(exc)) from exc
+            # Embeddings report only input (prompt) tokens.
+            usage = getattr(response, "usage", None)
+            set_generation_usage(
+                model=self._model,
+                input_tokens=usage.prompt_tokens if usage else None,
+            )
+            # OpenAI returns ``data`` indexed by request order; truncate to ``dim``.
+            return [list(item.embedding[: self.dim]) for item in response.data]

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

@@ -85,6 +85,13 @@ except ImportError:  # pragma: no cover - only without the [otel] extra
     _OTEL_AVAILABLE = False
 
 
+def _has_active_span() -> bool:
+    """True when a valid span context is currently active (a parent exists)."""
+    if not _OTEL_AVAILABLE:
+        return False
+    return _otel_trace.get_current_span().get_span_context().is_valid
+
+
 @contextmanager
 def memory_span(
     name: str,
@@ -94,6 +101,7 @@ def memory_span(
     user_id: str | None = None,
     metadata: Mapping[str, Any] | None = None,
     tags: Sequence[str] = DEFAULT_TAGS,
+    nested_only: bool = False,
 ) -> Iterator[Any]:
     """Open a span named ``name`` and stamp the langfuse.* attributes.
 
@@ -105,7 +113,14 @@ def memory_span(
         metadata: Flat mapping → ``langfuse.trace.metadata.<key>``; None
             values are dropped rather than emitted as the string "None".
         tags: ``langfuse.trace.tags`` list.
+        nested_only: When True, only open a span if one is already active.
+            Calls that run outside any request trace (e.g. cascade-time
+            embedding during indexing) would otherwise each start a NEW root
+            trace — a per-chunk trace explosion — so they no-op instead.
     """
+    if nested_only and not _has_active_span():
+        yield _otel_trace.get_current_span() if _OTEL_AVAILABLE else None
+        return
     tracer = get_tracer("everos")
     with tracer.start_as_current_span(name) as span:
         span.set_attribute(LF_OBSERVATION_TYPE, observation_type)
@@ -181,15 +196,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/extract/pipeline/user_memory.py

@@ -13,12 +13,14 @@ Run inside ``service.memorize`` via ``asyncio.gather`` alongside
 
 from __future__ import annotations
 
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 from everalgo.types import MemCell as AlgoMemCell
 from everalgo.user_memory import EpisodeExtractor
 
 from everos.component.utils.datetime import from_timestamp, to_iso_format
+from everos.config import resolve_root
 from everos.core.observability.logging import get_logger
 from everos.core.observability.tracing import capture_output, memory_span
 from everos.memory import Episode, IngestResult, PipelineOutcome
@@ -36,6 +38,14 @@ logger = get_logger(__name__)
 _TRACK = "user_memory"
 
 
+def _root_relative(path: str) -> str:
+    """Path relative to the memory root, for telemetry (never the host abspath)."""
+    try:
+        return str(Path(path).relative_to(resolve_root()))
+    except ValueError:
+        return path
+
+
 class UserMemoryPipeline:
     """Per-sender Episode extraction on a list of pre-cut MemCells."""
 
@@ -152,8 +162,10 @@ class UserMemoryPipeline:
                         )
                     )
                     md_paths.append(md_path)
-                    # Written .md path (only when capture_content is on).
-                    capture_output(persist_span, md_path)
+                    # Written .md path, memory-root-relative (only when
+                    # capture_content is on) — never leak the host absolute
+                    # path to the telemetry backend.
+                    capture_output(persist_span, _root_relative(md_path))
                 await self._engine.emit(
                     EpisodeExtracted(
                         memcell_id=memcell_id,

src/everos/memory/search/agentic.py

@@ -202,19 +202,17 @@ async def search_episodes_agentic(
 
     # 6. cluster_scoped: narrows hybrid_full to top-K cluster member expansions.
     async def cluster_scoped(q: str, _k: int) -> list[Candidate]:
-        with memory_span(
-            "everos.search.recall",
-            observation_type="retriever",
-            metadata={"phase": "agentic_cluster_scoped"},
-        ):
-            return await acluster_retrieve(
-                q,
-                base_retrieve=hybrid_full,
-                base_candidates=_CLUSTER_BASE_CANDIDATES,
-                clusters=clusters,
-                all_docs=all_docs,
-                cluster_top_k=_CLUSTER_TOP_K,
-            )
+        # No recall span here: cluster_scoped delegates to hybrid_full, which
+        # owns the recall span (and does the embedding). Wrapping it too
+        # produced a duplicate, same-name everos.search.recall nested in itself.
+        return await acluster_retrieve(
+            q,
+            base_retrieve=hybrid_full,
+            base_candidates=_CLUSTER_BASE_CANDIDATES,
+            clusters=clusters,
+            all_docs=all_docs,
+            cluster_top_k=_CLUSTER_TOP_K,
+        )
 
     # 7. Cross-encoder rerank fn (2-arg RerankFn, no internal truncation).
     rerank_fn = build_rerank_fn(

src/everos/memory/search/manager.py

@@ -42,6 +42,7 @@ from everos.core.context import resolve_request_id
 from everos.core.observability.logging import get_logger
 from everos.core.observability.tracing import (
     capture_input,
+    capture_output,
     current_trace_ids,
     emit_recall_scores,
     memory_span,
@@ -139,6 +140,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."""
 
@@ -222,13 +231,29 @@ class SearchManager:
                     unprocessed_messages=unprocessed,
                 )
 
+            # Returned hits (ids only) — content, so only when capture_content
+            # is on. This is the point of a search trace: what came back.
+            capture_output(
+                span,
+                {
+                    "episodes": [e.id for e in data.episodes],
+                    "agent_cases": [c.id for c in data.agent_cases],
+                    "agent_skills": [s.id for s in data.agent_skills],
+                },
+            )
+
             # 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/memorize.py

@@ -30,6 +30,7 @@ from pydantic import BaseModel
 
 from everos.component.llm import get_llm_client
 from everos.config import load_settings
+from everos.core.context import resolve_request_id
 from everos.core.observability.logging import get_logger
 from everos.core.observability.tracing import memory_span
 from everos.core.persistence import MemoryRoot
@@ -176,7 +177,11 @@ async def memorize(
         span_name,
         observation_type="span",
         session_id=session_id,
-        metadata={"mode": mode, "is_final": is_final},
+        metadata={
+            "mode": mode,
+            "is_final": is_final,
+            "request_id": resolve_request_id(),
+        },
     ):
         async with asyncio.timeout(settings.memorize.session_lock_timeout_seconds):
             async with get_session_lock(session_id):
@@ -197,15 +202,19 @@ async def _memorize_locked(
 ) -> MemorizeResult:
     """Inner critical section — runs under the per-session lock."""
     ingested = await ingest_process(payload)
-    boundary = await prepare_cells(
-        ingested,
-        mode=mode,
-        is_final=is_final,
-        llm_client=get_llm_client(),
-        prompt_loader=_get_prompt_loader(),
-        hard_token_limit=boundary_cfg.hard_token_limit,
-        hard_msg_limit=boundary_cfg.hard_msg_limit,
-    )
+    # Boundary detection runs an LLM (everalgo detect_boundaries). Wrap it in a
+    # generation span so its token usage lands on a GENERATION observation
+    # (costed by Langfuse) instead of the SPAN-typed request root (dropped).
+    with memory_span("everos.memcell.boundary", observation_type="generation"):
+        boundary = await prepare_cells(
+            ingested,
+            mode=mode,
+            is_final=is_final,
+            llm_client=get_llm_client(),
+            prompt_loader=_get_prompt_loader(),
+            hard_token_limit=boundary_cfg.hard_token_limit,
+            hard_msg_limit=boundary_cfg.hard_msg_limit,
+        )
 
     if not boundary.cells:
         # Nothing went past the boundary stage — no pipelines to dispatch.

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,48 @@ 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_nested_only_span_skips_when_no_active_parent(
+    captured: InMemorySpanExporter,
+) -> None:
+    # A nested_only span with no active parent must NOT start a new root trace
+    # (this is what prevented cascade-time embeddings from exploding into one
+    # orphan trace per chunk).
+    with memory_span(
+        "everos.embedding", observation_type="embedding", nested_only=True
+    ):
+        pass
+    force_flush()
+    assert captured.get_finished_spans() == ()
+
+
+def test_nested_only_span_opens_under_active_parent(
+    captured: InMemorySpanExporter,
+) -> None:
+    with (
+        memory_span("everos.memory.search", observation_type="retriever"),
+        memory_span("everos.embedding", observation_type="embedding", nested_only=True),
+    ):
+        pass
+    force_flush()
+    names = {s.name for s in captured.get_finished_spans()}
+    assert "everos.embedding" in names
+    assert "everos.memory.search" in names
+
+
 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
@@ -1130,6 +1182,27 @@ async def test_search_captures_query_when_content_on(_search_spans: Any) -> None
     assert json.loads(attrs["langfuse.observation.input"])["query"] == "hi"
 
 
+async def test_search_captures_returned_hits_when_content_on(
+    _search_spans: Any,
+) -> None:
+    """capture_content on → the search span records the returned hit ids
+    (episodes/agent_cases/agent_skills), not just the query."""
+    from everos.core.observability.tracing import set_capture_content
+
+    set_capture_content(True)
+    try:
+        mgr = _build_manager(episode_sparse=[_episode_row("ep_1")])
+        await mgr.search(_user_req(method=SearchMethod.KEYWORD))
+    finally:
+        set_capture_content(False)
+    import json
+
+    attrs = _span_index(_search_spans)["everos.memory.search"].attributes
+    out = json.loads(attrs["langfuse.observation.output"])
+    assert out["episodes"] == ["ep_1"]
+    assert out["agent_cases"] == [] and out["agent_skills"] == []
+
+
 async def test_search_omits_query_when_content_off(_search_spans: Any) -> None:
     mgr = _build_manager(episode_sparse=[_episode_row("ep_1")])
     await mgr.search(_user_req(method=SearchMethod.KEYWORD))

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