eigent/backend/app/agent/toolkit/observable_todo_toolkit.py

# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========

import logging
from typing import Any

from camel.toolkits import FunctionTool, TodoToolkit
from camel.toolkits.todo_toolkit import TodoItem

from app.agent.toolkit.abstract_toolkit import AbstractToolkit
from app.service.task import ActionTodoStateData, Agents, get_task_lock
from app.utils.listen.toolkit_listen import _safe_put_queue

logger = logging.getLogger("observable_todo_toolkit")


class ObservableTodoToolkit(TodoToolkit, AbstractToolkit):
    """CAMEL TodoToolkit with Eigent UI change events.

    This intentionally keeps CAMEL's todo data model and `todo_write` API as
    the source of truth. Eigent only observes successful writes and emits an
    SSE-compatible action for the frontend.
    """

    agent_name: str = Agents.single_agent

    def __init__(
        self,
        api_task_id: str,
        task_id: str,
        agent_id: str | None = None,
        working_dir: str | None = None,
        timeout: float | None = None,
    ) -> None:
        super().__init__(working_dir=working_dir, timeout=timeout)
        self.api_task_id = api_task_id
        self.task_id = task_id
        self.agent_id = agent_id

    def todo_write(self, todos: list[TodoItem]) -> str:
        """Create or update the current task todo list.

        Use this tool to track task progress with concise todo items. Each
        todo should include content, active_form, and status fields.

        Args:
            todos (list[TodoItem]): The full ordered todo list to store.

        Returns:
            str: A message indicating whether the todo list was updated.
        """
        result = super().todo_write(todos)
        if not result.startswith("[ERROR]"):
            self.emit_todo_state()
        return result

    def emit_todo_state(self) -> None:
        try:
            task_lock = get_task_lock(self.api_task_id)
        except Exception:
            logger.warning(
                "Could not emit todo_state because task lock is missing",
                extra={"project_id": self.api_task_id},
            )
            return

        data = {
            "project_id": self.api_task_id,
            "task_id": self.task_id,
            "agent_id": self.agent_id,
            "todos": self.serialized_todos(),
        }
        _safe_put_queue(task_lock, ActionTodoStateData(data=data))

    def serialized_todos(self) -> list[dict[str, Any]]:
        serialized: list[dict[str, Any]] = []
        for index, item in enumerate(self.todos, start=1):
            serialized.append(
                {
                    "id": f"todo_{index}",
                    "content": item.content,
                    "active_form": item.active_form,
                    "status": item.status,
                }
            )
        return serialized

    def get_tools(self) -> list[FunctionTool]:
        tools = [FunctionTool(self.todo_write)]
        for tool in tools:
            try:
                tool._toolkit_name = self.toolkit_name()
            except Exception:
                pass
        return tools

    @classmethod
    def toolkit_name(cls) -> str:
        return "TodoToolkit"