feat: fix inspection tool

Author: clfhhc

TRASH-FILES.md

@@ -99,9 +99,24 @@ def resolve_runner(runner: Runner | None) -> Runner:
     return get_default_runner()
 
 
+def is_runner_registered() -> bool:
+    """Return ``True`` if a default :class:`Runner` is set for the current thread.
+
+    **Debug / test only** — not part of the public API. Used by
+    ``harnessProfile`` to surface ``runner_registered=False`` in diagnostic
+    output so missing-Runner stalls are immediately visible.
+    """
+    try:
+        r = _runner_tls.runner
+    except AttributeError:
+        return False
+    return r is not None
+
+
 __all__ = [
     "Runner",
     "get_default_runner",
+    "is_runner_registered",
     "resolve_runner",
     "set_default_runner",
 ]

TRASH/EmitStrategy.md

@@ -138,6 +138,7 @@
     cached,
     empty,
     first_value_from,
+    first_where,
     for_each,
     from_any,
     from_async_iter,
@@ -250,6 +251,7 @@
     "log_slice",
     "empty",
     "first_value_from",
+    "first_where",
     "for_each",
     "from_any",
     "from_async_iter",

TRASH/emit_with_batch.md

@@ -543,6 +543,12 @@ def first_value_from(
     On ``COMPLETE`` without prior ``DATA``, raises :exc:`StopIteration`. With
     *timeout*, raises :exc:`TimeoutError` if no terminal message arrives in time.
 
+    **Important:** This subscribes to *source* and waits for a **future**
+    emission. It does NOT read the cached value — data that has already
+    flowed is gone. You must call this **before** the upstream emits, or
+    use ``source.get()`` / ``source.status`` to read already-cached state.
+    See COMPOSITION-GUIDE §2 (subscription ordering).
+
     Args:
         source: The node to await the first value from.
         timeout: Optional timeout in seconds.
@@ -602,6 +608,75 @@ def sink(msgs: Messages) -> None:
     return got[0]
 
 
+def first_where(
+    source: Node[Any],
+    predicate: Callable[[Any], bool],
+    *,
+    timeout: float | None = None,
+) -> Any:
+    """Block until the first ``DATA`` value satisfying *predicate* arrives.
+
+    Subscribes directly and checks *predicate* on each ``DATA`` emission.
+    No polling. Use in tests and bridging code where you need to wait for
+    a specific value synchronously.
+
+    **Important:** This only captures **future** emissions — data that has
+    already flowed through the node is gone and will not be seen. You must
+    call this **before** the upstream emits. For already-cached values, use
+    ``source.get()`` / ``source.status`` instead. See COMPOSITION-GUIDE §2.
+
+    Args:
+        source: The node to observe.
+        predicate: Called with each DATA value; returns ``True`` to accept.
+        timeout: Optional timeout in seconds.
+
+    Returns:
+        The first DATA payload where ``predicate(value)`` is ``True``.
+
+    Example:
+        ```python
+        val = first_where(strategy.node, lambda snap: len(snap) > 0, timeout=5.0)
+        ```
+    """
+    got: list[Any | None] = [None]
+    err_box: list[BaseException | Any | None] = [None]
+    done = threading.Event()
+
+    def sink(msgs: Messages) -> None:
+        for m in msgs:
+            t = m[0]
+            if t is MessageType.DATA:
+                v = _msg_val(m)
+                if got[0] is None and predicate(v):
+                    got[0] = v
+                    done.set()
+            elif t is MessageType.ERROR:
+                err_box[0] = _msg_val(m)
+                done.set()
+            elif t is MessageType.COMPLETE:
+                done.set()
+
+    unsub = source.subscribe(sink)
+    try:
+        if timeout is None:
+            done.wait()
+        elif not done.wait(timeout):
+            msg = "first_where timed out"
+            raise TimeoutError(msg)
+    finally:
+        unsub()
+
+    err = err_box[0]
+    if err is not None:
+        if isinstance(err, BaseException):
+            raise err
+        raise RuntimeError(str(err))
+    if got[0] is None:
+        msg = "first_where: source completed without a matching value"
+        raise StopIteration(msg)
+    return got[0]
+
+
 # --- multicast ----------------------------------------------------------------
 
 

src/graphrefly/core/runner.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
+from graphrefly.core.runner import is_runner_registered
 from graphrefly.graph.profile import graph_profile
 
 if TYPE_CHECKING:
@@ -36,6 +37,8 @@ class HarnessProfileResult:
     strategy_entries: int
     total_retries: int
     total_reingestions: int
+    # Runner diagnostic (session doc: surfaces registered=False immediately)
+    runner_registered: bool
 
 
 # ---------------------------------------------------------------------------
@@ -77,4 +80,5 @@ def harness_profile(
         strategy_entries=strategy_entries,
         total_retries=harness.total_retries.get() or 0,
         total_reingestions=harness.total_reingestions.get() or 0,
+        runner_registered=is_runner_registered(),
     )