feat: fix stratify

Author: clfhhc

docs/optimizations.md

@@ -155,6 +155,8 @@ Union-find over node identity merges components when nodes list dependencies at
 - **Feedback effect perpetually dirty status (Phase 8.1/8.2, noted 2026-04-06 QA — documented, accepted):** The `__feedback_effect_<condition>` node forwards DIRTY via default dispatch but consumes DATA. The effect stays in "dirty" status between emissions. **By design** — the status is technically correct. Effects don't need to settle. These nodes now carry `meta._internal: True` and can be filtered from `describe()` output when undesired.
 - **Reduction primitives cross-language parity (Phase 8.1, noted 2026-04-06, QA 2026-04-06, updated 8.2 2026-04-06):** Both TS and PY implement `stratify`, `funnel`, `feedback`, `budget_gate`/`budgetGate`, and `scorer` in `patterns/reduction`. All five follow the orchestration factory pattern (`_base_meta`, `_register_step`). Key alignment: (1) `stratify` buffers DIRTY until DATA arrives — on classifier miss, emits `[DIRTY, RESOLVED]` to preserve spec §1.3.1 (both). (2) `funnel` now uses `bridge()` from `core/bridge` for inter-stage wiring — graph-visible effect nodes (`__bridge_<from>→<stage>_input`) replace the old subscribe-based bridges. Visible in `describe()`, participates in two-phase push. (3) `feedback` counter node is source of truth (resettable via `graph.set()`); feedback wiring now uses a graph-registered effect node (`__feedback_effect_<condition>`) instead of raw subscribe. Counter name is `__feedback_<condition>` to support multiple loops per graph. (4) `budget_gate`/`budgetGate` force-flushes all buffered items on terminal regardless of budget; sends RESUME before terminal if paused; forwards constraint ERROR downstream, silences constraint COMPLETE, forwards unknown constraint types via default. (5) `scorer` coerces `None`/`undefined` to 0 before multiplication (no TypeError/NaN divergence). TS `ScoredItem` is a plain object; PY `ScoredItem` is a class with `__slots__` and `__eq__`. Meta keys: `reduction: True`, `reduction_type: "<name>"`. Both repos: 22 tests each.
 
+- **`stratify` two-dep gating (Phase 8.1/9.0, noted roadmap, resolved 2026-04-06, QA 2026-04-06):** `stratify` `on_message` previously intercepted only source messages (dep 0) and let rules messages (dep 1) fall through to default bitmask handling. When both source and rules updated in the same `batch()`, source DATA was classified against stale rules (rules DATA hadn't drained yet), and a spurious second emission occurred from default settlement of dep 1. Fix: `on_message` now intercepts **all** messages from both deps, implementing manual two-dep gating (same diamond-resolution pattern as `budget_gate`). Classification deferred until all dirty deps settle. Rules-only changes produce no downstream emission ("future items only"). QA hardening: (1) PAUSE/RESUME/INVALIDATE from rules dep swallowed (internal impl detail, not leaked downstream). (2) `classify` exceptions caught and treated as non-match (branch not killed). (3) `complete_when_deps_complete=False` explicit (branch lifecycle driven by source terminal, not auto-complete from both deps). Both TS and PY.
+
 - **Cleanup wrapper `{"cleanup": fn, "value"?: v}` (all phases, noted 2026-04-06, resolved 2026-04-06):** Node fn can now return `{"cleanup": callable, "value": v}` to explicitly separate cleanup from data. When returned: `cleanup` is registered; if `"value"` key is present, it's emitted as data. Plain callable returns remain cleanup for backward compat. Exported as `CleanupResult<T>` (TS) / documented dict pattern (PY). Resolves the documented limitation where returning a callable as data was silently consumed as cleanup.
 
 ---

docs/roadmap.md

@@ -26,7 +26,7 @@ These fix protocol violations identified post-8.2 ship. Must land before §9.2 a
 
 - [x] Rearchitect `feedback()` as graph-visible bridge node (8.2 → 9.0) — replaces subscribe-based shortcut; enables proper DIRTY→DATA two-phase on reentry/counter; resolves bare-DATA protocol gap
 - [x] Rearchitect `funnel()` bridges as graph-visible nodes (8.2 → 9.0) — replaces subscribe forwarding; resolves §5.9 imperative trigger violation + teardown leak
-- [ ] `stratify` two-dep gating (8.2 → 9.0) — gate classification on both source and rules settling (eliminates stale-rules race)
+- [x] `stratify` two-dep gating (8.2 → 9.0) — gate classification on both source and rules settling (eliminates stale-rules race)
 
 ### Wave 2 (Python): Audit & accountability parity (Weeks 4-9)
 
@@ -555,7 +555,7 @@ Pre-wired graphs for common "info → action" domains. Users fork/extend.
 - [x] `data_quality_graph(opts)` → Graph — DB/API ingest → schema validation → anomaly detection → drift alerting → remediation suggestions
 - [x] Rearchitect `feedback()` as graph-visible bridge node (replaces subscribe-based shortcut; enables proper DIRTY→DATA two-phase on reentry/counter; resolves bare-DATA protocol gap)
 - [x] Rearchitect `funnel()` bridges as graph-visible nodes (replaces subscribe forwarding; resolves §5.9 imperative trigger violation + teardown leak)
-- [ ] `stratify` two-dep gating: gate classification on both source and rules settling (eliminates stale-rules race when both updated in same `batch()`)
+- [x] `stratify` two-dep gating: gate classification on both source and rules settling (eliminates stale-rules race when both updated in same `batch()`)
 
 ### 8.3 — LLM graph composition
 

src/graphrefly/core/__init__.py

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

src/graphrefly/core/node.py

@@ -106,13 +106,25 @@ def _status_after_message(status: NodeStatus, msg: Message) -> NodeStatus:
 type Message = tuple[Any, Any] | tuple[Any]
 
 
+_CLEANUP_RESULT = "__graphrefly_cleanup_result__"
+"""Branded key that marks a :func:`cleanup_result` wrapper — prevents duck-type
+collisions with domain dicts that happen to have a ``cleanup`` key."""
+
+
+def cleanup_result(cleanup: Callable[[], None], value: Any = _SENTINEL) -> dict[str, Any]:
+    """Create a branded cleanup-result wrapper.
+
+    >>> node([dep], lambda vals, _: cleanup_result(release, computed))
+    """
+    r: dict[str, Any] = {_CLEANUP_RESULT: True, "cleanup": cleanup}
+    if value is not _SENTINEL:
+        r["value"] = value
+    return r
+
+
 def _is_cleanup_result(value: object) -> bool:
-    """Check for explicit cleanup wrapper: ``{"cleanup": fn}`` or ``{"cleanup": fn, "value": v}``."""
-    return (
-        isinstance(value, dict)
-        and "cleanup" in value
-        and callable(value["cleanup"])
-    )
+    """Check for branded cleanup wrapper created by :func:`cleanup_result`."""
+    return isinstance(value, dict) and value.get(_CLEANUP_RESULT) is True
 
 
 def _is_cleanup_fn(value: object) -> bool:

src/graphrefly/patterns/reduction.py

@@ -151,55 +151,105 @@ def _add_branch(
 ) -> None:
     """Add a stratify branch to the graph.
 
+    Two-dep gating: intercepts messages from **both** source (dep 0) and rules
+    (dep 1).  Classification is deferred until all dirty deps have settled,
+    eliminating the stale-rules race when both are updated in the same
+    ``batch()``.
+
     Protocol: DIRTY is buffered until DATA arrives. If the classifier matches,
     emit [DIRTY, DATA]. If not, emit [DIRTY, RESOLVED] so downstream exits
     dirty status cleanly (spec §1.3.1). Source RESOLVED forwards as RESOLVED.
+    Rules-only changes produce no downstream emission ("future items only").
     """
     branch_name = f"branch/{rule.name}"
-    pending_dirty = [False]
+    _no_value = object()
+
+    # Per-branch two-dep gating state
+    _source_dirty = [False]
+    _rules_dirty = [False]
+    _source_phase2 = [False]  # source delivered DATA or RESOLVED this cycle
+    _source_value: list[Any] = [_no_value]  # DATA payload, or _no_value for RESOLVED
+    _pending_dirty = [False]  # owe downstream a DIRTY
+
+    def _resolve(actions: Any) -> None:
+        if _source_phase2[0]:
+            _source_phase2[0] = False
+            value = _source_value[0]
+            _source_value[0] = _no_value
+            if value is not _no_value:
+                # Source emitted DATA — classify with settled rules
+                current_rules: list[StratifyRule] = rules_node.get() or []
+                current_rule = next((r for r in current_rules if r.name == rule.name), None)
+                try:
+                    matches = current_rule is not None and current_rule.classify(value)
+                except Exception:
+                    matches = False
+                if matches:
+                    _pending_dirty[0] = False
+                    actions.emit(value)
+                else:
+                    if _pending_dirty[0]:
+                        _pending_dirty[0] = False
+                        actions.down([(MessageType.DIRTY,), (MessageType.RESOLVED,)])
+            else:
+                # Source RESOLVED (unchanged)
+                if _pending_dirty[0]:
+                    _pending_dirty[0] = False
+                    actions.down([(MessageType.DIRTY,), (MessageType.RESOLVED,)])
+                else:
+                    actions.down([(MessageType.RESOLVED,)])
+        # else: rules-only change — no reclassification ("future items only")
 
     def on_message(msg: Any, dep_index: int, actions: Any) -> bool:
-        # Only intercept source messages (dep 0)
-        if dep_index != 0:
-            return False
-
         t = msg[0]
-        if t is MessageType.DATA:
-            value = msg[1]
-            current_rules: list[StratifyRule] = rules_node.get() or []
-            current_rule = next((r for r in current_rules if r.name == rule.name), None)
-            if current_rule is not None and current_rule.classify(value):
-                # Match: emit (actions.emit handles DIRTY+DATA wrapping)
-                pending_dirty[0] = False
-                actions.emit(value)
-            else:
-                # No match: emit DIRTY + RESOLVED so downstream exits dirty
-                if pending_dirty[0]:
-                    pending_dirty[0] = False
-                    actions.down([(MessageType.DIRTY,), (MessageType.RESOLVED,)])
-            return True
+
+        # --- DIRTY (phase 1) ---
         if t is MessageType.DIRTY:
-            pending_dirty[0] = True
+            if dep_index == 0:
+                _source_dirty[0] = True
+                _pending_dirty[0] = True
+            else:
+                _rules_dirty[0] = True
             return True
-        if t is MessageType.RESOLVED:
-            # Source unchanged — forward RESOLVED (with buffered DIRTY if any)
-            if pending_dirty[0]:
-                pending_dirty[0] = False
-                actions.down([(MessageType.DIRTY,), (MessageType.RESOLVED,)])
+
+        # --- Phase 2 (DATA / RESOLVED) ---
+        if t is MessageType.DATA or t is MessageType.RESOLVED:
+            if dep_index == 0:
+                _source_dirty[0] = False
+                _source_phase2[0] = True
+                _source_value[0] = msg[1] if t is MessageType.DATA else _no_value
             else:
-                actions.down([(MessageType.RESOLVED,)])
+                _rules_dirty[0] = False
+
+            # Wait for all dirty deps to settle
+            if _source_dirty[0] or _rules_dirty[0]:
+                return True
+
+            _resolve(actions)
             return True
+
+        # --- Terminal ---
         if t in (MessageType.COMPLETE, MessageType.ERROR, MessageType.TEARDOWN):
-            pending_dirty[0] = False
-            actions.down([msg])
+            _source_dirty[0] = False
+            _rules_dirty[0] = False
+            _source_phase2[0] = False
+            _source_value[0] = _no_value
+            _pending_dirty[0] = False
+            if dep_index == 0:
+                actions.down([msg])
+            # Rules terminal: swallow (branch stays alive)
             return True
-        return False
+
+        # Swallow PAUSE/RESUME/INVALIDATE from rules dep (internal impl detail);
+        # forward unknown source messages via default handling.
+        return dep_index == 1
 
     filter_node = node(
         [source, rules_node],
         lambda _d, _a: None,
         on_message=on_message,
         describe_kind="operator",
+        complete_when_deps_complete=False,
         meta=_base_meta("stratify_branch", {"branch": rule.name}),
     )
 

tests/test_reduction.py

@@ -112,6 +112,104 @@ def _sink(msgs: list) -> None:
         source.down([(MessageType.COMPLETE,)])
         assert completed[0] is True
 
+    def test_two_dep_gating_batch_source_and_rules(self) -> None:
+        """Stale-rules race: both source and rules update in same batch."""
+        source = state("x")
+        rules = [StratifyRule("match", classify=lambda v: v == "a")]
+
+        g = stratify("gate", source, rules)
+
+        seen: list[str] = []
+
+        def _sink(msgs: list) -> None:
+            for msg in msgs:
+                if msg[0] is MessageType.DATA:
+                    seen.append(msg[1])
+
+        g.resolve("branch/match").subscribe(_sink)
+
+        # Batch: update rules to match "b" AND send source "b" simultaneously.
+        # Without two-dep gating, classification would use old rules (match "a")
+        # and "b" would be dropped.  With gating, both settle first.
+        from graphrefly.core.protocol import batch
+
+        with batch():
+            g.set("rules", [StratifyRule("match", classify=lambda v: v == "b")])
+            source.down([(MessageType.DATA, "b")])
+
+        assert seen == ["b"]
+
+    def test_source_only_update_still_works(self) -> None:
+        """Regression: source-only updates must still classify correctly."""
+        source = state(0)
+        rules = [StratifyRule("pos", classify=lambda v: v > 0)]
+
+        g = stratify("src-only", source, rules)
+        seen: list[int] = []
+
+        def _sink(msgs: list) -> None:
+            for msg in msgs:
+                if msg[0] is MessageType.DATA:
+                    seen.append(msg[1])
+
+        g.resolve("branch/pos").subscribe(_sink)
+
+        source.down([(MessageType.DATA, 5)])
+        source.down([(MessageType.DATA, -1)])
+        source.down([(MessageType.DATA, 3)])
+        assert seen == [5, 3]
+
+    def test_rules_only_update_no_downstream(self) -> None:
+        """Rules-only change should not trigger reclassification."""
+        source = state("a")
+        rules = [StratifyRule("match", classify=lambda v: v == "a")]
+
+        g = stratify("rules-only", source, rules)
+
+        emissions: list[object] = []
+
+        def _sink(msgs: list) -> None:
+            for msg in msgs:
+                emissions.append(msg)
+
+        g.resolve("branch/match").subscribe(_sink)
+
+        # Initial source DATA
+        source.down([(MessageType.DATA, "a")])
+        initial_count = len(emissions)
+
+        # Change rules only — should NOT produce any new downstream messages
+        g.set("rules", [StratifyRule("match", classify=lambda v: v == "b")])
+        assert len(emissions) == initial_count
+
+    def test_both_settle_source_resolved(self) -> None:
+        """Source DIRTY→RESOLVED + rules DATA in same batch → RESOLVED downstream."""
+        source = state("x")
+        rules = [StratifyRule("match", classify=lambda v: v == "x")]
+
+        g = stratify("resolved", source, rules)
+
+        data_seen: list[object] = []
+        resolved_count = [0]
+
+        def _sink(msgs: list) -> None:
+            for msg in msgs:
+                if msg[0] is MessageType.DATA:
+                    data_seen.append(msg[1])
+                elif msg[0] is MessageType.RESOLVED:
+                    resolved_count[0] += 1
+
+        g.resolve("branch/match").subscribe(_sink)
+
+        # Initial emission
+        source.down([(MessageType.DATA, "x")])
+        assert data_seen == ["x"]
+
+        # Now update rules in isolation — source not involved
+        g.set("rules", [StratifyRule("match", classify=lambda v: v == "y")])
+        # No new data should appear
+        assert data_seen == ["x"]
+
 
 # ---------------------------------------------------------------------------
 # funnel