feat(events): add graph event filters

.gitignore

@@ -16,6 +16,7 @@ dist/
 
 # Plugin Daemon Slim
 .slim
+
 # Unit test / coverage reports
 htmlcov/
 .tox/
@@ -30,3 +31,7 @@ coverage.xml
 .hypothesis/
 .pytest_cache/
 cover/
+
+# IDE configs
+.idea/
+.vscode/

examples/slim_llm/README.md

@@ -9,7 +9,7 @@ start -> llm -> answer
 ## Files
 
 - `graph.yml`: the DSL graph
-- `dsl.py`: imports `graph.yml` with `graphon.dsl.loads()`
+- `dsl.py`: imports `graph.yml` and streams response events with `ResponseStreamFilter`
 - `code.py`: builds the graph with Python code
 - `settings.py`: shared credentials and Slim setup
 - `credentials.example.json`: credentials template

examples/slim_llm/dsl.py

@@ -2,8 +2,8 @@
 
 import argparse
 import sys
+from collections.abc import Callable
 from pathlib import Path
-from typing import Any
 
 if __package__ in {None, ""}:
     sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
@@ -15,10 +15,20 @@
     use_local_slim_binary,
 )
 from graphon.dsl import loads
+from graphon.filters import (
+    GraphEventFilterContext,
+    ResponseStreamFilter,
+    filter_graph_events,
+)
 from graphon.graph_events.graph import GraphRunSucceededEvent
+from graphon.graph_events.node import NodeRunStreamChunkEvent
 
 
-def run(query: str) -> str:
+def run(
+    query: str,
+    *,
+    on_stream_chunk: Callable[[str], None] | None = None,
+) -> str:
     use_local_slim_binary()
     engine = loads(
         GRAPH_FILE.read_text(encoding="utf-8"),
@@ -27,8 +37,19 @@ def run(query: str) -> str:
         start_inputs={"query": query},
     )
 
-    events = list(engine.run())
-    final_event: Any = events[-1] if events else None
+    events = filter_graph_events(
+        engine.run(),
+        context=GraphEventFilterContext.from_engine(engine),
+        filters=[ResponseStreamFilter()],
+    )
+    final_event: GraphRunSucceededEvent | None = None
+    for event in events:
+        if isinstance(event, NodeRunStreamChunkEvent):
+            if on_stream_chunk is not None:
+                on_stream_chunk(event.chunk)
+        elif isinstance(event, GraphRunSucceededEvent):
+            final_event = event
+
     if not isinstance(final_event, GraphRunSucceededEvent):
         msg = f"Workflow did not succeed: {type(final_event).__name__}"
         raise TypeError(msg)
@@ -45,7 +66,23 @@ def main() -> int:
     parser.add_argument("query", nargs="?", default=DEFAULT_QUERY)
     args = parser.parse_args()
 
-    sys.stdout.write(f"{run(args.query)}\n")
+    saw_stream = False
+    last_chunk = ""
+
+    def write_stream_chunk(chunk: str) -> None:
+        nonlocal last_chunk, saw_stream
+        saw_stream = True
+        last_chunk = chunk
+        sys.stdout.write(chunk)
+        sys.stdout.flush()
+
+    answer = run(args.query, on_stream_chunk=write_stream_chunk)
+    if saw_stream:
+        if not last_chunk.endswith("\n"):
+            sys.stdout.write("\n")
+    else:
+        sys.stdout.write(f"{answer}\n")
+
     return 0
 
 

examples/slim_llm/graph.yml

@@ -16,7 +16,7 @@ graph:
       title: OpenAI
       model:
         provider: langgenius/openai/openai
-        name: gpt-5.4
+        name: gpt-4o-mini
         mode: chat
         completion_params: {}
       prompt_template:

examples/slim_llm/settings.py

@@ -18,7 +18,7 @@
     "592c8252795b5f75807de2d609a03196ed02596b409f7642b4a07548c7ff57ef"
 )
 OPENAI_PROVIDER = "openai"
-OPENAI_MODEL = "gpt-5.4"
+OPENAI_MODEL = "gpt-4o-mini"
 DEFAULT_QUERY = "Reply with only the word Graphon."
 
 

src/graphon/filters/__init__.py

@@ -0,0 +1,15 @@
+from graphon.graph_engine.filters import (
+    GraphEventFilter,
+    GraphEventFilterContext,
+    ResponseStreamFilter,
+    ResumableGraphEventFilter,
+    filter_graph_events,
+)
+
+__all__ = [
+    "GraphEventFilter",
+    "GraphEventFilterContext",
+    "ResponseStreamFilter",
+    "ResumableGraphEventFilter",
+    "filter_graph_events",
+]

src/graphon/graph_engine/__init__.py

@@ -1,4 +1,19 @@
 from .config import GraphEngineConfig
+from .filters import (
+    GraphEventFilter,
+    GraphEventFilterContext,
+    ResponseStreamFilter,
+    ResumableGraphEventFilter,
+    filter_graph_events,
+)
 from .graph_engine import GraphEngine
 
-__all__ = ["GraphEngine", "GraphEngineConfig"]
+__all__ = [
+    "GraphEngine",
+    "GraphEngineConfig",
+    "GraphEventFilter",
+    "GraphEventFilterContext",
+    "ResponseStreamFilter",
+    "ResumableGraphEventFilter",
+    "filter_graph_events",
+]

src/graphon/graph_engine/event_management/event_handlers.py

@@ -36,7 +36,6 @@
 from graphon.runtime.graph_runtime_state import (
     GraphExecutionProtocol,
     GraphRuntimeState,
-    ResponseStreamCoordinatorProtocol,
 )
 
 from ..error_handler import ErrorHandler
@@ -60,7 +59,6 @@ def __init__(
         graph: Graph,
         graph_runtime_state: GraphRuntimeState,
         graph_execution: GraphExecutionProtocol,
-        response_coordinator: ResponseStreamCoordinatorProtocol,
         event_collector: EventManager,
         edge_processor: EdgeProcessor,
         state_manager: GraphStateManager,
@@ -72,7 +70,6 @@ def __init__(
             graph: The workflow graph
             graph_runtime_state: Runtime state with variable pool
             graph_execution: Graph execution aggregate
-            response_coordinator: Response stream coordinator
             event_collector: Event manager for collecting events
             edge_processor: Edge processor for edge traversal
             state_manager: Unified state manager
@@ -82,7 +79,6 @@ def __init__(
         self._graph = graph
         self._graph_runtime_state = graph_runtime_state
         self._graph_execution = graph_execution
-        self._response_coordinator = response_coordinator
         self._event_collector = event_collector
         self._edge_processor = edge_processor
         self._state_manager = state_manager
@@ -144,9 +140,6 @@ def _(self, event: NodeRunStartedEvent) -> None:
         node_execution.mark_started(event.id)
         self._graph_runtime_state.increment_node_run_steps()
 
-        # Track in response coordinator for stream ordering
-        self._response_coordinator.track_node_execution(event.node_id, event.id)
-
         # Collect the event only for the first attempt; retries remain silent
         if is_initial_attempt:
             self._event_collector.collect(event)
@@ -159,12 +152,7 @@ def _(self, event: NodeRunStreamChunkEvent) -> None:
             event: The stream chunk event
 
         """
-        # Process with response coordinator
-        streaming_events = list(self._response_coordinator.intercept_event(event))
-
-        # Collect all events
-        for stream_event in streaming_events:
-            self._event_collector.collect(stream_event)
+        self._event_collector.collect(event)
 
     @_dispatch.register
     def _(self, event: NodeRunVariableUpdatedEvent) -> None:
@@ -201,27 +189,20 @@ def _(self, event: NodeRunSucceededEvent) -> None:
         # Store outputs in variable pool
         self._store_node_outputs(event.node_id, event.node_run_result.outputs)
 
-        # Forward to response coordinator and emit streaming events
-        streaming_events = self._response_coordinator.intercept_event(event)
-        for stream_event in streaming_events:
-            self._event_collector.collect(stream_event)
-
         # Process edges and get ready nodes
         node = self._graph.nodes[event.node_id]
         if node.execution_type == NodeExecutionType.BRANCH:
-            ready_nodes, edge_streaming_events = (
-                self._edge_processor.handle_branch_completion(
-                    event.node_id,
-                    event.node_run_result.edge_source_handle,
-                )
+            ready_nodes, edge_events = self._edge_processor.handle_branch_completion(
+                event.node_id,
+                event.node_run_result.edge_source_handle,
             )
         else:
-            ready_nodes, edge_streaming_events = (
-                self._edge_processor.process_node_success(event.node_id)
+            ready_nodes, edge_events = self._edge_processor.process_node_success(
+                event.node_id
             )
 
-        # Collect streaming events from edge processing
-        for edge_event in edge_streaming_events:
+        # Collect traversal events from edge processing
+        for edge_event in edge_events:
             self._event_collector.collect(edge_event)
 
         # Enqueue ready nodes
@@ -304,21 +285,19 @@ def _(self, event: NodeRunExceptionEvent) -> None:
         node = self._graph.nodes[event.node_id]
 
         if node.error_strategy == ErrorStrategy.DEFAULT_VALUE:
-            ready_nodes, edge_streaming_events = (
-                self._edge_processor.process_node_success(event.node_id)
+            ready_nodes, edge_events = self._edge_processor.process_node_success(
+                event.node_id
             )
         elif node.error_strategy == ErrorStrategy.FAIL_BRANCH:
-            ready_nodes, edge_streaming_events = (
-                self._edge_processor.handle_branch_completion(
-                    event.node_id,
-                    event.node_run_result.edge_source_handle,
-                )
+            ready_nodes, edge_events = self._edge_processor.handle_branch_completion(
+                event.node_id,
+                event.node_run_result.edge_source_handle,
             )
         else:
             msg = f"Unsupported error strategy: {node.error_strategy}"
             raise NotImplementedError(msg)
 
-        for edge_event in edge_streaming_events:
+        for edge_event in edge_events:
             self._event_collector.collect(edge_event)
 
         for node_id in ready_nodes:

src/graphon/graph_engine/filters/__init__.py

@@ -0,0 +1,15 @@
+from graphon.graph_engine.filters.base import (
+    GraphEventFilter,
+    GraphEventFilterContext,
+    ResumableGraphEventFilter,
+)
+from graphon.graph_engine.filters.chain import filter_graph_events
+from graphon.graph_engine.filters.response_stream import ResponseStreamFilter
+
+__all__ = [
+    "GraphEventFilter",
+    "GraphEventFilterContext",
+    "ResponseStreamFilter",
+    "ResumableGraphEventFilter",
+    "filter_graph_events",
+]

src/graphon/graph_engine/filters/base.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from abc import abstractmethod
+from collections.abc import Iterable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Protocol
+
+from graphon.graph.graph import Graph
+from graphon.graph_events.base import GraphEngineEvent
+from graphon.runtime.graph_runtime_state_protocol import ReadOnlyGraphRuntimeState
+from graphon.runtime.read_only_wrappers import ReadOnlyGraphRuntimeStateWrapper
+
+if TYPE_CHECKING:
+    from graphon.graph_engine.graph_engine import GraphEngine
+
+
+@dataclass(frozen=True)
+class GraphEventFilterContext:
+    """Run-scoped context available to graph event filters."""
+
+    graph: Graph
+    runtime_state: ReadOnlyGraphRuntimeState
+
+    @classmethod
+    def from_engine(cls, engine: GraphEngine) -> GraphEventFilterContext:
+        return cls(
+            graph=engine.graph,
+            runtime_state=ReadOnlyGraphRuntimeStateWrapper(
+                engine.graph_runtime_state,
+            ),
+        )
+
+
+class GraphEventFilter(Protocol):
+    """Event-to-event transform used outside GraphEngine execution."""
+
+    @property
+    @abstractmethod
+    def filter_id(self) -> str:
+        """Stable identifier for diagnostics and external state storage."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def initialize(self, context: GraphEventFilterContext) -> None:
+        """Bind run-scoped context before events are processed."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def on_event(self, event: GraphEngineEvent) -> Iterable[GraphEngineEvent]:
+        """Transform one input event into zero or more output events."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def flush(self) -> Iterable[GraphEngineEvent]:
+        """Emit buffered events after the upstream source is exhausted."""
+        raise NotImplementedError
+
+
+class ResumableGraphEventFilter(GraphEventFilter, Protocol):
+    """Optional filter protocol for output-layer resume state."""
+
+    @abstractmethod
+    def dumps(self) -> str:
+        """Serialize this filter's private state."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def loads(self, data: str) -> None:
+        """Restore this filter's private state."""
+        raise NotImplementedError