From 1586152c44e7e77431cb10d2f7812ff4d51c9f5a Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 12 Sep 2026 14:49:47 +0800 Subject: [PATCH] feat(agent): support updating subagent tasks --- app/agent/middleware/subagents.py | 55 +++++++++++++++++++++--- docs/refactor/agent-codex-parity-plan.md | 1 + tests/test_agent_subagents.py | 39 +++++++++++++++++ 3 files changed, 90 insertions(+), 5 deletions(-) diff --git a/app/agent/middleware/subagents.py b/app/agent/middleware/subagents.py index b35f48dd4..00f323002 100644 --- a/app/agent/middleware/subagents.py +++ b/app/agent/middleware/subagents.py @@ -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 = "子代理任务控制器正在关闭，不能再启动新任务。" diff --git a/docs/refactor/agent-codex-parity-plan.md b/docs/refactor/agent-codex-parity-plan.md index a05da219f..4b061f89f 100644 --- a/docs/refactor/agent-codex-parity-plan.md +++ b/docs/refactor/agent-codex-parity-plan.md @@ -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 配对。 diff --git a/tests/test_agent_subagents.py b/tests/test_agent_subagents.py index 2b5378c00..a7b15dae7 100644 --- a/tests/test_agent_subagents.py +++ b/tests/test_agent_subagents.py @@ -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(): """管道模式应顺序执行子代理，并把上一步结果作为下一步私有上下文。"""