From bce43626f872fd3dfb47190956bb6622f052df67 Mon Sep 17 00:00:00 2001 From: "Luis Mendes (CoreRasurae)" Date: Tue, 4 Aug 2026 20:08:27 +0000 Subject: [PATCH] test(langgraph): capture parallel dispatch node-flag regression (#97) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a failing-first Behave regression test proving issue #97: PureLangGraph fires every candidate next-node concurrently via asyncio.gather/asyncio.create_task whenever the graph-level parallel_execution flag is 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 while the rest run sequentially. Three scenarios cover all three concurrent-dispatch sites named in issue #97, each driving a shared TimingRecorderAgent test double (features/mocks/) that records start/end timestamps and asserting the two sibling agents' execution intervals never overlap: - execute() (line 1078), via a non-agent FUNCTION trigger node. - execute_stream()'s non-AGENT branch (line 1965), via the same FUNCTION trigger node, driven through the streaming entrypoint. - execute_stream()'s intermediate-AGENT branch (line 1860), via an AGENT trigger node with two further sibling AGENT edges. Each assertion currently fails (both siblings start concurrently) while the bug is present. Tagged @tdd_issue, @tdd_issue_97, and @tdd_expected_fail per the project's TDD issue-capture workflow, so nox -s unit_tests stays green via the existing TddExpectedFailPolicy inversion hook. The fix itself is delivered separately by issue #97 on a bugfix/ branch. Also hardens the shared Then step to fail with a clear AssertionError (rather than a raw KeyError) if either sibling agent recorded no timing at all, and drops two unused test-only attributes and a no-op Background step. ISSUES CLOSED: #98 --- features/mocks/timing_recorder_agent.py | 44 ++++++ .../pure_graph_parallel_node_gate.feature | 42 ++++++ .../pure_graph_parallel_node_gate_steps.py | 141 ++++++++++++++++++ 3 files changed, 227 insertions(+) create mode 100644 features/mocks/timing_recorder_agent.py create mode 100644 features/pure_graph_parallel_node_gate.feature create mode 100644 features/steps/pure_graph_parallel_node_gate_steps.py diff --git a/features/mocks/timing_recorder_agent.py b/features/mocks/timing_recorder_agent.py new file mode 100644 index 0000000..e858ba3 --- /dev/null +++ b/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"] diff --git a/features/pure_graph_parallel_node_gate.feature b/features/pure_graph_parallel_node_gate.feature new file mode 100644 index 0000000..e7be422 --- /dev/null +++ b/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) + diff --git a/features/steps/pure_graph_parallel_node_gate_steps.py b/features/steps/pure_graph_parallel_node_gate_steps.py new file mode 100644 index 0000000..3302e87 --- /dev/null +++ b/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." + ) -- 2.52.0