feat: consolidate inspection tools

Author: clfhhc

docs/optimizations.md

@@ -34,6 +34,10 @@ Cross-cutting rules for reactive/async integration (especially `patterns.ai`, LL
 | **`Node` resolution without `get()`** | When blocking until first `DATA`, prefer `node.get()` when it already holds a settled value, then subscribe only if still pending — avoids hangs when the node does not replay `DATA` to new subscribers. | — |
 | **Passing plain strings through `fromAny` (TypeScript)** | `fromAny` treats strings as iterables (one `DATA` per character). For tool handlers that return plain strings, return the string directly; use `fromAny` only for `Node` / `AsyncIterable` / Promise-like after await. | — |
 
+- **`batch_id` in `ObserveEvent` timeline fields (decided 2026-04-08):** Added `batch_id?: number` / `batch_id: int` to `ObserveEvent` in both repos. Increments once per subscribe-callback invocation; all messages in one delivery share the same `batch_id`. Useful for correlating events that arrived together. TS: added to `_createObserveResult`, `_createObserveResultForAll`, and both fallback paths. PY: already present; now documented.
+
+- **`ObserveResult.completedCleanly` ambiguous in graph-wide mode (noted 2026-04-08, carried from TS):** Same issue as TS. In graph-wide observation, `completed_cleanly` and `errored` can both be `True` simultaneously if one node completes cleanly before another errors. Single-node mode is unaffected. Options: (A) rename to `any_completed_cleanly` / `any_errored`; (B) add `all_completed_cleanly`; (C) reset on any ERROR. Pending decision.
+
 ---
 
 ## Deferred follow-ups

docs/roadmap.md

@@ -47,6 +47,24 @@ TS has this shipped; Python needs parity for production reduction pipelines.
 - [ ] PyPI publish: `graphrefly-py` (7 → 9.3)
 - [ ] Docs site at `py.graphrefly.dev` (7 → 9.3)
 
+### Inspection Tool Consolidation (cross-cutting, TS + PY)
+
+PY consolidation — reduce inspection surface from 14+ tools to 9 with clear, non-overlapping responsibilities.
+
+**Design reference:** `archive/docs/SESSION-inspection-consolidation.md` (in graphrefly-ts)
+
+#### PY consolidation (DONE 2026-04-08)
+
+- [x] Merge `spy()` into `observe(format=)` — `format="pretty"|"json"` on `observe()` replaces `spy()`
+- [x] Add `trace()` (write + read), absorb `annotate()` + `trace_log()`
+- [x] Unexport `describe_node`, `meta_snapshot` from public API (internal to `core.meta`)
+- [x] `harness_trace()` pipeline stage tracer (`patterns/harness/trace.py`)
+- [x] Runner `__repr__` with `_scheduled`/`_completed` counters on `AsyncioRunner` and `TrioRunner`
+
+#### PY parity items already done
+
+- [x] `Graph.diff()` — already implemented
+
 ### Deferred (post-Wave 3)
 
 - §7.2 Showcase demos (Pyodide/WASM lab) — after TS demos prove the pattern

src/graphrefly/__init__.py

@@ -37,7 +37,6 @@
     defer_down,
     defer_set,
     derived,
-    describe_node,
     dispatch_messages,
     down_with_batch,
     dynamic_node,
@@ -47,7 +46,6 @@
     is_batching,
     is_phase2_message,
     is_v1,
-    meta_snapshot,
     monotonic_ns,
     node,
     normalize_actor,
@@ -74,7 +72,6 @@
     GraphDiffResult,
     GraphObserveSource,
     ObserveResult,
-    SpyHandle,
     TraceEntry,
     reachable,
 )
@@ -93,7 +90,6 @@
     "NodeVersionInfo",
     "ObserveResult",
     "PATH_SEP",
-    "SpyHandle",
     "TraceEntry",
     "V0",
     "V1",
@@ -138,13 +134,11 @@
     "compose_guards",
     "defer_down",
     "defer_set",
-    "describe_node",
     "dispatch_messages",
     "down_with_batch",
     "ensure_registered",
     "is_batching",
     "is_phase2_message",
-    "meta_snapshot",
     "node",
     "normalize_actor",
     "partition_for_batch",

src/graphrefly/compat/asyncio_runner.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -29,10 +30,12 @@ async def main():
         asyncio.run(main())
     """
 
-    __slots__ = ("_loop",)
+    __slots__ = ("_loop", "_scheduled", "_completed")
 
     def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
         self._loop = loop
+        self._scheduled = 0
+        self._completed = 0
 
     @classmethod
     def from_running(cls) -> AsyncioRunner:
@@ -50,11 +53,12 @@ def schedule(
         on_error: Callable[[BaseException], None],
     ) -> Callable[[], None]:
         task: asyncio.Task[Any] | None = None
-        cancelled = False
+        cancelled = threading.Event()
 
         def _create_task() -> None:
             nonlocal task
-            if cancelled:
+            if cancelled.is_set():
+                self._completed += 1
                 coro.close()
                 return
 
@@ -71,19 +75,34 @@ async def _wrapper() -> None:
                     on_error(err)
                 else:
                     on_result(result)
+                finally:
+                    self._completed += 1
 
             task = self._loop.create_task(_wrapper())
 
-        # Thread-safe: schedule task creation on the event loop.
-        self._loop.call_soon_threadsafe(_create_task)
+        # Increment eagerly on the calling thread so __repr__ is always consistent.
+        self._scheduled += 1
+        try:
+            self._loop.call_soon_threadsafe(_create_task)
+        except RuntimeError:
+            # Loop closed — cannot schedule; close the coroutine and balance the counter.
+            self._completed += 1
+            coro.close()
 
         def cancel() -> None:
-            nonlocal cancelled
-            cancelled = True
+            cancelled.set()
             if task is not None:
                 task.cancel()
 
         return cancel
 
+    def __repr__(self) -> str:
+        pending = self._scheduled - self._completed
+        running = self._loop.is_running()
+        return (
+            f"AsyncioRunner(scheduled={self._scheduled}, completed={self._completed}, "
+            f"pending={pending}, loop_running={running})"
+        )
+
 
 __all__ = ["AsyncioRunner"]

src/graphrefly/compat/trio_runner.py

@@ -8,6 +8,7 @@
     from graphrefly.compat.trio_runner import TrioRunner
     from graphrefly.core.runner import set_default_runner
 
+
     async def main():
         async with trio.open_nursery() as nursery:
             runner = TrioRunner(nursery)
@@ -19,6 +20,7 @@ async def main():
 
 from __future__ import annotations
 
+import threading
 from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
@@ -34,10 +36,12 @@ class TrioRunner:
     Cancel scopes provide best-effort cancellation.
     """
 
-    __slots__ = ("_nursery",)
+    __slots__ = ("_nursery", "_scheduled", "_completed")
 
     def __init__(self, nursery: trio.Nursery) -> None:
         self._nursery = nursery
+        self._scheduled = 0
+        self._completed = 0
 
     def schedule(
         self,
@@ -48,11 +52,12 @@ def schedule(
         import trio as _trio
 
         cancel_scope = _trio.CancelScope()
-        cancelled = False
+        cancelled = threading.Event()
 
         async def _wrapper() -> None:
             with cancel_scope:
-                if cancelled:
+                if cancelled.is_set():
+                    self._completed += 1
                     coro.close()
                     return
                 try:
@@ -67,15 +72,24 @@ async def _wrapper() -> None:
                     on_error(err)
                 else:
                     on_result(result)
+                finally:
+                    self._completed += 1
 
+        self._scheduled += 1
         self._nursery.start_soon(_wrapper)
 
         def cancel() -> None:
-            nonlocal cancelled
-            cancelled = True
+            cancelled.set()
             cancel_scope.cancel()
 
         return cancel
 
+    def __repr__(self) -> str:
+        pending = self._scheduled - self._completed
+        return (
+            f"TrioRunner(scheduled={self._scheduled}, completed={self._completed}, "
+            f"pending={pending})"
+        )
+
 
 __all__ = ["TrioRunner"]

src/graphrefly/core/__init__.py

@@ -17,7 +17,6 @@
     record_mutation,
     system_actor,
 )
-from graphrefly.core.meta import describe_node, meta_snapshot
 from graphrefly.core.node import (
     NO_VALUE,
     Node,
@@ -112,7 +111,6 @@
     "compose_guards",
     "defer_down",
     "defer_set",
-    "describe_node",
     "dispatch_messages",
     "down_with_batch",
     "ensure_registered",
@@ -121,7 +119,6 @@
     "is_terminal_message",
     "message_tier",
     "propagates_to_meta",
-    "meta_snapshot",
     "node",
     "normalize_actor",
     "partition_for_batch",

src/graphrefly/core/node.py

@@ -394,11 +394,11 @@ def _handle_local_lifecycle(self, messages: Messages) -> None:
                     continue
                 if lock is not None:
                     with lock:
-                        self._cached = m[1]  # type: ignore[misc]
+                        self._cached = m[1]
                 else:
-                    self._cached = m[1]  # type: ignore[misc]
+                    self._cached = m[1]
                 if self._versioning is not None:
-                    advance_version(self._versioning, m[1], self._hash_fn)  # type: ignore[misc]
+                    advance_version(self._versioning, m[1], self._hash_fn)
             if t is MessageType.INVALIDATE:
                 # GRAPHREFLY-SPEC §1.2: clear cached state; do not auto-emit from here.
                 if self._cleanup is not None:

src/graphrefly/extra/cascading_cache.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections import OrderedDict
-from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar, runtime_checkable
+from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar, cast, runtime_checkable
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -137,10 +137,20 @@ def _tier_has_save(tier: CacheTier) -> bool:
     return callable(getattr(tier, "save", None))
 
 
+def _tier_save(tier: CacheTier, key: str, value: Any) -> None:
+    """Call ``tier.save(key, value)`` — caller must guard with ``_tier_has_save`` first."""
+    cast("Any", tier).save(key, value)
+
+
 def _tier_has_clear(tier: CacheTier) -> bool:
     return callable(getattr(tier, "clear", None))
 
 
+def _tier_clear(tier: CacheTier, key: str) -> None:
+    """Call ``tier.clear(key)`` — caller must guard with ``_tier_has_clear`` first."""
+    cast("Any", tier).clear(key)
+
+
 class CascadingCache(Generic[V]):  # noqa: UP046
     """Multi-tier cache where each entry is a ``state()`` node.
 
@@ -174,7 +184,7 @@ def _promote(self, key: str, value: V, hit_tier: int) -> None:
         for i in range(hit_tier):
             tier = self._tiers[i]
             if _tier_has_save(tier):
-                tier.save(key, value)
+                _tier_save(tier, key, value)
 
     def _cascade(self, key: str, nd: Node[Any]) -> None:
         for tier_index, tier in enumerate(self._tiers):
@@ -202,10 +212,10 @@ def _evict_if_needed(self) -> None:
                         # Demote to deepest tier with save before evicting
                         for i in range(len(self._tiers) - 1, -1, -1):
                             if _tier_has_save(self._tiers[i]):
-                                self._tiers[i].save(victim, value)
+                                _tier_save(self._tiers[i], victim, value)
                                 for j in range(i):
                                     if _tier_has_clear(self._tiers[j]):
-                                        self._tiers[j].clear(victim)
+                                        _tier_clear(self._tiers[j], victim)
                                 break
                     nd.down([(MessageType.TEARDOWN,)])
                     del self._entries[victim]
@@ -239,9 +249,9 @@ def save(self, key: str, value: V) -> None:
         if self._write_through:
             for tier in self._tiers:
                 if _tier_has_save(tier):
-                    tier.save(key, value)
+                    _tier_save(tier, key, value)
         elif self._tiers and _tier_has_save(self._tiers[0]):
-            self._tiers[0].save(key, value)
+            _tier_save(self._tiers[0], key, value)
         if key in self._entries:
             self._entries[key].down([(MessageType.DATA, value)])
             if self._eviction is not None:
@@ -273,7 +283,7 @@ def delete(self, key: str) -> None:
                 self._eviction.delete(key)
         for tier in self._tiers:
             if _tier_has_clear(tier):
-                tier.clear(key)
+                _tier_clear(tier, key)
 
     def has(self, key: str) -> bool:
         """Check if a key is in the in-memory entries."""

src/graphrefly/extra/data_structures.py

@@ -145,9 +145,7 @@ def _prune_expired_internal(self) -> bool:
         """Remove expired keys from ``_store``; return whether anything changed."""
         now = _mono_now()
         expired = [
-            k
-            for k, (_, exp) in self._store.entries.items()
-            if exp is not None and exp <= now
+            k for k, (_, exp) in self._store.entries.items() if exp is not None and exp <= now
         ]
         if not expired:
             return False

src/graphrefly/graph/__init__.py

@@ -25,7 +25,6 @@
     GraphDiffResult,
     GraphObserveSource,
     ObserveResult,
-    SpyHandle,
     TraceEntry,
     reachable,
 )
@@ -56,7 +55,6 @@
     "NodeProfile",
     "ObserveResult",
     "PATH_SEP",
-    "SpyHandle",
     "TraceEntry",
     "WALEntry",
     "create_dag_cbor_codec",