feat(tracing): trace scheduled Runtime processing (#1243)

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 <http://localhost:6006>, 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 <model>` | 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 <model>` spans under the operation that triggered them.

docs/zh/docs/how-to/trace-with-phoenix.md

@@ -66,15 +66,19 @@ Memory extraction 发生在 flush 阶段，而不是捕获阶段。
 ## 查看 trace
 
 打开 <http://localhost:6006>，选择 `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 <model>` | 一次发往模型 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 <model>`
 span 挂在触发它的操作之下。

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()

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:

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

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",

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"]