scheduler: Spawn inherits WithObserve boundaries via GetBoundaries(k) (ADR-DOE-CORE-001)

.semgrep.yaml

@@ -4087,3 +4087,18 @@ rules:
       include:
         - /packages/doeff-conductor/src/**
         - /packages/doeff-agents/src/**
+
+  - id: doeff-scheduler-spawn-must-capture-boundaries
+    languages: [generic]
+    severity: ERROR
+    message: >
+      Issue scheduler-spawn-drops-observer-boundary / ADR-DOE-CORE-001: the
+      scheduler must capture the FULL boundary stack at the spawn site via
+      get_inner_boundaries (handlers AND WithObserve observers, nesting order
+      preserved). get_inner_handlers drops observer boundaries, silently
+      breaking subtree observation (e.g. OpenTelemetry effect->span
+      instrumentation) inside Spawn'd tasks.
+    pattern-regex: 'get_inner_handlers'
+    paths:
+      include:
+        - /packages/doeff-core-effects/doeff_core_effects/scheduler.py

docs/adr/defadr_doeff_core_001_spawn_boundary_inheritance.hy

@@ -0,0 +1,52 @@
+;;; Executable ADR: Spawn は spawn-site の boundary stack(handler+observer)を継承する。
+
+(require doeff-adr.macros [defadr defsemgrep rule law])
+(import doeff-adr.macros [fact interpretation counterexample])
+
+
+(defadr ADR-DOE-CORE-001
+  :title "Spawn は handler と同様に WithObserve 境界も捕捉・再インストールする(interleave 順序保存)"
+  :status "accepted"
+  :scope ["doeff-core-effects" "scheduler.Spawn" "doeff-vm.GetBoundaries" "WithObserve"]
+  :problem
+    [(fact
+       "scheduled() の内側に置いた WithObserve は Spawn した子タスクが perform する効果を一切観測しない。同じ位置の handler(WithHandler)は子に継承されるのに、observer だけが脱落する(silent drop — fail-fast 原則違反)。"
+       :evidence "issues/scheduler-spawn-drops-observer-boundary.md 再現 @ cacc4b85; tests/test_spawn_observer_inheritance.py は修正前 HEAD で 5 件 fail")
+     (fact
+       "機構: Spawn は get_inner_handlers(= GetHandlers(k))で prompt boundary のみを捕捉して子タスクに再インストールする。intercept(observer)boundary は捕捉対象外。observer 呼び出しは fiber 親チェーン走査であり、Spawn 子 fiber は scheduler の評価文脈を親に持つため、spawn-site〜scheduler 間の observer boundary はチェーン上に存在しない。"
+       :evidence "packages/doeff-core-effects/doeff_core_effects/scheduler.py Spawn 分岐; doeff/handler_utils.py; packages/doeff-vm-core/src/vm/step.rs call_all_observers")
+     (fact
+       "実務影響: doeff-traverse parallel worker 内の slog が observer ベース出力に出ない(proboscis-ema, 2026-04)。OpenTelemetry の effect→span 自動計装をサブツリー限定で書くと、Spawn を含む区間で計装が『付いているのに見えない』。"
+       :evidence "issue 動機セクション")]
+  :context
+    [(interpretation
+       "観測の期待値として handler と observer の継承規則は対称であるべき。非対称を正当化する仕様文書は存在しなかった(issue 論点 1)。")
+     (interpretation
+       "相対順序も意味論の一部: handler(WithObserve(obs, body)) の配置では handler 本体が emit する効果は obs に不可視。フラットな『observer を最外へ積む』再インストールは過剰観測(重複 span 等)を生む。よって捕捉・再インストールは interleaved(innermost-first)で順序を保存する。")
+     (interpretation
+       "primitive は opt-in クエリ GetBoundaries(k)。GetHandlers(k) と対称で、継続を消費しない(multi-shot 安全)。既存 GetHandlers とその多数の利用箇所(try_handler, doeff-traverse 等)は不変(issue 論点 2 の opt-in 案を採用しつつ、scheduler は既定で継承する)。")]
+  :decision
+    [(rule R1 "doeff-vm bridge に GetBoundaries(k) を追加する: spawn-site から catching handler までの [\"handler\"|\"observer\", callable] を innermost-first で返す。catching handler の prompt boundary を最終エントリとして含む(GetHandlers と対称)。")
+     (rule R2 "doeff.handler_utils.get_inner_boundaries(k) は GetBoundaries(k) の最終エントリ(呼び手自身)を落として返す。最終エントリが handler 種でなければ RuntimeError(fail-fast)。")
+     (rule R3 "scheduler の Spawn は get_inner_boundaries で boundary stack を捕捉し、子タスク再構築時に kind に応じて WithHandler / WithObserve を innermost-first で再インストールする。handler と observer の相対 nesting 順序を保存する。")
+     (rule R4 "再インストールされた boundary は子タスクの program が所有し、タスク終端(完了・失敗・キャンセル)で _release_task_refs により解放される。scheduler 内部効果(wrap_task の TaskCompleted 等)は boundary の外側で perform され、サブツリー observer から観測されない(issue 論点 3)。")
+     (rule R5 "Mask boundary は本 ADR の対象外(従来どおり Spawn 非継承)。継承が必要になった場合は GetBoundaries の kind 追加として本 ADR を改訂する。")]
+  :laws
+    [(law spawn-observer-inheritance
+       :statement "spawn_site_boundary(observer or handler) => reinstalled_on_child preserving_relative_order"
+       :counterexamples
+         [(counterexample "scheduled(WithObserve(obs, h(main))) で子タスクの効果が observed に含まれない")
+          (counterexample "observer をフラットに最外へ積み、handler 本体が emit する効果まで観測してしまう")
+          (counterexample "Spawn 経路が get_inner_handlers に退行し observer だけ脱落する")])]
+  :enforcement
+    [(defsemgrep scheduler-spawn-boundary-capture
+       "doeff-scheduler-spawn-must-capture-boundaries"
+       [{"relative-path" "packages/doeff-core-effects/doeff_core_effects/scheduler.py"
+         "source" "inner_handlers = yield get_inner_handlers(k)\n"}]
+       [{"relative-path" "packages/doeff-core-effects/doeff_core_effects/handlers.py"
+         "source" "inner_hs = yield get_inner_handlers(k)\n"}])]
+  :plans ["docs/adr/defadr_doeff_core_001_spawn_boundary_inheritance.hy"
+          "tests/test_spawn_observer_inheritance.py"
+          "packages/doeff-vm-core/src/continuation.rs (boundary_callables)"
+          "packages/doeff-vm/src/do_expr.rs (GetBoundaries pyclass)"
+          "packages/doeff-core-effects/doeff_core_effects/scheduler.py (Spawn boundary capture)"])

doeff/__init__.py

@@ -21,6 +21,7 @@
 from doeff.mcp import McpToolDef as McpToolDef
 from doeff.program import Apply as Apply
 from doeff.program import Expand as Expand
+from doeff.program import GetBoundaries as GetBoundaries
 from doeff.program import GetExecutionContext as GetExecutionContext
 from doeff.program import GetHandlers as GetHandlers
 from doeff.program import GetOuterHandlers as GetOuterHandlers
@@ -103,7 +104,7 @@ def WithObserve(observer, body):  # noqa: N802 - public compatibility constructo
 _DOEXPR_TYPES = (
     Pure, Perform, Resume, Transfer, Apply, Expand, Pass,
     WithHandlerType, WithObserveRaw, ResumeThrow, TransferThrow,
-    GetTraceback, GetExecutionContext, GetHandlers, GetOuterHandlers,
+    GetTraceback, GetExecutionContext, GetHandlers, GetBoundaries, GetOuterHandlers,
 )
 
 

doeff/handler_utils.py

@@ -3,7 +3,7 @@
 """
 
 from doeff.do import do
-from doeff.program import GetHandlers
+from doeff.program import GetBoundaries, GetHandlers
 
 
 @do
@@ -14,6 +14,10 @@ def get_inner_handlers(k):
     the effect — i.e., the caller). Returns the remaining inner handlers
     (innermost first).
 
+    NOTE: this captures prompt (WithHandler) boundaries only — WithObserve
+    observer boundaries are NOT included. Use get_inner_boundaries when the
+    reinstalled program must keep observers attached (e.g. Spawn).
+
     Usage in a handler:
         inner_hs = yield get_inner_handlers(k)
         for h in inner_hs:
@@ -23,3 +27,36 @@ def get_inner_handlers(k):
     if all_hs:
         return all_hs[:-1]
     return []
+
+
+@do
+def get_inner_boundaries(k):
+    """Get the interleaved boundary stack from a continuation, excluding the caller.
+
+    Yields GetBoundaries(k) — every handler (WithHandler) and observer
+    (WithObserve) boundary between the yield site and the handler that caught
+    the effect, innermost first — and drops the last entry (the catching
+    handler itself, i.e. the caller). Returns a list of
+    ("handler" | "observer", callable) tuples.
+
+    Raises RuntimeError if the captured chain does not terminate at a handler
+    boundary — the catching handler must always be the outermost entry.
+
+    Usage in a handler (reinstall preserving nesting order):
+        boundaries = yield get_inner_boundaries(k)
+        for kind, boundary_callable in boundaries:
+            if kind == "handler":
+                prog = handler(boundary_callable)(prog)
+            else:
+                prog = WithObserve(boundary_callable, prog)
+    """
+    all_boundaries = yield GetBoundaries(k)
+    if not all_boundaries:
+        return []
+    last_kind, _last_callable = all_boundaries[-1]
+    if last_kind != "handler":
+        raise RuntimeError(
+            "get_inner_boundaries: expected the catching handler as the last "
+            f"chain entry, got kind={last_kind!r}"
+        )
+    return [(kind, boundary_callable) for kind, boundary_callable in all_boundaries[:-1]]

doeff/program.py

@@ -9,6 +9,7 @@
 
 from doeff_vm import Apply as Apply
 from doeff_vm import Expand as Expand
+from doeff_vm import GetBoundaries as GetBoundaries
 from doeff_vm import GetExecutionContext as GetExecutionContext
 from doeff_vm import GetHandlers as GetHandlers
 from doeff_vm import GetOuterHandlers as GetOuterHandlers

packages/doeff-core-effects/doeff_core_effects/scheduler.py

@@ -21,20 +21,31 @@ def main():
     run(scheduled(main()))
 """
 
+from doeff_vm import Callable as _VmCallable
 from doeff_vm import EffectBase, Err, Ok, TailEval
+from doeff_vm import WithObserve as _WithObserveRaw
 
 from doeff.do import do
-from doeff.handler_utils import get_inner_handlers
+from doeff.handler_utils import get_inner_boundaries
 from doeff.program import Pass, Perform, Pure, Resume, Transfer
 from doeff.program import handler as _program_handler
 
 
+def _reinstall_boundary(prog, kind, boundary_callable):
+    """Re-wrap prog with one boundary captured at the spawn site."""
+    if kind == "handler":
+        return _program_handler(boundary_callable)(prog)
+    if kind == "observer":
+        return _WithObserveRaw(_VmCallable(boundary_callable), prog)
+    raise RuntimeError(f"unknown boundary kind: {kind!r}")
+
+
 def _enrich_exception_traceback(exc, task_meta=None, vm_ctx=None):
     """Build doeff traceback from VM execution context + task metadata.
 
     vm_ctx: from GetExecutionContext — fiber chain at error site (before unwinding).
             Contains ["frame", ...] and ["handler", ...] entries from the live fiber chain.
-    task_meta: scheduler task metadata with inner_handlers.
+    task_meta: scheduler task metadata with inner_boundaries.
     """
     entries = []
 
@@ -302,12 +313,12 @@ def enqueue_resume(owner_tid, cont, value, priority=PRIORITY_NORMAL):
     def enqueue_raise(owner_tid, cont, error, priority=PRIORITY_NORMAL):
         enqueue(("raise", owner_tid, cont, error), priority)
 
-    def alloc_task(program, priority=PRIORITY_NORMAL, inner_handlers=None):
+    def alloc_task(program, priority=PRIORITY_NORMAL, inner_boundaries=None):
         tid = fresh_id()
         tasks[tid] = {
             "status": "pending", "result": None,
             "program": program, "priority": priority,
-            "inner_handlers": inner_handlers or [],
+            "inner_boundaries": inner_boundaries or [],
         }
         return tid
 
@@ -389,9 +400,11 @@ def pick_next():  # noqa: PLR0912 - scheduler dispatch loop has one branch per r
                         continue  # skip cancelled tasks
                     tasks[tid]["status"] = "running"
                     prog = tasks[tid].pop("program")
-                    # Re-wrap task with inner handlers captured at spawn site
-                    for h in tasks[tid].pop("inner_handlers", []):
-                        prog = _program_handler(h)(prog)
+                    # Re-wrap task with the boundary stack (handlers AND
+                    # observers) captured at the spawn site, innermost first —
+                    # preserves handler/observer nesting order.
+                    for kind, boundary_callable in tasks[tid].pop("inner_boundaries", []):
+                        prog = _reinstall_boundary(prog, kind, boundary_callable)
                     return make_handler(tid)(wrap_task(tid, prog))
                 if entry[0] == "resume":
                     _, owner_tid, cont, value = entry
@@ -434,14 +447,14 @@ def _release_task_refs(tid):
         """Drop heavy references from a terminal task.
 
         Keeps status/result (needed by Wait/Gather) but releases the
-        program, inner_handlers, and spawn_site closures that pin large
+        program, inner_boundaries, and spawn_site closures that pin large
         Python object graphs into memory.
         """
         t = tasks.get(tid)
         if t is None:
             return
         t.pop("program", None)
-        t.pop("inner_handlers", None)
+        t.pop("inner_boundaries", None)
         t.pop("spawn_site", None)
 
     def resume_with_waitable_result(owner_tid, waiter_k, key):
@@ -573,8 +586,10 @@ def raw_handler(effect, k):
     def handle_scheduler_effect(current_tid, effect, k):  # noqa: PLR0911, PLR0912, PLR0915 - baseline cleanup keeps existing control flow unchanged
         drain()
         if isinstance(effect, Spawn):
-            # Capture inner handlers from continuation (between yield site and scheduler).
-            inner_handlers = yield get_inner_handlers(k)
+            # Capture the boundary stack (handlers AND WithObserve observers)
+            # from the continuation (between yield site and scheduler) so the
+            # spawned task keeps both — with nesting order preserved.
+            inner_boundaries = yield get_inner_boundaries(k)
 
             # Capture spawn site from continuation's traceback
             from doeff.program import GetTraceback
@@ -585,7 +600,7 @@ def handle_scheduler_effect(current_tid, effect, k):  # noqa: PLR0911, PLR0912,
                 if isinstance(f, (list, tuple)) and len(f) >= 3:
                     spawn_site = f"{f[0]}  {f[1]}:{f[2]}"
 
-            tid = alloc_task(effect.program, effect.priority, inner_handlers=inner_handlers)
+            tid = alloc_task(effect.program, effect.priority, inner_boundaries=inner_boundaries)
             tasks[tid]["spawn_site"] = spawn_site
             enqueue(("new", tid), effect.priority)
             enqueue_resume(current_tid, k, Task(tid))  # spawner resumes at normal priority

packages/doeff-vm-core/src/continuation.rs

@@ -39,6 +39,16 @@ pub struct DetachedFiber {
     pub fiber: Fiber,
 }
 
+/// Kind of callable boundary found while walking a detached fiber chain.
+/// `Handler` = prompt boundary (WithHandler), `Observer` = intercept
+/// boundary (WithObserve). Mask boundaries carry no callable and are
+/// not reported.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BoundaryKind {
+    Handler,
+    Observer,
+}
+
 /// A body-through-boundary fiber chain owned by `Continuation`.
 #[derive(Debug)]
 pub struct DetachedFiberChain {
@@ -126,6 +136,30 @@ impl DetachedFiberChain {
         handlers
     }
 
+    /// Collect the interleaved handler/observer boundary stack, innermost
+    /// first (chain head toward parents). The catching handler's prompt
+    /// boundary terminates the chain and is included as the last entry,
+    /// symmetric with `handler_callables`.
+    pub fn boundary_callables(&self) -> Vec<(BoundaryKind, CallableRef)> {
+        let mut boundaries = Vec::new();
+        let mut cursor = Some(self.head);
+
+        while let Some(fid) = cursor {
+            let Some(fiber) = self.fiber(fid) else { break };
+            if let Some(handler) = &fiber.handler {
+                if let Some(prompt) = handler.prompt_boundary() {
+                    boundaries.push((BoundaryKind::Handler, prompt.handler.clone()));
+                }
+                if let Some(intercept) = handler.intercept_boundary() {
+                    boundaries.push((BoundaryKind::Observer, intercept.interceptor.clone()));
+                }
+            }
+            cursor = fiber.parent;
+        }
+
+        boundaries
+    }
+
     pub fn collect_rich_context(&self) -> Vec<Value> {
         let mut raw: Vec<(String, String, u32)> = Vec::new();
         let mut first_boundary: Option<FiberId> = None;
@@ -319,6 +353,12 @@ impl Continuation {
         self.chain.as_ref().map(DetachedFiberChain::handler_callables)
     }
 
+    pub fn boundary_callables(&self) -> Option<Vec<(BoundaryKind, CallableRef)>> {
+        self.chain
+            .as_ref()
+            .map(DetachedFiberChain::boundary_callables)
+    }
+
     pub fn collect_rich_context(&self) -> Option<Vec<Value>> {
         self.chain.as_ref().map(DetachedFiberChain::collect_rich_context)
     }

packages/doeff-vm-core/src/lib.rs

@@ -49,7 +49,9 @@ mod vm_tests;
 #[cfg(feature = "python_bridge")]
 pub use arena::FiberArena;
 #[cfg(feature = "python_bridge")]
-pub use continuation::{Continuation, OwnedControlContinuation, PendingContinuation, PyK};
+pub use continuation::{
+    BoundaryKind, Continuation, OwnedControlContinuation, PendingContinuation, PyK,
+};
 #[cfg(feature = "python_bridge")]
 pub use do_ctrl::DoCtrl;
 #[cfg(feature = "python_bridge")]

packages/doeff-vm/doeff_vm/__init__.py

@@ -28,6 +28,7 @@
 GetTraceback = _ext.GetTraceback
 GetExecutionContext = _ext.GetExecutionContext
 GetHandlers = _ext.GetHandlers
+GetBoundaries = _ext.GetBoundaries
 GetOuterHandlers = _ext.GetOuterHandlers
 TailEval = _ext.TailEval
 

packages/doeff-vm/doeff_vm/__init__.pyi

@@ -130,6 +130,11 @@ class GetHandlers:
     def __init__(self, k: K) -> None: ...
     def __repr__(self) -> str: ...
 
+class GetBoundaries:
+    continuation: K
+    def __init__(self, k: K) -> None: ...
+    def __repr__(self) -> str: ...
+
 class GetOuterHandlers:
     def __init__(self) -> None: ...
     def __repr__(self) -> str: ...

packages/doeff-vm/src/do_expr.rs

@@ -356,6 +356,32 @@ impl PyGetHandlers {
     }
 }
 
+/// GetBoundaries(k) — extract the interleaved handler/observer boundary
+/// stack from the continuation's fiber chain, innermost first. Each entry is
+/// a `["handler" | "observer", callable]` pair. The catching handler is
+/// included as the last entry, symmetric with GetHandlers(k).
+///
+/// Used by the scheduler to reinstall the full spawn-site boundary stack —
+/// handlers AND WithObserve observers, preserving their relative nesting
+/// order — on Spawn'd tasks (issue scheduler-spawn-drops-observer-boundary).
+#[pyclass(name = "GetBoundaries", frozen, dict, module = "doeff_vm.doeff_vm")]
+pub struct PyGetBoundaries {
+    #[pyo3(get)]
+    pub continuation: Py<PyK>,
+}
+
+#[pymethods]
+impl PyGetBoundaries {
+    #[new]
+    fn new(k: Py<PyK>) -> Self {
+        Self { continuation: k }
+    }
+
+    fn __repr__(&self) -> &'static str {
+        "GetBoundaries(k)"
+    }
+}
+
 /// GetOuterHandlers — extract handlers installed ABOVE the current handler.
 ///
 /// When a handler catches an effect, its segment's parent is detached from the