features/mocks/timing_recorder_agent.py

@@ -0,0 +1,44 @@
+"""Test double for observing whether sibling graph nodes overlap in time.
+
+Used by the pure-graph parallel-dispatch regression scenario (issue #97,
+captured by the TDD test in issue #98) to prove whether two sibling
+next-nodes actually run concurrently or strictly one after another.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, List, Optional, Tuple
+
+from cleveractors.agents.base import Agent
+
+TimingEvent = Tuple[str, str, float]
+
+
+class TimingRecorderAgent(Agent):
+    """Agent double that records its own start/end timestamps and sleeps.
+
+    Every instance sharing the same ``event_log`` list lets a test
+    reconstruct the wall-clock interval each agent occupied, and thus
+    detect whether any two agents' intervals overlapped (concurrent
+    execution) or not (sequential execution).
+    """
+
+    def __init__(
+        self, name: str, event_log: List[TimingEvent], delay_seconds: float
+    ) -> None:
+        super().__init__(name=name)
+        self._event_log = event_log
+        self._delay_seconds = delay_seconds
+
+    async def process_message(
+        self, message: str, context: Optional[dict[str, Any]] = None
+    ) -> str:
+        loop = asyncio.get_running_loop()
+        self._event_log.append((self.name, "start", loop.time()))
+        await asyncio.sleep(self._delay_seconds)
+        self._event_log.append((self.name, "end", loop.time()))
+        return f"{self.name} finished"
+
+    def get_capabilities(self) -> List[str]:
+        return ["timing-recording"]

features/pure_graph_parallel_node_gate.feature

@@ -0,0 +1,42 @@
+Feature: Pure-graph concurrent dispatch honors each node's own parallel flag
+  As a graph author relying on the Actor Configuration Standard
+  I want sibling next-nodes that do not declare parallel: true to execute
+  strictly sequentially, even when the graph-level parallel_execution flag
+  is on
+  So that a graph drawing a sequential fan-out does not silently run its
+  branches concurrently
+
+  # Regression test for issue #97:
+  # The three concurrent-dispatch sites in PureLangGraph
+  # (src/cleveractors/langgraph/pure_graph.py, execute() at line 1078 and
+  # both execute_stream() branches at lines 1860 and 1965) fire every
+  # candidate next-node concurrently via asyncio.gather/asyncio.create_task
+  # whenever the graph-level `parallel_execution: true` (the default) and
+  # there are 2+ candidates, without ever consulting each node's own
+  # `parallel` flag (NodeConfig.parallel / can_execute_parallel()). This
+  # violates Actor Configuration Standard §6.7 and §12.3, which require that
+  # only the subset of next-nodes with `parallel: true` run concurrently;
+  # the rest MUST run sequentially. All three dispatch sites are exercised
+  # below so the permanent regression guard covers each of them.
+
+  @tdd_issue @tdd_issue_97 @tdd_expected_fail
+  Scenario: Sibling next-nodes without parallel: true execute sequentially, not concurrently
+    Given a trigger node with two outgoing edges to agents "left" and "right", neither marked parallel (pg)
+    And graph-level parallel_execution is enabled (pg)
+    When the graph is executed with message "go" (pg)
+    Then agents "left" and "right" should not have overlapped in execution time (pg)
+
+  @tdd_issue @tdd_issue_97 @tdd_expected_fail
+  Scenario: Sibling next-nodes without parallel: true execute sequentially under streaming, non-agent trigger
+    Given a trigger node with two outgoing edges to agents "left" and "right", neither marked parallel (pg)
+    And graph-level parallel_execution is enabled (pg)
+    When the graph is streamed with message "go" (pg)
+    Then agents "left" and "right" should not have overlapped in execution time (pg)
+
+  @tdd_issue @tdd_issue_97 @tdd_expected_fail
+  Scenario: Sibling next-nodes without parallel: true execute sequentially under streaming, agent trigger
+    Given an agent trigger node with two outgoing edges to agents "left" and "right", neither marked parallel (pg)
+    And graph-level parallel_execution is enabled (pg)
+    When the graph is streamed with message "go" (pg)
+    Then agents "left" and "right" should not have overlapped in execution time (pg)
+

features/steps/pure_graph_parallel_node_gate_steps.py

@@ -0,0 +1,141 @@
+"""Step definitions for the parallel-dispatch node-flag regression (issue #97).
+
+Builds a pure graph where a single trigger node (either a non-agent
+FUNCTION node or, for one scenario, an intermediate AGENT node) has two
+unconditional edges to two agent nodes, neither of which declares
+``parallel: true``, with ``parallel_execution: true`` at the graph level
+(the default). Drives both target agents through a shared timing log so
+the ``Then`` step can detect whether their execution intervals overlapped.
+The three scenarios in ``pure_graph_parallel_node_gate.feature`` exercise
+each of the three concurrent-dispatch sites in ``pure_graph.py``: the
+non-streaming ``execute()`` path, and both ``execute_stream()`` branches
+(the non-AGENT branch, reached via the same non-agent trigger, and the
+intermediate-AGENT branch, reached via the agent trigger).
+"""
+
+from __future__ import annotations
+
+from typing import List
+
+from behave import given, then, when
+from behave.api.async_step import async_run_until_complete
+from features.mocks.timing_recorder_agent import TimingEvent, TimingRecorderAgent
+
+from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
+from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
+
+AGENT_DELAY_SECONDS = 0.05
+
+
+@given(
+    'a trigger node with two outgoing edges to agents "{left}" and '
+    '"{right}", neither marked parallel (pg)'
+)
+def step_trigger_with_two_sibling_agents(context, left, right):
+    context.event_log: List[TimingEvent] = []
+    left_agent = TimingRecorderAgent(left, context.event_log, AGENT_DELAY_SECONDS)
+    right_agent = TimingRecorderAgent(right, context.event_log, AGENT_DELAY_SECONDS)
+
+    config = PureGraphConfig(
+        name="sibling_dispatch_graph",
+        entry_point="trigger",
+        nodes={
+            "trigger": NodeConfig(
+                name="trigger", type=NodeType.FUNCTION, function="validate"
+            ),
+            left: NodeConfig(
+                name=left, type=NodeType.AGENT, agent=left, parallel=False
+            ),
+            right: NodeConfig(
+                name=right, type=NodeType.AGENT, agent=right, parallel=False
+            ),
+        },
+        edges=[
+            Edge(source="trigger", target=left),
+            Edge(source="trigger", target=right),
+        ],
+    )
+    context.graph = PureLangGraph(config, agents={left: left_agent, right: right_agent})
+
+
+@given(
+    'an agent trigger node with two outgoing edges to agents "{left}" and '
+    '"{right}", neither marked parallel (pg)'
+)
+def step_agent_trigger_with_two_sibling_agents(context, left, right):
+    context.event_log: List[TimingEvent] = []
+    trigger_agent = TimingRecorderAgent(
+        "trigger", context.event_log, AGENT_DELAY_SECONDS
+    )
+    left_agent = TimingRecorderAgent(left, context.event_log, AGENT_DELAY_SECONDS)
+    right_agent = TimingRecorderAgent(right, context.event_log, AGENT_DELAY_SECONDS)
+
+    config = PureGraphConfig(
+        name="sibling_dispatch_graph_agent_trigger",
+        entry_point="trigger",
+        nodes={
+            "trigger": NodeConfig(
+                name="trigger", type=NodeType.AGENT, agent="trigger", parallel=False
+            ),
+            left: NodeConfig(
+                name=left, type=NodeType.AGENT, agent=left, parallel=False
+            ),
+            right: NodeConfig(
+                name=right, type=NodeType.AGENT, agent=right, parallel=False
+            ),
+        },
+        edges=[
+            Edge(source="trigger", target=left),
+            Edge(source="trigger", target=right),
+        ],
+    )
+    context.graph = PureLangGraph(
+        config,
+        agents={"trigger": trigger_agent, left: left_agent, right: right_agent},
+    )
+
+
+@given("graph-level parallel_execution is enabled (pg)")
+def step_parallel_execution_enabled(context):
+    context.graph.config.parallel_execution = True
+
+
+@when('the graph is executed with message "{message}" (pg)')
+@async_run_until_complete
+async def step_execute_graph(context, message):
+    await context.graph.execute(message)
+
+
+@when('the graph is streamed with message "{message}" (pg)')
+@async_run_until_complete
+async def step_stream_graph(context, message):
+    async for _ in context.graph.execute_stream(message):
+        pass
+
+
+@then('agents "{left}" and "{right}" should not have overlapped in execution time (pg)')
+def step_assert_no_overlap(context, left, right):
+    intervals: dict[str, tuple[float, float]] = {}
+    for name, kind, timestamp in context.event_log:
+        start, end = intervals.get(name, (timestamp, timestamp))
+        if kind == "start":
+            intervals[name] = (timestamp, end)
+        else:
+            intervals[name] = (start, timestamp)
+
+    assert left in intervals and right in intervals, (
+        f"Expected both {left!r} and {right!r} to have executed and recorded "
+        f"timing, but only recorded intervals for: {sorted(intervals)}"
+    )
+
+    left_start, left_end = intervals[left]
+    right_start, right_end = intervals[right]
+
+    overlapped = left_start < right_end and right_start < left_end
+    assert not overlapped, (
+        f"Expected {left!r} and {right!r} to run strictly sequentially (no "
+        "overlap) since neither declares parallel: true, but their "
+        f"intervals overlapped: {left}=({left_start}, {left_end}), "
+        f"{right}=({right_start}, {right_end}) — the graph dispatched both "
+        "concurrently, ignoring each node's own parallel flag."
+    )