subtask-orchestration

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Subtask Orchestration

子任务编排

Coordinate parallel work across repositories by creating subtasks, launching workspace sessions, and collecting results via task descriptions.

通过创建子任务、启动工作区会话并通过任务描述收集结果，在多个代码仓库间协调并行工作。

Core Concept

核心概念

Main Task Agent
     │
     ├─► create_task (subtask A) ─► start_workspace_session ─► Agent A executes
     │                                                              │
     ├─► create_task (subtask B) ─► start_workspace_session ─► Agent B executes
     │                                                              │
     └─► poll list_tasks + get_task ◄─── update_task (results) ────┘

Communication channel: Task

description

field serves as the data exchange medium between main task and subtasks.

Main Task Agent
     │
     ├─► create_task (subtask A) ─► start_workspace_session ─► Agent A executes
     │                                                              │
     ├─► create_task (subtask B) ─► start_workspace_session ─► Agent B executes
     │                                                              │
     └─► poll list_tasks + get_task ◄─── update_task (results) ────┘

沟通渠道：任务的

description

字段作为主任务与子任务之间的数据交换媒介。

Main Task Workflow

主任务工作流

1. Create Subtask

1. 创建子任务

IMPORTANT: Always append the SUBTASK_REPORT_INSTRUCTIONS to your task description.

create_task(
  project_id: "<current_project_id>",
  title: "Subtask: <specific work>",
  description: "<your task instructions here>\n\n" + SUBTASK_REPORT_INSTRUCTIONS
)

SUBTASK_REPORT_INSTRUCTIONS (copy this exactly into every subtask description):

---

重要提示：务必将SUBTASK_REPORT_INSTRUCTIONS附加到任务描述中。

create_task(
  project_id: "<current_project_id>",
  title: "Subtask: <specific work>",
  description: "<your task instructions here>\n\n" + SUBTASK_REPORT_INSTRUCTIONS
)

SUBTASK_REPORT_INSTRUCTIONS（完全复制到每个子任务描述中）：

---

MANDATORY: Report Results Before Completion

强制要求：完成前报告结果

You are running as a subtask. The main task depends on your results.

Before finishing, you MUST execute these steps:

Get your task identity:
```
context = get_context()
```

Update your task description with results:

update_task(
  task_id: context.task_id,
  description: "<your result report>"
)

你作为一个子任务运行。主任务依赖于你的结果。

完成前，你必须执行以下步骤：

获取你的任务标识：
```
context = get_context()
```

更新你的任务描述以添加结果：

update_task(
  task_id: context.task_id,
  description: "<your result report>"
)

Result Report Format

结果报告格式

Status

状态

[SUCCESS / FAILED / PARTIAL]

Summary

摘要

<1-2 sentence summary>

<1-2句话总结>

Completed Work

已完成工作

<item 1>
<item 2>

<项1>
<项2>

Outputs

输出

<file paths, PR links, artifacts>

<文件路径、PR链接、工件>

Notes

备注

This is NON-NEGOTIABLE. Main task polls your description for results. No report = invisible work.


**Record the returned `task_id` for each subtask.**

<给主任务的问题或信息>

此要求不可协商。主任务会轮询你的描述获取结果。无报告则工作视为未完成。


**记录每个子任务返回的`task_id`。**

2. Launch Workspace Session

2. 启动工作区会话

start_workspace_session(
  task_id: "<subtask_task_id>",
  executor: "CLAUDE_CODE",  // or: AMP, GEMINI, CODEX, OPENCODE, CURSOR_AGENT, QWEN_CODE, COPILOT, DROID
  repos: [{ repo_id: "<target_repo_id>", base_branch: "main" }]
)

start_workspace_session(
  task_id: "<subtask_task_id>",
  executor: "CLAUDE_CODE",  // 或：AMP, GEMINI, CODEX, OPENCODE, CURSOR_AGENT, QWEN_CODE, COPILOT, DROID
  repos: [{ repo_id: "<target_repo_id>", base_branch: "main" }]
)

3. Poll for Results (CRITICAL: Check Both Status AND Description)

3. 轮询结果（关键：同时检查状态和描述）

Use
list_tasks
to check execution status, then
get_task
for details:

undefined

使用
list_tasks
检查执行状态，再用
get_task
获取详情：

undefined

Step 1: Check task status via list_tasks

步骤1：通过list_tasks检查任务状态

tasks = list_tasks(project_id: "<project_id>") subtask = tasks.find(t => t.id == subtask_id)

Step 2: Determine subtask state

步骤2：确定子任务状态

if subtask.has_in_progress_attempt: # Still running - wait and poll again

elif subtask.last_attempt_failed: # FAILED! Workspace session crashed or setup script failed # Do NOT wait - mark as failed immediately

else: # Not running, not failed - check description for results result = get_task(task_id: subtask_id) if "## Status" in result.description: # Subtask reported results else: # Subtask completed but didn't report (edge case)

undefined

if subtask.has_in_progress_attempt: # 仍在运行 - 等待并再次轮询

elif subtask.last_attempt_failed: # 失败！工作区会话崩溃或设置脚本失败 # 不要等待 - 立即标记为失败

else: # 未运行、未失败 - 检查描述获取结果 result = get_task(task_id: subtask_id) if "## Status" in result.description: # 子任务已报告结果 else: # 子任务已完成但未报告（边缘情况）

undefined

Subtask State Matrix

子任务状态矩阵

`has_in_progress_attempt`	`last_attempt_failed`	Description has results	State
`true`	`false`	No	Running - wait
`false`	`true`	No	Failed - workspace crashed, don't wait
`false`	`false`	Yes	Completed - collect results
`false`	`false`	No	Completed but no report - check manually

`has_in_progress_attempt`	`last_attempt_failed`	描述包含结果	状态
`true`	`false`	否	运行中 - 等待
`false`	`true`	否	失败 - 工作区崩溃，无需等待
`false`	`false`	是	已完成 - 收集结果
`false`	`false`	否	已完成但无报告 - 手动检查

4. Aggregate Results

4. 汇总结果

After all subtasks complete (or fail), summarize outcomes.

所有子任务完成（或失败）后，总结结果。

Complete Example

完整示例

python

SUBTASK_REPORT_INSTRUCTIONS = """
---

python

SUBTASK_REPORT_INSTRUCTIONS = """
---

MANDATORY: Report Results Before Completion

You are running as a subtask. The main task depends on your results.

Before finishing, you MUST execute these steps:

Get your task identity:
```
context = get_context()
```

Update your task description with results:

update_task(
  task_id: context.task_id,
  description: "<your result report>"
)

You are running as a subtask. The main task depends on your results.

Before finishing, you MUST execute these steps:

Get your task identity:
```
context = get_context()
```

Update your task description with results:

update_task(
  task_id: context.task_id,
  description: "<your result report>"
)

Result Report Format

Status

[SUCCESS / FAILED / PARTIAL]

Summary

<1-2 sentence summary>

Completed Work

<item 1>
<item 2>

<item 1>
<item 2>

Outputs

<file paths, PR links, artifacts>

<file paths, PR links, artifacts>

Notes

This is NON-NEGOTIABLE. Main task polls your description for results. No report = invisible work. """

This is NON-NEGOTIABLE. Main task polls your description for results. No report = invisible work. """

Create and launch subtasks

创建并启动子任务

subtask_ids = [] for work_item in work_items: task = create_task( project_id: project_id, title: f"Subtask: {work_item.name}", description: f"{work_item.instructions}\n\n{SUBTASK_REPORT_INSTRUCTIONS}" ) subtask_ids.append(task.task_id)

start_workspace_session(
    task_id: task.task_id,
    executor: "CLAUDE_CODE",
    repos: [{ repo_id: work_item.repo_id, base_branch: "main" }]
)

start_workspace_session(
    task_id: task.task_id,
    executor: "CLAUDE_CODE",
    repos: [{ repo_id: work_item.repo_id, base_branch: "main" }]
)

Poll with failure detection

轮询并检测失败

pending = set(subtask_ids) results = {} failed = {}

while pending: tasks = list_tasks(project_id: project_id) task_map = {t.id: t for t in tasks}

for task_id in list(pending):
    task_status = task_map.get(task_id)
    
    if task_status.has_in_progress_attempt:
        continue  # Still running
        
    if task_status.last_attempt_failed:
        failed[task_id] = "Workspace session failed"
        pending.remove(task_id)
        continue
        
    result = get_task(task_id)
    if "## Status" in result.description:
        results[task_id] = result.description
        pending.remove(task_id)

print(f"Completed: {len(results)}, Failed: {len(failed)}")

undefined

pending = set(subtask_ids) results = {} failed = {}

while pending: tasks = list_tasks(project_id: project_id) task_map = {t.id: t for t in tasks}

for task_id in list(pending):
    task_status = task_map.get(task_id)
    
    if task_status.has_in_progress_attempt:
        continue  # 仍在运行
        
    if task_status.last_attempt_failed:
        failed[task_id] = "Workspace session failed"
        pending.remove(task_id)
        continue
        
    result = get_task(task_id)
    if "## Status" in result.description:
        results[task_id] = result.description
        pending.remove(task_id)

print(f"Completed: {len(results)}, Failed: {len(failed)}")

undefined

MCP Tools Reference

MCP工具参考

Tool	Role	Purpose
`get_context`	Subtask	Get own task_id, project_id, workspace_id
`create_task`	Main	Create subtask with instructions
`start_workspace_session`	Main	Launch subtask workspace
`list_tasks`	Main	Check execution status (has_in_progress_attempt, last_attempt_failed)
`get_task`	Main	Get subtask description for results
`update_task`	Subtask	Write results to own description
`list_repos`	Main	Get available repo IDs

工具	角色	用途
`get_context`	子任务	获取自身的task_id、project_id、workspace_id
`create_task`	主任务	创建包含指令的子任务
`start_workspace_session`	主任务	启动子任务工作区
`list_tasks`	主任务	检查执行状态（has_in_progress_attempt、last_attempt_failed）
`get_task`	主任务	获取子任务描述以获取结果
`update_task`	子任务	将结果写入自身描述
`list_repos`	主任务	获取可用的仓库ID

Error Handling

错误处理

Startup failure (

last_attempt_failed = true

Workspace session failed to start (setup script error, agent crash)
Do NOT wait - immediately mark subtask as failed
Check vibe-kanban UI for error logs

Subtask timeout:

```
has_in_progress_attempt = false
```
but no results after long time
Agent may have exited without reporting
Check workspace logs in UI

Missing report:

Subtask completed but description unchanged
Agent didn't follow SUBTASK_REPORT_INSTRUCTIONS
Manually check workspace output

启动失败（

last_attempt_failed = true

）：

工作区会话启动失败（设置脚本错误、Agent崩溃）
不要等待 - 立即将子任务标记为失败
查看vibe-kanban UI获取错误日志

子任务超时：

```
has_in_progress_attempt = false
```
但长时间无结果
Agent可能未报告就退出
在UI中查看工作区日志

缺失报告：

子任务已完成但描述未更改
Agent未遵循SUBTASK_REPORT_INSTRUCTIONS
手动检查工作区输出