Added endpoints for info about tasks to dashboard backend

src/serena/agent.py

@@ -198,6 +198,14 @@ class SerenaAgent:
         """
         return self._task_executor.get_current_tasks()
 
+    def get_last_executed_task(self) -> TaskExecutor.TaskInfo | None:
+        """
+        Gets the last executed task.
+
+        :return: the last executed task info or None if no task has been executed yet
+        """
+        return self._task_executor.get_last_executed_task()
+
     def get_language_server_manager(self) -> LanguageServerManager | None:
         if self._active_project is not None:
             return self._active_project.language_server_manager

src/serena/dashboard.py

@@ -2,7 +2,7 @@ import os
 import socket
 import threading
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Self
 
 from flask import Flask, Response, request, send_from_directory
 from pydantic import BaseModel
@@ -10,6 +10,7 @@ from sensai.util import logging
 
 from serena.analytics import ToolUsageStats
 from serena.constants import SERENA_DASHBOARD_DIR
+from serena.task_executor import TaskExecutor
 from serena.util.logging import MemoryLogHandler
 
 if TYPE_CHECKING:
@@ -85,6 +86,22 @@ class RequestDeleteMemory(BaseModel):
     memory_name: str
 
 
+class QueuedExecution(BaseModel):
+    task_id: int
+    is_running: bool
+    name: str
+    finished_successfully: bool
+
+    @classmethod
+    def from_task_info(cls, task_info: TaskExecutor.TaskInfo) -> Self:
+        return cls(
+            task_id=task_info.task_id,
+            is_running=task_info.is_running,
+            name=task_info.name,
+            finished_successfully=task_info.finished_successfully(),
+        )
+
+
 class SerenaDashboardAPI:
     log = logging.getLogger(__qualname__)
 
@@ -225,6 +242,24 @@ class SerenaDashboardAPI:
             except Exception as e:
                 return {"status": "error", "message": str(e)}
 
+        @self._app.route("/queued_task_executions", methods=["GET"])
+        def get_queued_executions() -> dict[str, Any]:
+            try:
+                current_executions = self._agent.get_current_tasks()
+                response = [QueuedExecution.from_task_info(task_info).model_dump() for task_info in current_executions]
+                return {"queued_executions": response, "status": "success"}
+            except Exception as e:
+                return {"status": "error", "message": str(e)}
+
+        @self._app.route("/last_execution", methods=["GET"])
+        def get_last_execution() -> dict[str, Any]:
+            try:
+                last_execution_info = self._agent.get_last_executed_task()
+                response = QueuedExecution.from_task_info(last_execution_info).model_dump() if last_execution_info is not None else None
+                return {"last_execution": response, "status": "success"}
+            except Exception as e:
+                return {"status": "error", "message": str(e)}
+
     def _get_log_messages(self, request_log: RequestLog) -> ResponseLog:
         all_messages = self._memory_log_handler.get_log_messages()
         requested_messages = all_messages[request_log.start_idx :] if request_log.start_idx <= len(all_messages) else []

src/serena/task_executor.py

@@ -5,7 +5,7 @@ from collections.abc import Callable
 from concurrent.futures import Future
 from dataclasses import dataclass
 from threading import Thread
-from typing import Generic, Optional, TypeVar
+from typing import Generic, TypeVar
 
 from sensai.util import logging
 from sensai.util.logging import LogTime
@@ -22,10 +22,11 @@ class TaskExecutor:
         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[TaskExecutor.Task] = None
+        self._task_executor_current_task: TaskExecutor.Task | None = None
+        self._task_executor_last_executed_task_info: TaskExecutor.TaskInfo | None = None
 
     class Task(ToStringMixin, Generic[T]):
-        def __init__(self, function: Callable[[], T], name: str, logged: bool = True, timeout: Optional[float] = None):
+        def __init__(self, function: Callable[[], T], name: str, logged: bool = True, timeout: float | None = None):
             """
             :param function: the function representing the task to execute
             :param name: the name of the task
@@ -126,6 +127,8 @@ class TaskExecutor:
             task.wait_until_done(timeout=task.timeout)
             with self._task_executor_lock:
                 self._task_executor_current_task = None
+                if task.logged:
+                    self._task_executor_last_executed_task_info = self.TaskInfo.from_task(task, is_running=False)
 
     @dataclass
     class TaskInfo:
@@ -140,6 +143,9 @@ class TaskExecutor:
         unique identifier of the task
         """
 
+        def finished_successfully(self) -> bool:
+            return self.future.done() and not self.future.cancelled() and self.future.exception() is None
+
         @staticmethod
         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))
@@ -147,7 +153,7 @@ class TaskExecutor:
         def cancel(self) -> None:
             self.future.cancel()
 
-    def get_current_tasks(self) -> list["TaskExecutor.TaskInfo"]:
+    def get_current_tasks(self) -> list[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).
@@ -200,3 +206,12 @@ class TaskExecutor:
         """
         task_obj = self.issue_task(task, name=name, logged=logged, timeout=timeout)
         return task_obj.result()
+
+    def get_last_executed_task(self) -> TaskInfo | None:
+        """
+        Gets information about the last executed task.
+
+        :return: TaskInfo of the last executed task, or None if no task has been executed yet.
+        """
+        with self._task_executor_lock:
+            return self._task_executor_last_executed_task_info