From b57367f5f382bf6d79764596ab892e489c0f41c7 Mon Sep 17 00:00:00 2001 From: Kairo-J Date: Mon, 17 Aug 2026 17:14:25 +0800 Subject: [PATCH 1/3] feat(tracing): add memory read-path spans Add framework-neutral Runtime tracing backed by ServerTracing. Trace memory search, reranking, experience recall, and context preparation. Cover span hierarchy, privacy, cancellation, and tracing failure isolation. Closes #1241 --- .../builtin/runtime/application.py | 169 ++++++--- .../builtin/runtime/composition.py | 45 +++ src/powercontext/builtin/runtime/protocols.py | 20 ++ src/powercontext/server/factory.py | 1 + src/powercontext/server/tracing.py | 37 +- tests/e2e/test_builtin_runtime.py | 113 ++++++ tests/e2e/test_observability.py | 323 +++++++++++++++++- tests/test_server_tracing.py | 95 ++++++ 8 files changed, 756 insertions(+), 47 deletions(-) diff --git a/src/powercontext/builtin/runtime/application.py b/src/powercontext/builtin/runtime/application.py index f6342e923..a479a40ba 100644 --- a/src/powercontext/builtin/runtime/application.py +++ b/src/powercontext/builtin/runtime/application.py @@ -4,8 +4,8 @@ import asyncio import logging -from collections.abc import AsyncIterator, Awaitable, Callable -from contextlib import asynccontextmanager +from collections.abc import AsyncIterator, Awaitable, Callable, Mapping +from contextlib import AbstractContextManager, asynccontextmanager, nullcontext from datetime import UTC, datetime from pathlib import Path from time import perf_counter @@ -89,7 +89,13 @@ SourceReceipt, ) from powercontext.builtin.runtime.prepared_context import PreparedContextBuild, PreparedContextBuilder -from powercontext.builtin.runtime.protocols import BuiltinTriggers, PowerContextProvider +from powercontext.builtin.runtime.protocols import ( + BuiltinTriggers, + PowerContextProvider, + RuntimeSpan, + RuntimeTracing, + TraceAttribute, +) from powercontext.builtin.runtime.readiness import ( ReadinessCheckStatus, RuntimeReadiness, @@ -114,6 +120,13 @@ logger = logging.getLogger(__name__) +_MEMORY_SEARCH_STAGE = "memory.search" +_MEMORY_SEARCH_REQUESTED_MODE = "powercontext.memory.search.requested_mode" +_MEMORY_SEARCH_LIMIT = "powercontext.memory.search.limit" +_MEMORY_SEARCH_MEMORY_PRESENT = "powercontext.memory.search.memory_present" +_MEMORY_SEARCH_MODE = "powercontext.memory.search.mode" +_MEMORY_SEARCH_RESULT_COUNT = "powercontext.memory.search.result_count" + ScopeIds = Callable[[], Awaitable[tuple[str, ...]]] ReviewServiceFactory = Callable[[str], ReviewService] GenerationServiceFactory = Callable[[str], ReviewedGenerationService] @@ -261,32 +274,72 @@ async def prepare(self, request: PrepareContextRequest, /) -> PreparedContext: self._runtime._context(self.scope_id, embedding_purpose=ModelUsagePurpose.MEMORY_RECALL) as context, self._runtime._lock(self.scope_id), ): - service = context.artifacts.memory - current = await _head_or_none(service, context.artifacts.memory_artifact_id) - memory_hits = () - if current is not None: - result = await service.search( - request.query, - memories=(current,), - limit=builder.memory_candidate_limit, - mode="auto", + with self._runtime._stage( + _MEMORY_SEARCH_STAGE, + attributes={ + _MEMORY_SEARCH_REQUESTED_MODE: "auto", + _MEMORY_SEARCH_LIMIT: builder.memory_candidate_limit, + }, + ) as span: + service = context.artifacts.memory + current = await _head_or_none(service, context.artifacts.memory_artifact_id) + memory_hits = () + if current is not None: + result = await service.search( + request.query, + memories=(current,), + limit=builder.memory_candidate_limit, + mode="auto", + ) + memory_hits = result.hits + if span is not None: + attributes: dict[str, TraceAttribute] = { + _MEMORY_SEARCH_MEMORY_PRESENT: current is not None, + _MEMORY_SEARCH_RESULT_COUNT: len(memory_hits), + } + if current is not None: + attributes[_MEMORY_SEARCH_MODE] = result.mode + span.set_attributes(attributes) + + experience_recall = self._runtime._experience_recall + with self._runtime._stage( + "experience.search", + attributes={ + "powercontext.experience.search.configured": experience_recall is not None, + "powercontext.experience.search.limit": builder.experience_candidate_limit, + }, + ) as span: + experience_hits = ( + () + if experience_recall is None + else await experience_recall( + self.scope_id, + request.query, + builder.experience_candidate_limit, + ) ) - memory_hits = result.hits - experience_hits = ( - () - if self._runtime._experience_recall is None - else await self._runtime._experience_recall( - self.scope_id, - request.query, - builder.experience_candidate_limit, + if span is not None: + span.set_attributes({"powercontext.experience.search.result_count": len(experience_hits)}) + + with self._runtime._stage( + "context.prepare", + attributes={ + "powercontext.context.prepare.memory_candidate_count": len(memory_hits), + "powercontext.context.prepare.experience_candidate_count": len(experience_hits), + }, + ) as span: + build = builder.build_result( + request=request, + memory_ref=None if current is None else current.as_ref(), + hits=memory_hits, + experience_hits=experience_hits, ) - ) - build = builder.build_result( - request=request, - memory_ref=None if current is None else current.as_ref(), - hits=memory_hits, - experience_hits=experience_hits, - ) + if span is not None: + span.set_attributes({ + "powercontext.context.prepare.selected_count": len(build.origins), + "powercontext.context.prepare.status": build.context.status, + "powercontext.context.prepare.content_bytes": build.context.content_bytes, + }) if self._runtime._recall_token_estimator is not None: try: measurement = await self._runtime._recall_token_estimator(self.scope_id, build) @@ -653,22 +706,40 @@ async def search(self, request: SearchMemoryRequest, /) -> MemorySearchPage: generation_purpose=ModelUsagePurpose.MEMORY_RECALL, embedding_purpose=ModelUsagePurpose.MEMORY_RECALL, ) as context: - service = context.artifacts.memory - current = await _head_or_none(service, context.artifacts.memory_artifact_id) - if current is None: - return MemorySearchPage(memory_ref=None, mode=None) - result = await service.search( - request.query, - memories=(current,), - limit=request.limit, - mode=request.mode, - ) - return MemorySearchPage( - memory_ref=current.as_ref(), - mode=result.mode, - hits=result.hits, - rerank=result.rerank, - ) + with self._runtime._stage( + _MEMORY_SEARCH_STAGE, + attributes={ + _MEMORY_SEARCH_REQUESTED_MODE: request.mode, + _MEMORY_SEARCH_LIMIT: request.limit, + }, + ) as span: + service = context.artifacts.memory + current = await _head_or_none(service, context.artifacts.memory_artifact_id) + if current is None: + if span is not None: + span.set_attributes({ + _MEMORY_SEARCH_MEMORY_PRESENT: False, + _MEMORY_SEARCH_RESULT_COUNT: 0, + }) + return MemorySearchPage(memory_ref=None, mode=None) + result = await service.search( + request.query, + memories=(current,), + limit=request.limit, + mode=request.mode, + ) + if span is not None: + span.set_attributes({ + _MEMORY_SEARCH_MEMORY_PRESENT: True, + _MEMORY_SEARCH_MODE: result.mode, + _MEMORY_SEARCH_RESULT_COUNT: len(result.hits), + }) + return MemorySearchPage( + memory_ref=current.as_ref(), + mode=result.mode, + hits=result.hits, + rerank=result.rerank, + ) async def list(self, *, include_inactive: bool = False) -> MemoryEntriesPage: async with self._runtime._context(self.scope_id) as context: @@ -914,6 +985,7 @@ def __init__( recall_token_estimator: RecallTokenEstimator | None = None, readiness: RuntimeReadinessChecks | None = None, clock: Clock | None = None, + tracing: RuntimeTracing | None = None, ) -> None: if source_window_limit < 1: raise _RuntimeConfigurationError("source_window_limit") @@ -929,6 +1001,7 @@ def __init__( self._recall_token_estimator = recall_token_estimator self._readiness = RuntimeReadinessChecks() if readiness is None else readiness self._clock = _utc_now if clock is None else clock + self._tracing = tracing self.source_window_limit = source_window_limit self._locks: dict[str, asyncio.Lock] = {} self._processor_lock = asyncio.Lock() @@ -1114,6 +1187,16 @@ async def _context( def _lock(self, scope_id: str) -> asyncio.Lock: return self._locks.setdefault(validate_scope_id(scope_id), asyncio.Lock()) + def _stage( + self, + name: str, + *, + attributes: Mapping[str, TraceAttribute], + ) -> AbstractContextManager[RuntimeSpan | None]: + if self._tracing is None: + return nullcontext(None) + return self._tracing.stage(name, attributes=attributes) + def _review(self, scope_id: str) -> ReviewService: if self._review_service is None: raise _RuntimeStateError("review") diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index 7435de29d..9f018175a 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -18,6 +18,8 @@ CandidatePipeline, DefaultMemoryEvidenceProjector, MemoryCapabilities, + MemoryHit, + MemoryRerankDecision, MemoryReranker, ) from powercontext.builtin.artifacts.skill import CodexSkillProvider, ExternalSkillProvider, SkillGenerator @@ -43,6 +45,7 @@ from powercontext.builtin.runtime.application import BuiltinRuntime from powercontext.builtin.runtime.config import BuiltinConfig, ExternalSkillsConfig, InferenceConfig, RuntimeConfig from powercontext.builtin.runtime.models import MemorySearchMode, RuntimeCapabilities +from powercontext.builtin.runtime.protocols import RuntimeTracing from powercontext.builtin.runtime.readiness import ( READINESS_PROBE_TIMEOUT_SECONDS, CachedReadinessProbe, @@ -100,6 +103,44 @@ def project_source(self, source: Source, /) -> JsonValue: return super().project_source(source) +class _TracingMemoryReranker: + """Trace one configured reranker without exposing Memory content.""" + + def __init__(self, delegate: MemoryReranker, tracing: RuntimeTracing) -> None: + self._delegate = delegate + self._tracing = tracing + + @property + def policy_id(self) -> str: + return self._delegate.policy_id + + @policy_id.setter + def policy_id(self, value: str) -> None: + self._delegate.policy_id = value + + async def rerank( + self, + query: str, + candidates: tuple[MemoryHit, ...], + limit: int, + /, + ) -> MemoryRerankDecision: + with self._tracing.stage( + "memory.rerank", + attributes={ + "powercontext.memory.rerank.candidate_count": len(candidates), + "powercontext.memory.rerank.limit": limit, + }, + ) as span: + decision = await self._delegate.rerank(query, candidates, limit) + span.set_attributes({ + "powercontext.memory.rerank.selected_count": len(decision.selected_ranks), + "powercontext.memory.rerank.discarded_rank_count": decision.discarded_rank_count, + "powercontext.memory.rerank.used_fallback": decision.used_fallback, + }) + return decision + + @asynccontextmanager async def open_builtin_runtime( config: BuiltinConfig, @@ -115,6 +156,7 @@ async def open_builtin_runtime( token_estimator: TokenEstimator | None = None, memory_reranker: MemoryReranker | None = None, instrumentation: InstrumentationSettings | None = None, + tracing: RuntimeTracing | None = None, ) -> AsyncIterator[BuiltinRuntime]: """Open the selected database, inference adapters, and built-in runtime.""" @@ -145,6 +187,8 @@ async def open_builtin_runtime( configured_skill = generated_skill if skill_generator is None else skill_generator configured_handoff = generated_handoff if handoff_pipeline is None else handoff_pipeline configured_reranker = generated_reranker if memory_reranker is None else memory_reranker + if configured_reranker is not None and tracing is not None: + configured_reranker = _TracingMemoryReranker(configured_reranker, tracing) configured_embedding_source = ( await _embedding_model(config.inference, resources, instrumentation) if embedding_model is None @@ -210,6 +254,7 @@ async def open_builtin_runtime( statistics_service=contexts.statistics, recall_token_estimator=contexts.estimate_recall_tokens, readiness=RuntimeReadinessChecks(readiness_probes), + tracing=tracing, ) ) if config.handoff_report.enabled: diff --git a/src/powercontext/builtin/runtime/protocols.py b/src/powercontext/builtin/runtime/protocols.py index 6d680ef46..b98fc1a7a 100644 --- a/src/powercontext/builtin/runtime/protocols.py +++ b/src/powercontext/builtin/runtime/protocols.py @@ -2,6 +2,8 @@ from __future__ import annotations +from collections.abc import Mapping +from contextlib import AbstractContextManager from typing import Protocol, TypeVar from powercontext.builtin.artifacts.handoff import ActivateHandoff, HandoffActivation @@ -12,6 +14,24 @@ SourcesT = TypeVar("SourcesT", covariant=True) ArtifactsT = TypeVar("ArtifactsT", covariant=True) TriggersT = TypeVar("TriggersT", covariant=True) +TraceAttribute = str | bool | int | float + + +class RuntimeSpan(Protocol): + """Record bounded attributes for one internal Runtime stage.""" + + def set_attributes(self, attributes: Mapping[str, TraceAttribute], /) -> None: ... + + +class RuntimeTracing(Protocol): + """Create framework-neutral spans for internal Runtime stages.""" + + def stage( + self, + name: str, + *, + attributes: Mapping[str, TraceAttribute], + ) -> AbstractContextManager[RuntimeSpan]: ... class PowerContextProvider(Protocol[SourcesT, ArtifactsT, TriggersT]): diff --git a/src/powercontext/server/factory.py b/src/powercontext/server/factory.py index 9db26bf1f..13057c46c 100644 --- a/src/powercontext/server/factory.py +++ b/src/powercontext/server/factory.py @@ -80,6 +80,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: handoff_pipeline=handoff_pipeline, embedding_model=embedding_model, instrumentation=resolved_tracing.instrumentation, + tracing=resolved_tracing, ) as runtime: readiness_probe.bind(runtime) app.state.application = runtime diff --git a/src/powercontext/server/tracing.py b/src/powercontext/server/tracing.py index 58850d5f0..0374637a5 100644 --- a/src/powercontext/server/tracing.py +++ b/src/powercontext/server/tracing.py @@ -3,7 +3,8 @@ from __future__ import annotations import asyncio -from contextlib import suppress +from collections.abc import Iterator, Mapping +from contextlib import contextmanager, suppress from importlib import import_module from typing import TYPE_CHECKING, Any @@ -33,6 +34,8 @@ "install 'powercontext[server,tracing-otlp]' before enabling tracing" ) +_TraceAttribute = str | bool | int | float + class ServerTracing: """Create failure-isolated spans with one configured tracer provider.""" @@ -70,6 +73,34 @@ def start_span( context=context, ) + @contextmanager + def stage( + self, + name: str, + *, + attributes: Mapping[str, _TraceAttribute], + ) -> Iterator[_ActiveSpan]: + """Trace one bounded Runtime stage without changing its behavior.""" + + span = self.start_span( + name, + kind=SpanKind.INTERNAL, + attributes={ + **attributes, + "powercontext.operation.name": name, + "powercontext.operation.unit": "stage", + }, + ) + try: + yield span + except asyncio.CancelledError as error: + span.finish("cancelled", error=error) + raise + except BaseException as error: + span.finish("failure", error=error) + raise + span.finish("success") + def shutdown(self) -> None: with suppress(Exception): self.provider.shutdown() @@ -106,7 +137,7 @@ def start( def request_id(self) -> str: return request_id_from_span(self.span) - def set_attributes(self, attributes: dict[str, Any]) -> None: + def set_attributes(self, attributes: Mapping[str, Any]) -> None: if self.span is not None: with suppress(Exception): self.span.set_attributes(attributes) @@ -116,7 +147,7 @@ def finish( outcome: str, *, error: BaseException | None = None, - attributes: dict[str, Any] | None = None, + attributes: Mapping[str, Any] | None = None, ) -> None: if self.finished: return diff --git a/tests/e2e/test_builtin_runtime.py b/tests/e2e/test_builtin_runtime.py index ea32101b7..833a57095 100644 --- a/tests/e2e/test_builtin_runtime.py +++ b/tests/e2e/test_builtin_runtime.py @@ -2,16 +2,21 @@ import asyncio import json +from collections.abc import Iterator, Mapping +from contextlib import contextmanager from powercontext.builtin.artifacts.memory import MemoryCandidateRequest, MemoryEntryInput, MemoryRerankDecision from powercontext.builtin.inference import InferenceUsage from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.builtin.runtime import ( BuiltinConfig, + BuiltinRuntime, CaptureSource, PrepareContextRequest, RememberMemoryRequest, + RuntimeCapabilities, SearchMemoryRequest, + open_builtin_contexts, open_builtin_runtime, ) from powercontext.builtin.sources import ContentSource @@ -32,9 +37,11 @@ async def extract(self, request: MemoryCandidateRequest, /) -> tuple[MemoryEntry def test_builtin_runtime_uses_the_selected_sqlite_database() -> None: async def scenario() -> None: + tracing = _RecordingTracing() async with open_builtin_runtime( BuiltinConfig(database=SQLiteConfig()), candidate_pipeline=_ContentCandidatePipeline(), + tracing=tracing, ) as runtime: captured = await runtime.sources.for_scope("project").capture( CaptureSource( @@ -68,6 +75,7 @@ async def scenario() -> None: assert no_memory.content is None assert no_match.status == "empty" assert no_match.content is None + assert not any(name == "memory.rerank" for name, _attributes in tracing.stages) asyncio.run(scenario()) @@ -106,3 +114,108 @@ async def scenario() -> None: assert all(page.rerank is not None for page in pages) asyncio.run(scenario()) + + +_TraceValue = str | bool | int | float + + +class _RecordingSpan: + def __init__(self, attributes: dict[str, _TraceValue]) -> None: + self.attributes = attributes + + def set_attributes(self, attributes: Mapping[str, _TraceValue], /) -> None: + self.attributes.update(attributes) + + +class _RecordingTracing: + def __init__(self) -> None: + self.stages: list[tuple[str, dict[str, _TraceValue]]] = [] + + @contextmanager + def stage( + self, + name: str, + *, + attributes: Mapping[str, _TraceValue], + ) -> Iterator[_RecordingSpan]: + recorded = dict(attributes) + self.stages.append((name, recorded)) + yield _RecordingSpan(recorded) + + +class _DynamicPolicyReranker: + policy_id = "test.dynamic-rerank.v1" + + async def rerank(self, query, candidates, limit, /) -> MemoryRerankDecision: + self.policy_id = "test.dynamic-rerank.v2" + return MemoryRerankDecision( + selected_ranks=(1,), + usage=InferenceUsage(requests=1), + discarded_rank_count=2, + ) + + +def test_traced_reranker_preserves_dynamic_policy_and_reports_rank_discards() -> None: + async def scenario() -> None: + tracing = _RecordingTracing() + reranker = _DynamicPolicyReranker() + async with open_builtin_runtime( + BuiltinConfig(), + memory_reranker=reranker, + tracing=tracing, + ) as runtime: + memory = runtime.memory.for_scope("traced-reranker") + await memory.remember( + RememberMemoryRequest(entries=(MemoryEntryInput(kind="fact", text="Trace this fact."),)) + ) + page = await memory.search(SearchMemoryRequest(query="trace", mode="fts", limit=1)) + + assert page.rerank is not None + assert page.rerank.policy_id == "test.dynamic-rerank.v2" + rerank_attributes = next(attributes for name, attributes in tracing.stages if name == "memory.rerank") + assert rerank_attributes == { + "powercontext.memory.rerank.candidate_count": 1, + "powercontext.memory.rerank.limit": 1, + "powercontext.memory.rerank.selected_count": 1, + "powercontext.memory.rerank.discarded_rank_count": 2, + "powercontext.memory.rerank.used_fallback": False, + } + + asyncio.run(scenario()) + + +def test_context_trace_is_stable_without_memory_or_experience_recall() -> None: + async def scenario() -> None: + tracing = _RecordingTracing() + async with ( + open_builtin_contexts(BuiltinConfig()) as contexts, + BuiltinRuntime( + provider=contexts, + capabilities=RuntimeCapabilities(memory_extraction=False, memory_search_modes=("fts",)), + tracing=tracing, + ) as runtime, + ): + prepared = await runtime.context.for_scope("empty-traced-context").prepare( + PrepareContextRequest(query="private empty query") + ) + + assert prepared.status == "empty" + stages = dict(tracing.stages) + memory_search = stages["memory.search"] + assert memory_search["powercontext.memory.search.requested_mode"] == "auto" + assert isinstance(memory_search["powercontext.memory.search.limit"], int) + assert memory_search["powercontext.memory.search.memory_present"] is False + assert memory_search["powercontext.memory.search.result_count"] == 0 + experience_search = stages["experience.search"] + assert experience_search["powercontext.experience.search.configured"] is False + assert isinstance(experience_search["powercontext.experience.search.limit"], int) + assert experience_search["powercontext.experience.search.result_count"] == 0 + assert stages["context.prepare"] == { + "powercontext.context.prepare.memory_candidate_count": 0, + "powercontext.context.prepare.experience_candidate_count": 0, + "powercontext.context.prepare.selected_count": 0, + "powercontext.context.prepare.status": "empty", + "powercontext.context.prepare.content_bytes": 0, + } + + asyncio.run(scenario()) diff --git a/tests/e2e/test_observability.py b/tests/e2e/test_observability.py index c61836813..aa9c16b28 100644 --- a/tests/e2e/test_observability.py +++ b/tests/e2e/test_observability.py @@ -12,16 +12,62 @@ from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import SpanKind +from pydantic_ai import Embedder +from pydantic_ai.embeddings import TestEmbeddingModel from pydantic_ai.models import Model from pydantic_ai.models.test import TestModel +from powercontext.builtin.artifacts.memory import EmbeddingProfile +from powercontext.builtin.inference.pydantic_ai import PydanticAIEmbeddingModel from powercontext.builtin.persistence.sqlite import SQLiteConfig -from powercontext.builtin.runtime.config import InferenceConfig +from powercontext.builtin.runtime.config import InferenceConfig, RuntimeConfig from powercontext.server.factory import create_server_app from powercontext.server.logging import OperationalContextFilter from powercontext.server.settings import McpConfig, ServerSettings from powercontext.server.tracing import ServerTracing +_STAGE_ATTRIBUTE_KEYS = { + "memory.search": { + "powercontext.operation.name", + "powercontext.operation.unit", + "powercontext.operation.outcome", + "powercontext.memory.search.requested_mode", + "powercontext.memory.search.limit", + "powercontext.memory.search.memory_present", + "powercontext.memory.search.mode", + "powercontext.memory.search.result_count", + }, + "memory.rerank": { + "powercontext.operation.name", + "powercontext.operation.unit", + "powercontext.operation.outcome", + "powercontext.memory.rerank.candidate_count", + "powercontext.memory.rerank.limit", + "powercontext.memory.rerank.selected_count", + "powercontext.memory.rerank.discarded_rank_count", + "powercontext.memory.rerank.used_fallback", + }, + "experience.search": { + "powercontext.operation.name", + "powercontext.operation.unit", + "powercontext.operation.outcome", + "powercontext.experience.search.configured", + "powercontext.experience.search.limit", + "powercontext.experience.search.result_count", + }, + "context.prepare": { + "powercontext.operation.name", + "powercontext.operation.unit", + "powercontext.operation.outcome", + "powercontext.context.prepare.memory_candidate_count", + "powercontext.context.prepare.experience_candidate_count", + "powercontext.context.prepare.selected_count", + "powercontext.context.prepare.status", + "powercontext.context.prepare.content_bytes", + }, +} + def test_observability_signals_correlate_without_counting_the_mcp_bridge(caplog, tmp_path) -> None: exporter = InMemorySpanExporter() @@ -140,6 +186,239 @@ def test_inference_spans_join_the_operation_trace_only_when_instrumented(monkeyp assert not any(_is_inference_span(span) for span in uninstrumented) +def test_memory_read_stage_spans_are_bounded_and_nested(monkeypatch, tmp_path) -> None: + # Resolve the configured test model without consulting the environment or a real provider. + monkeypatch.setattr( + "pydantic_ai.models.infer_model", + lambda model: model if isinstance(model, Model) else TestModel(custom_output_text='{"selected_ranks":[1]}'), + ) + + exporter = InMemorySpanExporter() + provider = TracerProvider(shutdown_on_exit=False) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + app = create_server_app( + settings=ServerSettings( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'memory-read-tracing.db'}"), + runtime=RuntimeConfig(memory_rerank_enabled=True), + inference=InferenceConfig(generation_model="test"), + mcp=McpConfig(enabled=False), + ), + tracing=ServerTracing(provider, instrumented=True), + ) + scope_id = "project:private-trace-scope" + memory_content = "Private trace sentinel evidence." + query = "private trace sentinel" + no_match_query = "unmatched giraffe phrase" + + with TestClient(app) as client: + remembered = client.post( + "/v1/memory/remember", + json={"scope_id": scope_id, "kind": "fact", "text": memory_content}, + ) + searched = client.post( + "/v1/memory/search", + json={"scope_id": scope_id, "query": query, "limit": 1, "mode": "fts"}, + ) + no_match = client.post( + "/v1/memory/search", + json={"scope_id": scope_id, "query": no_match_query, "limit": 1, "mode": "fts"}, + ) + no_memory = client.post( + "/v1/memory/search", + json={ + "scope_id": "project:private-empty-search-scope", + "query": query, + "limit": 1, + "mode": "fts", + }, + ) + prepared = client.post( + "/v1/context/prepare", + json={"scope_id": scope_id, "query": query}, + ) + empty = client.post( + "/v1/context/prepare", + json={"scope_id": "project:private-empty-scope", "query": query}, + ) + + assert remembered.status_code == 200 + assert searched.status_code == 200 + assert searched.json()["hits"] + assert no_match.status_code == 200 + assert no_match.json()["hits"] == [] + assert no_memory.status_code == 200 + assert no_memory.json()["hits"] == [] + assert prepared.status_code == 200 + assert prepared.json()["status"] == "ready" + assert empty.status_code == 200 + assert empty.json()["status"] == "empty" + + spans = list(exporter.get_finished_spans()) + search_applications = [span for span in spans if span.name == "powercontext search_memory"] + prepare_applications = [span for span in spans if span.name == "powercontext prepare_context"] + assert len(search_applications) == 3 + assert len(prepare_applications) == 2 + + search_applications_by_result = { + ( + bool( + (_only_child(spans, application, "memory.search").attributes or {}).get( + "powercontext.memory.search.memory_present" + ) + ), + (_only_child(spans, application, "memory.search").attributes or {})[ + "powercontext.memory.search.result_count" + ], + ): application + for application in search_applications + } + search_application = search_applications_by_result[(True, 1)] + search = _only_child(spans, search_application, "memory.search") + search_attributes = dict(search.attributes or {}) + assert search_attributes == { + "powercontext.operation.name": "memory.search", + "powercontext.operation.unit": "stage", + "powercontext.memory.search.requested_mode": "fts", + "powercontext.memory.search.limit": 1, + "powercontext.memory.search.memory_present": True, + "powercontext.memory.search.mode": "fts", + "powercontext.memory.search.result_count": 1, + "powercontext.operation.outcome": "success", + } + rerank = _only_child(spans, search, "memory.rerank") + assert dict(rerank.attributes or {}) == { + "powercontext.operation.name": "memory.rerank", + "powercontext.operation.unit": "stage", + "powercontext.memory.rerank.candidate_count": 1, + "powercontext.memory.rerank.limit": 1, + "powercontext.memory.rerank.selected_count": 1, + "powercontext.memory.rerank.discarded_rank_count": 0, + "powercontext.memory.rerank.used_fallback": False, + "powercontext.operation.outcome": "success", + } + invoke_agent = _only_child(spans, rerank, "invoke_agent memory_rerank") + chat = _only_child_with_prefix(spans, invoke_agent, "chat ") + assert {span.context.trace_id for span in (search_application, search, rerank, invoke_agent, chat)} == { + search_application.context.trace_id + } + + no_match_application = search_applications_by_result[(True, 0)] + no_match_search = _only_child(spans, no_match_application, "memory.search") + assert (no_match_search.attributes or {})["powercontext.memory.search.mode"] == "fts" + assert not _children(spans, no_match_search, "memory.rerank") + no_memory_application = search_applications_by_result[(False, 0)] + no_memory_search = _only_child(spans, no_memory_application, "memory.search") + assert "powercontext.memory.search.mode" not in (no_memory_search.attributes or {}) + assert not _children(spans, no_memory_search, "memory.rerank") + + prepared_by_memory_presence = { + bool( + (_only_child(spans, application, "memory.search").attributes or {}).get( + "powercontext.memory.search.memory_present" + ) + ): application + for application in prepare_applications + } + ready_application = prepared_by_memory_presence[True] + empty_application = prepared_by_memory_presence[False] + + ready_memory = _only_child(spans, ready_application, "memory.search") + assert (ready_memory.attributes or {})["powercontext.memory.search.result_count"] == 1 + assert _only_child(spans, ready_memory, "memory.rerank") + ready_experience = _only_child(spans, ready_application, "experience.search") + assert (ready_experience.attributes or {})["powercontext.experience.search.configured"] is True + assert (ready_experience.attributes or {})["powercontext.experience.search.result_count"] == 0 + ready_context = _only_child(spans, ready_application, "context.prepare") + assert (ready_context.attributes or {})["powercontext.context.prepare.memory_candidate_count"] == 1 + assert (ready_context.attributes or {})["powercontext.context.prepare.experience_candidate_count"] == 0 + assert (ready_context.attributes or {})["powercontext.context.prepare.selected_count"] == 1 + assert (ready_context.attributes or {})["powercontext.context.prepare.status"] == "ready" + ready_content_bytes = (ready_context.attributes or {})["powercontext.context.prepare.content_bytes"] + assert isinstance(ready_content_bytes, int) + assert ready_content_bytes > 0 + + empty_memory = _only_child(spans, empty_application, "memory.search") + empty_memory_attributes = dict(empty_memory.attributes or {}) + assert empty_memory_attributes["powercontext.memory.search.memory_present"] is False + assert empty_memory_attributes["powercontext.memory.search.result_count"] == 0 + assert "powercontext.memory.search.mode" not in empty_memory_attributes + assert not _children(spans, empty_memory, "memory.rerank") + empty_experience = _only_child(spans, empty_application, "experience.search") + assert (empty_experience.attributes or {})["powercontext.experience.search.result_count"] == 0 + empty_context = _only_child(spans, empty_application, "context.prepare") + assert (empty_context.attributes or {})["powercontext.context.prepare.selected_count"] == 0 + assert (empty_context.attributes or {})["powercontext.context.prepare.status"] == "empty" + assert (empty_context.attributes or {})["powercontext.context.prepare.content_bytes"] == 0 + + for span in spans: + allowed_keys = _STAGE_ATTRIBUTE_KEYS.get(span.name) + if allowed_keys is None: + continue + attributes = dict(span.attributes or {}) + assert attributes.keys() <= allowed_keys + assert all(isinstance(value, str | bool | int | float) for value in attributes.values()) + + exported = _exported_span_data(spans) + assert scope_id not in exported + assert "project:private-empty-scope" not in exported + assert "project:private-empty-search-scope" not in exported + assert memory_content not in exported + assert query not in exported + assert no_match_query not in exported + + +def test_embedding_span_joins_memory_search_stage_without_recording_text() -> None: + exporter = InMemorySpanExporter() + provider = TracerProvider(shutdown_on_exit=False) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracing = ServerTracing(provider, instrumented=True) + instrumentation = tracing.instrumentation + assert instrumentation is not None + embedding_model = PydanticAIEmbeddingModel( + embedder=Embedder(TestEmbeddingModel(dimensions=3), instrument=instrumentation), + profile=EmbeddingProfile( + profile_id="trace-test-v1", + model="test", + dimension=3, + distance="l2", + normalization="unit", + ), + ) + private_text = "private embedding sentinel" + + async def scenario() -> None: + application = tracing.start_span( + "powercontext search_memory", + kind=SpanKind.INTERNAL, + attributes={ + "powercontext.operation.name": "search_memory", + "powercontext.operation.unit": "application", + }, + ) + try: + with tracing.stage( + "memory.search", + attributes={ + "powercontext.memory.search.requested_mode": "vector", + "powercontext.memory.search.limit": 1, + }, + ): + await embedding_model.embed((private_text,)) + except BaseException as error: + application.finish("failure", error=error) + raise + application.finish("success") + + asyncio.run(scenario()) + + spans = list(exporter.get_finished_spans()) + application = next(span for span in spans if span.name == "powercontext search_memory") + search = _only_child(spans, application, "memory.search") + embedding = _only_child_with_prefix(spans, search, "embeddings ") + assert embedding.context.trace_id == application.context.trace_id + assert private_text not in _exported_span_data(spans) + + def _flush_memory_spans(database_path: Path, *, instrumented: bool) -> list[ReadableSpan]: exporter = InMemorySpanExporter() provider = TracerProvider(shutdown_on_exit=False) @@ -168,3 +447,45 @@ def _flush_memory_spans(database_path: Path, *, instrumented: bool) -> list[Read def _is_inference_span(span: ReadableSpan) -> bool: return span.instrumentation_scope is not None and span.instrumentation_scope.name == "pydantic-ai" + + +def _children(spans: list[ReadableSpan], parent: ReadableSpan, name: str) -> list[ReadableSpan]: + return [ + span + for span in spans + if span.name == name and span.parent is not None and span.parent.span_id == parent.context.span_id + ] + + +def _only_child(spans: list[ReadableSpan], parent: ReadableSpan, name: str) -> ReadableSpan: + children = _children(spans, parent, name) + assert len(children) == 1 + return children[0] + + +def _only_child_with_prefix(spans: list[ReadableSpan], parent: ReadableSpan, prefix: str) -> ReadableSpan: + children = [ + span + for span in spans + if span.name.startswith(prefix) and span.parent is not None and span.parent.span_id == parent.context.span_id + ] + assert len(children) == 1 + return children[0] + + +def _exported_span_data(spans: list[ReadableSpan]) -> str: + return json.dumps( + [ + { + "name": span.name, + "attributes": dict(span.attributes or {}), + "events": [{"name": event.name, "attributes": dict(event.attributes or {})} for event in span.events], + "status": { + "code": str(span.status.status_code), + "description": span.status.description, + }, + } + for span in spans + ], + default=str, + ) diff --git a/tests/test_server_tracing.py b/tests/test_server_tracing.py index fedc4e172..f9db435c0 100644 --- a/tests/test_server_tracing.py +++ b/tests/test_server_tracing.py @@ -9,6 +9,7 @@ from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import SpanKind, StatusCode from powercontext.client import PowerContextClient from powercontext.server.factory import create_server_app @@ -137,6 +138,100 @@ def test_inference_instrumentation_records_no_content() -> None: assert instrumentation.include_model_request_parameters is False +def test_runtime_stage_records_attributes_and_inherits_current_span() -> None: + tracing, exporter = _tracing() + parent = tracing.start_span( + "powercontext search_memory", + kind=SpanKind.INTERNAL, + attributes={}, + ) + + with tracing.stage( + "memory.search", + attributes={"powercontext.memory.search.limit": 10}, + ) as stage: + stage.set_attributes({"powercontext.memory.search.result_count": 2}) + parent.finish("success") + + spans = {span.name: span for span in exporter.get_finished_spans()} + application_span = spans["powercontext search_memory"] + stage_span = spans["memory.search"] + assert stage_span.parent is not None + assert stage_span.parent.span_id == application_span.context.span_id + assert stage_span.kind is SpanKind.INTERNAL + assert stage_span.attributes is not None + assert stage_span.attributes["powercontext.operation.name"] == "memory.search" + assert stage_span.attributes["powercontext.operation.unit"] == "stage" + assert stage_span.attributes["powercontext.operation.outcome"] == "success" + assert stage_span.attributes["powercontext.memory.search.limit"] == 10 + assert stage_span.attributes["powercontext.memory.search.result_count"] == 2 + + +def test_runtime_stage_records_failure_and_reraises_same_error() -> None: + tracing, exporter = _tracing() + error = ValueError("sensitive query") + + with pytest.raises(ValueError) as raised, tracing.stage("memory.search", attributes={}): + raise error + + assert raised.value is error + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].attributes is not None + assert spans[0].attributes["powercontext.operation.outcome"] == "failure" + assert spans[0].attributes["error.type"] == "ValueError" + assert "sensitive query" not in str(spans[0].attributes) + assert spans[0].status.status_code is StatusCode.ERROR + + +def test_runtime_stage_records_cancellation_and_reraises_same_error() -> None: + tracing, exporter = _tracing() + error = asyncio.CancelledError() + + with pytest.raises(asyncio.CancelledError) as raised, tracing.stage("memory.search", attributes={}): + raise error + + assert raised.value is error + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].attributes is not None + assert spans[0].attributes["powercontext.operation.outcome"] == "cancelled" + assert spans[0].attributes["error.type"] == "CancelledError" + assert spans[0].status.status_code is StatusCode.UNSET + + +def test_runtime_stage_finishes_and_detaches_for_base_exceptions() -> None: + tracing, exporter = _tracing() + + with pytest.raises(SystemExit), tracing.stage("memory.search", attributes={}): + raise SystemExit + with tracing.stage("experience.search", attributes={}): + pass + + spans = {span.name: span for span in exporter.get_finished_spans()} + failed = spans["memory.search"] + following = spans["experience.search"] + assert failed.attributes is not None + assert failed.attributes["powercontext.operation.outcome"] == "failure" + assert failed.attributes["error.type"] == "SystemExit" + assert following.parent is None + + +def test_runtime_stage_isolates_tracer_failure(monkeypatch) -> None: + tracing, exporter = _tracing() + + class BrokenTracer: + def start_span(self, *_args: object, **_kwargs: object) -> None: + raise RuntimeError + + monkeypatch.setattr(tracing, "tracer", BrokenTracer()) + + with tracing.stage("memory.search", attributes={}) as stage: + stage.set_attributes({"powercontext.memory.search.result_count": 0}) + + assert exporter.get_finished_spans() == () + + def test_tracing_export_requires_the_otlp_extra(monkeypatch) -> None: module = "opentelemetry.exporter.otlp.proto.http.trace_exporter" monkeypatch.setitem(sys.modules, module, None) From 099d3b4d7c8f712435f8393b00edb3fc2e1a12ec Mon Sep 17 00:00:00 2001 From: Kairo-J Date: Tue, 18 Aug 2026 12:30:30 +0800 Subject: [PATCH 2/3] fix(tracing): refine memory read-path observability - Rename the context stage from context.prepare to context.build. - Keep reranker policy_id as a regular attribute. - Prevent readiness embedding probes from exporting inference spans. - Preserve operational tracing under ALWAYS_ON sampling and injected models. - Replace synthetic tracing tests with real exported vector-search coverage. - Document the stage vocabulary and read-path spans in English and Chinese. --- docs/en/docs/how-to/trace-with-phoenix.md | 9 + .../en/rfcs/0046_observability_foundations.md | 4 + docs/zh/docs/how-to/trace-with-phoenix.md | 9 + .../zh/rfcs/0046_observability_foundations.md | 4 + .../builtin/inference/pydantic_ai.py | 11 +- .../builtin/runtime/application.py | 12 +- .../builtin/runtime/composition.py | 69 +++-- src/powercontext/server/factory.py | 22 +- src/powercontext/server/tracing.py | 55 +++- tests/e2e/test_builtin_runtime.py | 113 -------- tests/e2e/test_observability.py | 259 ++++++++++++++---- tests/test_server_tracing.py | 17 -- 12 files changed, 348 insertions(+), 236 deletions(-) diff --git a/docs/en/docs/how-to/trace-with-phoenix.md b/docs/en/docs/how-to/trace-with-phoenix.md index c51462260..425adb61c 100644 --- a/docs/en/docs/how-to/trace-with-phoenix.md +++ b/docs/en/docs/how-to/trace-with-phoenix.md @@ -77,6 +77,15 @@ Open , select the `default` project, and open the most re | `invoke_agent memory_extraction` | One PowerContext generation task. The name identifies the purpose, not the model. | | `chat ` | One request to the model provider, with token usage and latency. | +Memory read operations add the following internal stage spans beneath their application operation: + +| Span | Meaning | +| --- | --- | +| `memory.search` | Memory lookup for `search_memory` or `prepare_context`; embedding and reranking spans, when present, are nested beneath it. | +| `memory.rerank` | One actual reranker call; model-backed reranking nests `invoke_agent memory_rerank` beneath it. | +| `experience.search` | Experience recall during `prepare_context`; emitted even when recall is not configured. | +| `context.build` | The synchronous step that selects and renders the final prepared context from recalled candidates. | + The other PowerContext generation tasks appear under the same convention: `experience_incubation`, `experience_generation`, `skill_generation`, `handoff_generation`, and `memory_rerank`. When an embedding model is configured, embedding calls appear as `embeddings ` spans under the operation that triggered them. diff --git a/docs/en/rfcs/0046_observability_foundations.md b/docs/en/rfcs/0046_observability_foundations.md index 3d2ceb1e4..003505f4e 100644 --- a/docs/en/rfcs/0046_observability_foundations.md +++ b/docs/en/rfcs/0046_observability_foundations.md @@ -80,9 +80,13 @@ PowerContext distinguishes these units: | --- | --- | | Transport request | One external HTTP or MCP protocol request | | Application operation | One stable PowerContext operation, such as `search_memory` | +| Runtime stage | One bounded internal step within an application operation, such as `memory.search` | | Background activation | One manual or scheduled Source processing activation | | Dependency call | One outbound call to another service or provider | +Runtime stage spans use `stage` as their `powercontext.operation.unit` value. They expose internal latency without +creating another application operation. + A direct HTTP call produces one external request and one application operation. An MCP tool call also produces one external request and one application operation. Its internal HTTP bridge does not count as a second external request. diff --git a/docs/zh/docs/how-to/trace-with-phoenix.md b/docs/zh/docs/how-to/trace-with-phoenix.md index 14d303bdb..6043fcd67 100644 --- a/docs/zh/docs/how-to/trace-with-phoenix.md +++ b/docs/zh/docs/how-to/trace-with-phoenix.md @@ -75,6 +75,15 @@ Memory extraction 发生在 flush 阶段，而不是捕获阶段。 | `invoke_agent memory_extraction` | 一次 PowerContext generation 任务。名字标识用途，不是模型名。 | | `chat ` | 一次发往模型 provider 的请求，包含 token 用量和耗时。 | +Memory 读取操作还会在 application operation 之下添加以下内部 stage span： + +| Span | 含义 | +| --- | --- | +| `memory.search` | `search_memory` 或 `prepare_context` 中的 Memory 查询；存在 embedding 或 reranking span 时，它们嵌套在其下。 | +| `memory.rerank` | 一次实际 reranker 调用；使用模型的 reranking 会在其下嵌套 `invoke_agent memory_rerank`。 | +| `experience.search` | `prepare_context` 中的 Experience recall；未配置 recall 时也会产生。 | +| `context.build` | 根据召回候选同步选择并渲染最终 prepared context 的步骤。 | + 其他 generation 任务遵循同样的命名约定：`experience_incubation`、`experience_generation`、`skill_generation`、 `handoff_generation` 和 `memory_rerank`。配置了 embedding model 时，embedding 调用会作为 `embeddings ` span 挂在触发它的操作之下。 diff --git a/docs/zh/rfcs/0046_observability_foundations.md b/docs/zh/rfcs/0046_observability_foundations.md index f0aea5107..d626cbddb 100644 --- a/docs/zh/rfcs/0046_observability_foundations.md +++ b/docs/zh/rfcs/0046_observability_foundations.md @@ -77,9 +77,13 @@ PowerContext 区分以下工作单元： | --- | --- | | Transport request | 一次外部 HTTP 或 MCP protocol request | | Application operation | 一项稳定的 PowerContext operation，例如 `search_memory` | +| Runtime stage | application operation 中一个有界的内部步骤，例如 `memory.search` | | Background activation | 一次手动或定时 Source processing activation | | Dependency call | 一次对其他 service 或 provider 的 outbound call | +Runtime stage span 的 `powercontext.operation.unit` 值为 `stage`。它们用于展示内部耗时，但不会产生新的 +application operation。 + 直接 HTTP call 会产生一次 external request 和一次 application operation。MCP tool call 同样产生一次 external request 和一次 application operation。其内部 HTTP bridge 不计为第二次 external request。 diff --git a/src/powercontext/builtin/inference/pydantic_ai.py b/src/powercontext/builtin/inference/pydantic_ai.py index a264e7b9e..86b0b1175 100644 --- a/src/powercontext/builtin/inference/pydantic_ai.py +++ b/src/powercontext/builtin/inference/pydantic_ai.py @@ -4,7 +4,8 @@ import asyncio from collections.abc import Sequence -from typing import Generic, TypeVar, cast +from copy import copy +from typing import Generic, Self, TypeVar, cast from pydantic import BaseModel, Field @@ -172,6 +173,14 @@ def __init__( self._batch_size = batch_size self._limits = InferenceLimits() if limits is None else limits + def _without_instrumentation(self) -> Self: + """Copy this adapter for readiness without changing operational tracing.""" + + adapter = copy(self) + adapter._embedder = copy(self._embedder) + adapter._embedder.instrument = False + return adapter + async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult: """Embed documents and validate order, count, dimension, and finite values.""" diff --git a/src/powercontext/builtin/runtime/application.py b/src/powercontext/builtin/runtime/application.py index a479a40ba..e3b9ea319 100644 --- a/src/powercontext/builtin/runtime/application.py +++ b/src/powercontext/builtin/runtime/application.py @@ -322,10 +322,10 @@ async def prepare(self, request: PrepareContextRequest, /) -> PreparedContext: span.set_attributes({"powercontext.experience.search.result_count": len(experience_hits)}) with self._runtime._stage( - "context.prepare", + "context.build", attributes={ - "powercontext.context.prepare.memory_candidate_count": len(memory_hits), - "powercontext.context.prepare.experience_candidate_count": len(experience_hits), + "powercontext.context.build.memory_candidate_count": len(memory_hits), + "powercontext.context.build.experience_candidate_count": len(experience_hits), }, ) as span: build = builder.build_result( @@ -336,9 +336,9 @@ async def prepare(self, request: PrepareContextRequest, /) -> PreparedContext: ) if span is not None: span.set_attributes({ - "powercontext.context.prepare.selected_count": len(build.origins), - "powercontext.context.prepare.status": build.context.status, - "powercontext.context.prepare.content_bytes": build.context.content_bytes, + "powercontext.context.build.selected_count": len(build.origins), + "powercontext.context.build.status": build.context.status, + "powercontext.context.build.content_bytes": build.context.content_bytes, }) if self._runtime._recall_token_estimator is not None: try: diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index 9f018175a..21c66b20b 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -109,14 +109,7 @@ class _TracingMemoryReranker: def __init__(self, delegate: MemoryReranker, tracing: RuntimeTracing) -> None: self._delegate = delegate self._tracing = tracing - - @property - def policy_id(self) -> str: - return self._delegate.policy_id - - @policy_id.setter - def policy_id(self, value: str) -> None: - self._delegate.policy_id = value + self.policy_id = delegate.policy_id async def rerank( self, @@ -189,11 +182,21 @@ async def open_builtin_runtime( configured_reranker = generated_reranker if memory_reranker is None else memory_reranker if configured_reranker is not None and tracing is not None: configured_reranker = _TracingMemoryReranker(configured_reranker, tracing) - configured_embedding_source = ( - await _embedding_model(config.inference, resources, instrumentation) - if embedding_model is None - else embedding_model - ) + if embedding_model is None: + configured_embedding_source, readiness_embedding = await _embedding_models( + config.inference, + resources, + instrumentation, + ) + else: + from powercontext.builtin.inference.pydantic_ai import PydanticAIEmbeddingModel + + configured_embedding_source = embedding_model + readiness_embedding = ( + embedding_model._without_instrumentation() + if isinstance(embedding_model, PydanticAIEmbeddingModel) + else embedding_model + ) configured_embedding = ( None if configured_embedding_source is None else UsageReportingEmbeddingModel(configured_embedding_source) ) @@ -227,9 +230,9 @@ async def open_builtin_runtime( probe=generation_readiness, blocking=False, ) - if configured_embedding_source is not None: + if readiness_embedding is not None: readiness_probes["inference.embedding"] = ReadinessProbeDefinition( - probe=_embedding_readiness_probe(configured_embedding_source), + probe=_embedding_readiness_probe(readiness_embedding), blocking=False, ) runtime = await resources.enter_async_context( @@ -498,13 +501,13 @@ async def probe_generation() -> None: ) -async def _embedding_model( +async def _embedding_models( settings: InferenceConfig, resources: AsyncExitStack, instrumentation: InstrumentationSettings | None, -) -> EmbeddingModel | None: +) -> tuple[EmbeddingModel | None, EmbeddingModel | None]: if settings.embedding_model is None: - return None + return None, None from pydantic_ai import Embedder from pydantic_ai.embeddings import infer_embedding_model @@ -523,18 +526,26 @@ def provider_factory(provider_name: str) -> Provider[object]: model = infer_embedding_model(settings.embedding_model, provider_factory=provider_factory) for provider in providers: await resources.enter_async_context(provider) - return PydanticAIEmbeddingModel( - embedder=Embedder(model, instrument=instrumentation), - batch_size=settings.embedding_batch_size, - profile=EmbeddingProfile( - profile_id=_required(settings.embedding_profile_id), - model=settings.embedding_model, - dimension=_required(settings.embedding_dimension), - distance="l2", - normalization=settings.embedding_normalization, - ), - limits=InferenceLimits(timeout_seconds=settings.embedding_timeout_seconds), + profile = EmbeddingProfile( + profile_id=_required(settings.embedding_profile_id), + model=settings.embedding_model, + dimension=_required(settings.embedding_dimension), + distance="l2", + normalization=settings.embedding_normalization, ) + limits = InferenceLimits(timeout_seconds=settings.embedding_timeout_seconds) + + def adapter(instrument: InstrumentationSettings | bool | None) -> EmbeddingModel: + return PydanticAIEmbeddingModel( + embedder=Embedder(model, instrument=instrument), + batch_size=settings.embedding_batch_size, + profile=profile, + limits=limits, + ) + + # Readiness runs outside an application operation, so use the same provider model + # without instrumentation to avoid exporting an orphan inference span. + return adapter(instrumentation), adapter(False) def _embedding_readiness_probe(model: EmbeddingModel) -> ReadinessProbe: diff --git a/src/powercontext/server/factory.py b/src/powercontext/server/factory.py index 13057c46c..0fd9c4232 100644 --- a/src/powercontext/server/factory.py +++ b/src/powercontext/server/factory.py @@ -64,7 +64,7 @@ def create_server_app( resolved_tracing = ServerTracing.context_only() if tracing is None else tracing if metrics is not None: metrics.set_ready(False) - readiness_probe = _ServerReadinessProbe(metrics) + readiness_probe = _ServerReadinessProbe(metrics, tracing=resolved_tracing) @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: @@ -163,8 +163,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: class _ServerReadinessProbe: - def __init__(self, metrics: ServerMetrics | None) -> None: + def __init__(self, metrics: ServerMetrics | None, *, tracing: ServerTracing) -> None: self._metrics = metrics + self._tracing = tracing self._runtime: BuiltinRuntime | None = None self._last_status: ReadinessStatus | None = None @@ -178,14 +179,15 @@ def unbind(self) -> None: self._metrics.set_ready(False) async def __call__(self) -> ReadinessResponse: - runtime = self._runtime - if runtime is None: - response = ReadinessResponse( - status=ReadinessStatus.NOT_READY, - checks={"runtime": "not_ready"}, - ) - else: - response = await self._check(runtime) + with self._tracing._suppress_readiness_spans(): + runtime = self._runtime + if runtime is None: + response = ReadinessResponse( + status=ReadinessStatus.NOT_READY, + checks={"runtime": "not_ready"}, + ) + else: + response = await self._check(runtime) self._observe(response.status) return response diff --git a/src/powercontext/server/tracing.py b/src/powercontext/server/tracing.py index 0374637a5..eca028951 100644 --- a/src/powercontext/server/tracing.py +++ b/src/powercontext/server/tracing.py @@ -5,6 +5,7 @@ import asyncio from collections.abc import Iterator, Mapping from contextlib import contextmanager, suppress +from contextvars import ContextVar from importlib import import_module from typing import TYPE_CHECKING, Any @@ -37,13 +38,34 @@ _TraceAttribute = str | bool | int | float +class _SuppressibleTracer(Tracer): + """Delegate inference spans unless the current task is a readiness probe.""" + + def __init__(self, delegate: Tracer, suppressed: ContextVar[bool]) -> None: + self._delegate = delegate + self._suppressed = suppressed + self._noop = trace.NoOpTracer() + + def start_span(self, name: str, *args: Any, **kwargs: Any) -> Span: + return self._selected().start_span(name, *args, **kwargs) + + def start_as_current_span(self, name: str, *args: Any, **kwargs: Any) -> Any: + return self._selected().start_as_current_span(name, *args, **kwargs) + + def _selected(self) -> Tracer: + return self._noop if self._suppressed.get() else self._delegate + + class ServerTracing: """Create failure-isolated spans with one configured tracer provider.""" def __init__(self, provider: TracerProvider, *, instrumented: bool = False) -> None: self.provider = provider self.tracer = provider.get_tracer(_INSTRUMENTATION_NAME) - self.instrumentation = _inference_instrumentation(provider) if instrumented else None + self._inference_suppressed = ContextVar("powercontext_inference_suppressed", default=False) + self.instrumentation = ( + _inference_instrumentation(provider, self._inference_suppressed) if instrumented else None + ) @classmethod def context_only(cls) -> ServerTracing: @@ -101,6 +123,28 @@ def stage( raise span.finish("success") + @contextmanager + def _suppress_readiness_spans(self) -> Iterator[None]: + """Run readiness work under an unsampled context without inference spans.""" + + inference_token = self._inference_suppressed.set(True) + context_token: Token[Context] | None = None + try: + parent = trace.NonRecordingSpan( + trace.SpanContext( + trace_id=_ID_GENERATOR.generate_trace_id(), + span_id=_ID_GENERATOR.generate_span_id(), + is_remote=False, + trace_flags=trace.TraceFlags(0), + ) + ) + context_token = otel_context.attach(set_span_in_context(parent)) + yield + finally: + if context_token is not None: + otel_context.detach(context_token) + self._inference_suppressed.reset(inference_token) + def shutdown(self) -> None: with suppress(Exception): self.provider.shutdown() @@ -289,22 +333,27 @@ def configure_server_tracing(config: TracingConfig) -> ServerTracing: return ServerTracing(provider, instrumented=config.enabled) -def _inference_instrumentation(provider: TracerProvider) -> InstrumentationSettings | None: +def _inference_instrumentation( + provider: TracerProvider, + suppressed: ContextVar[bool], +) -> InstrumentationSettings | None: """Bind Pydantic AI spans to the Server provider without recording any content.""" try: settings_type = import_module("pydantic_ai.models.instrumented").InstrumentationSettings # Prompts, responses, Memory content, and vectors stay out of spans (RFC 0016); # model request parameters carry the full instructions, so they are excluded too. - return settings_type( + settings = settings_type( tracer_provider=provider, include_content=False, include_binary_content=False, include_model_request_parameters=False, ) + settings.tracer = _SuppressibleTracer(settings.tracer, suppressed) except Exception: # Tracing setup must never break the Runtime; an unavailable adapter just stays uninstrumented. return None + return settings def request_id_from_span(span: Span | None = None) -> str: diff --git a/tests/e2e/test_builtin_runtime.py b/tests/e2e/test_builtin_runtime.py index 833a57095..ea32101b7 100644 --- a/tests/e2e/test_builtin_runtime.py +++ b/tests/e2e/test_builtin_runtime.py @@ -2,21 +2,16 @@ import asyncio import json -from collections.abc import Iterator, Mapping -from contextlib import contextmanager from powercontext.builtin.artifacts.memory import MemoryCandidateRequest, MemoryEntryInput, MemoryRerankDecision from powercontext.builtin.inference import InferenceUsage from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.builtin.runtime import ( BuiltinConfig, - BuiltinRuntime, CaptureSource, PrepareContextRequest, RememberMemoryRequest, - RuntimeCapabilities, SearchMemoryRequest, - open_builtin_contexts, open_builtin_runtime, ) from powercontext.builtin.sources import ContentSource @@ -37,11 +32,9 @@ async def extract(self, request: MemoryCandidateRequest, /) -> tuple[MemoryEntry def test_builtin_runtime_uses_the_selected_sqlite_database() -> None: async def scenario() -> None: - tracing = _RecordingTracing() async with open_builtin_runtime( BuiltinConfig(database=SQLiteConfig()), candidate_pipeline=_ContentCandidatePipeline(), - tracing=tracing, ) as runtime: captured = await runtime.sources.for_scope("project").capture( CaptureSource( @@ -75,7 +68,6 @@ async def scenario() -> None: assert no_memory.content is None assert no_match.status == "empty" assert no_match.content is None - assert not any(name == "memory.rerank" for name, _attributes in tracing.stages) asyncio.run(scenario()) @@ -114,108 +106,3 @@ async def scenario() -> None: assert all(page.rerank is not None for page in pages) asyncio.run(scenario()) - - -_TraceValue = str | bool | int | float - - -class _RecordingSpan: - def __init__(self, attributes: dict[str, _TraceValue]) -> None: - self.attributes = attributes - - def set_attributes(self, attributes: Mapping[str, _TraceValue], /) -> None: - self.attributes.update(attributes) - - -class _RecordingTracing: - def __init__(self) -> None: - self.stages: list[tuple[str, dict[str, _TraceValue]]] = [] - - @contextmanager - def stage( - self, - name: str, - *, - attributes: Mapping[str, _TraceValue], - ) -> Iterator[_RecordingSpan]: - recorded = dict(attributes) - self.stages.append((name, recorded)) - yield _RecordingSpan(recorded) - - -class _DynamicPolicyReranker: - policy_id = "test.dynamic-rerank.v1" - - async def rerank(self, query, candidates, limit, /) -> MemoryRerankDecision: - self.policy_id = "test.dynamic-rerank.v2" - return MemoryRerankDecision( - selected_ranks=(1,), - usage=InferenceUsage(requests=1), - discarded_rank_count=2, - ) - - -def test_traced_reranker_preserves_dynamic_policy_and_reports_rank_discards() -> None: - async def scenario() -> None: - tracing = _RecordingTracing() - reranker = _DynamicPolicyReranker() - async with open_builtin_runtime( - BuiltinConfig(), - memory_reranker=reranker, - tracing=tracing, - ) as runtime: - memory = runtime.memory.for_scope("traced-reranker") - await memory.remember( - RememberMemoryRequest(entries=(MemoryEntryInput(kind="fact", text="Trace this fact."),)) - ) - page = await memory.search(SearchMemoryRequest(query="trace", mode="fts", limit=1)) - - assert page.rerank is not None - assert page.rerank.policy_id == "test.dynamic-rerank.v2" - rerank_attributes = next(attributes for name, attributes in tracing.stages if name == "memory.rerank") - assert rerank_attributes == { - "powercontext.memory.rerank.candidate_count": 1, - "powercontext.memory.rerank.limit": 1, - "powercontext.memory.rerank.selected_count": 1, - "powercontext.memory.rerank.discarded_rank_count": 2, - "powercontext.memory.rerank.used_fallback": False, - } - - asyncio.run(scenario()) - - -def test_context_trace_is_stable_without_memory_or_experience_recall() -> None: - async def scenario() -> None: - tracing = _RecordingTracing() - async with ( - open_builtin_contexts(BuiltinConfig()) as contexts, - BuiltinRuntime( - provider=contexts, - capabilities=RuntimeCapabilities(memory_extraction=False, memory_search_modes=("fts",)), - tracing=tracing, - ) as runtime, - ): - prepared = await runtime.context.for_scope("empty-traced-context").prepare( - PrepareContextRequest(query="private empty query") - ) - - assert prepared.status == "empty" - stages = dict(tracing.stages) - memory_search = stages["memory.search"] - assert memory_search["powercontext.memory.search.requested_mode"] == "auto" - assert isinstance(memory_search["powercontext.memory.search.limit"], int) - assert memory_search["powercontext.memory.search.memory_present"] is False - assert memory_search["powercontext.memory.search.result_count"] == 0 - experience_search = stages["experience.search"] - assert experience_search["powercontext.experience.search.configured"] is False - assert isinstance(experience_search["powercontext.experience.search.limit"], int) - assert experience_search["powercontext.experience.search.result_count"] == 0 - assert stages["context.prepare"] == { - "powercontext.context.prepare.memory_candidate_count": 0, - "powercontext.context.prepare.experience_candidate_count": 0, - "powercontext.context.prepare.selected_count": 0, - "powercontext.context.prepare.status": "empty", - "powercontext.context.prepare.content_bytes": 0, - } - - asyncio.run(scenario()) diff --git a/tests/e2e/test_observability.py b/tests/e2e/test_observability.py index aa9c16b28..a557d79bc 100644 --- a/tests/e2e/test_observability.py +++ b/tests/e2e/test_observability.py @@ -12,13 +12,22 @@ from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from opentelemetry.trace import SpanKind +from opentelemetry.sdk.trace.sampling import ALWAYS_ON from pydantic_ai import Embedder from pydantic_ai.embeddings import TestEmbeddingModel from pydantic_ai.models import Model +from pydantic_ai.models.instrumented import InstrumentationSettings from pydantic_ai.models.test import TestModel - -from powercontext.builtin.artifacts.memory import EmbeddingProfile +from sqlalchemy.ext.asyncio import AsyncConnection + +from powercontext.artifacts import ArtifactRef +from powercontext.builtin.artifacts.memory import ( + EmbeddingProfile, + MemoryCapabilities, + MemoryProjection, + MemorySearchChannels, + MemorySearchRequest, +) from powercontext.builtin.inference.pydantic_ai import PydanticAIEmbeddingModel from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.builtin.runtime.config import InferenceConfig, RuntimeConfig @@ -56,18 +65,80 @@ "powercontext.experience.search.limit", "powercontext.experience.search.result_count", }, - "context.prepare": { + "context.build": { "powercontext.operation.name", "powercontext.operation.unit", "powercontext.operation.outcome", - "powercontext.context.prepare.memory_candidate_count", - "powercontext.context.prepare.experience_candidate_count", - "powercontext.context.prepare.selected_count", - "powercontext.context.prepare.status", - "powercontext.context.prepare.content_bytes", + "powercontext.context.build.memory_candidate_count", + "powercontext.context.build.experience_candidate_count", + "powercontext.context.build.selected_count", + "powercontext.context.build.status", + "powercontext.context.build.content_bytes", }, } +_VECTOR_PROFILE = EmbeddingProfile( + profile_id="trace-test-v1", + model="test", + dimension=3, + distance="l2", + normalization="unit", +) + + +class _VectorMemoryIndex: + """Expose deterministic vector capability without a platform extension.""" + + capabilities = MemoryCapabilities( + fts=False, + vector=True, + embedding_profile=_VECTOR_PROFILE, + ) + tables = () + + async def initialize(self, _connection: AsyncConnection, /) -> None: + pass + + async def replace( + self, + _connection: AsyncConnection, + _scope_id: str, + _memory_ref: ArtifactRef, + _projections: tuple[MemoryProjection, ...], + /, + ) -> None: + pass + + async def search( + self, + _connection: AsyncConnection, + _scope_id: str, + request: MemorySearchRequest, + /, + ) -> MemorySearchChannels: + assert request.mode == "vector" + assert request.query_vector is not None + return MemorySearchChannels() + + async def vector_complete( + self, + _connection: AsyncConnection, + _scope_id: str, + _memories: tuple[ArtifactRef, ...], + profile: EmbeddingProfile, + /, + ) -> bool: + return profile == _VECTOR_PROFILE + + async def hydrate( + self, + _connection: AsyncConnection, + _scope_id: str, + projections: tuple[MemoryProjection, ...], + /, + ) -> tuple[MemoryProjection, ...]: + return projections + def test_observability_signals_correlate_without_counting_the_mcp_bridge(caplog, tmp_path) -> None: exporter = InMemorySpanExporter() @@ -190,7 +261,7 @@ def test_memory_read_stage_spans_are_bounded_and_nested(monkeypatch, tmp_path) - # Resolve the configured test model without consulting the environment or a real provider. monkeypatch.setattr( "pydantic_ai.models.infer_model", - lambda model: model if isinstance(model, Model) else TestModel(custom_output_text='{"selected_ranks":[1]}'), + lambda model: model if isinstance(model, Model) else TestModel(custom_output_text='{"selected_ranks":[99,1]}'), ) exporter = InMemorySpanExporter() @@ -292,7 +363,7 @@ def test_memory_read_stage_spans_are_bounded_and_nested(monkeypatch, tmp_path) - "powercontext.memory.rerank.candidate_count": 1, "powercontext.memory.rerank.limit": 1, "powercontext.memory.rerank.selected_count": 1, - "powercontext.memory.rerank.discarded_rank_count": 0, + "powercontext.memory.rerank.discarded_rank_count": 1, "powercontext.memory.rerank.used_fallback": False, "powercontext.operation.outcome": "success", } @@ -328,12 +399,12 @@ def test_memory_read_stage_spans_are_bounded_and_nested(monkeypatch, tmp_path) - ready_experience = _only_child(spans, ready_application, "experience.search") assert (ready_experience.attributes or {})["powercontext.experience.search.configured"] is True assert (ready_experience.attributes or {})["powercontext.experience.search.result_count"] == 0 - ready_context = _only_child(spans, ready_application, "context.prepare") - assert (ready_context.attributes or {})["powercontext.context.prepare.memory_candidate_count"] == 1 - assert (ready_context.attributes or {})["powercontext.context.prepare.experience_candidate_count"] == 0 - assert (ready_context.attributes or {})["powercontext.context.prepare.selected_count"] == 1 - assert (ready_context.attributes or {})["powercontext.context.prepare.status"] == "ready" - ready_content_bytes = (ready_context.attributes or {})["powercontext.context.prepare.content_bytes"] + ready_context = _only_child(spans, ready_application, "context.build") + assert (ready_context.attributes or {})["powercontext.context.build.memory_candidate_count"] == 1 + assert (ready_context.attributes or {})["powercontext.context.build.experience_candidate_count"] == 0 + assert (ready_context.attributes or {})["powercontext.context.build.selected_count"] == 1 + assert (ready_context.attributes or {})["powercontext.context.build.status"] == "ready" + ready_content_bytes = (ready_context.attributes or {})["powercontext.context.build.content_bytes"] assert isinstance(ready_content_bytes, int) assert ready_content_bytes > 0 @@ -345,10 +416,10 @@ def test_memory_read_stage_spans_are_bounded_and_nested(monkeypatch, tmp_path) - assert not _children(spans, empty_memory, "memory.rerank") empty_experience = _only_child(spans, empty_application, "experience.search") assert (empty_experience.attributes or {})["powercontext.experience.search.result_count"] == 0 - empty_context = _only_child(spans, empty_application, "context.prepare") - assert (empty_context.attributes or {})["powercontext.context.prepare.selected_count"] == 0 - assert (empty_context.attributes or {})["powercontext.context.prepare.status"] == "empty" - assert (empty_context.attributes or {})["powercontext.context.prepare.content_bytes"] == 0 + empty_context = _only_child(spans, empty_application, "context.build") + assert (empty_context.attributes or {})["powercontext.context.build.selected_count"] == 0 + assert (empty_context.attributes or {})["powercontext.context.build.status"] == "empty" + assert (empty_context.attributes or {})["powercontext.context.build.content_bytes"] == 0 for span in spans: allowed_keys = _STAGE_ATTRIBUTE_KEYS.get(span.name) @@ -367,56 +438,130 @@ def test_memory_read_stage_spans_are_bounded_and_nested(monkeypatch, tmp_path) - assert no_match_query not in exported -def test_embedding_span_joins_memory_search_stage_without_recording_text() -> None: +def test_vector_search_exports_embedding_under_memory_search_without_recording_text(monkeypatch, tmp_path) -> None: + monkeypatch.setattr( + "pydantic_ai.embeddings.infer_embedding_model", + lambda _model, **_kwargs: TestEmbeddingModel(dimensions=3), + ) + monkeypatch.setattr( + "powercontext.builtin.runtime.composition.SQLiteMemoryFTSIndex", + _VectorMemoryIndex, + ) + exporter = InMemorySpanExporter() provider = TracerProvider(shutdown_on_exit=False) provider.add_span_processor(SimpleSpanProcessor(exporter)) - tracing = ServerTracing(provider, instrumented=True) - instrumentation = tracing.instrumentation - assert instrumentation is not None - embedding_model = PydanticAIEmbeddingModel( - embedder=Embedder(TestEmbeddingModel(dimensions=3), instrument=instrumentation), - profile=EmbeddingProfile( - profile_id="trace-test-v1", - model="test", - dimension=3, - distance="l2", - normalization="unit", + app = create_server_app( + settings=ServerSettings( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'vector-tracing.db'}"), + inference=InferenceConfig( + embedding_model="test", + embedding_profile_id=_VECTOR_PROFILE.profile_id, + embedding_dimension=_VECTOR_PROFILE.dimension, + ), + mcp=McpConfig(enabled=False), ), + tracing=ServerTracing(provider, instrumented=True), ) + scope_id = "project:private-vector-scope" + memory_content = "Private vector memory sentinel." private_text = "private embedding sentinel" - async def scenario() -> None: - application = tracing.start_span( - "powercontext search_memory", - kind=SpanKind.INTERNAL, - attributes={ - "powercontext.operation.name": "search_memory", - "powercontext.operation.unit": "application", - }, + with TestClient(app) as client: + remembered = client.post( + "/v1/memory/remember", + json={"scope_id": scope_id, "kind": "fact", "text": memory_content}, + ) + searched = client.post( + "/v1/memory/search", + json={"scope_id": scope_id, "query": private_text, "limit": 1, "mode": "vector"}, ) - try: - with tracing.stage( - "memory.search", - attributes={ - "powercontext.memory.search.requested_mode": "vector", - "powercontext.memory.search.limit": 1, - }, - ): - await embedding_model.embed((private_text,)) - except BaseException as error: - application.finish("failure", error=error) - raise - application.finish("success") - asyncio.run(scenario()) + assert remembered.status_code == 200 + assert searched.status_code == 200 + assert searched.json()["mode"] == "vector" + assert searched.json()["hits"] == [] spans = list(exporter.get_finished_spans()) - application = next(span for span in spans if span.name == "powercontext search_memory") + applications = [span for span in spans if span.name == "powercontext search_memory"] + assert len(applications) == 1 + application = applications[0] search = _only_child(spans, application, "memory.search") embedding = _only_child_with_prefix(spans, search, "embeddings ") + assert dict(search.attributes or {}) == { + "powercontext.operation.name": "memory.search", + "powercontext.operation.unit": "stage", + "powercontext.memory.search.requested_mode": "vector", + "powercontext.memory.search.limit": 1, + "powercontext.memory.search.memory_present": True, + "powercontext.memory.search.mode": "vector", + "powercontext.memory.search.result_count": 0, + "powercontext.operation.outcome": "success", + } + assert embedding.name == "embeddings test" assert embedding.context.trace_id == application.context.trace_id - assert private_text not in _exported_span_data(spans) + assert not any(_is_inference_span(span) and span.parent is None for span in spans) + exported = _exported_span_data(spans) + assert scope_id not in exported + assert memory_content not in exported + assert private_text not in exported + + +def test_injected_always_on_embedding_skips_readiness_but_traces_vector_search(monkeypatch, tmp_path) -> None: + monkeypatch.setattr( + "powercontext.builtin.runtime.composition.SQLiteMemoryFTSIndex", + _VectorMemoryIndex, + ) + + exporter = InMemorySpanExporter() + provider = TracerProvider(sampler=ALWAYS_ON, shutdown_on_exit=False) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracing = ServerTracing(provider, instrumented=True) + embedding_model = PydanticAIEmbeddingModel( + embedder=Embedder( + TestEmbeddingModel(dimensions=3), + instrument=InstrumentationSettings(tracer_provider=provider), + ), + profile=_VECTOR_PROFILE, + ) + + app = create_server_app( + settings=ServerSettings( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'readiness-tracing.db'}"), + mcp=McpConfig(enabled=False), + ), + embedding_model=embedding_model, + tracing=tracing, + ) + scope_id = "project:injected-always-on" + + with TestClient(app) as client: + readiness = client.get("/health/ready") + readiness_spans = list(exporter.get_finished_spans()) + assert not [span for span in readiness_spans if span.parent is None] + assert not any(_is_inference_span(span) for span in readiness_spans) + + remembered = client.post( + "/v1/memory/remember", + json={"scope_id": scope_id, "kind": "fact", "text": "Private injected vector memory."}, + ) + exporter.clear() + searched = client.post( + "/v1/memory/search", + json={"scope_id": scope_id, "query": "private injected query", "limit": 1, "mode": "vector"}, + ) + + assert readiness.status_code == 200 + assert readiness.json()["checks"]["inference.embedding"] == "ready" + assert remembered.status_code == 200 + assert searched.status_code == 200 + assert searched.json()["mode"] == "vector" + spans = list(exporter.get_finished_spans()) + application = next(span for span in spans if span.name == "powercontext search_memory") + search = _only_child(spans, application, "memory.search") + embedding = _only_child_with_prefix(spans, search, "embeddings ") + assert embedding.name == "embeddings test" + assert [span for span in spans if _is_inference_span(span)] == [embedding] def _flush_memory_spans(database_path: Path, *, instrumented: bool) -> list[ReadableSpan]: diff --git a/tests/test_server_tracing.py b/tests/test_server_tracing.py index f9db435c0..aa9a15fef 100644 --- a/tests/test_server_tracing.py +++ b/tests/test_server_tracing.py @@ -200,23 +200,6 @@ def test_runtime_stage_records_cancellation_and_reraises_same_error() -> None: assert spans[0].status.status_code is StatusCode.UNSET -def test_runtime_stage_finishes_and_detaches_for_base_exceptions() -> None: - tracing, exporter = _tracing() - - with pytest.raises(SystemExit), tracing.stage("memory.search", attributes={}): - raise SystemExit - with tracing.stage("experience.search", attributes={}): - pass - - spans = {span.name: span for span in exporter.get_finished_spans()} - failed = spans["memory.search"] - following = spans["experience.search"] - assert failed.attributes is not None - assert failed.attributes["powercontext.operation.outcome"] == "failure" - assert failed.attributes["error.type"] == "SystemExit" - assert following.parent is None - - def test_runtime_stage_isolates_tracer_failure(monkeypatch) -> None: tracing, exporter = _tracing() From 4b989751420d405a70801213b8ad38a1d4c000ea Mon Sep 17 00:00:00 2001 From: Kairo-J Date: Tue, 18 Aug 2026 16:20:12 +0800 Subject: [PATCH 3/3] fix(tracing): isolate readiness tracing failures - Make readiness tracing setup and cleanup best-effort. - Prevent tracing failures from aborting Server startup. - Preserve readiness errors and cancellation behavior. - Cover attach failures through startup and the readiness endpoint. --- src/powercontext/server/tracing.py | 14 ++++++++++---- tests/test_server_tracing.py | 26 ++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 4 deletions(-) diff --git a/src/powercontext/server/tracing.py b/src/powercontext/server/tracing.py index eca028951..b6fc67bda 100644 --- a/src/powercontext/server/tracing.py +++ b/src/powercontext/server/tracing.py @@ -127,9 +127,11 @@ def stage( def _suppress_readiness_spans(self) -> Iterator[None]: """Run readiness work under an unsampled context without inference spans.""" - inference_token = self._inference_suppressed.set(True) + inference_token = None context_token: Token[Context] | None = None - try: + with suppress(Exception): + inference_token = self._inference_suppressed.set(True) + with suppress(Exception): parent = trace.NonRecordingSpan( trace.SpanContext( trace_id=_ID_GENERATOR.generate_trace_id(), @@ -139,11 +141,15 @@ def _suppress_readiness_spans(self) -> Iterator[None]: ) ) context_token = otel_context.attach(set_span_in_context(parent)) + try: yield finally: if context_token is not None: - otel_context.detach(context_token) - self._inference_suppressed.reset(inference_token) + with suppress(Exception): + otel_context.detach(context_token) + if inference_token is not None: + with suppress(Exception): + self._inference_suppressed.reset(inference_token) def shutdown(self) -> None: with suppress(Exception): diff --git a/tests/test_server_tracing.py b/tests/test_server_tracing.py index aa9a15fef..0a498160b 100644 --- a/tests/test_server_tracing.py +++ b/tests/test_server_tracing.py @@ -11,6 +11,7 @@ from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from opentelemetry.trace import SpanKind, StatusCode +from powercontext.builtin.persistence.sqlite import SQLiteConfig from powercontext.client import PowerContextClient from powercontext.server.factory import create_server_app from powercontext.server.settings import ( @@ -215,6 +216,31 @@ def start_span(self, *_args: object, **_kwargs: object) -> None: assert exporter.get_finished_spans() == () +def test_readiness_ignores_tracing_setup_failure(monkeypatch, tmp_path) -> None: + tracing, _ = _tracing(instrumented=True) + app = create_server_app( + settings=ServerSettings( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'runtime.db'}"), + mcp=McpConfig(enabled=False), + metrics=MetricsConfig(enabled=False), + logging=ServerLoggingConfig(access=False), + ), + scheduler_path=tmp_path / "scheduler.db", + tracing=tracing, + ) + + def fail_attach(_context: object) -> None: + raise RuntimeError + + monkeypatch.setattr("powercontext.server.tracing.otel_context.attach", fail_attach) + + with TestClient(app) as client: + response = client.get("/health/ready") + + assert response.status_code == 200 + assert response.json()["status"] == "ready" + + def test_tracing_export_requires_the_otlp_extra(monkeypatch) -> None: module = "opentelemetry.exporter.otlp.proto.http.trace_exporter" monkeypatch.setitem(sys.modules, module, None)