From b252bd7a5d87b64efe403cf471e6dd6ab8333fbb Mon Sep 17 00:00:00 2001 From: guozhihao-224 Date: Sun, 16 Aug 2026 23:34:03 +0800 Subject: [PATCH] feat(tracing): enhance tracing for scheduled operations and memory flush - Updated documentation to reflect the addition of a new `memory.flush` span in the tracing output. - Implemented minimal OpenTelemetry-free tracing hooks in `tracing.py` for domain code. - Added tracing for scheduled source window processing and experience incubation, capturing success, failure, and cancellation outcomes. - Introduced a `DomainTracer` to adapt server tracing for domain-specific spans. - Enhanced tests to verify the correct tracing behavior for scheduled operations and memory flush events. --- docs/en/docs/how-to/trace-with-phoenix.md | 6 +- docs/zh/docs/how-to/trace-with-phoenix.md | 6 +- .../builtin/runtime/application.py | 103 +++++++++++++++- .../builtin/runtime/composition.py | 4 + src/powercontext/server/factory.py | 3 +- src/powercontext/server/tracing.py | 32 ++++- src/powercontext/tracing.py | 30 +++++ tests/builtin/runtime/test_scheduler.py | 110 ++++++++++++++++++ tests/e2e/test_observability.py | 76 +++++++++++- 9 files changed, 359 insertions(+), 11 deletions(-) create mode 100644 src/powercontext/tracing.py diff --git a/docs/en/docs/how-to/trace-with-phoenix.md b/docs/en/docs/how-to/trace-with-phoenix.md index c51462260..d7cf4fdfa 100644 --- a/docs/en/docs/how-to/trace-with-phoenix.md +++ b/docs/en/docs/how-to/trace-with-phoenix.md @@ -68,15 +68,19 @@ Memory extraction runs during the flush, not during capture. ## Read the trace Open , select the `default` project, and open the most recent trace for -`powercontext-server`. The flush produces four nested spans in one trace: +`powercontext-server`. The flush produces five nested spans in one trace: | Span | Meaning | | --- | --- | | `HTTP flush_memory` | The inbound HTTP request. `powercontext.request.id` matches the `X-PowerContext-Request-ID` response header. | | `powercontext flush_memory` | The application operation, independent of the transport that invoked it. | +| `memory.flush` | One Source-window flush, also emitted by a scheduled Source-window activation. | | `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. | +A scheduled Source-window activation produces its own root trace (`scheduled.process_source_window` with the same +`memory.flush` underneath), independent of any HTTP or MCP request. Scheduled spans carry no `powercontext.request.id`. + 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/zh/docs/how-to/trace-with-phoenix.md b/docs/zh/docs/how-to/trace-with-phoenix.md index 14d303bdb..31f580225 100644 --- a/docs/zh/docs/how-to/trace-with-phoenix.md +++ b/docs/zh/docs/how-to/trace-with-phoenix.md @@ -66,15 +66,19 @@ Memory extraction 发生在 flush 阶段，而不是捕获阶段。 ## 查看 trace 打开 ，选择 `default` project，打开 `powercontext-server` 最新的一条 trace。这次 flush -在同一条 trace 中产生四层嵌套 span： +在同一条 trace 中产生五层嵌套 span： | Span | 含义 | | --- | --- | | `HTTP flush_memory` | 入站 HTTP 请求。`powercontext.request.id` 与响应头 `X-PowerContext-Request-ID` 一致。 | | `powercontext flush_memory` | application 操作，与调用它的 transport 无关。 | +| `memory.flush` | 一次 Source-window flush，scheduled Source-window activation 也会发出它。 | | `invoke_agent memory_extraction` | 一次 PowerContext generation 任务。名字标识用途，不是模型名。 | | `chat ` | 一次发往模型 provider 的请求，包含 token 用量和耗时。 | +scheduled Source-window activation 会产生自己的 root trace（`scheduled.process_source_window`，下面带同样的 +`memory.flush`），与任何 HTTP 或 MCP 请求无关。Scheduled span 不携带 `powercontext.request.id`。 + 其他 generation 任务遵循同样的命名约定：`experience_incubation`、`experience_generation`、`skill_generation`、 `handoff_generation` 和 `memory_rerank`。配置了 embedding model 时，embedding 调用会作为 `embeddings ` span 挂在触发它的操作之下。 diff --git a/src/powercontext/builtin/runtime/application.py b/src/powercontext/builtin/runtime/application.py index f6342e923..9e31d08c5 100644 --- a/src/powercontext/builtin/runtime/application.py +++ b/src/powercontext/builtin/runtime/application.py @@ -111,6 +111,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler from powercontext.builtin.handoff_report.application import HandoffReportApplication + from powercontext.tracing import Span, Tracer logger = logging.getLogger(__name__) @@ -371,7 +372,24 @@ async def incubate(self, /, *, limit: int | None = None) -> ExperienceIncubation ), self._runtime._lock(self.scope_id), ): - return await incubator(self.scope_id, window_limit) + span = _start_operation_span(self._runtime, "experience.incubation") + try: + result = await incubator(self.scope_id, window_limit) + except asyncio.CancelledError: + _finish_span(span, "cancelled") + raise + except Exception as error: + _finish_span(span, "failure", error=error) + raise + _finish_span( + span, + "success" if result.processed else "noop", + attributes={ + "source_count": result.source_count, + "candidate_count": result.candidate_count, + }, + ) + return result class ExperienceApplication: @@ -766,7 +784,21 @@ async def flush(self, /, *, limit: int | None = None) -> MemoryFlushResult: ) as context: window_limit = self._runtime.source_window_limit if limit is None else limit async with self._runtime._lock(self.scope_id): - return await context.triggers.flush(limit=window_limit) + span = _start_operation_span(self._runtime, "memory.flush") + try: + result = await context.triggers.flush(limit=window_limit) + except asyncio.CancelledError: + _finish_span(span, "cancelled") + raise + except Exception as error: + _finish_span(span, "failure", error=error) + raise + _finish_span( + span, + "success" if result.processed else "noop", + attributes={"source_count": result.source_count}, + ) + return result async def cursor(self) -> SourceCursor: async with self._runtime._context(self.scope_id) as context: @@ -798,6 +830,11 @@ async def run(self) -> None: if self._runtime._closing or self._runtime._closed: return started_at = perf_counter() + span = _start_background_span( + self._runtime, + "scheduled.process_source_window", + operation="process_source_window", + ) try: result = await self._runtime.memory.for_scope(scope_id).flush() except asyncio.CancelledError: @@ -806,6 +843,7 @@ async def run(self) -> None: operation="process_source_window", started_at=started_at, ) + _finish_span(span, "cancelled") raise except Exception as error: _log_scheduled_processing( @@ -814,13 +852,16 @@ async def run(self) -> None: started_at=started_at, error=error, ) + _finish_span(span, "failure", error=error) else: + outcome = "success" if result.processed else "noop" _log_scheduled_processing( - "success" if result.processed else "noop", + outcome, operation="process_source_window", started_at=started_at, source_count=result.source_count, ) + _finish_span(span, outcome, attributes={"source_count": result.source_count}) class ScheduledExperienceProcessor: @@ -838,6 +879,11 @@ async def run(self) -> None: if self._runtime._closing or self._runtime._closed: return started_at = perf_counter() + span = _start_background_span( + self._runtime, + "scheduled.incubate_experience_candidates", + operation="incubate_experience_candidates", + ) try: result = await self._runtime.experience.for_scope(scope_id).incubate() except asyncio.CancelledError: @@ -846,6 +892,7 @@ async def run(self) -> None: operation="incubate_experience_candidates", started_at=started_at, ) + _finish_span(span, "cancelled") raise except Exception as error: _log_scheduled_processing( @@ -854,14 +901,24 @@ async def run(self) -> None: started_at=started_at, error=error, ) + _finish_span(span, "failure", error=error) else: + outcome = "success" if result.processed else "noop" _log_scheduled_processing( - "success" if result.processed else "noop", + outcome, operation="incubate_experience_candidates", started_at=started_at, source_count=result.source_count, candidate_count=result.candidate_count, ) + _finish_span( + span, + outcome, + attributes={ + "source_count": result.source_count, + "candidate_count": result.candidate_count, + }, + ) def _log_scheduled_processing( @@ -894,6 +951,42 @@ def _log_scheduled_processing( ) +def _start_background_span(runtime: BuiltinRuntime, name: str, *, operation: str) -> Span | None: + tracer = runtime._tracer + if tracer is None: + return None + return tracer.start_root_span( + name, + attributes={ + "powercontext.operation.name": operation, + "powercontext.operation.unit": "background", + }, + ) + + +def _start_operation_span(runtime: BuiltinRuntime, name: str) -> Span | None: + tracer = runtime._tracer + if tracer is None: + return None + # note(guozhihao-224): boundary spans appear under both HTTP and scheduled roots, so they carry + # no operation identity; outcome and bounded counts are set on finish. + return tracer.start_span(name, attributes={}) + + +def _finish_span( + span: Span | None, + outcome: str, + *, + error: BaseException | None = None, + attributes: dict[str, object] | None = None, +) -> None: + if span is None: + return + if attributes: + span.set_attributes(attributes) + span.finish(outcome, error=error) + + class BuiltinRuntime: """Add business-specific operations over composed built-in contexts.""" @@ -914,6 +1007,7 @@ def __init__( recall_token_estimator: RecallTokenEstimator | None = None, readiness: RuntimeReadinessChecks | None = None, clock: Clock | None = None, + tracer: Tracer | None = None, ) -> None: if source_window_limit < 1: raise _RuntimeConfigurationError("source_window_limit") @@ -929,6 +1023,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._tracer = tracer self.source_window_limit = source_window_limit self._locks: dict[str, asyncio.Lock] = {} self._processor_lock = asyncio.Lock() diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index 7435de29d..eed0b8be9 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -58,6 +58,8 @@ if TYPE_CHECKING: from pydantic_ai.models.instrumented import InstrumentationSettings + from powercontext.tracing import Tracer + ValueT = TypeVar("ValueT") @@ -115,6 +117,7 @@ async def open_builtin_runtime( token_estimator: TokenEstimator | None = None, memory_reranker: MemoryReranker | None = None, instrumentation: InstrumentationSettings | None = None, + tracing: Tracer | None = None, ) -> AsyncIterator[BuiltinRuntime]: """Open the selected database, inference adapters, and built-in runtime.""" @@ -210,6 +213,7 @@ async def open_builtin_runtime( statistics_service=contexts.statistics, recall_token_estimator=contexts.estimate_recall_tokens, readiness=RuntimeReadinessChecks(readiness_probes), + tracer=tracing, ) ) if config.handoff_report.enabled: diff --git a/src/powercontext/server/factory.py b/src/powercontext/server/factory.py index 9db26bf1f..9e8fe2f3e 100644 --- a/src/powercontext/server/factory.py +++ b/src/powercontext/server/factory.py @@ -30,7 +30,7 @@ from powercontext.server.metrics import CONTENT_TYPE_LATEST, HttpMetricsMiddleware, ServerMetrics from powercontext.server.middleware import StaticBearerMiddleware from powercontext.server.settings import ServerSettings -from powercontext.server.tracing import HttpTracingMiddleware, ServerTracing +from powercontext.server.tracing import DomainTracer, HttpTracingMiddleware, ServerTracing from powercontext.server.web import mount_web_ui logger = logging.getLogger(__name__) @@ -80,6 +80,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: handoff_pipeline=handoff_pipeline, embedding_model=embedding_model, instrumentation=resolved_tracing.instrumentation, + tracing=DomainTracer(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..9d76bc3bd 100644 --- a/src/powercontext/server/tracing.py +++ b/src/powercontext/server/tracing.py @@ -20,7 +20,7 @@ from opentelemetry.trace import Span, SpanKind, Status, StatusCode, Tracer, set_span_in_context from starlette.types import ASGIApp, Message, Receive, Scope, Send -from powercontext.server.context import bind_request_id, is_internal_bridge, reset_request_id +from powercontext.server.context import bind_request_id, current_request_id, is_internal_bridge, reset_request_id from powercontext.server.settings import TracingConfig if TYPE_CHECKING: @@ -138,6 +138,35 @@ def finish( self.span.end() +class DomainTracer: + """Adapt ServerTracing to the domain Tracer protocol for built-in spans.""" + + def __init__(self, tracing: ServerTracing) -> None: + self._tracing = tracing + + def start_span(self, name: str, *, attributes: dict[str, object]) -> _ActiveSpan: + span_attributes = dict(attributes) + request_id = current_request_id() + # note(guozhihao-224): only child spans join a request; scheduled roots are fresh traces with no request id. + if request_id is not None: + span_attributes["powercontext.request.id"] = request_id + return self._tracing.start_span( + name, + kind=SpanKind.INTERNAL, + attributes=span_attributes, + context=None, + ) + + def start_root_span(self, name: str, *, attributes: dict[str, object]) -> _ActiveSpan: + # note(guozhihao-224): fresh empty context keeps scheduled activations as independent trace roots. + return self._tracing.start_span( + name, + kind=SpanKind.INTERNAL, + attributes=dict(attributes), + context=Context(), + ) + + class HttpTracingMiddleware: """Trace external HTTP requests while excluding infrastructure and the MCP bridge.""" @@ -302,6 +331,7 @@ def _request_id_attribute(scope: Scope) -> dict[str, str]: __all__ = [ + "DomainTracer", "HttpTracingMiddleware", "McpTracingMiddleware", "ServerTracing", diff --git a/src/powercontext/tracing.py b/src/powercontext/tracing.py new file mode 100644 index 000000000..97c39ffb8 --- /dev/null +++ b/src/powercontext/tracing.py @@ -0,0 +1,30 @@ +"""Minimal OpenTelemetry-free tracing hooks for domain code. + +Domain code (the built-in Runtime and artifact services) must not depend on an OpenTelemetry +implementation. These protocols describe the small surface those services need to record bounded +spans; the ready-to-run Server adapts its ServerTracing to this protocol and injects it through +composition. When no tracer is injected, domain code creates no spans and behavior is unchanged. +""" + +from __future__ import annotations + +from typing import Protocol + + +class Span(Protocol): + """One failure-isolated span opened by domain code.""" + + def set_attributes(self, attributes: dict[str, object]) -> None: ... + + def finish(self, outcome: str, *, error: BaseException | None = None) -> None: ... + + +class Tracer(Protocol): + """Start bounded spans without inheriting any concrete tracing implementation.""" + + def start_span(self, name: str, *, attributes: dict[str, object]) -> Span | None: ... + + def start_root_span(self, name: str, *, attributes: dict[str, object]) -> Span | None: ... + + +__all__ = ["Span", "Tracer"] diff --git a/tests/builtin/runtime/test_scheduler.py b/tests/builtin/runtime/test_scheduler.py index 9055ebd77..56efa1391 100644 --- a/tests/builtin/runtime/test_scheduler.py +++ b/tests/builtin/runtime/test_scheduler.py @@ -5,6 +5,9 @@ import sqlite3 import pytest +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 powercontext import PowerContext from powercontext.builtin.runtime import ( @@ -21,6 +24,8 @@ scheduler_database_path, ) from powercontext.builtin.sources import SourceCursor +from powercontext.server.tracing import DomainTracer, ServerTracing +from powercontext.tracing import Tracer class _Provider: @@ -67,6 +72,24 @@ async def __call__(self, scope_id: str, limit: int) -> ExperienceIncubationResul ) +class _FailingTriggers: + async def flush(self, *, limit: int) -> MemoryFlushResult: + del limit + raise RuntimeError("scheduled failure") # noqa: TRY003 + + async def cursor(self) -> SourceCursor: + return SourceCursor() + + +class _CancellingTriggers: + async def flush(self, *, limit: int) -> MemoryFlushResult: + del limit + raise asyncio.CancelledError + + async def cursor(self) -> SourceCursor: + return SourceCursor() + + async def _scope_ids() -> tuple[str, ...]: return ("scheduled",) @@ -76,15 +99,24 @@ def _runtime( *, scope_ids=_scope_ids, experience_incubator: _ScheduledExperience | None = None, + tracer: Tracer | None = None, ) -> BuiltinRuntime: return BuiltinRuntime( provider=_Provider(PowerContext(sources=object(), artifacts=object(), triggers=triggers)), # type: ignore[arg-type] capabilities=RuntimeCapabilities(memory_extraction=True, memory_search_modes=("fts",)), scope_ids=scope_ids, experience_incubator=experience_incubator, + tracer=tracer, ) +def _tracing() -> tuple[DomainTracer, InMemorySpanExporter]: + exporter = InMemorySpanExporter() + provider = TracerProvider(shutdown_on_exit=False) + provider.add_span_processor(SimpleSpanProcessor(exporter)) + return DomainTracer(ServerTracing(provider)), exporter + + def _stored_jobs(database) -> list[tuple[str, float | None]]: with sqlite3.connect(scheduler_database_path(database)) as connection: return connection.execute("SELECT id, next_run_time FROM powercontext_scheduler_jobs ORDER BY id").fetchall() @@ -226,3 +258,81 @@ async def scenario() -> None: await second.close() asyncio.run(scenario()) + + +def test_scheduled_source_window_emits_a_root_trace() -> None: + async def scenario() -> None: + tracer, exporter = _tracing() + runtime = _runtime(_ScheduledTriggers(), tracer=tracer) + assert runtime.processor is not None + await runtime.processor.run() + + spans = exporter.get_finished_spans() + root = next(span for span in spans if span.name == "scheduled.process_source_window") + flush = next(span for span in spans if span.name == "memory.flush") + assert root.attributes is not None + assert flush.attributes is not None + assert root.parent is None + assert root.attributes["powercontext.operation.outcome"] == "noop" + assert root.attributes["powercontext.operation.unit"] == "background" + assert root.attributes["source_count"] == 0 + assert flush.parent is not None + assert flush.parent.span_id == root.context.span_id + assert flush.attributes["source_count"] == 0 + assert all("scope_id" not in (span.attributes or {}) for span in spans) + + asyncio.run(scenario()) + + +def test_scheduled_experience_emits_a_root_trace() -> None: + async def scenario() -> None: + tracer, exporter = _tracing() + runtime = _runtime(_ScheduledTriggers(), experience_incubator=_ScheduledExperience(), tracer=tracer) + assert runtime.experience_processor is not None + await runtime.experience_processor.run() + + spans = exporter.get_finished_spans() + root = next(span for span in spans if span.name == "scheduled.incubate_experience_candidates") + incubation = next(span for span in spans if span.name == "experience.incubation") + assert root.attributes is not None + assert incubation.attributes is not None + assert root.parent is None + assert root.attributes["powercontext.operation.outcome"] == "success" + assert root.attributes["source_count"] == 1 + assert root.attributes["candidate_count"] == 1 + assert incubation.parent is not None + assert incubation.parent.span_id == root.context.span_id + assert incubation.attributes["candidate_count"] == 1 + assert all("scope_id" not in (span.attributes or {}) for span in spans) + + asyncio.run(scenario()) + + +def test_scheduled_source_window_records_a_failure_outcome() -> None: + async def scenario() -> None: + tracer, exporter = _tracing() + runtime = _runtime(_FailingTriggers(), tracer=tracer) + assert runtime.processor is not None + await runtime.processor.run() + + root = next(span for span in exporter.get_finished_spans() if span.name == "scheduled.process_source_window") + assert root.attributes is not None + assert root.attributes["powercontext.operation.outcome"] == "failure" + assert root.attributes["error.type"] == "RuntimeError" + + asyncio.run(scenario()) + + +def test_scheduled_source_window_records_a_cancelled_outcome() -> None: + async def scenario() -> None: + tracer, exporter = _tracing() + runtime = _runtime(_CancellingTriggers(), tracer=tracer) + assert runtime.processor is not None + with pytest.raises(asyncio.CancelledError): + await runtime.processor.run() + + root = next(span for span in exporter.get_finished_spans() if span.name == "scheduled.process_source_window") + assert root.attributes is not None + assert root.attributes["powercontext.operation.outcome"] == "cancelled" + + asyncio.run(scenario()) diff --git a/tests/e2e/test_observability.py b/tests/e2e/test_observability.py index c61836813..5eb62c9dc 100644 --- a/tests/e2e/test_observability.py +++ b/tests/e2e/test_observability.py @@ -4,6 +4,7 @@ import json import logging from pathlib import Path +from time import monotonic import httpx from fastapi.testclient import TestClient @@ -16,7 +17,7 @@ from pydantic_ai.models.test import TestModel 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 @@ -125,16 +126,19 @@ def test_inference_spans_join_the_operation_trace_only_when_instrumented(monkeyp transport = next(span for span in instrumented if span.name == "HTTP flush_memory") application = next(span for span in instrumented if span.name == "powercontext flush_memory") + memory_flush = next(span for span in instrumented if span.name == "memory.flush") invoke_agent = next(span for span in instrumented if span.name == "invoke_agent memory_extraction") chat = next(span for span in instrumented if span.name.startswith("chat ")) assert application.parent is not None assert application.parent.span_id == transport.context.span_id + assert memory_flush.parent is not None + assert memory_flush.parent.span_id == application.context.span_id assert invoke_agent.parent is not None - assert invoke_agent.parent.span_id == application.context.span_id + assert invoke_agent.parent.span_id == memory_flush.context.span_id assert chat.parent is not None assert chat.parent.span_id == invoke_agent.context.span_id - assert {span.context.trace_id for span in (transport, application, invoke_agent, chat)} == { + assert {span.context.trace_id for span in (transport, application, memory_flush, invoke_agent, chat)} == { transport.context.trace_id } assert not any(_is_inference_span(span) for span in uninstrumented) @@ -168,3 +172,69 @@ 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 test_scheduled_source_window_emits_a_root_trace(monkeypatch, tmp_path) -> None: + monkeypatch.setattr( + "pydantic_ai.models.infer_model", + lambda model: model if isinstance(model, Model) else TestModel(custom_output_text='{"candidates":[]}'), + ) + 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 / 'scheduled.db'}"), + inference=InferenceConfig(generation_model="test"), + runtime=RuntimeConfig(schedule_seconds=0.02), + mcp=McpConfig(enabled=False), + ), + scheduler_path=tmp_path / "scheduler.db", + tracing=ServerTracing(provider), + ) + scope_id = "project:scheduled-tracing" + + async def scenario() -> None: + async with ( + app.router.lifespan_context(app), + httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://testserver", + ) as transport, + ): + captured = await transport.post( + "/v1/sources/content", + json={"scope_id": scope_id, "source_id": "task-1", "content": "bounded evidence"}, + ) + assert captured.status_code == 202 + deadline = monotonic() + 3 + while monotonic() < deadline: + if any( + span.name == "scheduled.process_source_window" + and (span.attributes or {}).get("source_count") == 1 + for span in exporter.get_finished_spans() + ): + return + await asyncio.sleep(0.02) + raise AssertionError("scheduled Source-window activation did not process the captured source") # noqa: TRY003 + + asyncio.run(scenario()) + + spans = exporter.get_finished_spans() + root = next( + span + for span in spans + if span.name == "scheduled.process_source_window" and (span.attributes or {}).get("source_count") == 1 + ) + flush = next( + span + for span in spans + if span.name == "memory.flush" and span.parent is not None and span.parent.span_id == root.context.span_id + ) + assert root.attributes is not None + assert flush.attributes is not None + assert root.parent is None + assert root.attributes["powercontext.operation.outcome"] == "success" + assert root.attributes["powercontext.operation.unit"] == "background" + assert flush.attributes["source_count"] == 1 + assert all("scope_id" not in (span.attributes or {}) for span in spans)