Loading...
Loading...
Orchestrate parallel subtasks across multiple repositories using vibe-kanban MCP. Use this skill when: (1) Work needs to be distributed across multiple repos, (2) Parallel execution of independent tasks is needed, (3) You need to create subtasks and collect their results, (4) Multi-agent coordination via kanban task descriptions. Triggers: "subtask", "multi-repo", "parallel tasks", "distribute work", "orchestrate".
npx skill4agent add apocalypseyun/skills subtask-orchestrationMain 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) ────┘descriptioncreate_task(
project_id: "<current_project_id>",
title: "Subtask: <specific work>",
description: "<your task instructions here>\n\n" + 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:**
1. Get your task identity:
2. Update your task description with results:
### Result Report Format
## Status
[SUCCESS / FAILED / PARTIAL]
## Summary
<1-2 sentence summary>
## Completed Work
- <item 1>
- <item 2>
## Outputs
- <file paths, PR links, artifacts>
## Notes
<issues or info for main task>
---
**This is NON-NEGOTIABLE.** Main task polls your description for results. No report = invisible work.task_idstart_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" }]
)list_tasksget_task# Step 1: Check task status via list_tasks
tasks = list_tasks(project_id: "<project_id>")
subtask = tasks.find(t => t.id == subtask_id)
# Step 2: Determine subtask state
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) | | Description has results | State |
|---|---|---|---|
| | No | Running - wait |
| | No | Failed - workspace crashed, don't wait |
| | Yes | Completed - collect results |
| | No | Completed but no report - check manually |
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:**
1. Get your task identity:
2. Update your task description with results:
### Result Report Format
## Status
[SUCCESS / FAILED / PARTIAL]
## Summary
<1-2 sentence summary>
## Completed Work
- <item 1>
- <item 2>
## Outputs
- <file paths, PR links, artifacts>
## Notes
<issues or info for main task>
---
**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" }]
)
# 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)}")| Tool | Role | Purpose |
|---|---|---|
| Subtask | Get own task_id, project_id, workspace_id |
| Main | Create subtask with instructions |
| Main | Launch subtask workspace |
| Main | Check execution status (has_in_progress_attempt, last_attempt_failed) |
| Main | Get subtask description for results |
| Subtask | Write results to own description |
| Main | Get available repo IDs |
last_attempt_failed = truehas_in_progress_attempt = false