diff --git a/docs/en/docs/how-to/trace-with-phoenix.md b/docs/en/docs/how-to/trace-with-phoenix.md
index 4abf210a6..8fd72800c 100644
--- a/docs/en/docs/how-to/trace-with-phoenix.md
+++ b/docs/en/docs/how-to/trace-with-phoenix.md
@@ -68,12 +68,13 @@ 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` | The Runtime stage that processes the Source window. Inference spans nest beneath it when extraction runs. |
 | `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. |
 
@@ -84,9 +85,11 @@ take the write lock, so they emit no `scope.lock` span:
 | --- | --- |
 | `scope.context` | Resolving the scope's context from the configured provider; near zero for the built-in provider, visible when a provider does I/O here. |
 | `scope.lock` | Waiting for the scope write lock, ending the moment it is acquired. `powercontext.scope.lock.contended` reports whether another operation already held it. |
+| `memory.flush` | One Source-window flush for `flush_memory` or a scheduled activation. |
 | `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. |
+| `experience.incubation` | One Experience incubation run for a scheduled activation. |
 | `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`,
@@ -97,6 +100,23 @@ Spans are exported in batches, so allow a few seconds before refreshing. An MCP
 `MCP mcp.tools.call` in place of the `HTTP` span. Readiness probes are deliberately not traced, so health checks do not
 create single-span traces.
 
+## Scheduled background spans
+
+When a scheduler interval is configured (`schedule_seconds` or `experience_schedule_seconds`), each scheduled activation
+starts its own trace instead of joining an unrelated request trace. The activation is the root span with
+`powercontext.operation.unit` set to `background`:
+
+| Span | Meaning |
+| --- | --- |
+| `scheduled.process_source_window` | One scheduled Source-window activation. Its outcome is `success`, `noop`, `failure`, or `cancelled`. |
+| `scheduled.incubate_experience_candidates` | One scheduled Experience incubation activation, with the same outcome vocabulary. |
+| `memory.flush` | The flush run beneath a Source-window activation; it also appears under `HTTP flush_memory`. |
+| `experience.incubation` | The incubation run beneath an Experience activation. |
+
+Scheduled roots record only bounded counts — `powercontext.background.source_count` and
+`powercontext.background.candidate_count` — and never a `scope_id`, request ID, or Memory content. Inference spans created
+by a scheduled activation are nested beneath its root in the same trace.
+
 ## What is not exported
 
 PowerContext configures inference instrumentation to exclude content. Spans carry model identifiers, token usage,
diff --git a/docs/en/rfcs/0046_observability_foundations.md b/docs/en/rfcs/0046_observability_foundations.md
index 003505f4e..6d6354cb6 100644
--- a/docs/en/rfcs/0046_observability_foundations.md
+++ b/docs/en/rfcs/0046_observability_foundations.md
@@ -87,6 +87,9 @@ PowerContext distinguishes these units:
 Runtime stage spans use `stage` as their `powercontext.operation.unit` value. They expose internal latency without
 creating another application operation.
 
+Background activations use `background` as their `powercontext.operation.unit` value. Scheduled activations start their
+own trace rather than inheriting an inbound request context.
+
 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 245189a61..bca419641 100644
--- a/docs/zh/docs/how-to/trace-with-phoenix.md
+++ b/docs/zh/docs/how-to/trace-with-phoenix.md
@@ -66,12 +66,13 @@ 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 的 Runtime stage。extraction 跑起来时，推理 span 嵌套在它下面。 |
 | `invoke_agent memory_extraction` | 一次 PowerContext generation 任务。名字标识用途，不是模型名。 |
 | `chat <model>` | 一次发往模型 provider 的请求，包含 token 用量和耗时。 |
 
@@ -82,9 +83,11 @@ scope 相关操作还会在 application operation 之下添加以下内部 stage
 | --- | --- |
 | `scope.context` | 从配置的 provider 解析该 scope 的 context；内建 provider 下接近零，provider 在此做 I/O 时才可见。 |
 | `scope.lock` | 等待该 scope 的写锁，在获取到锁的瞬间结束。`powercontext.scope.lock.contended` 表示进入时是否已被其他操作持有。 |
+| `memory.flush` | 一次 Source-window flush，出现在 `flush_memory` 或定时激活之下。 |
 | `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 时也会产生。 |
+| `experience.incubation` | 一次 Experience incubation，出现在定时激活之下。 |
 | `context.build` | 根据召回候选同步选择并渲染最终 prepared context 的步骤。 |
 
 其他 generation 任务遵循同样的命名约定：`experience_incubation`、`experience_generation`、`skill_generation`、
@@ -94,6 +97,22 @@ span 挂在触发它的操作之下。
 span 是批量导出的，刷新前请稍等几秒。MCP 请求会用 `MCP mcp.tools.call` 取代 `HTTP` span。readiness 探活被有意
 排除在 trace 之外，因此健康检查不会产生只含单个 span 的 trace。
 
+## 定时后台 span
+
+配置了定时器间隔（`schedule_seconds` 或 `experience_schedule_seconds`）后，每次定时激活都会开启一条**独立**的
+trace，而不会并入无关的请求 trace。激活本身是根 span，其 `powercontext.operation.unit` 为 `background`：
+
+| Span | 含义 |
+| --- | --- |
+| `scheduled.process_source_window` | 一次定时 Source-window 激活，outcome 为 `success`、`noop`、`failure` 或 `cancelled`。 |
+| `scheduled.incubate_experience_candidates` | 一次定时 Experience incubation 激活，outcome 词表相同。 |
+| `memory.flush` | Source-window 激活下的 flush 执行；它也会出现在 `HTTP flush_memory` 之下。 |
+| `experience.incubation` | Experience 激活下的 incubation 执行。 |
+
+定时根 span 只记录有界的计数——`powercontext.background.source_count` 与
+`powercontext.background.candidate_count`——绝不携带 `scope_id`、request ID 或 Memory 内容。定时激活产生的推理
+span 会作为其根 span 的子节点嵌套在同一条 trace 中。
+
 ## 哪些内容不会被导出
 
 PowerContext 在配置推理 instrumentation 时关闭了内容记录。span 只携带模型标识、token 用量、耗时和错误类别；
diff --git a/docs/zh/rfcs/0046_observability_foundations.md b/docs/zh/rfcs/0046_observability_foundations.md
index d626cbddb..5f1f15712 100644
--- a/docs/zh/rfcs/0046_observability_foundations.md
+++ b/docs/zh/rfcs/0046_observability_foundations.md
@@ -84,6 +84,9 @@ PowerContext 区分以下工作单元：
 Runtime stage span 的 `powercontext.operation.unit` 值为 `stage`。它们用于展示内部耗时，但不会产生新的
 application operation。
 
+Background activation 的 `powercontext.operation.unit` 值为 `background`。定时 activation 会开启独立 trace，而不是
+继承入站请求上下文。
+
 直接 HTTP call 会产生一次 external request 和一次 application operation。MCP tool call 同样产生一次
 external request 和一次 application operation。其内部 HTTP bridge 不计为第二次 external request。
 
diff --git a/src/powercontext/builtin/runtime/application.py b/src/powercontext/builtin/runtime/application.py
index 94eb7e8ee..750b7d20e 100644
--- a/src/powercontext/builtin/runtime/application.py
+++ b/src/powercontext/builtin/runtime/application.py
@@ -492,7 +492,15 @@ async def incubate(self, /, *, limit: int | None = None) -> ExperienceIncubation
             ),
             self._runtime._locked(self.scope_id),
         ):
-            return await incubator(self.scope_id, window_limit)
+            with self._runtime._stage("experience.incubation", attributes={}) as span:
+                result = await incubator(self.scope_id, window_limit)
+                if span is not None:
+                    span.set_attributes({
+                        "powercontext.experience.incubation.source_count": result.source_count,
+                        "powercontext.experience.incubation.candidate_count": result.candidate_count,
+                    })
+                    span.set_outcome("success" if result.processed else "noop")
+                return result
 
 
 class ExperienceApplication:
@@ -1042,7 +1050,12 @@ 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._locked(self.scope_id):
-                return await context.triggers.flush(limit=window_limit)
+                with self._runtime._stage("memory.flush", attributes={}) as span:
+                    result = await context.triggers.flush(limit=window_limit)
+                    if span is not None:
+                        span.set_attributes({"powercontext.memory.flush.source_count": result.source_count})
+                        span.set_outcome("success" if result.processed else "noop")
+                    return result
 
     async def cursor(self) -> SourceCursor:
         async with self._runtime._context(self.scope_id) as context:
@@ -1074,29 +1087,39 @@ async def run(self) -> None:
                 if self._runtime._closing or self._runtime._closed:
                     return
                 started_at = perf_counter()
-                try:
-                    result = await self._runtime.memory.for_scope(scope_id).flush()
-                except asyncio.CancelledError:
-                    _log_scheduled_processing(
-                        "cancelled",
-                        operation="process_source_window",
-                        started_at=started_at,
-                    )
-                    raise
-                except Exception as error:
-                    _log_scheduled_processing(
-                        "failure",
-                        operation="process_source_window",
-                        started_at=started_at,
-                        error=error,
-                    )
-                else:
-                    _log_scheduled_processing(
-                        "success" if result.processed else "noop",
-                        operation="process_source_window",
-                        started_at=started_at,
-                        source_count=result.source_count,
-                    )
+                with self._runtime._background(
+                    "scheduled.process_source_window",
+                    operation="process_source_window",
+                ) as span:
+                    try:
+                        result = await self._runtime.memory.for_scope(scope_id).flush()
+                    except asyncio.CancelledError:
+                        _log_scheduled_processing(
+                            "cancelled",
+                            operation="process_source_window",
+                            started_at=started_at,
+                        )
+                        raise
+                    except Exception as error:
+                        _log_scheduled_processing(
+                            "failure",
+                            operation="process_source_window",
+                            started_at=started_at,
+                            error=error,
+                        )
+                        if span is not None:
+                            span.set_outcome("failure")
+                    else:
+                        outcome = "success" if result.processed else "noop"
+                        _log_scheduled_processing(
+                            outcome,
+                            operation="process_source_window",
+                            started_at=started_at,
+                            source_count=result.source_count,
+                        )
+                        if span is not None:
+                            span.set_outcome(outcome)
+                            span.set_attributes({"powercontext.background.source_count": result.source_count})
 
 
 class ScheduledExperienceProcessor:
@@ -1114,30 +1137,43 @@ async def run(self) -> None:
                 if self._runtime._closing or self._runtime._closed:
                     return
                 started_at = perf_counter()
-                try:
-                    result = await self._runtime.experience.for_scope(scope_id).incubate()
-                except asyncio.CancelledError:
-                    _log_scheduled_processing(
-                        "cancelled",
-                        operation="incubate_experience_candidates",
-                        started_at=started_at,
-                    )
-                    raise
-                except Exception as error:
-                    _log_scheduled_processing(
-                        "failure",
-                        operation="incubate_experience_candidates",
-                        started_at=started_at,
-                        error=error,
-                    )
-                else:
-                    _log_scheduled_processing(
-                        "success" if result.processed else "noop",
-                        operation="incubate_experience_candidates",
-                        started_at=started_at,
-                        source_count=result.source_count,
-                        candidate_count=result.candidate_count,
-                    )
+                with self._runtime._background(
+                    "scheduled.incubate_experience_candidates",
+                    operation="incubate_experience_candidates",
+                ) as span:
+                    try:
+                        result = await self._runtime.experience.for_scope(scope_id).incubate()
+                    except asyncio.CancelledError:
+                        _log_scheduled_processing(
+                            "cancelled",
+                            operation="incubate_experience_candidates",
+                            started_at=started_at,
+                        )
+                        raise
+                    except Exception as error:
+                        _log_scheduled_processing(
+                            "failure",
+                            operation="incubate_experience_candidates",
+                            started_at=started_at,
+                            error=error,
+                        )
+                        if span is not None:
+                            span.set_outcome("failure")
+                    else:
+                        outcome = "success" if result.processed else "noop"
+                        _log_scheduled_processing(
+                            outcome,
+                            operation="incubate_experience_candidates",
+                            started_at=started_at,
+                            source_count=result.source_count,
+                            candidate_count=result.candidate_count,
+                        )
+                        if span is not None:
+                            span.set_outcome(outcome)
+                            span.set_attributes({
+                                "powercontext.background.source_count": result.source_count,
+                                "powercontext.background.candidate_count": result.candidate_count,
+                            })
 
 
 def _log_scheduled_processing(
@@ -1451,6 +1487,16 @@ def _stage(
             return nullcontext(None)
         return self._tracing.stage(name, attributes=attributes)
 
+    def _background(
+        self,
+        name: str,
+        *,
+        operation: str,
+    ) -> AbstractContextManager[RuntimeSpan | None]:
+        if self._tracing is None:
+            return nullcontext(None)
+        return self._tracing.background(name, operation=operation, attributes={})
+
     def _review(self, scope_id: str) -> ReviewService:
         if self._review_service is None:
             raise _RuntimeStateError("review")
diff --git a/src/powercontext/builtin/runtime/protocols.py b/src/powercontext/builtin/runtime/protocols.py
index af3a66e3b..480b019a1 100644
--- a/src/powercontext/builtin/runtime/protocols.py
+++ b/src/powercontext/builtin/runtime/protocols.py
@@ -32,10 +32,12 @@
 
 
 class RuntimeSpan(Protocol):
-    """Record bounded attributes for one internal Runtime stage."""
+    """Record bounded attributes and a deferred outcome for one internal Runtime stage."""
 
     def set_attributes(self, attributes: Mapping[str, TraceAttribute], /) -> None: ...
 
+    def set_outcome(self, outcome: str, /) -> None: ...
+
 
 class RuntimeTracing(Protocol):
     """Create framework-neutral spans for internal Runtime stages."""
@@ -47,6 +49,14 @@ def stage(
         attributes: Mapping[str, TraceAttribute],
     ) -> AbstractContextManager[RuntimeSpan]: ...
 
+    def background(
+        self,
+        name: str,
+        *,
+        operation: str,
+        attributes: Mapping[str, TraceAttribute],
+    ) -> AbstractContextManager[RuntimeSpan]: ...
+
 
 class PowerContextProvider(Protocol[SourcesT, ArtifactsT, TriggersT]):
     """Resolve an already composed context without transferring lifecycle ownership."""
diff --git a/src/powercontext/server/tracing.py b/src/powercontext/server/tracing.py
index b903c8732..2521a3058 100644
--- a/src/powercontext/server/tracing.py
+++ b/src/powercontext/server/tracing.py
@@ -138,7 +138,47 @@ def stage(
         except BaseException as error:
             span.finish("failure", error=error)
             raise
-        span.finish("success")
+        span.finish(span.outcome or "success")
+
+    @contextmanager
+    def background(
+        self,
+        name: str,
+        *,
+        operation: str,
+        attributes: Mapping[str, _TraceAttribute],
+    ) -> Iterator[_ActiveSpan]:
+        """Trace one scheduled activation as an independent trace root."""
+
+        isolation_token: Token[Context] | None = None
+        with suppress(Exception):
+            # Keep an empty context attached even if the root span fails to start, so
+            # child stages cannot join an ambient HTTP/MCP trace.
+            isolation_token = otel_context.attach(Context())
+        try:
+            span = self.start_span(
+                name,
+                kind=SpanKind.INTERNAL,
+                context=Context(),
+                attributes={
+                    **attributes,
+                    "powercontext.operation.name": operation,
+                    "powercontext.operation.unit": "background",
+                },
+            )
+            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(span.outcome or "success")
+        finally:
+            if isolation_token is not None:
+                with suppress(Exception):
+                    otel_context.detach(isolation_token)
 
     @contextmanager
     def _suppress_readiness_spans(self) -> Iterator[None]:
@@ -178,6 +218,14 @@ def __init__(self, span: Span | None, token: Token[Context] | None) -> None:
         self.span = span
         self.token = token
         self.finished = False
+        self._outcome: str | None = None
+
+    @property
+    def outcome(self) -> str | None:
+        return self._outcome
+
+    def set_outcome(self, outcome: str) -> None:
+        self._outcome = outcome
 
     @classmethod
     def start(
diff --git a/tests/builtin/runtime/test_scheduler.py b/tests/builtin/runtime/test_scheduler.py
index 855e8c8f8..dd94a4635 100644
--- a/tests/builtin/runtime/test_scheduler.py
+++ b/tests/builtin/runtime/test_scheduler.py
@@ -15,11 +15,15 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import logging
 import sqlite3
 from typing import Any
 
 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 (
@@ -36,6 +40,7 @@
     scheduler_database_path,
 )
 from powercontext.builtin.sources import SourceCursor
+from powercontext.server.tracing import ServerTracing
 
 
 class _Provider:
@@ -48,17 +53,18 @@ async def get(self, scope_id: str, /) -> PowerContext[Any, Any, Any]:
 
 
 class _ScheduledTriggers:
-    def __init__(self) -> None:
+    def __init__(self, *, source_count: int = 0) -> None:
         self.dispatched = asyncio.Event()
+        self.source_count = source_count
 
     async def flush(self, *, limit: int) -> MemoryFlushResult:
         del limit
         self.dispatched.set()
         return MemoryFlushResult(
             previous_cursor=0,
-            high_watermark=0,
-            current_cursor=0,
-            source_count=0,
+            high_watermark=self.source_count,
+            current_cursor=self.source_count,
+            source_count=self.source_count,
             memory_ref=None,
         )
 
@@ -66,6 +72,29 @@ async def cursor(self) -> SourceCursor:
         return SourceCursor()
 
 
+class _FailingTriggers:
+    async def flush(self, *, limit: int) -> MemoryFlushResult:
+        del limit
+        raise RuntimeError("flush failed")  # noqa: TRY003
+
+    async def cursor(self) -> SourceCursor:
+        return SourceCursor()
+
+
+class _BlockingTriggers:
+    def __init__(self) -> None:
+        self.entered = asyncio.Event()
+
+    async def flush(self, *, limit: int) -> MemoryFlushResult:
+        del limit
+        self.entered.set()
+        await asyncio.Event().wait()
+        raise AssertionError("unreachable")
+
+    async def cursor(self) -> SourceCursor:
+        return SourceCursor()
+
+
 class _ScheduledExperience:
     def __init__(self) -> None:
         self.dispatched = asyncio.Event()
@@ -82,6 +111,35 @@ async def __call__(self, scope_id: str, limit: int) -> ExperienceIncubationResul
         )
 
 
+class _NoopExperience:
+    async def __call__(self, scope_id: str, limit: int) -> ExperienceIncubationResult:
+        del scope_id, limit
+        return ExperienceIncubationResult(
+            previous_cursor=0,
+            high_watermark=0,
+            current_cursor=0,
+            source_count=0,
+            candidate_count=0,
+        )
+
+
+class _FailingExperience:
+    async def __call__(self, scope_id: str, limit: int) -> ExperienceIncubationResult:
+        del scope_id, limit
+        raise RuntimeError("incubation failed")  # noqa: TRY003
+
+
+class _BlockingExperience:
+    def __init__(self) -> None:
+        self.entered = asyncio.Event()
+
+    async def __call__(self, scope_id: str, limit: int) -> ExperienceIncubationResult:
+        del scope_id, limit
+        self.entered.set()
+        await asyncio.Event().wait()
+        raise AssertionError("unreachable")
+
+
 async def _scope_ids() -> tuple[str, ...]:
     return ("scheduled",)
 
@@ -90,13 +148,31 @@ def _runtime(
     triggers: object,
     *,
     scope_ids=_scope_ids,
-    experience_incubator: _ScheduledExperience | None = None,
+    experience_incubator: (
+        _ScheduledExperience | _NoopExperience | _FailingExperience | _BlockingExperience | None
+    ) = None,
+    tracing: ServerTracing | 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,
+        tracing=tracing,
+    )
+
+
+def _tracing() -> tuple[ServerTracing, InMemorySpanExporter]:
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider(shutdown_on_exit=False)
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    return ServerTracing(provider), exporter
+
+
+def _scope_id_leak(spans) -> str:
+    return json.dumps(
+        [{"name": span.name, "attributes": dict(span.attributes or {})} for span in spans],
+        default=str,
     )
 
 
@@ -222,6 +298,183 @@ async def scenario() -> None:
     assert "scope_id" not in vars(record)
 
 
+async def _private_scope_ids() -> tuple[str, ...]:
+    return ("project:private-scheduled-scope",)
+
+
+def test_scheduled_processor_records_root_and_flush_spans() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        runtime = _runtime(_ScheduledTriggers(), tracing=tracing, scope_ids=_private_scope_ids)
+        assert runtime.processor is not None
+        await runtime.processor.run()
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.process_source_window"]
+    flush = spans["memory.flush"]
+    assert root.parent is None
+    assert flush.parent is not None and flush.parent.span_id == root.context.span_id
+    assert root.attributes is not None
+    assert root.attributes["powercontext.operation.name"] == "process_source_window"
+    assert root.attributes["powercontext.operation.unit"] == "background"
+    assert root.attributes["powercontext.operation.outcome"] == "noop"
+    assert root.attributes["powercontext.background.source_count"] == 0
+    assert flush.attributes is not None
+    assert flush.attributes["powercontext.operation.unit"] == "stage"
+    assert flush.attributes["powercontext.operation.outcome"] == "noop"
+    assert flush.attributes["powercontext.memory.flush.source_count"] == 0
+    assert "project:private-scheduled-scope" not in _scope_id_leak(exporter.get_finished_spans())
+
+
+def test_scheduled_processor_records_success_outcome() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        runtime = _runtime(_ScheduledTriggers(source_count=3), tracing=tracing)
+        assert runtime.processor is not None
+        await runtime.processor.run()
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.process_source_window"]
+    flush = spans["memory.flush"]
+    assert root.attributes is not None and root.attributes["powercontext.operation.outcome"] == "success"
+    assert root.attributes["powercontext.background.source_count"] == 3
+    assert flush.attributes is not None and flush.attributes["powercontext.operation.outcome"] == "success"
+    assert flush.attributes["powercontext.memory.flush.source_count"] == 3
+
+
+def test_scheduled_processor_records_failure_and_swallows_error() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        runtime = _runtime(_FailingTriggers(), tracing=tracing)
+        assert runtime.processor is not None
+        await runtime.processor.run()
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.process_source_window"]
+    flush = spans["memory.flush"]
+    assert root.attributes is not None and root.attributes["powercontext.operation.outcome"] == "failure"
+    assert flush.attributes is not None and flush.attributes["powercontext.operation.outcome"] == "failure"
+
+
+def test_scheduled_processor_records_cancellation() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        triggers = _BlockingTriggers()
+        runtime = _runtime(triggers, tracing=tracing)
+        assert runtime.processor is not None
+        task = asyncio.create_task(runtime.processor.run())
+        await triggers.entered.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.process_source_window"]
+    assert root.attributes is not None and root.attributes["powercontext.operation.outcome"] == "cancelled"
+
+
+def test_scheduled_experience_records_root_and_incubation_spans() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        runtime = _runtime(
+            _ScheduledTriggers(),
+            experience_incubator=_ScheduledExperience(),
+            tracing=tracing,
+            scope_ids=_private_scope_ids,
+        )
+        assert runtime.experience_processor is not None
+        await runtime.experience_processor.run()
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.incubate_experience_candidates"]
+    incubation = spans["experience.incubation"]
+    assert root.parent is None
+    assert incubation.parent is not None and incubation.parent.span_id == root.context.span_id
+    assert root.attributes is not None
+    assert root.attributes["powercontext.operation.name"] == "incubate_experience_candidates"
+    assert root.attributes["powercontext.operation.unit"] == "background"
+    assert root.attributes["powercontext.operation.outcome"] == "success"
+    assert root.attributes["powercontext.background.source_count"] == 1
+    assert root.attributes["powercontext.background.candidate_count"] == 1
+    assert incubation.attributes is not None
+    assert incubation.attributes["powercontext.experience.incubation.source_count"] == 1
+    assert incubation.attributes["powercontext.experience.incubation.candidate_count"] == 1
+    assert "project:private-scheduled-scope" not in _scope_id_leak(exporter.get_finished_spans())
+
+
+def test_scheduled_experience_records_noop_outcome() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        runtime = _runtime(_ScheduledTriggers(), experience_incubator=_NoopExperience(), tracing=tracing)
+        assert runtime.experience_processor is not None
+        await runtime.experience_processor.run()
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.incubate_experience_candidates"]
+    incubation = spans["experience.incubation"]
+    assert root.attributes is not None and root.attributes["powercontext.operation.outcome"] == "noop"
+    assert root.attributes["powercontext.background.source_count"] == 0
+    assert root.attributes["powercontext.background.candidate_count"] == 0
+    assert incubation.attributes is not None and incubation.attributes["powercontext.operation.outcome"] == "noop"
+    assert incubation.attributes["powercontext.experience.incubation.source_count"] == 0
+    assert incubation.attributes["powercontext.experience.incubation.candidate_count"] == 0
+
+
+def test_scheduled_experience_records_failure_and_swallows_error() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        runtime = _runtime(_ScheduledTriggers(), experience_incubator=_FailingExperience(), tracing=tracing)
+        assert runtime.experience_processor is not None
+        await runtime.experience_processor.run()
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.incubate_experience_candidates"]
+    incubation = spans["experience.incubation"]
+    assert root.attributes is not None and root.attributes["powercontext.operation.outcome"] == "failure"
+    assert incubation.attributes is not None and incubation.attributes["powercontext.operation.outcome"] == "failure"
+
+
+def test_scheduled_experience_records_cancellation() -> None:
+    tracing, exporter = _tracing()
+
+    async def scenario() -> None:
+        incubator = _BlockingExperience()
+        runtime = _runtime(_ScheduledTriggers(), experience_incubator=incubator, tracing=tracing)
+        assert runtime.experience_processor is not None
+        task = asyncio.create_task(runtime.experience_processor.run())
+        await incubator.entered.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    asyncio.run(scenario())
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    root = spans["scheduled.incubate_experience_candidates"]
+    assert root.attributes is not None and root.attributes["powercontext.operation.outcome"] == "cancelled"
+
+
 def test_scheduler_requires_file_storage_and_one_live_owner(tmp_path) -> None:
     async def scenario() -> None:
         with pytest.raises(SchedulerConfigurationError):
diff --git a/tests/e2e/test_observability.py b/tests/e2e/test_observability.py
index c947465d1..cc667dc5f 100644
--- a/tests/e2e/test_observability.py
+++ b/tests/e2e/test_observability.py
@@ -19,6 +19,7 @@
 import logging
 import sqlite3
 from pathlib import Path
+from time import monotonic, sleep
 
 import httpx
 import pytest
@@ -39,6 +40,7 @@
 from powercontext.artifacts import ArtifactRef
 from powercontext.builtin.artifacts.memory import (
     EmbeddingProfile,
+    MemoryCandidateRequest,
     MemoryCapabilities,
     MemoryEntryInput,
     MemoryProjection,
@@ -105,6 +107,32 @@
         "powercontext.context.build.status",
         "powercontext.context.build.content_bytes",
     },
+    "memory.flush": {
+        "powercontext.operation.name",
+        "powercontext.operation.unit",
+        "powercontext.operation.outcome",
+        "powercontext.memory.flush.source_count",
+    },
+    "experience.incubation": {
+        "powercontext.operation.name",
+        "powercontext.operation.unit",
+        "powercontext.operation.outcome",
+        "powercontext.experience.incubation.source_count",
+        "powercontext.experience.incubation.candidate_count",
+    },
+    "scheduled.process_source_window": {
+        "powercontext.operation.name",
+        "powercontext.operation.unit",
+        "powercontext.operation.outcome",
+        "powercontext.background.source_count",
+    },
+    "scheduled.incubate_experience_candidates": {
+        "powercontext.operation.name",
+        "powercontext.operation.unit",
+        "powercontext.operation.outcome",
+        "powercontext.background.source_count",
+        "powercontext.background.candidate_count",
+    },
 }
 
 _VECTOR_PROFILE = EmbeddingProfile(
@@ -131,6 +159,9 @@ def __enter__(self) -> _StageTeardown:
     def set_attributes(self, _attributes: object, /) -> None:
         pass
 
+    def set_outcome(self, _outcome: str, /) -> None:
+        pass
+
     def __exit__(self, *_: object) -> None:
         if self._failing:
             raise _StageTeardownError
@@ -142,6 +173,9 @@ class _ScopeLockTeardownFailingTracing:
     def stage(self, name: str, **_: object) -> _StageTeardown:
         return _StageTeardown(failing=name == "scope.lock")
 
+    def background(self, name: str, **_: object) -> _StageTeardown:
+        return _StageTeardown(failing=name == "scope.lock")
+
 
 class _VectorMemoryIndex:
     """Expose deterministic vector capability without a platform extension."""
@@ -342,16 +376,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")
+    flush_stage = 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 flush_stage.parent is not None
+    assert flush_stage.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 == flush_stage.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, flush_stage, invoke_agent, chat)} == {
         transport.context.trace_id
     }
     assert not any(_is_inference_span(span) for span in uninstrumented)
@@ -638,6 +675,86 @@ async def scenario() -> bool:
     assert asyncio.run(scenario()) is False
 
 
+class _EmptyCandidatePipeline:
+    """Produce no Memory candidates so a scheduled flush advances the cursor without a model."""
+
+    async def extract(self, request: MemoryCandidateRequest, /) -> tuple[MemoryEntryInput, ...]:
+        del request
+        return ()
+
+
+class _EmptyExperiencePipeline:
+    """Produce no Experience candidates so a scheduled incubation advances the cursor without a model."""
+
+    async def incubate(self, sources: tuple[object, ...], /) -> tuple[()]:
+        del sources
+        return ()
+
+
+def test_scheduled_source_window_starts_an_independent_trace_root(tmp_path) -> None:
+    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-tracing.db'}"),
+            runtime=RuntimeConfig(schedule_seconds=0.02),
+            mcp=McpConfig(enabled=False),
+        ),
+        scheduler_path=tmp_path / "scheduler.db",
+        candidate_pipeline=_EmptyCandidatePipeline(),
+        tracing=ServerTracing(provider),
+    )
+    scope_id = "project:private-scheduled-trace"
+    content = "private scheduled evidence"
+
+    with TestClient(app) as client:
+        captured = client.post(
+            "/v1/sources/content",
+            json={"scope_id": scope_id, "source_id": "task-1", "content": content},
+        )
+        assert captured.status_code == 202
+        root = _wait_for_named_span(exporter, "scheduled.process_source_window", outcome="success")
+        spans = list(exporter.get_finished_spans())
+        flush = _only_child(spans, root, "memory.flush")
+        _assert_scheduled_background_trace(root, flush, spans, scope_id=scope_id, content=content)
+        assert (root.attributes or {})["powercontext.background.source_count"] == 1
+        assert (flush.attributes or {})["powercontext.memory.flush.source_count"] == 1
+
+
+def test_scheduled_experience_starts_an_independent_trace_root(tmp_path) -> None:
+    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-experience-tracing.db'}"),
+            runtime=RuntimeConfig(experience_schedule_seconds=0.02),
+            mcp=McpConfig(enabled=False),
+        ),
+        scheduler_path=tmp_path / "scheduler.db",
+        experience_pipeline=_EmptyExperiencePipeline(),
+        tracing=ServerTracing(provider),
+    )
+    scope_id = "project:private-scheduled-experience"
+    content = "private scheduled incubation evidence"
+
+    with TestClient(app) as client:
+        captured = client.post(
+            "/v1/sources/content",
+            json={"scope_id": scope_id, "source_id": "task-1", "content": content},
+        )
+        assert captured.status_code == 202
+        root = _wait_for_named_span(exporter, "scheduled.incubate_experience_candidates", outcome="success")
+        spans = list(exporter.get_finished_spans())
+        incubation = _only_child(spans, root, "experience.incubation")
+        _assert_scheduled_background_trace(root, incubation, spans, scope_id=scope_id, content=content)
+        assert (root.attributes or {})["powercontext.background.source_count"] == 1
+        assert (root.attributes or {})["powercontext.background.candidate_count"] == 0
+        assert (incubation.attributes or {})["powercontext.experience.incubation.source_count"] == 1
+        assert (incubation.attributes or {})["powercontext.experience.incubation.candidate_count"] == 0
+
+
 def test_vector_search_exports_embedding_under_memory_search_without_recording_text(monkeypatch, tmp_path) -> None:
     monkeypatch.setattr(
         "pydantic_ai.embeddings.infer_embedding_model",
@@ -802,6 +919,51 @@ def _children(spans: list[ReadableSpan], parent: ReadableSpan, name: str) -> lis
     ]
 
 
+def _wait_for_named_span(
+    exporter: InMemorySpanExporter,
+    name: str,
+    *,
+    outcome: str,
+    timeout: float = 3,
+) -> ReadableSpan:
+    deadline = monotonic() + timeout
+    while monotonic() < deadline:
+        for span in exporter.get_finished_spans():
+            if span.name == name and (span.attributes or {}).get("powercontext.operation.outcome") == outcome:
+                return span
+        sleep(0.02)
+    raise AssertionError(f"{name} span was not exported")  # noqa: TRY003
+
+
+def _assert_stage_attribute_keys(span: ReadableSpan) -> None:
+    allowed_keys = _STAGE_ATTRIBUTE_KEYS[span.name]
+    attributes = dict(span.attributes or {})
+    assert attributes.keys() <= allowed_keys
+    assert all(isinstance(value, str | bool | int | float) for value in attributes.values())
+
+
+def _assert_scheduled_background_trace(
+    root: ReadableSpan,
+    stage: ReadableSpan,
+    spans: list[ReadableSpan],
+    *,
+    scope_id: str,
+    content: str,
+) -> None:
+    assert root.parent is None
+    assert stage.parent is not None and stage.parent.span_id == root.context.span_id
+    assert (root.attributes or {})["powercontext.operation.unit"] == "background"
+    assert (root.attributes or {})["powercontext.operation.outcome"] == "success"
+    assert (stage.attributes or {})["powercontext.operation.unit"] == "stage"
+    _assert_stage_attribute_keys(root)
+    _assert_stage_attribute_keys(stage)
+    http_trace_ids = {span.context.trace_id for span in spans if span.name.startswith("HTTP ")}
+    assert root.context.trace_id not in http_trace_ids
+    exported = _exported_span_data(spans)
+    assert scope_id not in exported
+    assert content not in exported
+
+
 def _only_child(spans: list[ReadableSpan], parent: ReadableSpan, name: str) -> ReadableSpan:
     children = _children(spans, parent, name)
     assert len(children) == 1
diff --git a/tests/test_server_tracing.py b/tests/test_server_tracing.py
index b2e7ee6d2..04f477519 100644
--- a/tests/test_server_tracing.py
+++ b/tests/test_server_tracing.py
@@ -16,6 +16,7 @@
 
 import asyncio
 import sys
+from typing import Any
 
 import httpx
 import pytest
@@ -230,6 +231,83 @@ def start_span(self, *_args: object, **_kwargs: object) -> None:
     assert exporter.get_finished_spans() == ()
 
 
+def test_runtime_stage_records_explicit_noop_and_failure_outcomes() -> None:
+    tracing, exporter = _tracing()
+
+    with tracing.stage("memory.flush", attributes={}) as noop:
+        noop.set_outcome("noop")
+    with tracing.stage("memory.flush", attributes={}) as failed:
+        failed.set_outcome("failure")
+
+    spans = [span for span in exporter.get_finished_spans() if span.name == "memory.flush"]
+    assert len(spans) == 2
+    outcomes = {span.attributes["powercontext.operation.outcome"] for span in spans if span.attributes is not None}
+    assert outcomes == {"noop", "failure"}
+    failed_span = next(
+        span
+        for span in spans
+        if span.attributes is not None and span.attributes["powercontext.operation.outcome"] == "failure"
+    )
+    assert failed_span.status.status_code is StatusCode.ERROR
+
+
+def test_background_stage_starts_a_fresh_trace_outside_an_ambient_span() -> None:
+    tracing, exporter = _tracing()
+    ambient = tracing.start_span("HTTP flush_memory", kind=SpanKind.SERVER, attributes={})
+
+    with tracing.background(
+        "scheduled.process_source_window",
+        operation="process_source_window",
+        attributes={},
+    ) as stage:
+        stage.set_outcome("noop")
+        stage.set_attributes({"powercontext.background.source_count": 0})
+    ambient.finish("success")
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    ambient_span = spans["HTTP flush_memory"]
+    root = spans["scheduled.process_source_window"]
+    assert root.parent is None
+    assert root.context.trace_id != ambient_span.context.trace_id
+    assert root.kind is SpanKind.INTERNAL
+    attributes = root.attributes or {}
+    assert attributes["powercontext.operation.name"] == "process_source_window"
+    assert attributes["powercontext.operation.unit"] == "background"
+    assert attributes["powercontext.operation.outcome"] == "noop"
+    assert attributes["powercontext.background.source_count"] == 0
+
+
+def test_background_isolates_child_spans_when_root_start_fails(monkeypatch) -> None:
+    tracing, exporter = _tracing()
+    ambient = tracing.start_span("HTTP flush_memory", kind=SpanKind.SERVER, attributes={})
+    original = tracing.tracer.start_span
+
+    def fail_scheduled_root(name: str, **kwargs: Any) -> Any:
+        if name == "scheduled.process_source_window":
+            raise RuntimeError
+        return original(name, **kwargs)
+
+    monkeypatch.setattr(tracing.tracer, "start_span", fail_scheduled_root)
+
+    with (
+        tracing.background(
+            "scheduled.process_source_window",
+            operation="process_source_window",
+            attributes={},
+        ),
+        tracing.stage("memory.flush", attributes={}) as stage,
+    ):
+        stage.set_outcome("noop")
+    ambient.finish("success")
+
+    spans = {span.name: span for span in exporter.get_finished_spans()}
+    assert "scheduled.process_source_window" not in spans
+    ambient_span = spans["HTTP flush_memory"]
+    flush = spans["memory.flush"]
+    assert flush.parent is None
+    assert flush.context.trace_id != ambient_span.context.trace_id
+
+
 def test_readiness_ignores_tracing_setup_failure(monkeypatch, tmp_path) -> None:
     tracing, _ = _tracing(instrumented=True)
     app = create_server_app(