diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 0bc3d3e03..0f51917f3 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -73,3 +73,4 @@ Below are some of the specific details of various contributions. * HAL 9000 has contributed the TDD scenario for plan tree correction visual marking (PR #8671 / issue #8576): added a failing BDD scenario proving that corrected nodes (decisions with is_correction=True) are not visually distinguished in the plan tree output, formalizing Spec Requirement #7 as an executable specification. * HAL 9000 has contributed the `--clone-into` CLI argument for `container-instance`, the `CloneIntoHandler` module, the `devcontainer-instance` snapshot sandbox strategy, and the `ContainerLifecycleState.DISCOVERED` terminology alignment (PR #8304, issue #7555). * HAL 9000 has contributed the ACMS Context Tier Hydration documentation (PR #9208 / issue #6175): documented the `context_tier_hydrator` module in the ACMS Architecture section of the specification, covering its public interface, file listing strategy, budget limits, and fragment structure. +* HAL 9000 has contributed the agent task memory leak fix (#9044): replaced `list.remove` with `set.discard` as the done_callback for asyncio tasks in `Agent._tasks`, preventing unbounded memory growth in long-lived agents and ensuring safe concurrent task removal. diff --git a/features/agent_task_memory_leak_fix.feature b/features/agent_task_memory_leak_fix.feature new file mode 100644 index 000000000..e08e0db67 --- /dev/null +++ b/features/agent_task_memory_leak_fix.feature @@ -0,0 +1,44 @@ +@a2a @agents @memory_leak @agent_task_pruning +Feature: Agent Task Memory Leak Fix + As a CleverAgents developer + I want to ensure that completed tasks are removed from Agent._tasks + So that long-lived agent instances do not accumulate unbounded memory + + Background: + Given a test agent for memory leak testing + + @agent_task_pruning @memory_leak + Scenario: Completed tasks are removed from _tasks list after processing + When I send a message to the agent + And I wait for the task to complete + Then the _tasks list should be empty + And the message should have been processed + + @agent_task_pruning @memory_leak + Scenario: Multiple messages result in bounded _tasks list + When I send 10 messages to the agent + And I wait for all tasks to complete + Then the _tasks list should be empty + And all 10 messages should have been processed + + @agent_task_pruning @memory_leak + Scenario: Failed tasks are also removed from _tasks list + When I send a message that will fail to the agent + And I wait for the task to complete + Then the _tasks list should be empty + And the error should have been raised + + @agent_task_pruning @memory_leak + Scenario: _tasks list does not grow unboundedly during long-lived operation + When I send 100 messages to the agent in rapid succession + And I wait for all tasks to complete + Then the _tasks list should be empty + And the maximum _tasks list size should not exceed 10 + And all 100 messages should have been processed + + @agent_task_pruning @memory_leak + Scenario: Task removal is thread-safe with concurrent messages + When I send 50 messages to the agent concurrently + And I wait for all tasks to complete + Then the _tasks list should be empty + And all 50 messages should have been processed diff --git a/features/steps/agent_task_memory_leak_fix_steps.py b/features/steps/agent_task_memory_leak_fix_steps.py new file mode 100644 index 000000000..34c00ee37 --- /dev/null +++ b/features/steps/agent_task_memory_leak_fix_steps.py @@ -0,0 +1,235 @@ +"""Steps for testing Agent task memory leak fix.""" + +import asyncio +import threading +from collections.abc import Callable +from typing import Any + +from behave import given, then, when + +from cleveragents.agents.base import Agent + + +class TestAgent(Agent): + """Test agent for memory leak testing.""" + + def __init__( + self, + name: str, + config: dict[str, Any] | None = None, + fail: bool = False, + ): + self.fail = fail + self.processed_messages: list[Any] = [] + super().__init__(name, config) + + async def process_message( + self, message: Any, context: dict[str, Any] | None = None + ) -> Any: + """Process a message, optionally failing.""" + if self.fail: + raise RuntimeError("Intentional test failure") + self.processed_messages.append(message) + return f"processed:{message}" + + def get_capabilities(self) -> list[str]: + return ["test"] + + +class _BackgroundLoop: + """A persistent asyncio event loop running in a background thread. + + This is required because ``asyncio.create_task()`` (used inside + ``Agent._setup_processing_pipeline``) requires a *running* event loop. + Behave step functions are synchronous, so we keep a dedicated loop alive + in a daemon thread and submit all coroutines to it via + ``asyncio.run_coroutine_threadsafe``. + """ + + def __init__(self) -> None: + self._loop = asyncio.new_event_loop() + self._thread = threading.Thread( + target=self._loop.run_forever, + daemon=True, + name="behave-agent-test-loop", + ) + self._thread.start() + + @property + def loop(self) -> asyncio.AbstractEventLoop: + return self._loop + + def run(self, coro: Any) -> None: + """Submit *coro* to the background loop and block until it completes.""" + future = asyncio.run_coroutine_threadsafe(coro, self._loop) + future.result(timeout=10.0) + + def stop(self) -> None: + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(timeout=5.0) + + +# Module-level singleton — shared across all scenarios in a test run. +_BG_LOOP = _BackgroundLoop() + + +def _send_message(agent: Agent, message: Any) -> None: + """Send a message to *agent* from within the background event loop. + + ``Agent.send_message`` triggers ``asyncio.create_task`` synchronously + inside the RxPY ``on_next`` callback. That call requires a *running* + event loop, so we must invoke it from within the background loop thread. + """ + + async def _do_send() -> None: + agent.send_message(message) + + _BG_LOOP.run(_do_send()) + + +def _wait_for_tasks( + agent: Agent, + *, + timeout: float = 5.0, + interval: float = 0.05, + tracker: Callable[[int], None] | None = None, +) -> None: + """Drive the background event loop until the agent's task set is empty.""" + + async def _wait() -> None: + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + active = len(agent._tasks) + if tracker is not None: + tracker(active) + if active == 0: + break + await asyncio.sleep(interval) + + _BG_LOOP.run(_wait()) + + +@given("a test agent for memory leak testing") +def step_basic_agent_setup(context: Any) -> None: + """Set up a test agent for memory leak testing.""" + + # Instantiate the agent inside the background loop so that + # ``asyncio.create_task`` (called during ``_setup_processing_pipeline``) + # has a running event loop available. + async def _make_agent() -> TestAgent: + return TestAgent("test-agent") + + future = asyncio.run_coroutine_threadsafe(_make_agent(), _BG_LOOP.loop) + context.agent = future.result(timeout=5.0) + context.processed_count = 0 + context.max_tasks_size = 0 + context.error_raised = False + + +@when("I send a message to the agent") +def step_send_message(context: Any) -> None: + """Send a single message to the agent.""" + _send_message(context.agent, "test-message") + + +@when("I send {count:d} messages to the agent") +def step_send_multiple_messages(context: Any, count: int) -> None: + """Send multiple messages to the agent.""" + context.expected_count = count + for i in range(count): + _send_message(context.agent, f"message-{i}") + + +@when("I send {count:d} messages to the agent in rapid succession") +def step_send_multiple_messages_rapid(context: Any, count: int) -> None: + """Send multiple messages rapidly to the agent.""" + context.expected_count = count + for i in range(count): + _send_message(context.agent, f"message-{i}") + + +@when("I send {count:d} messages to the agent concurrently") +def step_send_multiple_messages_concurrent(context: Any, count: int) -> None: + """Send multiple messages concurrently to the agent.""" + context.expected_count = count + for i in range(count): + _send_message(context.agent, f"concurrent-message-{i}") + + +@when("I wait for the task to complete") +def step_wait_for_task(context: Any) -> None: + """Wait for a single task to complete.""" + _wait_for_tasks(context.agent) + + +@when("I wait for all tasks to complete") +def step_wait_for_all_tasks(context: Any) -> None: + """Wait for all tasks to complete.""" + + def _track(active: int) -> None: + if active > context.max_tasks_size: + context.max_tasks_size = active + + _wait_for_tasks(context.agent, tracker=_track) + + +@when("I send a message that will fail to the agent") +def step_send_failing_message(context: Any) -> None: + """Send a message to an agent that will fail.""" + + async def _make_fail_agent() -> TestAgent: + return TestAgent("fail-agent", fail=True) + + future = asyncio.run_coroutine_threadsafe(_make_fail_agent(), _BG_LOOP.loop) + context.fail_agent = future.result(timeout=5.0) + _send_message(context.fail_agent, "will-fail") + + +@then("the _tasks list should be empty") +def step_verify_tasks_empty(context: Any) -> None: + """Verify that the _tasks list is empty.""" + _wait_for_tasks(context.agent) + assert len(context.agent._tasks) == 0, ( + f"Expected _tasks to be empty, but found {len(context.agent._tasks)} tasks" + ) + + +@then("the message should have been processed") +def step_verify_message_processed(context: Any) -> None: + """Verify that the message was processed.""" + assert len(context.agent.processed_messages) > 0, ( + "Expected at least one message to be processed" + ) + + +@then("all {count:d} messages should have been processed") +def step_verify_all_messages_processed(context: Any, count: int) -> None: + """Verify that all messages were processed.""" + _wait_for_tasks(context.agent) + assert len(context.agent.processed_messages) == count, ( + f"Expected {count} messages to be processed, " + f"but found {len(context.agent.processed_messages)}" + ) + + +@then("the maximum _tasks list size should not exceed {max_size:d}") +def step_verify_max_tasks_size(context: Any, max_size: int) -> None: + """Verify that the maximum _tasks list size did not exceed a threshold.""" + assert context.max_tasks_size <= max_size, ( + f"Expected max _tasks size to be <= {max_size}, " + f"but found {context.max_tasks_size}" + ) + + +@then("the error should have been raised") +def step_verify_error_raised(context: Any) -> None: + """Verify that an error was raised during processing and tasks were cleaned up.""" + _wait_for_tasks(context.fail_agent) + assert len(context.fail_agent._tasks) == 0, ( + "Expected _tasks to be empty even after error" + ) + # Confirm that no messages were processed (the error prevented processing) + assert len(context.fail_agent.processed_messages) == 0, ( + "Expected no messages to be processed when agent is configured to fail" + ) diff --git a/src/cleveragents/agents/base.py b/src/cleveragents/agents/base.py index 69d7a5a2a..49c97bc08 100644 --- a/src/cleveragents/agents/base.py +++ b/src/cleveragents/agents/base.py @@ -25,13 +25,16 @@ class Agent(ABC): self.config = config or {} self.input_stream: Any = Subject() self.output_stream: Any = Subject() - self._tasks: list[asyncio.Task[Any]] = [] + self._tasks: set[asyncio.Task[Any]] = set() self._setup_processing_pipeline() def _setup_processing_pipeline(self) -> None: def _on_next(msg: Any) -> None: task = asyncio.create_task(self._process_wrapper(msg)) - self._tasks.append(task) + self._tasks.add(task) + # Add callback to remove task from set when it completes + # This prevents unbounded growth of _tasks set (memory leak fix) + task.add_done_callback(self._tasks.discard) self.input_stream.subscribe( on_next=_on_next,