feat: address optimizations

Author: clfhhc

docs/optimizations.md

@@ -8,7 +8,7 @@
 
 - **PY `ReactiveMapBundle` parity — `.get(key)`, `.has(key)`, `.size` (noted 2026-04-07):**
   - **Level A: DONE (2026-04-07).** Added `.get(key)`, `.has(key)`, `.size` to PY `ReactiveMapBundle` matching TS signatures. PY harness `strategy.py` updated to use `.get(key)` instead of Versioned navigation.
-  - **Level B: Deferred (post-1.0).** `ReactiveMapBundle.node` (TS) / `.data` (PY) emits `Versioned<{ map: ReadonlyMap<K,V> }>` / `Versioned(version, MappingProxyType)`. The `Versioned` wrapper is a protocol optimization (efficient RESOLVED dedup via version comparison) that leaks into composition code when using the node as a derived dep. **Proposed fix:** `.node` / `.data` emits the unwrapped map directly; version-based equality handled internally via `equals` option on the state node. Consumers see `ReadonlyMap<K,V>` / `MappingProxyType`, not `Versioned`. Breaking change — defer to post-1.0 audit of all `Versioned` usage.
+  - **Level B: DONE (PY, 2026-04-07).** Removed `Versioned` wrapper from all reactive bundle APIs (ReactiveMap, ReactiveLog, ReactiveList, ReactiveIndex). `.data` / `.entries` / `.items` / `.ordered` now emit unwrapped domain types (`MappingProxyType`, `tuple`, etc.). Internal version counter drives efficient equality without leaking into composition code (spec §5.12). All downstream consumers updated (messaging, cqrs, ai, domain-templates, composite, harness/strategy).
 
 - **Whole-repo `emit` → `down` audit + `up` / backpressure / `message_tier` sweep (all phases, noted 2026-04-07):**
   - **TS: DONE (2026-04-07).** Renames: `emitWithBatch` → `downWithBatch`, `_emitToSinks` → `_downToSinks`, `_emitAutoValue` → `_downAutoValue`, `_boundEmitToSinks` → `_boundDownToSinks`, `_emitSequential` → `_downSequential`, `emitLine` → `flushLine` (reactive-layout). Batch param `emit` → `sink`. `up()` audit: no asymmetries. `messageTier()` audit: already clean. `NodeActions.emit()` kept (different semantics from `actions.down()`). CQRS `CommandActions.emit()` kept (domain concept). Spec updated (`_emitAutoValue` → `_downAutoValue`).
@@ -42,38 +42,37 @@ Non-blocking items tracked for later. **Keep this section identical in both repo
 
 | Item | Notes |
 |------|-------|
-| **`lastDepValues` + `Object.is` / referential equality (resolved 2026-03-31 — keep + document)** | Default `Object.is` identity check is correct for the common immutable-value case. The `node({ equals })` option already exists for custom comparison. Document clearly that mutable dep values should use a custom `equals` function. No code change needed. |
+| **`lastDepValues` + `Object.is` / referential equality (resolved 2026-03-31 — documented)** | Default identity check is correct for the common immutable-value case. The `node(equals=...)` option already exists for custom comparison. Mutable dep values should use a custom `equals` function. **Documented in `node()` docstring (2026-04-07).** |
 | **`sideEffects: false` in `package.json`** | TypeScript package only. Safe while the library has no import-time side effects. Revisit if global registration or polyfills are added at module load. |
-| **JSDoc / docstrings on `node()` and public APIs** | `docs/docs-guidance.md`: JSDoc on new TS exports; docstrings on new Python public APIs. |
+| **JSDoc / docstrings on `node()` and public APIs** | `docs/docs-guidance.md`: JSDoc on new TS exports; docstrings on new Python public APIs. `node()` equals guidance added (2026-04-07). `flat_map` ERROR behavior documented (2026-04-07). `from_redis_stream` COMPLETE/disconnect documented (2026-04-07). |
 | **Roadmap §0.3 checkboxes** | Mark Phase 0.3 items when the team agrees the milestone is complete. |
 
 ### Factory teardown — `dispose()` pattern (D1/D2, noted 2026-04-07)
 
 | Item | Status | Notes |
 |------|--------|-------|
-| **Phase 4+ factories don't register internal nodes on the graph** | Proposed (2026-04-07) | `harness_loop` (and other factories) create internal nodes (`triage_node`, `_router` effect, `_retry_effect`, `execute_node`, `verify_node`, `with_latest_from` combinators, etc.) that are never added to the parent graph via `graph.add()`. When `graph.destroy()` fires, these nodes retain their subscriptions and closures — memory leak on repeated create/destroy cycles. Keepalive subscriptions (`_router.subscribe(...)`, `_retry_effect.subscribe(...)`) also leak because their unsub handles are discarded. |
-| **Proposed: `dispose()` convention** | Proposed (2026-04-07) | Factories collect all cleanup functions into a `_disposers: list[Callable[[], None]]` list. Override `destroy()` on the returned graph class to drain the list. Keepalive unsubs are just more entries in the same list. This is a Phase 0 primitive addition — a lightweight contract that any factory can adopt. Alternatives considered: (A) `graph.add()` all internal nodes — simple but pollutes the node registry with internals; (B) manual bookkeeping of unsub handles — fragile, easy to miss one. Option C (dispose list) is composable, private, and hard to get wrong. D2 (leaked keepalive handles) collapses into D1 — same list, same `destroy()` drain. |
+| **Phase 4+ factories don't register internal nodes on the graph** | **DONE (TS + PY, 2026-04-07)** | Added `Graph.addDisposer(fn)` / `Graph.add_disposer(fn)` — general-purpose disposer registration drained on `destroy()` **before** TEARDOWN signal. TS: Fixed `harnessLoop`, `strategyModel`, `agentMemory`, `feedback`, `gate`, `contentModerationGraph`, `funnel` bridge, `ChatStreamGraph`, `ToolRegistryGraph`. PY: Fixed `harness_loop`, `reduction.py`, `ChatStreamGraph`, `ToolRegistryGraph`, `AgentMemoryGraph`. Dead `_version` counter removed from all reactive bundles (TS + PY). |
 
 ### AI surface (Phase 4.4) — deferred optimizations
 
 | Item | Status | Notes |
 |------|--------|-------|
-| **Re-indexes entire store on every change** | Deferred | Decision: diff-based indexing using `Versioned` snapshot version field to track indexed entries. Deferred to after Phase 6 — current N is small enough that full re-index is acceptable pre-1.0. |
+| **Re-indexes entire store on every change** | Deferred | Decision: diff-based indexing using internal version counter to track indexed entries. Deferred to after Phase 6 — current N is small enough that full re-index is acceptable pre-1.0. |
 | **Budget packing always includes first item** | Documented behavior | The retrieval budget packer always includes the first ranked result even if it exceeds `maxTokens`. This is intentional "never return empty" semantics — a query that matches at least one entry always returns something. Callers who need strict budget enforcement should post-filter. |
 | **Retrieval pipeline auto-wires when vectors/KG enabled** | Documented behavior | When `embedFn` or `enableKnowledgeGraph` is set, the retrieval pipeline automatically wires vector search and KG expansion into the retrieval derived node. There is no explicit opt-in/opt-out per retrieval stage — the presence of the capability implies its use. Callers who need selective retrieval should use the individual nodes directly. |
 
 ### Tier 2 extra operators — deferred semantics
 
 | Item | Status | Notes |
 |------|--------|-------|
-| **`mergeMap` / `merge_map` + `ERROR`** | Documented limitation (2026-03-31) | When the outer stream or one inner emits `ERROR`, other inner subscriptions may keep running until they complete or unsubscribe. Rx-style "first error cancels all sibling inners" is **not** specified or implemented. Current behavior (inner errors don't cascade) is arguably more useful for parallel work — no change needed. Document in JSDoc/docstrings. |
+| **`mergeMap` / `merge_map` + `ERROR`** | **Documented (PY docstring, 2026-04-07)** | Inner errors propagate downstream but do not cancel sibling inners. Outer ERROR cancels all inners. Current behavior is intentional for parallel work. **Documented in `flat_map` docstring.** |
 
 ### Ingest adapters — deferred items
 
 | Item | Status | Notes |
 |------|--------|-------|
-| **`fromRedisStream` / `from_redis_stream` never emits COMPLETE** | Documented limitation (2026-04-03) | Long-lived stream consumers intentionally never complete. The consumer loop runs until teardown. This is expected behavior for persistent stream sources (same as Kafka). Document in JSDoc/docstrings. |
-| **`fromRedisStream` / `from_redis_stream` does not disconnect client** | Documented limitation (2026-04-03) | The caller owns the Redis client lifecycle. The adapter does not call `disconnect()` on teardown — the caller is responsible for closing the connection. Same contract as `fromKafka` (caller owns `consumer.connect()`/`disconnect()`). |
+| **`fromRedisStream` / `from_redis_stream` never emits COMPLETE** | **Documented (PY docstring, 2026-04-07)** | Long-lived stream consumers intentionally never complete. **Documented in `from_redis_stream` docstring.** |
+| **`fromRedisStream` / `from_redis_stream` does not disconnect client** | **Documented (PY docstring, 2026-04-07)** | The caller owns the Redis client lifecycle. **Documented in `from_redis_stream` docstring.** |
 | **PY `from_csv` / `from_ndjson` thread not joined on cleanup** | Documented limitation (2026-04-03) | Python file-ingest adapters run in a daemon thread. On teardown, `active[0] = False` signals the thread to exit but does not `join()` it. The daemon flag ensures the thread does not block process exit. A future optimization could add optional `join(timeout)` on cleanup for stricter resource control. |
 
 ### Intentional cross-language divergences

src/graphrefly/core/__init__.py

@@ -26,7 +26,6 @@
     NodeImpl,
     NodeStatus,
     SubscribeHints,
-    cleanup_result,
     node,
 )
 from graphrefly.core.protocol import (

src/graphrefly/core/node.py

@@ -976,6 +976,14 @@ def node(
         **kwargs: Any option key accepted by :class:`NodeImpl` (e.g. ``name``,
             ``initial``, ``equals``, ``guard``, ``thread_safe``).
 
+    **Equality (``equals``):** The default ``equals`` uses identity (``is``)
+    for immutable values.  If your node produces *mutable* dep values
+    (dicts, lists) that are modified in place, provide a custom ``equals``
+    function — otherwise downstream nodes may skip updates because the
+    identity check sees the *same* object::
+
+        node([source], compute_fn, equals=lambda a, b: a == b)
+
     Returns:
         A new :class:`NodeImpl` instance.
 

src/graphrefly/extra/__init__.py

@@ -110,7 +110,6 @@
     ReactiveListBundle,
     ReactiveLogBundle,
     ReactiveMapBundle,
-    Versioned,
     log_slice,
     pubsub,
     reactive_index,
@@ -214,7 +213,6 @@
     "ReactiveListBundle",
     "ReactiveLogBundle",
     "ReactiveMapBundle",
-    "Versioned",
     "NS_PER_MS",
     "NS_PER_SEC",
     "CacheTier",

src/graphrefly/extra/adapters.py

@@ -1848,6 +1848,15 @@ def from_redis_stream(
 
     Uses XREAD with BLOCK to reactively consume stream entries.
 
+    **Lifecycle notes:**
+
+    - This source intentionally **never emits** ``COMPLETE``. Long-lived
+      stream consumers are expected to run indefinitely; completion would
+      tear down downstream operators prematurely.
+    - The caller **owns the Redis client lifecycle**. ``from_redis_stream``
+      does not close or disconnect the client on teardown — call
+      ``client.close()`` yourself when the graph is destroyed.
+
     Args:
         client: Redis client instance with ``xread`` method (caller owns connection).
         key: Redis stream key.

src/graphrefly/extra/composite.py

@@ -113,11 +113,9 @@ class DistillBundle:
 
 
 def _snapshot_map(store: ReactiveMapBundle) -> Mapping[str, Any]:
-    snap = store.data.get()
+    snap = store.entries.get()
     if snap is None:
         return {}
-    if hasattr(snap, "value"):
-        return cast("Mapping[str, Any]", snap.value)
     return cast("Mapping[str, Any]", snap)
 
 
@@ -218,18 +216,18 @@ def _delete_keys(keys: list[str]) -> None:
         )
 
     compact = derived(
-        [store.data, context_node],
+        [store.entries, context_node],
         lambda deps, _a: _pack_compact(
-            deps[0].value if hasattr(deps[0], "value") else deps[0],
+            deps[0],
             deps[1],
             score,
             cost,
             budget,
         ),
     )
     size = derived(
-        [store.data],
-        lambda deps, _a: len(deps[0].value if hasattr(deps[0], "value") else deps[0]),
+        [store.entries],
+        lambda deps, _a: len(deps[0]) if deps[0] is not None else 0,
         initial=0,
     )
     compact.subscribe(lambda _msgs: None)
@@ -260,8 +258,9 @@ def _compute_evictions(
     evict: Callable[[str, Any], Any],
 ) -> list[str]:
     out: list[str] = []
-    current = get(store.data)
-    snapshot = current.value if hasattr(current, "value") else current
+    snapshot = get(store.entries)
+    if snapshot is None:
+        return out
     for key, mem in snapshot.items():
         verdict = evict(key, mem)
         if isinstance(verdict, Node):