From d59c748929666446daf9896f28ce6a082dfca305 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 17:05:15 +0100 Subject: [PATCH 01/15] Allow to cancel a task in the task queue via the Future, executing every task in a separate thread --- src/serena/agent.py | 27 +++++++++---- test/serena/test_serena_agent.py | 67 ++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 7 deletions(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index e4e1199d..d9015f55 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -211,18 +211,31 @@ class SerenaAgent: """ Execute the task, setting the result or exception on the future. """ - try: - with LogTime(self.name, logger=log): - result = self._function() - self.future.set_result(result) - except Exception as e: - log.error(f"Error during execution of {self.name}: {e}", exc_info=e) - self.future.set_exception(e) + + def run_task(): + try: + with LogTime(self.name, logger=log): + result = self._function() + if not self.future.done(): + self.future.set_result(result) + except Exception as e: + if not self.future.done(): + log.error(f"Error during execution of {self.name}: {e}", exc_info=e) + self.future.set_exception(e) + + thread = Thread(target=run_task, name=self.name) + thread.start() def _process_task_queue(self) -> None: while True: + log.debug("Waiting for task ...") task: SerenaAgent.Task = self._task_executor_queue.get() task.execute() + try: + # wait for task completion + task.future.result() + except: + pass def get_language_server_manager(self) -> LanguageServerManager | None: if self._active_project is not None: diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index 882b45be..90c053a0 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -61,6 +61,14 @@ def serena_agent(request: pytest.FixtureRequest, serena_config): return SerenaAgent(project=project_name, serena_config=serena_config) +@pytest.fixture +def basic_serena_agent(serena_config): + """ + Fixture for a basic SerenaAgent without a project + """ + return SerenaAgent(project=None, serena_config=serena_config) + + class TestSerenaAgent: @pytest.mark.parametrize( "serena_agent,symbol_name,expected_kind,expected_file", @@ -284,3 +292,62 @@ class TestSerenaAgent: symbols = json.loads(result) assert not symbols, f"Expected to find no symbols for {name_path}. Symbols found: {symbols}" + + @staticmethod + def _create_task(delay: float, name: str, exception: bool = False): + def fn(): + time.sleep(delay) + if exception: + raise ValueError(f"Task {name} failed") + return True + + fn.__name__ = name + return fn + + def test_task_queue_sequence(self, basic_serena_agent): + """ + Tests that a sequence of tasks is executed correctly + """ + future1 = basic_serena_agent.issue_task(self._create_task(1, "task1")) + future2 = basic_serena_agent.issue_task(self._create_task(1, "task2")) + assert future1.result() is True + assert future2.result() is True + + def test_task_queue_exception(self, basic_serena_agent): + """ + Tests that tasks that raise exceptions are handled correctly, i.e. that + * the exception is propagated, + * subsequent tasks are still executed. + """ + future1 = basic_serena_agent.issue_task(self._create_task(1, "task1", exception=True)) + future2 = basic_serena_agent.issue_task(self._create_task(1, "task2")) + have_exception = False + try: + assert future1.result() + except Exception as e: + assert isinstance(e, ValueError) + have_exception = True + assert have_exception + assert future2.result() is True + + def test_task_queue_cancel(self, basic_serena_agent): + """ + Tests that tasks that are cancelled are handled correctly, i.e. that + * subsequent tasks are executed as soon as cancellation ensues. + * the cancelled task raises CancelledError when result() is called. + """ + start_time = time.time() + future1 = basic_serena_agent.issue_task(self._create_task(10, "task1")) + future2 = basic_serena_agent.issue_task(self._create_task(1, "task2")) + time.sleep(1) + future1.cancel() + assert future2.result() is True + end_time = time.time() + assert (end_time - start_time) < 9, "Cancelled task did not stop in time" + have_cancelled_error = False + try: + future1.result() + except Exception as e: + assert e.__class__.__name__ == "CancelledError" + have_cancelled_error = True + assert have_cancelled_error From b5e5af54a5b6416e15cd58c4e5c31138e1fcdd49 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 17:57:57 +0100 Subject: [PATCH 02/15] Use an inspectable list as the task queue, adding a way to retrieve the current list of tasks (for the dashboard) --- src/serena/agent.py | 73 +++++++++++++++++++++++++++----- test/serena/test_serena_agent.py | 21 +++++++++ 2 files changed, 83 insertions(+), 11 deletions(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index d9015f55..5b89910d 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -8,11 +8,12 @@ import os import platform import sys import threading +import time import webbrowser from collections.abc import Callable from concurrent.futures import Future +from dataclasses import dataclass from logging import Logger -from queue import Queue from threading import Thread from typing import TYPE_CHECKING, Any, Optional, TypeVar @@ -157,11 +158,12 @@ class SerenaAgent: # create executor for starting the language server and running tools in another thread # This executor is used to achieve linear task execution, so it is important to use a single-threaded executor. - self._task_executor_queue: Queue = Queue() + self._task_executor_lock = threading.Lock() + self._task_executor_queue: list[SerenaAgent.Task] = [] self._task_executor_thread = Thread(target=self._process_task_queue, name="SerenaAgentTaskExecutor", daemon=True) self._task_executor_thread.start() - self._task_executor_lock = threading.Lock() self._task_executor_task_index = 1 + self._task_executor_current_task: Optional[SerenaAgent.Task] = None # Initialize the prompt factory self.prompt_factory = SerenaPromptFactory() @@ -207,12 +209,12 @@ class SerenaAgent: def _tostring_includes(self) -> list[str]: return ["name"] - def execute(self) -> None: + def start(self) -> None: """ - Execute the task, setting the result or exception on the future. + Executes the task in a separate thread, setting the result or exception on the future. """ - def run_task(): + def run_task() -> None: try: with LogTime(self.name, logger=log): result = self._function() @@ -228,14 +230,63 @@ class SerenaAgent: def _process_task_queue(self) -> None: while True: - log.debug("Waiting for task ...") - task: SerenaAgent.Task = self._task_executor_queue.get() - task.execute() + # obtain task from the queue + task: SerenaAgent.Task | None = None + with self._task_executor_lock: + if len(self._task_executor_queue) > 0: + task = self._task_executor_queue.pop(0) + if task is None: + time.sleep(0.1) + continue + + # start task execution asynchronously + with self._task_executor_lock: + self._task_executor_current_task = task + log.info("Starting execution of %s", task) + task.start() + + # wait for task completion try: - # wait for task completion task.future.result() except: pass + with self._task_executor_lock: + self._task_executor_current_task = None + + @dataclass + class TaskInfo: + name: str + is_running: bool + future: Future + """ + future for accessing the task's result + """ + task_id: int + """ + unique identifier of the task + """ + + @staticmethod + def from_task(task: "SerenaAgent.Task", is_running: bool) -> "SerenaAgent.TaskInfo": + return SerenaAgent.TaskInfo(name=task.name, is_running=is_running, future=task.future, task_id=id(task)) + + def cancel(self) -> None: + self.future.cancel() + + def get_current_tasks(self) -> list["SerenaAgent.TaskInfo"]: + """ + Gets the list of tasks currently running or queued for execution. + The function returns a list of thread-safe TaskInfo objects (specifically created for the caller). + + :return: the list of tasks in the execution order (running task first) + """ + tasks = [] + with self._task_executor_lock: + if self._task_executor_current_task is not None: + tasks.append(self.TaskInfo.from_task(self._task_executor_current_task, True)) + for task in self._task_executor_queue: + tasks.append(self.TaskInfo.from_task(task, False)) + return tasks def get_language_server_manager(self) -> LanguageServerManager | None: if self._active_project is not None: @@ -423,7 +474,7 @@ class SerenaAgent: self._task_executor_task_index += 1 log.info(f"Scheduling {task_name}") task = SerenaAgent.Task(function=task, name=task_name) - self._task_executor_queue.put(task) + self._task_executor_queue.append(task) return task.future def execute_task(self, task: Callable[[], T], name: str | None = None) -> T: diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index 90c053a0..72896e47 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -351,3 +351,24 @@ class TestSerenaAgent: assert e.__class__.__name__ == "CancelledError" have_cancelled_error = True assert have_cancelled_error + + def test_task_queue_cancellation_via_task_info(self, basic_serena_agent): + start_time = time.time() + basic_serena_agent.issue_task(self._create_task(10, "task1")) + basic_serena_agent.issue_task(self._create_task(1, "task2")) + task_infos = basic_serena_agent.get_current_tasks() + task_infos2 = basic_serena_agent.get_current_tasks() + + # test expected tasks + assert len(task_infos) == 2 + assert "task1" in task_infos[0].name + assert "task2" in task_infos[1].name + + # test task identifiers being stable + assert task_infos2[0].task_id == task_infos[0].task_id + + # test cancellation + task_infos[0].cancel() + assert task_infos[1].future.result() + end_time = time.time() + assert (end_time - start_time) < 9, "Cancelled task did not stop in time" From a85d54f3a56f67c78370c9d8df8b09591bcc488c Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 18:14:12 +0100 Subject: [PATCH 03/15] Allow logging of task management to be disabled --- src/serena/agent.py | 35 +++++++++++++++++++++++++---------- 1 file changed, 25 insertions(+), 10 deletions(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index 5b89910d..1cf21d9f 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -201,9 +201,15 @@ class SerenaAgent: process.join(timeout=1) class Task(ToStringMixin): - def __init__(self, function: Callable[[], Any], name: str): + def __init__(self, function: Callable[[], Any], name: str, logged: bool = True): + """ + :param function: the function representing the task to execute + :param name: the name of the task + :param logged: whether to log management of the task; if False, only errors will be logged + """ self.name = name self.future: concurrent.futures.Future = concurrent.futures.Future() + self.logged = logged self._function = function def _tostring_includes(self) -> list[str]: @@ -216,7 +222,7 @@ class SerenaAgent: def run_task() -> None: try: - with LogTime(self.name, logger=log): + with LogTime(self.name, logger=log, enabled=self.logged): result = self._function() if not self.future.done(): self.future.set_result(result) @@ -242,7 +248,8 @@ class SerenaAgent: # start task execution asynchronously with self._task_executor_lock: self._task_executor_current_task = task - log.info("Starting execution of %s", task) + if task.logged: + log.info("Starting execution of %s", task) task.start() # wait for task completion @@ -460,32 +467,40 @@ class SerenaAgent: log.info(f"Active tools ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}") - def issue_task(self, task: Callable[[], Any], name: str | None = None) -> Future: + def issue_task(self, task: Callable[[], Any], name: str | None = None, logged: bool = True) -> Future: """ Issue a task to the executor for asynchronous execution. It is ensured that tasks are executed in the order they are issued, one after another. :param task: the task to execute :param name: the name of the task for logging purposes; if None, use the task function's name + :param logged: whether to log management of the task; if False, only errors will be logged :return: a Future object representing the execution of the task """ with self._task_executor_lock: - task_name = f"Task-{self._task_executor_task_index}[{name or task.__name__}]" - self._task_executor_task_index += 1 - log.info(f"Scheduling {task_name}") - task = SerenaAgent.Task(function=task, name=task_name) + if logged: + task_prefix_name = f"Task-{self._task_executor_task_index}" + self._task_executor_task_index += 1 + else: + task_prefix_name = "BackgroundTask" + task_name = f"{task_prefix_name}:{name or task.__name__}" + if logged: + log.info(f"Scheduling {task_name}") + task = SerenaAgent.Task(function=task, name=task_name, logged=logged) self._task_executor_queue.append(task) return task.future - def execute_task(self, task: Callable[[], T], name: str | None = None) -> T: + def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True) -> T: """ Executes the given task synchronously via the agent's task executor. This is useful for tasks that need to be executed immediately and whose results are needed right away. :param task: the task to execute + :param name: the name of the task for logging purposes; if None, use the task function's name + :param logged: whether to log management of the task; if False, only errors will be logged :return: the result of the task execution """ - future = self.issue_task(task, name=name) + future = self.issue_task(task, name=name, logged=logged) return future.result() def is_using_language_server(self) -> bool: From 7bef2b21b53cc8c681b6ae32714f460f1be484ee Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 19:47:41 +0100 Subject: [PATCH 04/15] Handle cancellation before execution --- src/serena/agent.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/serena/agent.py b/src/serena/agent.py index 1cf21d9f..07128c2f 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -222,6 +222,10 @@ class SerenaAgent: def run_task() -> None: try: + if self.future.done(): + if self.logged: + log.info(f"Task {self.name} was already completed before start; skipping execution") + return with LogTime(self.name, logger=log, enabled=self.logged): result = self._function() if not self.future.done(): From 840ef02c716bdd2e9b6e3612f2fc17b716399396 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 20:05:33 +0100 Subject: [PATCH 05/15] SerenaAgent.get_current_tasks: Exclude completes/cancelled tasks --- src/serena/agent.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index 07128c2f..92ca0aec 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -238,6 +238,12 @@ class SerenaAgent: thread = Thread(target=run_task, name=self.name) thread.start() + def is_done(self) -> bool: + """ + :return: whether the task has completed (either successfully, with failure, or via cancellation) + """ + return self.future.done() + def _process_task_queue(self) -> None: while True: # obtain task from the queue @@ -296,7 +302,8 @@ class SerenaAgent: if self._task_executor_current_task is not None: tasks.append(self.TaskInfo.from_task(self._task_executor_current_task, True)) for task in self._task_executor_queue: - tasks.append(self.TaskInfo.from_task(task, False)) + if not task.is_done(): + tasks.append(self.TaskInfo.from_task(task, False)) return tasks def get_language_server_manager(self) -> LanguageServerManager | None: From 7aa1df2e414a6a648338c72237e012fc8e7e2041 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 20:43:44 +0100 Subject: [PATCH 06/15] Improve tests for task cancellation (testing before & during execution) --- test/serena/test_serena_agent.py | 56 ++++++++++++++++++++++---------- 1 file changed, 38 insertions(+), 18 deletions(-) diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index 72896e47..c88e620a 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -293,23 +293,25 @@ class TestSerenaAgent: symbols = json.loads(result) assert not symbols, f"Expected to find no symbols for {name_path}. Symbols found: {symbols}" - @staticmethod - def _create_task(delay: float, name: str, exception: bool = False): - def fn(): - time.sleep(delay) - if exception: - raise ValueError(f"Task {name} failed") - return True + class Task: + def __init__(self, delay: float, exception: bool = False): + self.delay = delay + self.exception = exception + self.did_run = False - fn.__name__ = name - return fn + def run(self): + self.did_run = True + time.sleep(self.delay) + if self.exception: + raise ValueError("Task failed") + return True def test_task_queue_sequence(self, basic_serena_agent): """ Tests that a sequence of tasks is executed correctly """ - future1 = basic_serena_agent.issue_task(self._create_task(1, "task1")) - future2 = basic_serena_agent.issue_task(self._create_task(1, "task2")) + future1 = basic_serena_agent.issue_task(self.Task(1).run, name="task1") + future2 = basic_serena_agent.issue_task(self.Task(1).run, name="task2") assert future1.result() is True assert future2.result() is True @@ -319,8 +321,8 @@ class TestSerenaAgent: * the exception is propagated, * subsequent tasks are still executed. """ - future1 = basic_serena_agent.issue_task(self._create_task(1, "task1", exception=True)) - future2 = basic_serena_agent.issue_task(self._create_task(1, "task2")) + future1 = basic_serena_agent.issue_task(self.Task(1, exception=True).run, name="task1") + future2 = basic_serena_agent.issue_task(self.Task(1).run, name="task2") have_exception = False try: assert future1.result() @@ -330,15 +332,15 @@ class TestSerenaAgent: assert have_exception assert future2.result() is True - def test_task_queue_cancel(self, basic_serena_agent): + def test_task_queue_cancel_current(self, basic_serena_agent): """ Tests that tasks that are cancelled are handled correctly, i.e. that * subsequent tasks are executed as soon as cancellation ensues. * the cancelled task raises CancelledError when result() is called. """ start_time = time.time() - future1 = basic_serena_agent.issue_task(self._create_task(10, "task1")) - future2 = basic_serena_agent.issue_task(self._create_task(1, "task2")) + future1 = basic_serena_agent.issue_task(self.Task(10).run, name="task1") + future2 = basic_serena_agent.issue_task(self.Task(1).run, name="task2") time.sleep(1) future1.cancel() assert future2.result() is True @@ -352,10 +354,28 @@ class TestSerenaAgent: have_cancelled_error = True assert have_cancelled_error + def test_task_queue_cancel_future(self, basic_serena_agent): + """ + Tests that when a future task is cancelled, it is never run at all + """ + task1 = self.Task(10) + task2 = self.Task(1) + future1 = basic_serena_agent.issue_task(task1.run, name="task1") + future2 = basic_serena_agent.issue_task(task2.run, name="task2") + time.sleep(1) + future2.cancel() + future1.cancel() + try: + future2.result() + except: + pass + assert task1.did_run + assert not task2.did_run + def test_task_queue_cancellation_via_task_info(self, basic_serena_agent): start_time = time.time() - basic_serena_agent.issue_task(self._create_task(10, "task1")) - basic_serena_agent.issue_task(self._create_task(1, "task2")) + basic_serena_agent.issue_task(self.Task(10).run, "task1") + basic_serena_agent.issue_task(self.Task(1).run, "task2") task_infos = basic_serena_agent.get_current_tasks() task_infos2 = basic_serena_agent.get_current_tasks() From aa21e9ffd09916dbb47782e48da07caab6b946a2 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:02:48 +0100 Subject: [PATCH 07/15] Add Task.wait_until_doen --- src/serena/agent.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index 92ca0aec..17ed378a 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -224,7 +224,7 @@ class SerenaAgent: try: if self.future.done(): if self.logged: - log.info(f"Task {self.name} was already completed before start; skipping execution") + log.info(f"Task {self.name} was already completed/cancelled; skipping execution") return with LogTime(self.name, logger=log, enabled=self.logged): result = self._function() @@ -244,6 +244,18 @@ class SerenaAgent: """ return self.future.done() + def wait_until_done(self, timeout: float | None = None) -> None: + """ + Waits until the task is done or the timeout is reached. + + :param timeout: the maximum time to wait in seconds, or None to wait indefinitely + :return: True if the task is done, False if the timeout was reached + """ + try: + self.future.result(timeout=timeout) + except: + pass + def _process_task_queue(self) -> None: while True: # obtain task from the queue @@ -259,14 +271,11 @@ class SerenaAgent: with self._task_executor_lock: self._task_executor_current_task = task if task.logged: - log.info("Starting execution of %s", task) + log.info("Starting execution of %s", task.name) task.start() # wait for task completion - try: - task.future.result() - except: - pass + task.wait_until_done() with self._task_executor_lock: self._task_executor_current_task = None From d7e294ed6fe0ee3787ad808511decdd5247c39e9 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:04:47 +0100 Subject: [PATCH 08/15] Split history src/serena/agent.py to src/serena/task_queue.py - rename file to target-name --- src/serena/{agent.py => task_queue.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/serena/{agent.py => task_queue.py} (100%) diff --git a/src/serena/agent.py b/src/serena/task_queue.py similarity index 100% rename from src/serena/agent.py rename to src/serena/task_queue.py From 6e34eb674ce1c893ebba5469497b695d469c3e88 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:04:48 +0100 Subject: [PATCH 09/15] Split history src/serena/agent.py to src/serena/task_queue.py - rename source-file to git-split-temp --- src/serena/agent.py => git-split-temp | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/serena/agent.py => git-split-temp (100%) diff --git a/src/serena/agent.py b/git-split-temp similarity index 100% rename from src/serena/agent.py rename to git-split-temp From 8ef2659910fc5fb8885335ff5f43206802f16146 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:04:49 +0100 Subject: [PATCH 10/15] Split history src/serena/agent.py to src/serena/task_queue.py - restore name of source-file --- git-split-temp => src/serena/agent.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename git-split-temp => src/serena/agent.py (100%) diff --git a/git-split-temp b/src/serena/agent.py similarity index 100% rename from git-split-temp rename to src/serena/agent.py From 8272e4e20c3ee13c13b235cad38a4982dc4db8e1 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:18:29 +0100 Subject: [PATCH 11/15] Factor out TaskExecutor from SerenaAgent --- src/serena/agent.py | 140 +--------- src/serena/task_queue.py | 554 +-------------------------------------- 2 files changed, 17 insertions(+), 677 deletions(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index 17ed378a..df5023dc 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -2,24 +2,18 @@ The Serena Model Context Protocol (MCP) Server """ -import concurrent.futures import multiprocessing import os import platform import sys -import threading -import time import webbrowser from collections.abc import Callable from concurrent.futures import Future -from dataclasses import dataclass from logging import Logger -from threading import Thread from typing import TYPE_CHECKING, Any, Optional, TypeVar from sensai.util import logging from sensai.util.logging import LogTime -from sensai.util.string import ToStringMixin from interprompt.jinja_template import JinjaTemplate from serena import serena_version @@ -30,6 +24,7 @@ from serena.dashboard import SerenaDashboardAPI from serena.ls_manager import LanguageServerManager from serena.project import Project from serena.prompt_factory import SerenaPromptFactory +from serena.task_queue import TaskExecutor from serena.tools import ActivateProjectTool, GetCurrentConfigTool, Tool, ToolMarker, ToolRegistry from serena.util.inspection import iter_subclasses from serena.util.logging import MemoryLogHandler @@ -158,12 +153,7 @@ class SerenaAgent: # create executor for starting the language server and running tools in another thread # This executor is used to achieve linear task execution, so it is important to use a single-threaded executor. - self._task_executor_lock = threading.Lock() - self._task_executor_queue: list[SerenaAgent.Task] = [] - self._task_executor_thread = Thread(target=self._process_task_queue, name="SerenaAgentTaskExecutor", daemon=True) - self._task_executor_thread.start() - self._task_executor_task_index = 1 - self._task_executor_current_task: Optional[SerenaAgent.Task] = None + self._task_executor = TaskExecutor("SerenaAgentTaskExecutor") # Initialize the prompt factory self.prompt_factory = SerenaPromptFactory() @@ -200,120 +190,14 @@ class SerenaAgent: process.start() process.join(timeout=1) - class Task(ToStringMixin): - def __init__(self, function: Callable[[], Any], name: str, logged: bool = True): - """ - :param function: the function representing the task to execute - :param name: the name of the task - :param logged: whether to log management of the task; if False, only errors will be logged - """ - self.name = name - self.future: concurrent.futures.Future = concurrent.futures.Future() - self.logged = logged - self._function = function - - def _tostring_includes(self) -> list[str]: - return ["name"] - - def start(self) -> None: - """ - Executes the task in a separate thread, setting the result or exception on the future. - """ - - def run_task() -> None: - try: - if self.future.done(): - if self.logged: - log.info(f"Task {self.name} was already completed/cancelled; skipping execution") - return - with LogTime(self.name, logger=log, enabled=self.logged): - result = self._function() - if not self.future.done(): - self.future.set_result(result) - except Exception as e: - if not self.future.done(): - log.error(f"Error during execution of {self.name}: {e}", exc_info=e) - self.future.set_exception(e) - - thread = Thread(target=run_task, name=self.name) - thread.start() - - def is_done(self) -> bool: - """ - :return: whether the task has completed (either successfully, with failure, or via cancellation) - """ - return self.future.done() - - def wait_until_done(self, timeout: float | None = None) -> None: - """ - Waits until the task is done or the timeout is reached. - - :param timeout: the maximum time to wait in seconds, or None to wait indefinitely - :return: True if the task is done, False if the timeout was reached - """ - try: - self.future.result(timeout=timeout) - except: - pass - - def _process_task_queue(self) -> None: - while True: - # obtain task from the queue - task: SerenaAgent.Task | None = None - with self._task_executor_lock: - if len(self._task_executor_queue) > 0: - task = self._task_executor_queue.pop(0) - if task is None: - time.sleep(0.1) - continue - - # start task execution asynchronously - with self._task_executor_lock: - self._task_executor_current_task = task - if task.logged: - log.info("Starting execution of %s", task.name) - task.start() - - # wait for task completion - task.wait_until_done() - with self._task_executor_lock: - self._task_executor_current_task = None - - @dataclass - class TaskInfo: - name: str - is_running: bool - future: Future - """ - future for accessing the task's result - """ - task_id: int - """ - unique identifier of the task - """ - - @staticmethod - def from_task(task: "SerenaAgent.Task", is_running: bool) -> "SerenaAgent.TaskInfo": - return SerenaAgent.TaskInfo(name=task.name, is_running=is_running, future=task.future, task_id=id(task)) - - def cancel(self) -> None: - self.future.cancel() - - def get_current_tasks(self) -> list["SerenaAgent.TaskInfo"]: + def get_current_tasks(self) -> list[TaskExecutor.TaskInfo]: """ Gets the list of tasks currently running or queued for execution. The function returns a list of thread-safe TaskInfo objects (specifically created for the caller). :return: the list of tasks in the execution order (running task first) """ - tasks = [] - with self._task_executor_lock: - if self._task_executor_current_task is not None: - tasks.append(self.TaskInfo.from_task(self._task_executor_current_task, True)) - for task in self._task_executor_queue: - if not task.is_done(): - tasks.append(self.TaskInfo.from_task(task, False)) - return tasks + return self._task_executor.get_current_tasks() def get_language_server_manager(self) -> LanguageServerManager | None: if self._active_project is not None: @@ -497,18 +381,7 @@ class SerenaAgent: :param logged: whether to log management of the task; if False, only errors will be logged :return: a Future object representing the execution of the task """ - with self._task_executor_lock: - if logged: - task_prefix_name = f"Task-{self._task_executor_task_index}" - self._task_executor_task_index += 1 - else: - task_prefix_name = "BackgroundTask" - task_name = f"{task_prefix_name}:{name or task.__name__}" - if logged: - log.info(f"Scheduling {task_name}") - task = SerenaAgent.Task(function=task, name=task_name, logged=logged) - self._task_executor_queue.append(task) - return task.future + return self._task_executor.issue_task(task, name=name, logged=logged) def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True) -> T: """ @@ -520,8 +393,7 @@ class SerenaAgent: :param logged: whether to log management of the task; if False, only errors will be logged :return: the result of the task execution """ - future = self.issue_task(task, name=name, logged=logged) - return future.result() + return self._task_executor.execute_task(task, name=name, logged=logged) def is_using_language_server(self) -> bool: """ diff --git a/src/serena/task_queue.py b/src/serena/task_queue.py index 17ed378a..9ea20139 100644 --- a/src/serena/task_queue.py +++ b/src/serena/task_queue.py @@ -3,202 +3,30 @@ The Serena Model Context Protocol (MCP) Server """ import concurrent.futures -import multiprocessing -import os -import platform -import sys import threading import time -import webbrowser from collections.abc import Callable from concurrent.futures import Future from dataclasses import dataclass -from logging import Logger from threading import Thread -from typing import TYPE_CHECKING, Any, Optional, TypeVar +from typing import Any, Optional, TypeVar from sensai.util import logging from sensai.util.logging import LogTime from sensai.util.string import ToStringMixin -from interprompt.jinja_template import JinjaTemplate -from serena import serena_version -from serena.analytics import RegisteredTokenCountEstimator, ToolUsageStats -from serena.config.context_mode import RegisteredContext, SerenaAgentContext, SerenaAgentMode -from serena.config.serena_config import SerenaConfig, ToolInclusionDefinition, ToolSet -from serena.dashboard import SerenaDashboardAPI -from serena.ls_manager import LanguageServerManager -from serena.project import Project -from serena.prompt_factory import SerenaPromptFactory -from serena.tools import ActivateProjectTool, GetCurrentConfigTool, Tool, ToolMarker, ToolRegistry -from serena.util.inspection import iter_subclasses -from serena.util.logging import MemoryLogHandler -from solidlsp.ls_config import Language - -if TYPE_CHECKING: - from serena.gui_log_viewer import GuiLogViewer - log = logging.getLogger(__name__) -TTool = TypeVar("TTool", bound="Tool") T = TypeVar("T") -SUCCESS_RESULT = "OK" -class ProjectNotFoundError(Exception): - pass - - -class AvailableTools: - def __init__(self, tools: list[Tool]): - """ - :param tools: the list of available tools - """ - self.tools = tools - self.tool_names = [tool.get_name_from_cls() for tool in tools] - self.tool_marker_names = set() - for marker_class in iter_subclasses(ToolMarker): - for tool in tools: - if isinstance(tool, marker_class): - self.tool_marker_names.add(marker_class.__name__) - - def __len__(self) -> int: - return len(self.tools) - - -class SerenaAgent: - def __init__( - self, - project: str | None = None, - project_activation_callback: Callable[[], None] | None = None, - serena_config: SerenaConfig | None = None, - context: SerenaAgentContext | None = None, - modes: list[SerenaAgentMode] | None = None, - memory_log_handler: MemoryLogHandler | None = None, - ): - """ - :param project: the project to load immediately or None to not load any project; may be a path to the project or a name of - an already registered project; - :param project_activation_callback: a callback function to be called when a project is activated. - :param serena_config: the Serena configuration or None to read the configuration from the default location. - :param context: the context in which the agent is operating, None for default context. - The context may adjust prompts, tool availability, and tool descriptions. - :param modes: list of modes in which the agent is operating (they will be combined), None for default modes. - The modes may adjust prompts, tool availability, and tool descriptions. - :param memory_log_handler: a MemoryLogHandler instance from which to read log messages; if None, a new one will be created - if necessary. - """ - # obtain serena configuration using the decoupled factory function - self.serena_config = serena_config or SerenaConfig.from_config_file() - - # project-specific instances, which will be initialized upon project activation - self._active_project: Project | None = None - - # adjust log level - serena_log_level = self.serena_config.log_level - if Logger.root.level != serena_log_level: - log.info(f"Changing the root logger level to {serena_log_level}") - Logger.root.setLevel(serena_log_level) - - def get_memory_log_handler() -> MemoryLogHandler: - nonlocal memory_log_handler - if memory_log_handler is None: - memory_log_handler = MemoryLogHandler(level=serena_log_level) - Logger.root.addHandler(memory_log_handler) - return memory_log_handler - - # open GUI log window if enabled - self._gui_log_viewer: Optional["GuiLogViewer"] = None - if self.serena_config.gui_log_window_enabled: - if platform.system() == "Darwin": - log.warning("GUI log window is not supported on macOS") - else: - # even importing on macOS may fail if tkinter dependencies are unavailable (depends on Python interpreter installation - # which uv used as a base, unfortunately) - from serena.gui_log_viewer import GuiLogViewer - - self._gui_log_viewer = GuiLogViewer("dashboard", title="Serena Logs", memory_log_handler=get_memory_log_handler()) - self._gui_log_viewer.start() - - # set the agent context - if context is None: - context = SerenaAgentContext.load_default() - self._context = context - - # instantiate all tool classes - self._all_tools: dict[type[Tool], Tool] = {tool_class: tool_class(self) for tool_class in ToolRegistry().get_all_tool_classes()} - tool_names = [tool.get_name_from_cls() for tool in self._all_tools.values()] - - # If GUI log window is enabled, set the tool names for highlighting - if self._gui_log_viewer is not None: - self._gui_log_viewer.set_tool_names(tool_names) - - token_count_estimator = RegisteredTokenCountEstimator[self.serena_config.token_count_estimator] - log.info(f"Will record tool usage statistics with token count estimator: {token_count_estimator.name}.") - self._tool_usage_stats = ToolUsageStats(token_count_estimator) - - # log fundamental information - log.info(f"Starting Serena server (version={serena_version()}, process id={os.getpid()}, parent process id={os.getppid()})") - log.info("Configuration file: %s", self.serena_config.config_file_path) - log.info("Available projects: {}".format(", ".join(self.serena_config.project_names))) - log.info(f"Loaded tools ({len(self._all_tools)}): {', '.join([tool.get_name_from_cls() for tool in self._all_tools.values()])}") - - self._check_shell_settings() - - # determine the base toolset defining the set of exposed tools (which e.g. the MCP shall see), - # limited by the Serena config, the context (which is fixed for the session) and JetBrains mode - tool_inclusion_definitions: list[ToolInclusionDefinition] = [self.serena_config, self._context] - if self._context.name == RegisteredContext.IDE_ASSISTANT.value: - tool_inclusion_definitions.extend(self._ide_assistant_context_tool_inclusion_definitions(project)) - if self.serena_config.jetbrains: - tool_inclusion_definitions.append(SerenaAgentMode.from_name_internal("jetbrains")) - - self._base_tool_set = ToolSet.default().apply(*tool_inclusion_definitions) - self._exposed_tools = AvailableTools([t for t in self._all_tools.values() if self._base_tool_set.includes_name(t.get_name())]) - log.info(f"Number of exposed tools: {len(self._exposed_tools)}") - - # create executor for starting the language server and running tools in another thread - # This executor is used to achieve linear task execution, so it is important to use a single-threaded executor. +class TaskExecutor: + def __init__(self, name: str): self._task_executor_lock = threading.Lock() - self._task_executor_queue: list[SerenaAgent.Task] = [] - self._task_executor_thread = Thread(target=self._process_task_queue, name="SerenaAgentTaskExecutor", daemon=True) + self._task_executor_queue: list[TaskExecutor.Task] = [] + self._task_executor_thread = Thread(target=self._process_task_queue, name=name, daemon=True) self._task_executor_thread.start() self._task_executor_task_index = 1 - self._task_executor_current_task: Optional[SerenaAgent.Task] = None - - # Initialize the prompt factory - self.prompt_factory = SerenaPromptFactory() - self._project_activation_callback = project_activation_callback - - # set the active modes - if modes is None: - modes = SerenaAgentMode.load_default_modes() - self._modes = modes - - self._active_tools: dict[type[Tool], Tool] = {} - self._update_active_tools() - - # activate a project configuration (if provided or if there is only a single project available) - if project is not None: - try: - self.activate_project_from_path_or_name(project) - except Exception as e: - log.error(f"Error activating project '{project}' at startup: {e}", exc_info=e) - - # start the dashboard (web frontend), registering its log handler - # should be the last thing to happen in the initialization since the dashboard - # may access various parts of the agent - if self.serena_config.web_dashboard: - self._dashboard_thread, port = SerenaDashboardAPI( - get_memory_log_handler(), tool_names, agent=self, tool_usage_stats=self._tool_usage_stats - ).run_in_thread() - dashboard_url = f"http://127.0.0.1:{port}/dashboard/index.html" - log.info("Serena web dashboard started at %s", dashboard_url) - if self.serena_config.web_dashboard_open_on_launch: - # open the dashboard URL in the default web browser (using a separate process to control - # output redirection) - process = multiprocessing.Process(target=self._open_dashboard, args=(dashboard_url,)) - process.start() - process.join(timeout=1) + self._task_executor_current_task: Optional[TaskExecutor.Task] = None class Task(ToStringMixin): def __init__(self, function: Callable[[], Any], name: str, logged: bool = True): @@ -259,7 +87,7 @@ class SerenaAgent: def _process_task_queue(self) -> None: while True: # obtain task from the queue - task: SerenaAgent.Task | None = None + task: TaskExecutor.Task | None = None with self._task_executor_lock: if len(self._task_executor_queue) > 0: task = self._task_executor_queue.pop(0) @@ -293,13 +121,13 @@ class SerenaAgent: """ @staticmethod - def from_task(task: "SerenaAgent.Task", is_running: bool) -> "SerenaAgent.TaskInfo": - return SerenaAgent.TaskInfo(name=task.name, is_running=is_running, future=task.future, task_id=id(task)) + def from_task(task: "TaskExecutor.Task", is_running: bool) -> "TaskExecutor.TaskInfo": + return TaskExecutor.TaskInfo(name=task.name, is_running=is_running, future=task.future, task_id=id(task)) def cancel(self) -> None: self.future.cancel() - def get_current_tasks(self) -> list["SerenaAgent.TaskInfo"]: + def get_current_tasks(self) -> list["TaskExecutor.TaskInfo"]: """ Gets the list of tasks currently running or queued for execution. The function returns a list of thread-safe TaskInfo objects (specifically created for the caller). @@ -315,178 +143,6 @@ class SerenaAgent: tasks.append(self.TaskInfo.from_task(task, False)) return tasks - def get_language_server_manager(self) -> LanguageServerManager | None: - if self._active_project is not None: - return self._active_project.language_server_manager - return None - - def get_language_server_manager_or_raise(self) -> LanguageServerManager: - language_server_manager = self.get_language_server_manager() - if language_server_manager is None: - raise Exception( - "The language server manager is not initialized, indicating a problem during project activation. " - "Inform the user, telling them to inspect Serena's logs in order to determine the issue. " - "IMPORTANT: Wait for further instructions before you continue!" - ) - return language_server_manager - - def get_context(self) -> SerenaAgentContext: - return self._context - - def get_tool_description_override(self, tool_name: str) -> str | None: - return self._context.tool_description_overrides.get(tool_name, None) - - def _check_shell_settings(self) -> None: - # On Windows, Claude Code sets COMSPEC to Git-Bash (often even with a path containing spaces), - # which causes all sorts of trouble, preventing language servers from being launched correctly. - # So we make sure that COMSPEC is unset if it has been set to bash specifically. - if platform.system() == "Windows": - comspec = os.environ.get("COMSPEC", "") - if "bash" in comspec: - os.environ["COMSPEC"] = "" # force use of default shell - log.info("Adjusting COMSPEC environment variable to use the default shell instead of '%s'", comspec) - - def _ide_assistant_context_tool_inclusion_definitions(self, project_root_or_name: str | None) -> list[ToolInclusionDefinition]: - """ - In the IDE assistant context, the agent is assumed to work on a single project, and we thus - want to apply that project's tool exclusions/inclusions from the get-go, limiting the set - of tools that will be exposed to the client. - Furthermore, we disable tools that are only relevant for project activation. - So if the project exists, we apply all the aforementioned exclusions. - - :param project_root_or_name: the project root path or project name - :return: - """ - tool_inclusion_definitions = [] - if project_root_or_name is not None: - # Note: Auto-generation is disabled, because the result must be returned instantaneously - # (project generation could take too much time), so as not to delay MCP server startup - # and provide responses to the client immediately. - project = self.load_project_from_path_or_name(project_root_or_name, autogenerate=False) - if project is not None: - tool_inclusion_definitions.append( - ToolInclusionDefinition( - excluded_tools=[ActivateProjectTool.get_name_from_cls(), GetCurrentConfigTool.get_name_from_cls()] - ) - ) - tool_inclusion_definitions.append(project.project_config) - return tool_inclusion_definitions - - def record_tool_usage(self, input_kwargs: dict, tool_result: str | dict, tool: Tool) -> None: - """ - Record the usage of a tool with the given input and output strings if tool usage statistics recording is enabled. - """ - tool_name = tool.get_name() - input_str = str(input_kwargs) - output_str = str(tool_result) - log.debug(f"Recording tool usage for tool '{tool_name}'") - self._tool_usage_stats.record_tool_usage(tool_name, input_str, output_str) - - @staticmethod - def _open_dashboard(url: str) -> None: - # Redirect stdout and stderr file descriptors to /dev/null, - # making sure that nothing can be written to stdout/stderr, even by subprocesses - null_fd = os.open(os.devnull, os.O_WRONLY) - os.dup2(null_fd, sys.stdout.fileno()) - os.dup2(null_fd, sys.stderr.fileno()) - os.close(null_fd) - - # open the dashboard URL in the default web browser - webbrowser.open(url) - - def get_project_root(self) -> str: - """ - :return: the root directory of the active project (if any); raises a ValueError if there is no active project - """ - project = self.get_active_project() - if project is None: - raise ValueError("Cannot get project root if no project is active.") - return project.project_root - - def get_exposed_tool_instances(self) -> list["Tool"]: - """ - :return: the tool instances which are exposed (e.g. to the MCP client). - Note that the set of exposed tools is fixed for the session, as - clients don't react to changes in the set of tools, so this is the superset - of tools that can be offered during the session. - If a client should attempt to use a tool that is dynamically disabled - (e.g. because a project is activated that disables it), it will receive an error. - """ - return list(self._exposed_tools.tools) - - def get_active_project(self) -> Project | None: - """ - :return: the active project or None if no project is active - """ - return self._active_project - - def get_active_project_or_raise(self) -> Project: - """ - :return: the active project or raises an exception if no project is active - """ - project = self.get_active_project() - if project is None: - raise ValueError("No active project. Please activate a project first.") - return project - - def set_modes(self, modes: list[SerenaAgentMode]) -> None: - """ - Set the current mode configurations. - - :param modes: List of mode names or paths to use - """ - self._modes = modes - self._update_active_tools() - - log.info(f"Set modes to {[mode.name for mode in modes]}") - - def get_active_modes(self) -> list[SerenaAgentMode]: - """ - :return: the list of active modes - """ - return list(self._modes) - - def _format_prompt(self, prompt_template: str) -> str: - template = JinjaTemplate(prompt_template) - return template.render(available_tools=self._exposed_tools.tool_names, available_markers=self._exposed_tools.tool_marker_names) - - def create_system_prompt(self) -> str: - available_markers = self._exposed_tools.tool_marker_names - log.info("Generating system prompt with available_tools=(see exposed tools), available_markers=%s", available_markers) - system_prompt = self.prompt_factory.create_system_prompt( - context_system_prompt=self._format_prompt(self._context.prompt), - mode_system_prompts=[self._format_prompt(mode.prompt) for mode in self._modes], - available_tools=self._exposed_tools.tool_names, - available_markers=available_markers, - ) - - # If a project is active at startup, append its activation message - if self._active_project is not None: - system_prompt += "\n\n" + self._active_project.get_activation_message() - - log.info("System prompt:\n%s", system_prompt) - return system_prompt - - def _update_active_tools(self) -> None: - """ - Update the active tools based on enabled modes and the active project. - The base tool set already takes the Serena configuration and the context into account - (as well as any internal modes that are not handled dynamically, such as JetBrains mode). - """ - tool_set = self._base_tool_set.apply(*self._modes) - if self._active_project is not None: - tool_set = tool_set.apply(self._active_project.project_config) - if self._active_project.project_config.read_only: - tool_set = tool_set.without_editing_tools() - - self._active_tools = { - tool_class: tool_instance - for tool_class, tool_instance in self._all_tools.items() - if tool_set.includes_name(tool_instance.get_name()) - } - - log.info(f"Active tools ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}") - def issue_task(self, task: Callable[[], Any], name: str | None = None, logged: bool = True) -> Future: """ Issue a task to the executor for asynchronous execution. @@ -506,7 +162,7 @@ class SerenaAgent: task_name = f"{task_prefix_name}:{name or task.__name__}" if logged: log.info(f"Scheduling {task_name}") - task = SerenaAgent.Task(function=task, name=task_name, logged=logged) + task = self.Task(function=task, name=task_name, logged=logged) self._task_executor_queue.append(task) return task.future @@ -522,191 +178,3 @@ class SerenaAgent: """ future = self.issue_task(task, name=name, logged=logged) return future.result() - - def is_using_language_server(self) -> bool: - """ - :return: whether this agent uses language server-based code analysis - """ - return not self.serena_config.jetbrains - - def _activate_project(self, project: Project) -> None: - log.info(f"Activating {project.project_name} at {project.project_root}") - self._active_project = project - self._update_active_tools() - - def init_language_server_manager() -> None: - # start the language server - with LogTime("Language server initialization", logger=log): - self.reset_language_server_manager() - - # initialize the language server in the background (if in language server mode) - if self.is_using_language_server(): - self.issue_task(init_language_server_manager) - - if self._project_activation_callback is not None: - self._project_activation_callback() - - def load_project_from_path_or_name(self, project_root_or_name: str, autogenerate: bool) -> Project | None: - """ - Get a project instance from a path or a name. - - :param project_root_or_name: the path to the project root or the name of the project - :param autogenerate: whether to autogenerate the project for the case where first argument is a directory - which does not yet contain a Serena project configuration file - :return: the project instance if it was found/could be created, None otherwise - """ - project_instance: Project | None = self.serena_config.get_project(project_root_or_name) - if project_instance is not None: - log.info(f"Found registered project '{project_instance.project_name}' at path {project_instance.project_root}") - elif autogenerate and os.path.isdir(project_root_or_name): - project_instance = self.serena_config.add_project_from_path(project_root_or_name) - log.info(f"Added new project {project_instance.project_name} for path {project_instance.project_root}") - return project_instance - - def activate_project_from_path_or_name(self, project_root_or_name: str) -> Project: - """ - Activate a project from a path or a name. - If the project was already registered, it will just be activated. - If the argument is a path at which no Serena project previously existed, the project will be created beforehand. - Raises ProjectNotFoundError if the project could neither be found nor created. - - :return: a tuple of the project instance and a Boolean indicating whether the project was newly - created - """ - project_instance: Project | None = self.load_project_from_path_or_name(project_root_or_name, autogenerate=True) - if project_instance is None: - raise ProjectNotFoundError( - f"Project '{project_root_or_name}' not found: Not a valid project name or directory. " - f"Existing project names: {self.serena_config.project_names}" - ) - self._activate_project(project_instance) - return project_instance - - def get_active_tool_classes(self) -> list[type["Tool"]]: - """ - :return: the list of active tool classes for the current project - """ - return list(self._active_tools.keys()) - - def get_active_tool_names(self) -> list[str]: - """ - :return: the list of names of the active tools for the current project - """ - return sorted([tool.get_name_from_cls() for tool in self.get_active_tool_classes()]) - - def tool_is_active(self, tool_class: type["Tool"] | str) -> bool: - """ - :param tool_class: the class or name of the tool to check - :return: True if the tool is active, False otherwise - """ - if isinstance(tool_class, str): - return tool_class in self.get_active_tool_names() - else: - return tool_class in self.get_active_tool_classes() - - def get_current_config_overview(self) -> str: - """ - :return: a string overview of the current configuration, including the active and available configuration options - """ - result_str = "Current configuration:\n" - result_str += f"Serena version: {serena_version()}\n" - result_str += f"Loglevel: {self.serena_config.log_level}, trace_lsp_communication={self.serena_config.trace_lsp_communication}\n" - if self._active_project is not None: - result_str += f"Active project: {self._active_project.project_name}\n" - else: - result_str += "No active project\n" - result_str += "Available projects:\n" + "\n".join(list(self.serena_config.project_names)) + "\n" - result_str += f"Active context: {self._context.name}\n" - - # Active modes - active_mode_names = [mode.name for mode in self.get_active_modes()] - result_str += "Active modes: {}\n".format(", ".join(active_mode_names)) + "\n" - - # Available but not active modes - all_available_modes = SerenaAgentMode.list_registered_mode_names() - inactive_modes = [mode for mode in all_available_modes if mode not in active_mode_names] - if inactive_modes: - result_str += "Available but not active modes: {}\n".format(", ".join(inactive_modes)) + "\n" - - # Active tools - result_str += "Active tools (after all exclusions from the project, context, and modes):\n" - active_tool_names = self.get_active_tool_names() - # print the tool names in chunks - chunk_size = 4 - for i in range(0, len(active_tool_names), chunk_size): - chunk = active_tool_names[i : i + chunk_size] - result_str += " " + ", ".join(chunk) + "\n" - - # Available but not active tools - all_tool_names = sorted([tool.get_name_from_cls() for tool in self._all_tools.values()]) - inactive_tool_names = [tool for tool in all_tool_names if tool not in active_tool_names] - if inactive_tool_names: - result_str += "Available but not active tools:\n" - for i in range(0, len(inactive_tool_names), chunk_size): - chunk = inactive_tool_names[i : i + chunk_size] - result_str += " " + ", ".join(chunk) + "\n" - - return result_str - - def reset_language_server_manager(self) -> None: - """ - Starts/resets the language server manager for the current project - """ - tool_timeout = self.serena_config.tool_timeout - if tool_timeout is None or tool_timeout < 0: - ls_timeout = None - else: - if tool_timeout < 10: - raise ValueError(f"Tool timeout must be at least 10 seconds, but is {tool_timeout} seconds") - ls_timeout = tool_timeout - 5 # the LS timeout is for a single call, it should be smaller than the tool timeout - - # instantiate and start the necessary language servers - self.get_active_project_or_raise().create_language_server_manager( - log_level=self.serena_config.log_level, - ls_timeout=ls_timeout, - trace_lsp_communication=self.serena_config.trace_lsp_communication, - ls_specific_settings=self.serena_config.ls_specific_settings, - ) - - def add_language(self, language: Language) -> None: - """ - Adds a new language to the active project, spawning the respective language server and updating the project configuration. - The addition is scheduled via the agent's task executor and executed synchronously, i.e. the method returns - when the addition is complete. - - :param language: the language to add - """ - self.execute_task(lambda: self.get_active_project_or_raise().add_language(language), name=f"AddLanguage:{language.value}") - - def remove_language(self, language: Language) -> None: - """ - Removes a language from the active project, shutting down the respective language server and updating the project configuration. - The removal is scheduled via the agent's task executor and executed asynchronously. - - :param language: the language to remove - """ - self.issue_task(lambda: self.get_active_project_or_raise().remove_language(language), name=f"RemoveLanguage:{language.value}") - - def get_tool(self, tool_class: type[TTool]) -> TTool: - return self._all_tools[tool_class] # type: ignore - - def print_tool_overview(self) -> None: - ToolRegistry().print_tool_overview(self._active_tools.values()) - - def __del__(self) -> None: - """ - Destructor to clean up the language server instance and GUI logger - """ - if not hasattr(self, "_is_initialized"): - return - log.info("SerenaAgent is shutting down ...") - if self._active_project is not None: - self._active_project.shutdown() - self._active_project = None - if self._gui_log_viewer: - log.info("Stopping the GUI log window ...") - self._gui_log_viewer.stop() - - def get_tool_by_name(self, tool_name: str) -> Tool: - tool_class = ToolRegistry().get_tool_class_by_name(tool_name) - return self.get_tool(tool_class) From 34dd560d2c8e515a42c5051a6a9cc60831adf95f Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:18:50 +0100 Subject: [PATCH 12/15] Rename module task_queue -> task_executor --- src/serena/agent.py | 2 +- src/serena/{task_queue.py => task_executor.py} | 0 2 files changed, 1 insertion(+), 1 deletion(-) rename src/serena/{task_queue.py => task_executor.py} (100%) diff --git a/src/serena/agent.py b/src/serena/agent.py index df5023dc..78adcdc3 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -24,7 +24,7 @@ from serena.dashboard import SerenaDashboardAPI from serena.ls_manager import LanguageServerManager from serena.project import Project from serena.prompt_factory import SerenaPromptFactory -from serena.task_queue import TaskExecutor +from serena.task_executor import TaskExecutor from serena.tools import ActivateProjectTool, GetCurrentConfigTool, Tool, ToolMarker, ToolRegistry from serena.util.inspection import iter_subclasses from serena.util.logging import MemoryLogHandler diff --git a/src/serena/task_queue.py b/src/serena/task_executor.py similarity index 100% rename from src/serena/task_queue.py rename to src/serena/task_executor.py From 04c9a680ead34a0948689e4410b3415b798b33d6 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:36:26 +0100 Subject: [PATCH 13/15] Improve timeout handling of tasks, * allowing a timeout to be passed at task construction * using the given timeout (if any) for the task queue itself and for result retrieval --- src/serena/agent.py | 17 ++++++----- src/serena/task_executor.py | 58 +++++++++++++++++++++++++------------ 2 files changed, 50 insertions(+), 25 deletions(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index 78adcdc3..aecc21d3 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -8,9 +8,8 @@ import platform import sys import webbrowser from collections.abc import Callable -from concurrent.futures import Future from logging import Logger -from typing import TYPE_CHECKING, Any, Optional, TypeVar +from typing import TYPE_CHECKING, Optional, TypeVar from sensai.util import logging from sensai.util.logging import LogTime @@ -371,7 +370,9 @@ class SerenaAgent: log.info(f"Active tools ({len(self._active_tools)}): {', '.join(self.get_active_tool_names())}") - def issue_task(self, task: Callable[[], Any], name: str | None = None, logged: bool = True) -> Future: + def issue_task( + self, task: Callable[[], T], name: str | None = None, logged: bool = True, timeout: float | None = None + ) -> TaskExecutor.Task[T]: """ Issue a task to the executor for asynchronous execution. It is ensured that tasks are executed in the order they are issued, one after another. @@ -379,11 +380,12 @@ class SerenaAgent: :param task: the task to execute :param name: the name of the task for logging purposes; if None, use the task function's name :param logged: whether to log management of the task; if False, only errors will be logged - :return: a Future object representing the execution of the task + :param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely + :return: the task object, through which the task's future result can be accessed """ - return self._task_executor.issue_task(task, name=name, logged=logged) + return self._task_executor.issue_task(task, name=name, logged=logged, timeout=timeout) - def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True) -> T: + def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True, timeout: float | None = None) -> T: """ Executes the given task synchronously via the agent's task executor. This is useful for tasks that need to be executed immediately and whose results are needed right away. @@ -391,9 +393,10 @@ class SerenaAgent: :param task: the task to execute :param name: the name of the task for logging purposes; if None, use the task function's name :param logged: whether to log management of the task; if False, only errors will be logged + :param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely :return: the result of the task execution """ - return self._task_executor.execute_task(task, name=name, logged=logged) + return self._task_executor.execute_task(task, name=name, logged=logged, timeout=timeout) def is_using_language_server(self) -> bool: """ diff --git a/src/serena/task_executor.py b/src/serena/task_executor.py index 9ea20139..bae3d325 100644 --- a/src/serena/task_executor.py +++ b/src/serena/task_executor.py @@ -1,7 +1,3 @@ -""" -The Serena Model Context Protocol (MCP) Server -""" - import concurrent.futures import threading import time @@ -9,7 +5,7 @@ from collections.abc import Callable from concurrent.futures import Future from dataclasses import dataclass from threading import Thread -from typing import Any, Optional, TypeVar +from typing import Generic, Optional, TypeVar from sensai.util import logging from sensai.util.logging import LogTime @@ -28,16 +24,18 @@ class TaskExecutor: self._task_executor_task_index = 1 self._task_executor_current_task: Optional[TaskExecutor.Task] = None - class Task(ToStringMixin): - def __init__(self, function: Callable[[], Any], name: str, logged: bool = True): + class Task(ToStringMixin, Generic[T]): + def __init__(self, function: Callable[[], T], name: str, logged: bool = True, timeout: Optional[float] = None): """ :param function: the function representing the task to execute :param name: the name of the task :param logged: whether to log management of the task; if False, only errors will be logged + :param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely """ self.name = name self.future: concurrent.futures.Future = concurrent.futures.Future() self.logged = logged + self.timeout = timeout self._function = function def _tostring_includes(self) -> list[str]: @@ -72,12 +70,34 @@ class TaskExecutor: """ return self.future.done() + def result(self, timeout: float | None = None) -> T: + """ + Blocks until the task is done or the timeout is reached, and returns the result. + If an exception occurred during task execution, it is raised here. + If the timeout is reached, a TimeoutError is raised (but the task is not cancelled). + If the task is cancelled, a CancelledError is raised. + + :param timeout: the maximum time to wait in seconds; if None, use the task's own timeout + (which may be None to wait indefinitely) + :return: True if the task is done, False if the timeout was reached + """ + return self.future.result(timeout=timeout) + + def cancel(self) -> None: + """ + Cancels the task. If it has not yet started, it will not be executed. + If it has already started, its future will be marked as cancelled and will raise a CancelledError + when its result is requested. + """ + self.future.cancel() + def wait_until_done(self, timeout: float | None = None) -> None: """ Waits until the task is done or the timeout is reached. + The task is done if it either completed successfully, failed with an exception, or was cancelled. - :param timeout: the maximum time to wait in seconds, or None to wait indefinitely - :return: True if the task is done, False if the timeout was reached + :param timeout: the maximum time to wait in seconds; if None, use the task's own timeout + (which may be None to wait indefinitely) """ try: self.future.result(timeout=timeout) @@ -103,7 +123,7 @@ class TaskExecutor: task.start() # wait for task completion - task.wait_until_done() + task.wait_until_done(timeout=task.timeout) with self._task_executor_lock: self._task_executor_current_task = None @@ -143,7 +163,7 @@ class TaskExecutor: tasks.append(self.TaskInfo.from_task(task, False)) return tasks - def issue_task(self, task: Callable[[], Any], name: str | None = None, logged: bool = True) -> Future: + def issue_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True, timeout: float | None = None) -> Task[T]: """ Issue a task to the executor for asynchronous execution. It is ensured that tasks are executed in the order they are issued, one after another. @@ -151,7 +171,8 @@ class TaskExecutor: :param task: the task to execute :param name: the name of the task for logging purposes; if None, use the task function's name :param logged: whether to log management of the task; if False, only errors will be logged - :return: a Future object representing the execution of the task + :param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely + :return: the task object, through which the task's future result can be accessed """ with self._task_executor_lock: if logged: @@ -162,11 +183,11 @@ class TaskExecutor: task_name = f"{task_prefix_name}:{name or task.__name__}" if logged: log.info(f"Scheduling {task_name}") - task = self.Task(function=task, name=task_name, logged=logged) - self._task_executor_queue.append(task) - return task.future + task_obj = self.Task(function=task, name=task_name, logged=logged, timeout=timeout) + self._task_executor_queue.append(task_obj) + return task_obj - def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True) -> T: + def execute_task(self, task: Callable[[], T], name: str | None = None, logged: bool = True, timeout: float | None = None) -> T: """ Executes the given task synchronously via the agent's task executor. This is useful for tasks that need to be executed immediately and whose results are needed right away. @@ -174,7 +195,8 @@ class TaskExecutor: :param task: the task to execute :param name: the name of the task for logging purposes; if None, use the task function's name :param logged: whether to log management of the task; if False, only errors will be logged + :param timeout: the maximum time to wait for task completion in seconds, or None to wait indefinitely :return: the result of the task execution """ - future = self.issue_task(task, name=name, logged=logged) - return future.result() + task_obj = self.issue_task(task, name=name, logged=logged, timeout=timeout) + return task_obj.result() From 854bb9771091e40051f1daa7427e74443460d14a Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:39:28 +0100 Subject: [PATCH 14/15] Handle TimeoutError in tool execution --- src/serena/tools/tools_base.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/serena/tools/tools_base.py b/src/serena/tools/tools_base.py index c4b0f300..a79f2911 100644 --- a/src/serena/tools/tools_base.py +++ b/src/serena/tools/tools_base.py @@ -286,8 +286,14 @@ class Tool(Component): return result - future = self.agent.issue_task(task, name=self.__class__.__name__) - return future.result(timeout=self.agent.serena_config.tool_timeout) + # execute the tool in the agent's task executor, with timeout + try: + task_exec = self.agent.issue_task(task, name=self.__class__.__name__) + return task_exec.result(timeout=self.agent.serena_config.tool_timeout) + except Exception as e: # typically TimeoutError (other exceptions caught in task) + msg = f"Error: {e.__class__.__name__} - {e}" + log.error(msg) + return msg class EditedFileContext: From fcd4e6d90cec8c15bde142c0867c8386074e0990 Mon Sep 17 00:00:00 2001 From: Dominik Jain Date: Thu, 30 Oct 2025 21:49:22 +0100 Subject: [PATCH 15/15] Move TaskExecutor tests to a separate test file --- src/serena/agent.py | 2 +- test/serena/test_serena_agent.py | 108 --------------------------- test/serena/test_task_executor.py | 119 ++++++++++++++++++++++++++++++ 3 files changed, 120 insertions(+), 109 deletions(-) create mode 100644 test/serena/test_task_executor.py diff --git a/src/serena/agent.py b/src/serena/agent.py index aecc21d3..fc4ab315 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -151,7 +151,7 @@ class SerenaAgent: log.info(f"Number of exposed tools: {len(self._exposed_tools)}") # create executor for starting the language server and running tools in another thread - # This executor is used to achieve linear task execution, so it is important to use a single-threaded executor. + # This executor is used to achieve linear task execution self._task_executor = TaskExecutor("SerenaAgentTaskExecutor") # Initialize the prompt factory diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index c88e620a..882b45be 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -61,14 +61,6 @@ def serena_agent(request: pytest.FixtureRequest, serena_config): return SerenaAgent(project=project_name, serena_config=serena_config) -@pytest.fixture -def basic_serena_agent(serena_config): - """ - Fixture for a basic SerenaAgent without a project - """ - return SerenaAgent(project=None, serena_config=serena_config) - - class TestSerenaAgent: @pytest.mark.parametrize( "serena_agent,symbol_name,expected_kind,expected_file", @@ -292,103 +284,3 @@ class TestSerenaAgent: symbols = json.loads(result) assert not symbols, f"Expected to find no symbols for {name_path}. Symbols found: {symbols}" - - class Task: - def __init__(self, delay: float, exception: bool = False): - self.delay = delay - self.exception = exception - self.did_run = False - - def run(self): - self.did_run = True - time.sleep(self.delay) - if self.exception: - raise ValueError("Task failed") - return True - - def test_task_queue_sequence(self, basic_serena_agent): - """ - Tests that a sequence of tasks is executed correctly - """ - future1 = basic_serena_agent.issue_task(self.Task(1).run, name="task1") - future2 = basic_serena_agent.issue_task(self.Task(1).run, name="task2") - assert future1.result() is True - assert future2.result() is True - - def test_task_queue_exception(self, basic_serena_agent): - """ - Tests that tasks that raise exceptions are handled correctly, i.e. that - * the exception is propagated, - * subsequent tasks are still executed. - """ - future1 = basic_serena_agent.issue_task(self.Task(1, exception=True).run, name="task1") - future2 = basic_serena_agent.issue_task(self.Task(1).run, name="task2") - have_exception = False - try: - assert future1.result() - except Exception as e: - assert isinstance(e, ValueError) - have_exception = True - assert have_exception - assert future2.result() is True - - def test_task_queue_cancel_current(self, basic_serena_agent): - """ - Tests that tasks that are cancelled are handled correctly, i.e. that - * subsequent tasks are executed as soon as cancellation ensues. - * the cancelled task raises CancelledError when result() is called. - """ - start_time = time.time() - future1 = basic_serena_agent.issue_task(self.Task(10).run, name="task1") - future2 = basic_serena_agent.issue_task(self.Task(1).run, name="task2") - time.sleep(1) - future1.cancel() - assert future2.result() is True - end_time = time.time() - assert (end_time - start_time) < 9, "Cancelled task did not stop in time" - have_cancelled_error = False - try: - future1.result() - except Exception as e: - assert e.__class__.__name__ == "CancelledError" - have_cancelled_error = True - assert have_cancelled_error - - def test_task_queue_cancel_future(self, basic_serena_agent): - """ - Tests that when a future task is cancelled, it is never run at all - """ - task1 = self.Task(10) - task2 = self.Task(1) - future1 = basic_serena_agent.issue_task(task1.run, name="task1") - future2 = basic_serena_agent.issue_task(task2.run, name="task2") - time.sleep(1) - future2.cancel() - future1.cancel() - try: - future2.result() - except: - pass - assert task1.did_run - assert not task2.did_run - - def test_task_queue_cancellation_via_task_info(self, basic_serena_agent): - start_time = time.time() - basic_serena_agent.issue_task(self.Task(10).run, "task1") - basic_serena_agent.issue_task(self.Task(1).run, "task2") - task_infos = basic_serena_agent.get_current_tasks() - task_infos2 = basic_serena_agent.get_current_tasks() - - # test expected tasks - assert len(task_infos) == 2 - assert "task1" in task_infos[0].name - assert "task2" in task_infos[1].name - - # test task identifiers being stable - assert task_infos2[0].task_id == task_infos[0].task_id - - # test cancellation - task_infos[0].cancel() - assert task_infos[1].future.result() - end_time = time.time() - assert (end_time - start_time) < 9, "Cancelled task did not stop in time" diff --git a/test/serena/test_task_executor.py b/test/serena/test_task_executor.py new file mode 100644 index 00000000..25c31c9d --- /dev/null +++ b/test/serena/test_task_executor.py @@ -0,0 +1,119 @@ +import time + +import pytest + +from serena.task_executor import TaskExecutor + + +@pytest.fixture +def executor(): + """ + Fixture for a basic SerenaAgent without a project + """ + return TaskExecutor("TestExecutor") + + +class Task: + def __init__(self, delay: float, exception: bool = False): + self.delay = delay + self.exception = exception + self.did_run = False + + def run(self): + self.did_run = True + time.sleep(self.delay) + if self.exception: + raise ValueError("Task failed") + return True + + +def test_task_executor_sequence(executor): + """ + Tests that a sequence of tasks is executed correctly + """ + future1 = executor.issue_task(Task(1).run, name="task1") + future2 = executor.issue_task(Task(1).run, name="task2") + assert future1.result() is True + assert future2.result() is True + + +def test_task_executor_exception(executor): + """ + Tests that tasks that raise exceptions are handled correctly, i.e. that + * the exception is propagated, + * subsequent tasks are still executed. + """ + future1 = executor.issue_task(Task(1, exception=True).run, name="task1") + future2 = executor.issue_task(Task(1).run, name="task2") + have_exception = False + try: + assert future1.result() + except Exception as e: + assert isinstance(e, ValueError) + have_exception = True + assert have_exception + assert future2.result() is True + + +def test_task_executor_cancel_current(executor): + """ + Tests that tasks that are cancelled are handled correctly, i.e. that + * subsequent tasks are executed as soon as cancellation ensues. + * the cancelled task raises CancelledError when result() is called. + """ + start_time = time.time() + future1 = executor.issue_task(Task(10).run, name="task1") + future2 = executor.issue_task(Task(1).run, name="task2") + time.sleep(1) + future1.cancel() + assert future2.result() is True + end_time = time.time() + assert (end_time - start_time) < 9, "Cancelled task did not stop in time" + have_cancelled_error = False + try: + future1.result() + except Exception as e: + assert e.__class__.__name__ == "CancelledError" + have_cancelled_error = True + assert have_cancelled_error + + +def test_task_executor_cancel_future(executor): + """ + Tests that when a future task is cancelled, it is never run at all + """ + task1 = Task(10) + task2 = Task(1) + future1 = executor.issue_task(task1.run, name="task1") + future2 = executor.issue_task(task2.run, name="task2") + time.sleep(1) + future2.cancel() + future1.cancel() + try: + future2.result() + except: + pass + assert task1.did_run + assert not task2.did_run + + +def test_task_executor_cancellation_via_task_info(executor): + start_time = time.time() + executor.issue_task(Task(10).run, "task1") + executor.issue_task(Task(1).run, "task2") + task_infos = executor.get_current_tasks() + task_infos2 = executor.get_current_tasks() + + # test expected tasks + assert len(task_infos) == 2 + assert "task1" in task_infos[0].name + assert "task2" in task_infos[1].name + + # test task identifiers being stable + assert task_infos2[0].task_id == task_infos[0].task_id + + # test cancellation + task_infos[0].cancel() + assert task_infos[1].future.result() + end_time = time.time() + assert (end_time - start_time) < 9, "Cancelled task did not stop in time"