Spaces:
Paused
Paused
| # -*- coding: utf-8 -*- | |
| """The task updated tool class.""" | |
| from typing import Literal | |
| from pydantic import BaseModel, Field | |
| from ._task_tool_base import _TaskToolBase | |
| from .._response import ToolChunk | |
| from ...state import AgentState | |
| from ...exception import DeveloperOrientedException | |
| from ...message import TextBlock, ToolResultState | |
| class _TaskUpdateParams(BaseModel): | |
| """The params of the update task.""" | |
| task_id: str = Field(description="The task id.") | |
| subject: str | None = Field( | |
| default=None, | |
| description="New subject for the task", | |
| ) | |
| description: str | None = Field( | |
| default=None, | |
| description="New description for the task", | |
| ) | |
| add_blocks: list[str] | None = Field( | |
| default=None, | |
| description="Task IDs that this task blocks", | |
| ) | |
| status: Literal[ | |
| "pending", | |
| "in_progress", | |
| "completed", | |
| "deleted", | |
| ] | None = Field( | |
| default=None, | |
| description="New status for the task", | |
| ) | |
| add_blocked_by: list[str] | None = Field( | |
| default=None, | |
| description="Task IDs that block this task", | |
| ) | |
| owner: str | None = Field( | |
| default=None, | |
| description="New owner for the task", | |
| ) | |
| metadata: dict | None = Field( | |
| default=None, | |
| description="Metadata keys to merge into the task. " | |
| "Set a key to null to delete it.", | |
| ) | |
| class TaskUpdate(_TaskToolBase): | |
| """The tool to update the agent task.""" | |
| name: str = "TaskUpdate" | |
| description: str = """Use this tool to update a task in the task list. | |
| ## When to Use This Tool | |
| **Mark tasks as resolved:** | |
| - When you have completed the work described in a task | |
| - When a task is no longer needed or has been superseded | |
| - IMPORTANT: Always mark your assigned tasks as resolved when you finish them | |
| - After resolving, call TaskList to find your next task | |
| - ONLY mark a task as completed when you have FULLY accomplished it | |
| - If you encounter errors, blockers, or cannot finish, keep the task as in_progress | |
| - When blocked, create a new task describing what needs to be resolved | |
| - Never mark a task as completed if: | |
| - Tests are failing | |
| - Implementation is partial | |
| - You encountered unresolved errors | |
| - You couldn't find necessary files or dependencies | |
| **Delete tasks:** | |
| - When a task is no longer relevant or was created in error | |
| - Setting status to `deleted` permanently removes the task | |
| **Update task details:** | |
| - When requirements change or become clearer | |
| - When establishing dependencies between tasks | |
| ## Fields You Can Update | |
| - **status**: The task status (see Status Workflow below) | |
| - **subject**: Change the task title (imperative form, e.g., "Run tests") | |
| - **description**: Change the task description | |
| - **owner**: Change the task owner (agent name) | |
| - **metadata**: Merge metadata keys into the task (set a key to null to delete it) | |
| - **add_blocks**: Mark tasks that cannot start until this one completes | |
| - **add_blocked_by**: Mark tasks that must complete before this one can start | |
| ## Status Workflow | |
| Status progresses: `pending` → `in_progress` → `completed` | |
| Use `deleted` to permanently remove a task. | |
| ## Staleness | |
| Make sure to read a task's latest state using `TaskGet` before updating it. | |
| ## Examples | |
| Mark task as in progress when starting work: | |
| ```json | |
| {"task_id": "1", "status": "in_progress"} | |
| ``` | |
| Mark task as completed after finishing work: | |
| ```json | |
| {"task_id": "1", "status": "completed"} | |
| ``` | |
| Delete a task: | |
| ```json | |
| {"task_id": "1", "status": "deleted"} | |
| ``` | |
| Claim a task by setting owner: | |
| ```json | |
| {"task_id": "1", "owner": "my-name"} | |
| ``` | |
| Set up task dependencies: | |
| ```json | |
| {"task_id": "2", "add_blocked_by": ["1"]} | |
| ```""" # noqa: E501 | |
| input_schema: dict = _TaskUpdateParams.model_json_schema() | |
| async def call( | |
| self, | |
| _agent_state: AgentState, | |
| task_id: str, | |
| subject: str | None = None, | |
| description: str | None = None, | |
| add_blocks: list[str] | None = None, | |
| status: Literal["pending", "completed", "in_progress", "deleted"] | |
| | None = None, | |
| add_blocked_by: list[str] | None = None, | |
| owner: str | None = None, | |
| metadata: dict | None = None, | |
| ) -> ToolChunk: | |
| """Update the agent task.""" | |
| if not isinstance(_agent_state, AgentState): | |
| # Expose error to the developer | |
| raise DeveloperOrientedException( | |
| f"Error: {self.name} requires AgentState to be provided, got " | |
| f"{_agent_state} instead.", | |
| ) | |
| index = None | |
| for i, task in enumerate(_agent_state.tasks_context.tasks): | |
| if task.id == task_id: | |
| index = i | |
| if index is None: | |
| return ToolChunk( | |
| content=[ | |
| TextBlock( | |
| text=f"TaskNotFoundError: " | |
| f"The task (id={task_id}) does not exist.", | |
| ), | |
| ], | |
| state=ToolResultState.ERROR, | |
| ) | |
| updated_fields = [] | |
| if subject: | |
| updated_fields.append("subject") | |
| _agent_state.tasks_context.tasks[index].subject = subject | |
| if description is not None: | |
| updated_fields.append("description") | |
| _agent_state.tasks_context.tasks[index].description = description | |
| existed_ids = [_.id for _ in _agent_state.tasks_context.tasks] | |
| if add_blocks: | |
| current_blocks = _agent_state.tasks_context.tasks[index].blocks | |
| new_blocks = [ | |
| _ | |
| for _ in add_blocks | |
| if _ not in current_blocks and _ in existed_ids | |
| ] | |
| if new_blocks: | |
| updated_fields.append("add_blocks") | |
| for block_id in new_blocks: | |
| self._update_block_relation( | |
| task_id, | |
| block_id, | |
| _agent_state, | |
| ) | |
| if add_blocked_by is not None: | |
| current_blocked_by = _agent_state.tasks_context.tasks[ | |
| index | |
| ].blocked_by | |
| new_blocked_by = [ | |
| _ | |
| for _ in add_blocked_by | |
| if _ not in current_blocked_by and _ in existed_ids | |
| ] | |
| if new_blocked_by: | |
| updated_fields.append("add_blocked_by") | |
| for blocked_by_id in new_blocked_by: | |
| self._update_block_relation( | |
| blocked_by_id, | |
| task_id, | |
| _agent_state, | |
| ) | |
| if status: | |
| if status == "deleted": | |
| # Permanently remove the task | |
| _agent_state.tasks_context.tasks.pop(index) | |
| # Remove task id from all the blocks and blocked_by | |
| for task in _agent_state.tasks_context.tasks: | |
| if task_id in task.blocks: | |
| task.blocks.remove(task_id) | |
| if task_id in task.blocked_by: | |
| task.blocked_by.remove(task_id) | |
| return ToolChunk( | |
| content=[ | |
| TextBlock( | |
| text=f"Task (id={task_id}) has been deleted.", | |
| ), | |
| ], | |
| ) | |
| # Update the status | |
| updated_fields.append("status") | |
| _agent_state.tasks_context.tasks[index].state = status | |
| if owner is not None: | |
| updated_fields.append("owner") | |
| _agent_state.tasks_context.tasks[index].owner = owner | |
| if metadata: | |
| updated_fields.append("metadata") | |
| for k, v in metadata.items(): | |
| if v is None: | |
| _agent_state.tasks_context.tasks[index].metadata.pop( | |
| k, | |
| None, | |
| ) | |
| else: | |
| _agent_state.tasks_context.tasks[index].metadata[k] = v | |
| if updated_fields: | |
| res = f'Update task (id={task_id}) {", ".join(updated_fields)}.' | |
| else: | |
| res = ( | |
| f"No updates were made to the task (id={task_id}). " | |
| f"Make sure you provided at least one field to update and " | |
| f"the values are correct." | |
| ) | |
| if _agent_state.tasks_context.tasks[index].state == "completed": | |
| res += ( | |
| "\n\nTask completed. Call TaskList now to find your next " | |
| "available task or see if your work unblocked others." | |
| ) | |
| return ToolChunk(content=[TextBlock(text=res)]) | |
| def _update_block_relation( | |
| block_id: str, | |
| blocked_by_id: str, | |
| _agent_state: AgentState, | |
| ) -> None: | |
| """Update the block relationship between the tasks. | |
| Args: | |
| block_id (`str`): | |
| The id of the task that blocks the other tasks. | |
| blocked_by_id (`str`): | |
| The id of the task blocked by the task of `block_id`. | |
| _agent_state (`AgentState`): | |
| The agent state to update. | |
| """ | |
| # Update the blocks | |
| for task in _agent_state.tasks_context.tasks: | |
| if task.id == block_id and blocked_by_id not in task.blocks: | |
| task.blocks.append(blocked_by_id) | |
| if task.id == blocked_by_id and block_id not in task.blocked_by: | |
| task.blocked_by.append(block_id) | |