feat: reusable harness patterns

Author: clfhhc

src/graphrefly/extra/__init__.py

@@ -135,6 +135,7 @@
     with_status,
 )
 from graphrefly.extra.sources import (
+    ReactiveCounterBundle,
     cached,
     empty,
     first_value_from,
@@ -146,8 +147,10 @@
     from_cron,
     from_iter,
     from_timer,
+    keepalive,
     never,
     of,
+    reactive_counter,
     replay,
     share,
     share_replay,
@@ -281,9 +284,12 @@
     "from_webhook",
     "from_iter",
     "from_timer",
+    "keepalive",
     "never",
     "of",
     "parse_prometheus_text",
+    "ReactiveCounterBundle",
+    "reactive_counter",
     "parse_statsd",
     "parse_syslog",
     "pubsub",

src/graphrefly/extra/sources.py

@@ -17,6 +17,7 @@
 
 from graphrefly.core.node import NO_VALUE, Node, NodeActions, node
 from graphrefly.core.protocol import Messages, MessageType
+from graphrefly.core.sugar import state
 
 
 def _source_initial_kwargs(source: Node[Any]) -> dict[str, Any]:
@@ -850,12 +851,92 @@ def on_msg(msg: tuple[Any, ...], _dep_index: int, actions: NodeActions) -> bool:
     "from_cron",
     "from_iter",
     "from_timer",
+    "keepalive",
     "never",
     "of",
+    "ReactiveCounterBundle",
+    "reactive_counter",
     "replay",
     "share",
     "share_replay",
     "throw_error",
     "to_array",
     "to_list",
 ]
+
+
+# ---------------------------------------------------------------------------
+# keepalive
+# ---------------------------------------------------------------------------
+
+
+def keepalive(n: Node[Any]) -> Any:
+    """Activate a compute node's upstream wiring without a real sink.
+
+    Derived/effect nodes are lazy — they don't compute until at least one
+    subscriber exists (COMPOSITION-GUIDE §5). ``keepalive`` subscribes with
+    an empty sink so the node stays wired for ``.get()`` and upstream
+    propagation.
+
+    Returns the unsubscribe handle.  Common usage::
+
+        graph.add_disposer(keepalive(node))
+    """
+    return n.subscribe(lambda _msgs: None)
+
+
+# ---------------------------------------------------------------------------
+# reactive_counter
+# ---------------------------------------------------------------------------
+
+
+class ReactiveCounterBundle:
+    """Typed bundle returned by :func:`reactive_counter`.
+
+    Attributes mirror the TS ``ReactiveCounterBundle`` type for cross-language parity.
+    """
+
+    __slots__ = ("_node", "_cap")
+
+    def __init__(self, counter: Node[Any], cap: int) -> None:
+        self._node = counter
+        self._cap = cap
+
+    @property
+    def node(self) -> Node[Any]:
+        """Reactive node holding the current count."""
+        return self._node
+
+    def increment(self) -> bool:
+        """Increment by 1. Returns ``False`` if cap would be exceeded."""
+        current = self._node.get()
+        if current is None:
+            current = 0
+        if current >= self._cap:
+            return False
+        self._node.down([(MessageType.DIRTY,), (MessageType.DATA, current + 1)])
+        return True
+
+    def get(self) -> int:
+        """Current count (synchronous read)."""
+        v = self._node.get()
+        return v if v is not None else 0
+
+    def at_cap(self) -> bool:
+        """Whether the counter has reached its cap."""
+        v = self._node.get()
+        return (v if v is not None else 0) >= self._cap
+
+
+def reactive_counter(cap: int) -> ReactiveCounterBundle:
+    """Reactive counter with a cap — the building block for circuit breakers.
+
+    Wraps a ``state(0)`` node with ``increment()`` that respects a maximum.
+    The ``node`` is subscribable and composable like any reactive node. When
+    the cap is reached, ``increment()`` returns ``False``.
+
+    Returns a :class:`ReactiveCounterBundle` with ``node``, ``increment``,
+    ``get``, and ``at_cap`` members.
+    """
+    counter = state(0)
+    return ReactiveCounterBundle(counter, cap)

src/graphrefly/patterns/_internal.py

@@ -0,0 +1,64 @@
+"""Shared internal utilities for the patterns layer.
+
+These are private helpers used across multiple pattern modules. They are NOT
+part of the public API.
+
+General-purpose reactive utilities (``keepalive``, ``reactive_counter``) live
+in ``extra.sources`` and are re-exported here for convenience.
+
+.. note:: Internal module — do not import from user code.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# Re-export general-purpose utilities from extra (canonical home).
+from graphrefly.extra.sources import keepalive, reactive_counter
+
+
+def domain_meta(
+    domain: str,
+    kind: str,
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build a domain metadata dict for pattern-layer nodes.
+
+    Each domain (orchestration, messaging, reduction, ai, cqrs,
+    domain_template) follows the same shape::
+
+        { "<domain>": True, "<domain>_type": "<kind>", ...extra }
+    """
+    out: dict[str, Any] = {domain: True, f"{domain}_type": kind}
+    if extra is not None:
+        out.update(extra)
+    return out
+
+
+def tracking_key(item: Any) -> str:
+    """Stable tracking key for an item with retry/reingestion decoration.
+
+    Uses ``related_to[0]`` if present (carries the original key forward
+    through retries and reingestions). Falls back to ``summary`` for
+    first-time items.
+    """
+    related = (
+        item.get("related_to") if isinstance(item, dict) else getattr(item, "related_to", None)
+    )
+    if related:
+        first = related[0] if isinstance(related, (list, tuple)) else None
+        if first is not None:
+            return str(first)
+    if isinstance(item, dict):
+        summary = item.get("summary", str(item))
+    else:
+        summary = getattr(item, "summary", str(item))
+    return str(summary)
+
+
+__all__ = [
+    "domain_meta",
+    "keepalive",
+    "reactive_counter",
+    "tracking_key",
+]

src/graphrefly/patterns/ai.py

@@ -26,6 +26,8 @@
 from graphrefly.extra.sources import first_value_from, from_any, from_timer
 from graphrefly.extra.tier2 import switch_map
 from graphrefly.graph.graph import Graph
+from graphrefly.patterns._internal import domain_meta
+from graphrefly.patterns._internal import keepalive as _keepalive
 from graphrefly.patterns.memory import (
     KnowledgeGraph,
     VectorIndex,
@@ -181,15 +183,7 @@ class MemoryTiers:
 
 
 def _ai_meta(kind: str, extra: dict[str, Any] | None = None) -> dict[str, Any]:
-    out: dict[str, Any] = {"ai": True, "ai_type": kind}
-    if extra:
-        out.update(extra)
-    return out
-
-
-def _keepalive(n: Any) -> Any:
-    """Subscribe to keep derived node wired; returns unsubscribe handle."""
-    return n.subscribe(lambda _msgs: None)
+    return domain_meta("ai", kind, extra)
 
 
 _DEFAULT_TIMEOUT = 30.0  # seconds

src/graphrefly/patterns/cqrs.py

@@ -23,6 +23,8 @@
 from graphrefly.core.sugar import derived, state
 from graphrefly.extra.data_structures import reactive_log
 from graphrefly.graph.graph import Graph
+from graphrefly.patterns._internal import domain_meta
+from graphrefly.patterns._internal import keepalive as _keepalive_raw
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Sequence
@@ -67,10 +69,7 @@ def _build_event_guard(allow: Any, deny: Any) -> None:
 
 
 def _cqrs_meta(kind: str, extra: dict[str, Any] | None = None) -> dict[str, Any]:
-    out: dict[str, Any] = {"cqrs": True, "cqrs_type": kind}
-    if extra:
-        out.update(extra)
-    return out
+    return domain_meta("cqrs", kind, extra)
 
 
 @dataclass(slots=True)
@@ -83,7 +82,7 @@ class _EventLogEntry:
 
 def _keepalive(n: Any) -> Callable[[], None]:
     """Keep dep wiring alive; returns unsubscribe handle for cleanup."""
-    return cast("Callable[[], None]", n.subscribe(lambda _msgs: None))
+    return cast("Callable[[], None]", _keepalive_raw(n))
 
 
 def _tuple_snapshot(raw: Any) -> tuple[Any, ...]:

src/graphrefly/patterns/domain_templates.py

@@ -24,6 +24,7 @@
 from graphrefly.core.sugar import derived, effect, state
 from graphrefly.extra.data_structures import reactive_log
 from graphrefly.graph.graph import Graph
+from graphrefly.patterns._internal import domain_meta
 from graphrefly.patterns.reduction import (
     StratifyRule,
     feedback,
@@ -37,10 +38,7 @@
 
 
 def _base_meta(kind: str, extra: dict[str, Any] | None = None) -> dict[str, Any]:
-    out: dict[str, Any] = {"domain_template": True, "template_type": kind}
-    if extra:
-        out.update(extra)
-    return out
+    return domain_meta("domain_template", kind, extra)
 
 
 def _is_tagged(value: Any, tag: str) -> bool:
@@ -50,10 +48,6 @@ def _is_tagged(value: Any, tag: str) -> bool:
     return value.get("type") == tag or value.get("kind") == tag
 
 
-def _keepalive(n: NodeImpl[Any]) -> Any:
-    return n.subscribe(lambda _msgs: None)
-
-
 # ---------------------------------------------------------------------------
 # 1. observability_graph
 # ---------------------------------------------------------------------------

src/graphrefly/patterns/harness/__init__.py

@@ -9,6 +9,7 @@
     EvalJudgeScore,
     EvalResult,
     EvalTaskResult,
+    create_intake_bridge,
     eval_intake_bridge,
 )
 from graphrefly.patterns.harness.loop import HarnessGraph, harness_loop
@@ -76,6 +77,7 @@
     "EvalJudgeScore",
     "EvalResult",
     "EvalTaskResult",
+    "create_intake_bridge",
     "eval_intake_bridge",
     # loop
     "HarnessGraph",

src/graphrefly/patterns/harness/bridge.py

@@ -17,6 +17,49 @@
 
 from .types import IntakeItem, Severity
 
+# ---------------------------------------------------------------------------
+# Generic intake bridge
+# ---------------------------------------------------------------------------
+
+
+def create_intake_bridge(
+    source: NodeImpl[Any],
+    intake_topic: TopicGraph,
+    parser: Any,
+    *,
+    name: str | None = None,
+) -> NodeImpl[Any]:
+    """Generic source→intake bridge factory.
+
+    Watches a source node for new values, passes each through a user-supplied
+    ``parser`` that produces zero or more ``IntakeItem`` instances, and publishes
+    them to the given intake topic.
+
+    This is the generalized pattern behind :func:`eval_intake_bridge`. Use it
+    for CI results, test failures, Slack messages, monitoring alerts, or any
+    domain where structured results should flow into a harness loop.
+
+    Args:
+        source: Reactive node emitting domain-specific data.
+        intake_topic: TopicGraph to publish IntakeItem entries to.
+        parser: ``(value: T) -> list[IntakeItem]``. Return empty list to skip.
+        name: Optional name for the effect node.
+
+    Returns:
+        The effect node (for lifecycle management).
+    """
+
+    def _bridge(deps: list[Any], _actions: Any) -> None:
+        value = deps[0]
+        if value is None:
+            return
+        items = parser(value)
+        for item in items:
+            intake_topic.publish(item)
+
+    return effect([source], _bridge, name=name or "intake-bridge")
+
+
 # ---------------------------------------------------------------------------
 # Generic eval result shape
 # ---------------------------------------------------------------------------

src/graphrefly/patterns/harness/loop.py

@@ -15,6 +15,7 @@
 from graphrefly.core.sugar import effect, state
 from graphrefly.extra.tier1 import merge, with_latest_from
 from graphrefly.graph.graph import Graph
+from graphrefly.patterns._internal import tracking_key as _tracking_key
 from graphrefly.patterns.ai import prompt_node
 from graphrefly.patterns.messaging import TopicGraph
 from graphrefly.patterns.orchestration import gate
@@ -36,26 +37,6 @@
     from collections.abc import Callable
 
 
-def _tracking_key(item: Any) -> str:
-    """Stable tracking key for an item.
-
-    Uses ``related_to[0]`` if the item is already a retry or reingestion
-    (carries the original key forward). Falls back to the raw summary.
-    """
-    related = (
-        item.get("related_to") if isinstance(item, dict) else getattr(item, "related_to", None)
-    )
-    if related:
-        first = related[0] if isinstance(related, (list, tuple)) else None
-        if first:
-            return str(first)
-    if isinstance(item, dict):
-        summary = item.get("summary", str(item))
-    else:
-        summary = getattr(item, "summary", str(item))
-    return str(summary)
-
-
 # ---------------------------------------------------------------------------
 # Default prompts
 # ---------------------------------------------------------------------------