refactor(engine)!: simplify execution architecture

[laipz8200]

Important

1. Make sure you have read our contribution guidelines
2. Search existing issues and pull requests to confirm this change is not a duplicate
3. Open or identify the issue this pull request resolves or advances
4. Use a Conventional Commits title for this pull request, and mark breaking changes with !
5. Remember that the pull request title will become the squash merge commit message
6. If CLA Assistant prompts you, sign CLA.md in the pull request conversation

Related Issue

Closes #248

Summary

This refactor reduces the engine's public vocabulary and collapses overlapping execution responsibilities. It intentionally removes compatibility aliases: downstream integrations must upgrade imports, constructor calls, extension hooks, and event field access together.

Architecture changes

• Replace GraphEngine and the graph_engine package with Engine and graphon.engine.
• Give each execution frame one graph and one RuntimeState; child frames receive only their container-scoped graph. Child-local variables never escape their frame; Loop Variable Assigner updates to selectors already owned by a parent are written back, while Iteration frames remain isolated.
• Replace graph traversal/state-manager layers with Scheduler, and keep frame creation/restoration in FrameRegistry.
• Use a fixed worker count and direct worker lifecycle management; dynamic worker scaling and GraphEngineConfig are removed.
• Collapse event management into event, command channels/processing into command, and use singular container_handler, filter, layer, and worker modules.
• Keep one command channel boundary, one event stream, and direct failure handling instead of manager/wrapper layers.

Downstream migration

Imports and public names

Before	After
graphon.graph_engine	graphon.engine
graphon.graph_events	graphon.engine_events
graphon.filters / old engine filter paths	graphon.engine.filter
old layers, command_channels, command_processing, container_handlers, and worker_management paths	singular modules under graphon.engine
GraphEngine	Engine
GraphEngineEvent	EngineEvent
GraphNodeEventBase	NodeEvent
NodeEventBase	NodeEventPayload
GraphRuntimeState	RuntimeState
GraphInitParams	InitParams
GraphEngineLayer	Layer
GraphEventFilter	EngineEventFilter
GraphEventFilterContext	EngineEventFilterContext
filter_graph_events(...)	filter_engine_events(...)
TaskEvent	NodeEventTask

Concrete GraphRun*Event, GraphEdge*Event, and NodeRun*Event names remain unchanged. There are no aliases for the old imports or type names.

Engine construction and identity

• Remove GraphEngineConfig; pass the fixed worker count as workers= when loading or constructing an engine. workers must be a positive integer; booleans and floats are rejected before runtime state is attached.
• Stop passing workflow_id to Engine. Set it on the root RuntimeState; restored and child states share its GraphExecution identity.
• The command channel is optional and defaults to InMemoryChannel.
• Replace engine.layer(layer) with engine.add_layer(layer). The method mutates the engine and returns None.

Events and containers

• Replace in_loop_id and in_iteration_id with one container_id field on engine events.
• container_id identifies the event's direct owning container. Top-level events use container_id == ""; nested events do not expose ancestor container IDs.
• Node implementations yield NodeEventPayload; Node.run() adds execution context and emits NodeEvent, which is part of the EngineEvent stream. container_id is defined by the common EngineEvent, so GraphRun lifecycle events expose the same field and use "" at the top level.
• Existing nested DSL nodes that contain both legacy loop_id and iteration_id remain readable when their container ancestry identifies the direct owner. Only genuinely ambiguous definitions must add canonical container_id.
• Any node referenced as a scope owner must materialize with NodeExecutionType.CONTAINER; ordinary nodes can no longer own silently skipped child scopes. Downstream custom container types remain supported through their node factory execution type.
• Iteration dependency extraction recognizes canonical container_id ownership as well as legacy iteration_id.
• Loop-variable selectors in the parent frame are authoritative. External UpdateVariablesCommand values processed after child write-back survive later rounds, terminal results, pauses, and resume; integrations must not reapply stale Loop result outputs to those selectors.
• Consumers that persist, filter, transform, or render events must update their payload schemas and field access. Concrete event payloads otherwise retain their existing fields.

Layers and filters

• Layer lifecycle methods have default no-op implementations; custom layers only override the hooks they use.
• Remove DebugLoggingLayer and GraphEngineLayerNotInitializedError; invalid layer use now raises the standard runtime error.
• Remove ResumableEngineEventFilter and filter_id; filters operate directly on the engine event stream.
• ResponseStreamFilter and ExecutionLimitsLayer use their canonical singular-module imports. Nonessential ExecutionLimitsLayer helpers and LimitType are no longer public API.

Commands and remote control

• CommandProcessor now receives frame_registry= instead of variable_pool=. Custom constructors must register the root frame before polling commands so workflow-level updates reach every live inherited scope.
• Remove GraphEngineManager. Downstream applications now own task IDs, Redis key naming, and command routing policy.
• Replace GraphEngineCommand / CommandType branching with the concrete discriminated command types such as AbortCommand, PauseCommand, and UpdateVariablesCommand.
• Remove the VariableUpdate wrapper; UpdateVariablesCommand.updates receives variables directly.
• During the rolling-deployment compatibility window, Redis command writers continue emitting the legacy :pending marker and { "value": variable } update wrapper. New readers do not require the marker and accept both wrapped and direct update values. Custom Redis pipelines must continue supporting set in addition to list read/delete, push, and expiry operations. The marker and wrapper can be removed after pre-refactor consumers have been retired for at least one command TTL.

Extension APIs

ContainerHandler.prepare_frame_event(...) now observes variable-update events after the source frame pool has been updated. Custom handlers can inspect or propagate the stored value without reapplying the event.

ContainerHandler hooks change as follows:

Before	After
start_await(...)	handle_request(...)
complete_frame(...)	complete_frame_if_ready(...)
should_collect(...)	should_emit(...)

FrameRegistry changes as follows:

Before	After
materialize_frame(...)	create(...)
materialize_child_frame(...)	create_child(...)
materialize_child_frame_from_state(...)	restore_child(...)
registry.get(frame_id)	registry[frame_id]

Execution frame fields change as follows:

Before	After
frame.graph_runtime_state	frame.state
frame.state_manager / frame.edge_processor	frame.scheduler
frame.error_handler	frame.failure_handler

Custom node factories must implement pure validate_node(...) -> NodeExecutionType preflight, return an isolated scoped copy from with_graph_config(...), and continue binding child runtime state through with_runtime_state(...). validate_node(...) must resolve the same concrete implementation and schema as create_node(...), return that implementation's execution type, and resolve static plugin and credential requirements without constructing a Node, Slim/tool runtime, or other execution collaborator. Factories that own InitParams must copy it with the passed graph_config before create_node(...) can invoke node constructors or post_init(); do not mutate a factory shared with parent or sibling frames. Custom ready queues must consume ReadyTask values and implement serialization through the new ready-queue module; ReadyQueueState is no longer public.

Removed internal boundaries

• Remove GraphStateManager, EdgeProcessor, graph_traversal, event_management, orchestration, error_handler, and dynamic worker-management modules.
• Move GraphExecution and NodeExecution to graphon.runtime.execution.
• Move the dispatcher, event stream/processor, node failure handler, worker pool, and scheduler to their direct owner modules.
• Keep WorkflowExecution temporarily for downstream compatibility, with a TODO marking its future removal. Replace the event stream's custom read/write lock with threading.Lock; downstream code must not import or depend on the removed ReadWriteLock.

Edge identity and persisted state

• Edge.id now preserves the public DSL edge ID and is unique within its owning graph instead of being replaced by a generated runtime edge_N ID. Missing DSL IDs still receive a deterministic edge_N fallback. Explicit IDs reserve matching fallback names only within the owning graph, so separate frame graphs may reuse the same edge ID.
• Edge traversal events now include frame_id. Consumers combining events from multiple frames must use (frame_id, edge_id) as the runtime identity; callers constructing traversal events must provide frame_id.
• RuntimeState snapshots advance from 2.0 to 3.0. A removable, validator-gated compatibility path translates legacy full-graph edge_N state only after the persisted graph is attached.
• ResponseStreamFilter snapshots advance from 1.0 to 2.0. Its isolated compatibility path translates legacy paths when the filter is initialized against the persisted graph. Programmatic Graph.new() graphs without graph_config can restore version 1 filter state only while their complete edge IDs remain the original contiguous edge_N sequence.
• Downstream snapshot envelopes do not need their own version bump when these snapshots are stored as opaque values. Resume implementations must rebuild the graph definition captured for the paused run before reserializing restored legacy state.

Persistence compatibility

• Runtime snapshots serialize field data and version tags rather than Python class names. Version 2 RuntimeState, root-only version 1 RuntimeState, and version 1 ResponseStreamFilter snapshots remain readable through the bounded compatibility paths described above.
• Version 1 runtime snapshots that contain a child-container node task cannot be reconstructed: that format did not persist the owning frame, container invocation, Loop round, Iteration item, or suspended continuation. They now fail while the graph is attached, before workers start, and the queued task is preserved. Downstream resume code should surface this as a non-resumable legacy snapshot instead of retrying it indefinitely.
• Any restored ResumeTask whose container invocation is missing now fails synchronously before workers start, with ready and deferred tasks returned to their original queues. Downstream resume code should treat this as corrupt or incompatible persisted state rather than a successful or retryable run.
• If pause races with abort or a fatal worker failure, the terminal abort/failure takes precedence and no pause snapshot is created from still-active frames. Consumers must persist resumable state only after receiving GraphRunPausedEvent.
• While the scheduler remains unfinished, an empty dispatcher queue is treated as transient and polled again; integrations no longer receive a synthetic stall failure based on queue-size sampling.
• Snapshots written before graph attachment remain attachable to any subsequently rebuilt valid graph.
• InitParams.model_dump() data is unchanged, although its generated schema title changes with the class name.
• Python imports, isinstance checks, schema-title assertions, and pickles that refer to removed class/module names are not compatible.
• Event persistence must migrate the former loop/iteration owner fields to container_id.

Dify adaptation scope

Dify should upgrade Graphon and migrate in one change. The main affected areas are workflow engine construction, runtime-state creation and restoration, layer registration, command/stop routing, event filters, persistence and response conversion, custom container handlers, and type annotations. In particular, all in_loop_id / in_iteration_id access in runners, persistence layers, and response converters must use direct-owner container_id semantics rather than mechanically preserving both ancestor fields.

Checklist

• This pull request links the issue it resolves or advances
• This pull request title follows Conventional Commits, and any breaking change is marked with !
• If CLA Assistant prompted me, I signed CLA.md in the pull request conversation

[dosubot]

📄 Knowledge review

✏️ Suggested updates

2 page suggestions need review.

Page	Library	Status
README /graphon/blob/main/src/graphon/graph_engine/command_channels/README.md	dify-plugin-sdks	⬆️ Pushed to this PR
README /graphon/blob/main/src/graphon/graph_engine/command_channels/README.md	graphon	⬆️ Pushed to this PR

---

[github-actions]

All contributors on this pull request have signed the CLA.
_{Posted by the CLA Assistant Lite bot.}

[gaoyue1989]

Hi @laipz8200 — thank you for driving this refactor. We run exported Dify DSL
workflows on Graphon as an independent service (related to #61), with an
interactive workload that has a hard end-to-end budget of 800 ms per turn
(IVR voice response). While profiling it with OpenTelemetry we found a class of
per-node handoff stalls that this rewrite touches directly — filing here since
the new dispatcher.py / worker/worker.py / event/stream.py are exactly
where it lives.

Finding: fixed polling adds 25–50 ms of dead time per node hop

Three fixed sleeps serialize every node transition:

Location (this branch)	Code	Effect
engine/worker/worker.py:132	self._stop_event.wait(0.1)	Worker idles up to 100 ms after its queue is empty; next ready task (queued by the dispatcher after the previous node's finished-event) waits for this poll
engine/dispatcher.py:123,163	self._dispatch_queue.get(timeout=0.1)	Completion events wait for the dispatcher poll before dependents are enqueued
engine/event/stream.py:127	time.sleep(0.001)	Consumer-side spin while pulling events

On a sequential chain these compound: 25–50 ms per node hop, regardless of
how fast the nodes themselves are.

Measured impact (OpenTelemetry span attribution, 110 samples × 2 rounds)

Production-shaped Chatflow: ~22 nodes/turn, remote plugin daemon, mock LLM
(instant), local-process sandbox — i.e. all real I/O removed, isolating pure
orchestration cost:

wall avg 808.9 ms = gap 508.4 ms (63%)   ← workflow.run − Σ(node spans)
                  + code 160.8  (20%)
                  + http  65.2  (8%)
                  + llm rpc 27.5 (3%)
state/SSE overhead ≈ 0.2 ms              ← not a factor

Per-category gap: 407–596 ms across scenario groups; P95 wall 1162 ms.

Two-line fix, verified

--- a/src/graphon/engine/worker/worker.py
+++ b/src/graphon/engine/worker/worker.py
@@ -129,7 +129,7 @@
             if not task_claimed:
-                self._stop_event.wait(0.1)
+                self._stop_event.wait(0.0005)
                 continue

--- a/src/graphon/engine/dispatcher.py
+++ b/src/graphon/engine/dispatcher.py
@@ (both call sites)
-            task = self._dispatch_queue.get(timeout=0.1)
+            task = self._dispatch_queue.get(timeout=0.002)

Results on the same 110-sample suite (branch coverage unchanged at 55/55):

Metric	Before	After	Δ
orchestration gap avg	508 ms	27–39 ms	−93 %
wall avg	808.9 ms	331.4 ms	−59 %
wall P95	1162 ms	541 ms	−54 %
Max	1184 ms	607 ms

Request

Since #249 is rewriting these exact modules: could the new scheduler /
dispatcher adopt event-driven wakeup (a threading.Condition notified on
enqueue, or blocking get() without long timeouts) instead of carrying the
0.1 s polls into the new architecture? A condition-based handoff removes the
stalls entirely without busy-spinning (the 1 ms consumer sleep can stay — its
cost is negligible).

Happy to contribute the patch against this branch, add an "interactive
latency" benchmark case to tests/engine/, or share raw trace data if useful.