refactor(engine)!: simplify execution architecture

README.md

@@ -8,9 +8,9 @@ protocols, and a runnable end-to-end example.
 
 ## Highlights
 
-- Queue-based `GraphEngine` orchestration with event-driven execution
+- Queue-based `Engine` orchestration with event-driven execution
 - Graph parsing, validation, and fluent graph building
-- Shared runtime state, variable pool, and workflow execution domain models
+- Shared runtime state, variable pool, and workflow execution state
 - Built-in node implementations for common workflow patterns
 - DSL import support with Slim-backed LLM nodes
 - HTTP, file, tool, and human-input integration protocols
@@ -89,9 +89,10 @@ For the exact credential shape and runtime notes, see
 At a high level, direct Graphon usage looks like this:
 
 1. Build or load a graph and instantiate nodes into a `Graph`.
-2. Prepare `GraphRuntimeState` and seed the `VariablePool`.
+2. Prepare `RuntimeState` with the workflow ID and seed the `VariablePool`.
 3. Configure model, file, HTTP, tool, or human-input adapters as needed.
-4. Run `GraphEngine` and consume emitted graph events.
+4. Run `Engine` and consume emitted engine events; a local command channel is
+   created automatically unless an external channel is supplied.
 5. Read final outputs from runtime state.
 
 For Dify DSL documents, use `graphon.dsl.loads()` to build the engine from the
@@ -128,14 +129,15 @@ planned as a separate follow-up.
 ## Project Layout
 
 - `src/graphon/graph`: graph structures, parsing, validation, and builders
-- `src/graphon/graph_engine`: orchestration, workers, command channels, and
-  layers
+- `src/graphon/engine`: dispatch, workers, commands, events, and layers
 - `src/graphon/runtime`: runtime state, read-only wrappers, and variable pool
 - `src/graphon/nodes`: built-in workflow node implementations
 - `src/graphon/model_runtime`: provider/model abstractions and shared model
   entities
 - `src/graphon/dsl`: DSL import support, including Slim-backed runtime adapters
-- `src/graphon/graph_events`: event models emitted during execution
+- `src/graphon/node_events`: payloads emitted by node implementations before
+  execution context is attached
+- `src/graphon/engine_events`: complete events emitted by the engine
 - `src/graphon/http`: HTTP client abstractions and default implementation
 - `src/graphon/file`: workflow file models and file runtime helpers
 - `src/graphon/protocols`: public protocol re-exports for integrations
@@ -149,10 +151,10 @@ planned as a separate follow-up.
   runnable Slim LLM example setup
 - [src/graphon/model_runtime/README.md](src/graphon/model_runtime/README.md):
   model runtime overview
-- [src/graphon/graph_engine/layers/README.md](src/graphon/graph_engine/layers/README.md):
+- [src/graphon/engine/layer/README.md](src/graphon/engine/layer/README.md):
   engine layer extension points
-- [src/graphon/graph_engine/command_channels/README.md](src/graphon/graph_engine/command_channels/README.md):
-  local and distributed command channels
+- [src/graphon/engine/command/README.md](src/graphon/engine/command/README.md):
+  command processing and local or distributed channels
 
 ## Development
 

examples/slim_llm/code.py

@@ -20,12 +20,11 @@
     use_local_slim_binary,
 )
 from graphon.dsl.slim import SlimLLM
-from graphon.entities.graph_init_params import GraphInitParams
+from graphon.engine import Engine
+from graphon.entities.graph_init_params import InitParams
 from graphon.file.enums import FileType
 from graphon.file.models import File
 from graphon.graph.graph import Graph
-from graphon.graph_engine.command_channels import InMemoryChannel
-from graphon.graph_engine.graph_engine import GraphEngine
 from graphon.model_runtime.entities.llm_entities import LLMMode
 from graphon.model_runtime.entities.message_entities import (
     PromptMessage,
@@ -42,7 +41,7 @@
 from graphon.nodes.llm.entities import ContextConfig
 from graphon.nodes.start import StartNode
 from graphon.nodes.start.entities import StartNodeData
-from graphon.runtime.graph_runtime_state import GraphRuntimeState
+from graphon.runtime.graph_runtime_state import RuntimeState
 from graphon.runtime.variable_pool import VariablePool
 
 
@@ -79,9 +78,13 @@ def run(query: str) -> str:
     use_local_slim_binary()
     credentials = load_credentials()
     workflow_id = "slim-llm-code-example"
-    graph_state = GraphRuntimeState(variable_pool=VariablePool(), start_at=time.time())
+    graph_state = RuntimeState(
+        workflow_id=workflow_id,
+        variable_pool=VariablePool(),
+        start_at=time.time(),
+    )
     graph_state.variable_pool.add(("start", "query"), query)
-    graph_init = GraphInitParams(
+    graph_init = InitParams(
         workflow_id=workflow_id,
         graph_config={"nodes": [], "edges": []},
         run_context={},
@@ -100,11 +103,9 @@ def run(query: str) -> str:
             parameters={},
         ),
     )
-    engine = GraphEngine(
-        workflow_id=workflow_id,
+    engine = Engine(
         graph=graph,
         graph_runtime_state=graph_state,
-        command_channel=InMemoryChannel(),
     )
 
     list(engine.run())
@@ -117,8 +118,8 @@ def run(query: str) -> str:
 
 def build_graph(
     *,
-    graph_init: GraphInitParams,
-    graph_state: GraphRuntimeState,
+    graph_init: InitParams,
+    graph_state: RuntimeState,
     llm: SlimLLM,
 ) -> Graph:
     start = StartNode(

examples/slim_llm/dsl.py

@@ -15,13 +15,13 @@
     use_local_slim_binary,
 )
 from graphon.dsl import loads
-from graphon.filters import (
-    GraphEventFilterContext,
+from graphon.engine.filter import (
+    EngineEventFilterContext,
     ResponseStreamFilter,
-    filter_graph_events,
+    filter_engine_events,
 )
-from graphon.graph_events.graph import GraphRunSucceededEvent
-from graphon.graph_events.node import NodeRunStreamChunkEvent
+from graphon.engine_events.graph import GraphRunSucceededEvent
+from graphon.engine_events.node import NodeRunStreamChunkEvent
 
 
 def run(
@@ -37,9 +37,9 @@ def run(
         start_inputs={"query": query},
     )
 
-    events = filter_graph_events(
+    events = filter_engine_events(
         engine.run(),
-        context=GraphEventFilterContext.from_engine(engine),
+        context=EngineEventFilterContext.from_engine(engine),
         filters=[ResponseStreamFilter()],
     )
     final_event: GraphRunSucceededEvent | None = None

src/graphon/dsl/importer.py

@@ -10,15 +10,14 @@
 import yaml
 from pydantic import ValidationError
 
-from graphon.entities.graph_init_params import GraphInitParams
+from graphon.engine import Engine
+from graphon.engine.command import CommandChannel
+from graphon.engine.container_handler import ContainerHandlerFactory
+from graphon.entities.graph_init_params import InitParams
 from graphon.enums import BuiltinNodeTypes
 from graphon.graph.graph import Graph
 from graphon.graph.validation import GraphValidationError
-from graphon.graph_engine.command_channels import CommandChannel, InMemoryChannel
-from graphon.graph_engine.config import GraphEngineConfig
-from graphon.graph_engine.container_handlers import ContainerHandlerFactory
-from graphon.graph_engine.graph_engine import GraphEngine
-from graphon.runtime.graph_runtime_state import GraphRuntimeState
+from graphon.runtime.graph_runtime_state import RuntimeState
 from graphon.runtime.variable_pool import VariablePool
 
 from .entities import (
@@ -94,9 +93,9 @@ def loads(
     run_context: Mapping[str, Any] | None = None,
     start_inputs: Mapping[str, Any] | None = None,
     command_channel: CommandChannel | None = None,
-    config: GraphEngineConfig | None = None,
+    workers: int = 5,
     container_handler_factories: Sequence[ContainerHandlerFactory] = (),
-) -> GraphEngine:
+) -> Engine:
     plan = inspect(dsl, source_kind=source_kind)
     if plan.load_status == LoadStatus.UNSUPPORTED:
         raise _dsl_error(
@@ -128,18 +127,18 @@ def loads(
         run_context=run_context or {},
         start_inputs=start_inputs or {},
     )
-    graph_init_params = GraphInitParams(
+    graph_init_params = InitParams(
         workflow_id=workflow_id,
         graph_config=graph_config,
         run_context=run_context or {},
         call_depth=0,
     )
-    graph_runtime_state = GraphRuntimeState(
+    graph_runtime_state = RuntimeState(
         variable_pool=variable_pool,
         start_at=time.time(),
+        workflow_id=workflow_id,
     )
     parsed_credentials = _parse_credentials(credentials)
-    engine_config = config or GraphEngineConfig()
     node_factory = SlimDslNodeFactory(
         graph_config=graph_config,
         graph_init_params=graph_init_params,
@@ -170,12 +169,11 @@ def loads(
             kind=plan.document.kind,
         ) from error
 
-    return GraphEngine(
-        workflow_id=workflow_id,
+    return Engine(
         graph=graph,
         graph_runtime_state=graph_runtime_state,
-        command_channel=command_channel or InMemoryChannel(),
-        config=engine_config,
+        command_channel=command_channel,
+        workers=workers,
         container_handler_factories=container_handler_factories,
     )
 

src/graphon/dsl/node_factory.py

@@ -66,7 +66,7 @@
 from graphon.nodes.variable_assigner.v2.node import (
     VariableAssignerNode as VariableAssignerNodeV2,
 )
-from graphon.runtime.graph_runtime_state import GraphRuntimeState
+from graphon.runtime.graph_runtime_state import RuntimeState
 from graphon.template_rendering import Jinja2TemplateRenderer, TemplateRenderError
 
 from .code_runtime import SandboxCodeExecutor
@@ -418,7 +418,7 @@ class _NodeBuildRequest:
 class SlimDslNodeFactory:
     graph_config: Mapping[str, Any]
     graph_init_params: Any
-    graph_runtime_state: GraphRuntimeState
+    graph_runtime_state: RuntimeState
     credentials: DslCredentials
     dependencies: list[DslDependency]
     slim_client_config: SlimClientConfig = field(init=False)
@@ -442,7 +442,7 @@ def __post_init__(self) -> None:
 
     def with_runtime_state(
         self,
-        graph_runtime_state: GraphRuntimeState,
+        graph_runtime_state: RuntimeState,
     ) -> SlimDslNodeFactory:
         return replace(self, graph_runtime_state=graph_runtime_state)
 

src/graphon/engine/__init__.py

@@ -0,0 +1,3 @@
+from .engine import Engine
+
+__all__ = ["Engine"]

src/graphon/graph_engine/command_channels/README.md → src/graphon/engine/command/README.md

@@ -1,9 +1,17 @@
-# Command Channels
+# Commands
 
-Channel implementations for external workflow control.
+Command processing and channels for external workflow control.
+
+The supported command union contains `AbortCommand`, `PauseCommand`, and
+`UpdateVariablesCommand`. Their `command_type` literals are used to deserialize
+commands received from distributed channels.
 
 ## Components
 
+### CommandProcessor
+
+Polls a command channel and applies commands to the current graph execution.
+
 ### InMemoryChannel
 
 Thread-safe in-memory queue for single-process deployments.
@@ -21,9 +29,11 @@ Redis-based queue for distributed deployments.
 ## Usage
 
 ```python
+from graphon.engine.command import AbortCommand, InMemoryChannel, RedisChannel
+
 # Local execution
 channel = InMemoryChannel()
-channel.send_command(AbortCommand(graph_id="workflow-123"))
+channel.send_command(AbortCommand(reason="stop"))
 
 # Distributed execution
 redis_channel = RedisChannel(

src/graphon/engine/command/__init__.py

@@ -0,0 +1,23 @@
+"""Engine command communication and processing."""
+
+from .builtin.in_memory import InMemoryChannel
+from .builtin.redis import RedisChannel
+from .entities import (
+    AbortCommand,
+    Command,
+    PauseCommand,
+    UpdateVariablesCommand,
+)
+from .processor import CommandProcessor
+from .protocol import CommandChannel
+
+__all__ = [
+    "AbortCommand",
+    "Command",
+    "CommandChannel",
+    "CommandProcessor",
+    "InMemoryChannel",
+    "PauseCommand",
+    "RedisChannel",
+    "UpdateVariablesCommand",
+]

src/graphon/engine/command/builtin/__init__.py

@@ -0,0 +1 @@
+"""Implementation modules for the command channels exported by the parent package."""

src/graphon/graph_engine/command_channels/in_memory_channel.py → src/graphon/engine/command/builtin/in_memory.py

@@ -7,29 +7,29 @@
 from queue import Empty, Queue
 from typing import final
 
-from ..entities.commands import GraphEngineCommand
+from ..entities import Command
 
 
 @final
 class InMemoryChannel:
     """In-memory command channel implementation using a thread-safe queue.
 
-    Each instance is dedicated to a single GraphEngine/workflow execution.
+    Each instance is dedicated to a single Engine/workflow execution.
     Suitable for local development, testing, and single-instance deployments.
     """
 
     def __init__(self) -> None:
         """Initialize the in-memory channel with a single queue."""
-        self._queue: Queue[GraphEngineCommand] = Queue()
+        self._queue: Queue[Command] = Queue()
 
-    def fetch_commands(self) -> list[GraphEngineCommand]:
+    def fetch_commands(self) -> list[Command]:
         """Fetch all pending commands from the queue.
 
         Returns:
             List of pending commands (drains the queue)
 
         """
-        commands: list[GraphEngineCommand] = []
+        commands: list[Command] = []
 
         # Drain all available commands from the queue
         while not self._queue.empty():
@@ -41,7 +41,7 @@ def fetch_commands(self) -> list[GraphEngineCommand]:
 
         return commands
 
-    def send_command(self, command: GraphEngineCommand) -> None:
+    def send_command(self, command: Command) -> None:
         """Send a command to this channel's queue.
 
         Args: