diff --git a/CHANGELOG.md b/CHANGELOG.md index 227c223..761c097 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/everos/component/embedding/openai_provider.py b/src/everos/component/embedding/openai_provider.py index 08d131d..b26b89e 100644 --- a/src/everos/component/embedding/openai_provider.py +++ b/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] diff --git a/src/everos/core/lifespan/tracing_lifespan.py b/src/everos/core/lifespan/tracing_lifespan.py index 29fc4f7..70b1fde 100644 --- a/src/everos/core/lifespan/tracing_lifespan.py +++ b/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 diff --git a/src/everos/core/observability/tracing/attributes.py b/src/everos/core/observability/tracing/attributes.py index 46d2f5a..d184cf7 100644 --- a/src/everos/core/observability/tracing/attributes.py +++ b/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.``; 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) diff --git a/src/everos/core/observability/tracing/provider.py b/src/everos/core/observability/tracing/provider.py index 95630cb..e58d501 100644 --- a/src/everos/core/observability/tracing/provider.py +++ b/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( diff --git a/src/everos/core/observability/tracing/scores.py b/src/everos/core/observability/tracing/scores.py index ac480eb..6256acb 100644 --- a/src/everos/core/observability/tracing/scores.py +++ b/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: diff --git a/src/everos/memory/extract/pipeline/user_memory.py b/src/everos/memory/extract/pipeline/user_memory.py index a58ee1b..e36532c 100644 --- a/src/everos/memory/extract/pipeline/user_memory.py +++ b/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, diff --git a/src/everos/memory/search/agentic.py b/src/everos/memory/search/agentic.py index aab634a..1b79e8a 100644 --- a/src/everos/memory/search/agentic.py +++ b/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( diff --git a/src/everos/memory/search/manager.py b/src/everos/memory/search/manager.py index b06e046..14c1d27 100644 --- a/src/everos/memory/search/manager.py +++ b/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. diff --git a/src/everos/service/memorize.py b/src/everos/service/memorize.py index d65fdfe..9fe806b 100644 --- a/src/everos/service/memorize.py +++ b/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. diff --git a/src/everos/service/search.py b/src/everos/service/search.py index e8369d6..cf9905f 100644 --- a/src/everos/service/search.py +++ b/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 diff --git a/tests/unit/test_core/test_observability/test_scores.py b/tests/unit/test_core/test_observability/test_scores.py index 6cd80f7..40ff0c7 100644 --- a/tests/unit/test_core/test_observability/test_scores.py +++ b/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: diff --git a/tests/unit/test_core/test_observability/test_span_attributes.py b/tests/unit/test_core/test_observability/test_span_attributes.py index ee3a89f..fe8970c 100644 --- a/tests/unit/test_core/test_observability/test_span_attributes.py +++ b/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) diff --git a/tests/unit/test_core/test_observability/test_tracing.py b/tests/unit/test_core/test_observability/test_tracing.py index 5e3518a..9431daa 100644 --- a/tests/unit/test_core/test_observability/test_tracing.py +++ b/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 diff --git a/tests/unit/test_memory/test_search/test_manager.py b/tests/unit/test_memory/test_search/test_manager.py index b396d46..f352575 100644 --- a/tests/unit/test_memory/test_search/test_manager.py +++ b/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)) diff --git a/tests/unit/test_service/test_search_llm_client.py b/tests/unit/test_service/test_search_llm_client.py new file mode 100644 index 0000000..ce748d5 --- /dev/null +++ b/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