Per-invocation isolation: concurrent invoke() on one assistant (WorkflowRun)

[GuanyiLi-Craig]

What & why

One EventDrivenWorkflow instance can now handle many invoke() calls concurrently.

Before this change, all runtime state — the quiescence tracker, the ready-queue, the stop flag, and the topic queues — lived on the shared workflow definition and was reset() at the start of every invoke(). Two concurrent invocations on the same instance therefore corrupted each other: the second invoke's reset drained the first's queues mid-flight, and a single stop() halted both. This is Defect #3 / runtime Gap 00 from the design notes.

Approach — "share by reference, isolate by allocation"

A new WorkflowRun owns everything that mutates during one invocation. The definition stays read-only during a run; each invoke() allocates a WorkflowRun with its own tracker, ready-queue, stop flag, and per-run topic queues (shallow topic copies with fresh queues, so the queue kind is preserved). Tools, LLM clients, and topology are shared by reference — there is no cloning of the node/tool graph, so concurrency stays cheap (allocation is O(nodes + topics)).

Changes

File	Change
event_driven_workflow.py	Slimmed to the immutable definition/facade. invoke() runs on a WorkflowRun registered in _active_runs; stop() forwards to in-flight runs.
workflow_run.py (new)	WorkflowRun state + shared topology/quiescence/commit helpers + thin init/run dispatch.
run_recovery.py (new)	Seeding + replay recovery (init_run).
parallel_engine.py / sequential_engine.py (new)	The two execution engines as free functions taking the run (WorkflowRun referenced only under TYPE_CHECKING → no import cycle).
utils.py	publish_events / get_node_input route I/O through the run's topics-by-name.

Tests

• tests/workflow/test_invocation_isolation.py — the characterization test: concurrent invokes on one instance complete independently (parallel + sequential). Fails on the pre-change runtime.
• tests/workflow/test_workflow_run.py — 22 unit tests for WorkflowRun internals (isolation invariants, helpers, error wrapping, stop-forwarding, recovery seeding).
• tests_integration/ — 3 concurrent invoke() examples (parallel single-node, sequential single-node, 3-node function-call workflow), each asserting per-request isolation with no cross-talk. Verified against real providers.
• .env.example — documents required API keys for the integration tests (read via dotenv).
• Retargeted 4 coupled unit tests from definition-private fields to a WorkflowRun.

Verification

• 692 unit tests pass (was 670)
• mypy grafi clean
• pre-commit (black / isort / flake8) clean
• Full integration suite green: 54 passed, 4 skipped (ollama/local), 0 failed

Notes / scope

• Public APIs (Assistant, EventDrivenWorkflow, Node, builders) are unchanged.
• The event store is still read from the global container (injected into the run); full dependency injection is deferred to a separate change.
• Unrelated: simple_function_call_assistant_async_example.py is a pre-existing flaky integration test (asserts a literal "bad" substring on free-form LLM output); its event-count accounting is unchanged by this PR (stable at 24). Not modified here.

🤖 Generated with Claude Code

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.