Spaces:
Paused
Paused
File size: 9,572 Bytes
9792ea7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 | # -*- 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)])
@staticmethod
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)
|