feat(agent): support updating subagent tasks

This commit is contained in:
jxxghp
2026-09-12 14:49:47 +08:00
parent 85a724b40c
commit 1586152c44
3 changed files with 90 additions and 5 deletions

View File

@@ -68,7 +68,7 @@ Delegation modes:
- Use `task` for one blocking subtask when you need the result immediately.
- Use `subagent_task` for two or more independent subtasks. Start them first
with `action=start` and a `tasks` array, then use `action=status`,
`action=wait`, or `action=cancel` with the returned task IDs.
`action=wait`, `action=update`, or `action=cancel` with the returned task IDs.
- Use `subagent_task` with `action=run` when you want to launch a bounded
batch and wait for the batch in one tool call.
- Use `subagent_task` with `action=pipeline` when later subtasks must use
@@ -95,7 +95,8 @@ SUBAGENT_CONTROL_DESCRIPTION = (
"Start and manage multiple MoviePilot subagent tasks asynchronously. "
"Use action=start with tasks=[{description}] to launch a batch "
"and get task IDs immediately. Use action=status to inspect tasks, action=wait "
"to wait for all or any task result, action=cancel to stop running tasks, and "
"to wait for all or any task result, action=update with task_id and a new description "
"to replace a running task after bounded cancellation, action=cancel to stop running tasks, and "
"action=run to launch a bounded batch and wait in one call. Use action=pipeline "
"to run tasks sequentially while passing each result as private context to the "
"next task."
@@ -177,9 +178,9 @@ class _SubAgentTaskSpec(BaseModel):
class _SubAgentControlInput(BaseModel):
"""异步子代理管控工具输入。"""
action: Literal["start", "status", "wait", "cancel", "run", "pipeline"] = Field(
action: Literal["start", "status", "wait", "cancel", "update", "run", "pipeline"] = Field(
default="start",
description="Task action: start, status, wait, cancel, run, or pipeline.",
description="Task action: start, status, wait, cancel, update, run, or pipeline.",
)
description: Optional[str] = Field(
default=None,
@@ -199,7 +200,7 @@ class _SubAgentControlInput(BaseModel):
)
task_id: Optional[str] = Field(
default=None,
description="Single task ID for status, wait, or cancel.",
description="Single task ID for status, wait, cancel, or update.",
)
wait_mode: Literal["all", "any"] = Field(
default="all",
@@ -982,6 +983,32 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
logger.info(f"子代理任务取消完成: tasks={len(cancellable_tasks)}")
return [record for record in records if record.task in pending]
async def _update_record(
self,
record: _SubAgentRuntimeTask,
*,
description: str,
terminal_sessions: Optional[list[SubAgentTerminalGrant]],
) -> tuple[bool, Optional[str]]:
"""有界取消旧任务后复用 task_id 启动更新后的子代理任务。"""
if not record.task.done():
pending = await self._cancel_records([record])
if pending:
return False, "旧子代理任务未能在取消窗口内收敛,更新被拒绝。"
record.description = description.strip()
record.terminal_sessions = terminal_sessions or []
record.started_at = None
record.finished_at = None
record.task = asyncio.create_task(
self._execute_managed_task(record), name=record.task_id,
)
record.task.add_done_callback(
lambda finished_task, finished_task_id=record.task_id: self._mark_task_finished(
finished_task_id, finished_task,
)
)
return True, None
def seal(self) -> None:
"""封住新的 detached 子代理提交,既有任务继续由记录表持有。"""
self._accepting_tasks = False
@@ -1195,6 +1222,24 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
) -> str:
"""管理异步子代理任务。"""
logger.info(f"收到子代理管控操作: action={action}")
if action == "update":
if not task_id or not description or not description.strip():
return self._json_response({
"success": False, "error": "update 必须提供 task_id 和新的 description。",
})
record = self._tasks.get(task_id)
if record is None:
return self._json_response({
"success": False, "error": f"未知子代理任务: {task_id}",
})
updated, error = await self._update_record(
record, description=description, terminal_sessions=terminal_sessions,
)
return self._json_response({
"success": updated, "action": action, "task_id": task_id,
"error": error, "tasks": [self._task_output(record)],
})
if action in {"start", "run", "pipeline"}:
if not self._accepting_tasks:
error = "子代理任务控制器正在关闭,不能再启动新任务。"

View File

@@ -119,4 +119,5 @@ uv run --locked --no-sync python -m scripts.evaluation \
- 浏览器重定向安全阶段:点击、填写、选择、脚本执行、内容读取和标签页切换后统一复核当前 URL;动作已经发生但重定向到未授权私网时返回 `unknown`,要求先核验页面状态,不直接重复动作。浏览器、文件、Shell、子代理相关回归共 157 项通过。
- 真实工具组合复核:同一本机 provider 的 `gpt-5.6-luna + max` 运行 `command_execution` live `/tmp/moviepilot-agent-round-luna-live-command-current-20260912.json` 通过(2 次模型调用、真实 Shell 回执、退出码 0);浏览器 live `/tmp/moviepilot-agent-round-luna-live-browser-current-20260912.json` 通过(6 次模型调用、导航/快照/`click_ref`/正文 4 次回执,独立页面 oracle 为 `BROWSER_OK`)。这证明结构化恢复合同和动作后 URL 校验已进入真实生产图;原生浏览器仍因用户保存权限保持单独 blocked。
- 子代理并行真实配对复核:使用官方 OAuth 的同一 `gpt-5.6-luna + max`、32 次调用上限、8192 输出和 300 秒超时,MoviePilot `/tmp/moviepilot-agent-round-luna-oauth-live-subagent-parallel-current-20260912.json` 与原生 `/tmp/moviepilot-agent-round-luna-native-subagent-parallel-current-20260912.json` 均通过;严格摘要 `/tmp/moviepilot-agent-round-luna-oauth-subagent-parallel-pair-current-20260912.json` 为 `pair_valid=true`、`both_passed=true`。双方真实派发两个通用只读子任务,完成订阅 73 与站点 11/13 核验,0 失败/重复/副作用;MoviePilot 17 次模型调用、223786 tokens、84.212 秒,原生 Codex 24 次、476349 tokens、125.645 秒。该轮证明子代理并发编排在真实生产图与原生 Harness 中均可收敛,不能外推终端分享能力。
- 子代理任务更新阶段:`subagent_task` 新增 `action=update`,要求指定 `task_id` 和新描述;宿主先在有界取消窗口内收口旧任务,再复用同一 task ID 启动更新任务并返回新状态。更新失败不会覆盖仍在途的旧任务;相关调度回归已通过。`send_input/resume` 仍需要 provider 原生会话协议,暂不伪装成已完成。
- 浏览器环境复核:启动本机 Chrome 后用同一官方 OAuth 配置重跑 `/tmp/moviepilot-agent-round-luna-oauth-native-browser-chrome-20260912.json`,原生已加载 Browser plugin、node_repl 并选择 Chrome;真实回环导航仍被保存的浏览器权限拒绝。该结果排除了“CLI 未登录或未启动 Chrome”的解释,下一次只有在用户允许对应回环地址后才继续导航、点击和 oracle 配对。

View File

@@ -385,6 +385,45 @@ def test_control_tool_starts_tasks_concurrently_and_waits():
asyncio.run(_run_test())
def test_control_tool_updates_task_with_same_id_after_bounded_cancel():
"""update 应取消旧任务、复用 task_id 并把新描述交给子代理。"""
async def _run_test():
model = FakeListChatModel(responses=["ok"])
middleware = SubAgentTaskControlMiddleware(
model=model,
profiles=subagent_module._builtin_subagent_profiles(),
tools=[],
)
calls = []
release = asyncio.Event()
async def _fake_run_task(self, *, description, subagent_type, task_id=None, terminal_sessions=None):
calls.append(description)
if description == "旧任务":
await release.wait()
return f"完成:{description}"
with patch.object(subagent_module._SubAgentAgentProvider, "run_task", new=_fake_run_task):
started = json.loads(await middleware._control_task(action="start", description="旧任务"))
task_id = started["tasks"][0]["task_id"]
await asyncio.sleep(0)
updated = json.loads(await middleware._control_task(
action="update", task_id=task_id, description="更新后的任务",
))
release.set()
await middleware._wait_records(
records=[middleware._tasks[task_id]], wait_mode="all", timeout_ms=1000,
)
assert updated["success"] is True
assert updated["task_id"] == task_id
assert updated["tasks"][0]["description"] == "更新后的任务"
assert calls == ["旧任务", "更新后的任务"]
asyncio.run(_run_test())
def test_control_tool_pipeline_passes_previous_results_to_next_step():
"""管道模式应顺序执行子代理,并把上一步结果作为下一步私有上下文。"""