custom-claude-skills/skills/dispatch/SKILL.md

---
name: dispatch
description: >
  Submit .agent-tasks.json to the control plane in dependency-ordered waves. Records
  cp_task_ids atomically. Use after /decompose to fire a task batch at the CP.
user_invocable: true
allowed-tools: Read, Write, Bash(python3 *), Bash(scripts/dispatch-task *), Bash(scripts/wait-for-tasks *), Bash(ls *), Bash(date *), Bash(cat *)
---

# /dispatch Skill

<command-name>dispatch</command-name>

You are dispatching a batch of tasks to the agent-runtimes control plane.

## Pre-gathered context

### Task file
!`cat .agent-tasks.json 2>/dev/null || echo "NO_TASK_FILE"`

### Available templates
!`ls task-templates/ 2>/dev/null || echo "NO_TEMPLATES_DIR"`

### Current date
!`date +%Y-%m-%dT%H:%M:%S`

---

## Instructions

Parse `$ARGUMENTS` for flags:
- `--wait-wave` → block until each wave reaches a terminal state before submitting the next
- `--continue-on-failure` → submit the next wave even if the current wave had failures (only relevant with `--wait-wave`)
- `--as-manifest` → translate `.agent-tasks.json` to a `POST /manifests` payload and print to stdout; do NOT submit tasks

### Step 1: Check task file exists

If the pre-gathered task file output is `NO_TASK_FILE`, say:
```
No .agent-tasks.json found. Run /decompose to create one.
```
Stop.

### Step 2: Validate token file (DS-10)

Run:
```bash
python3 -c "
import os, stat, sys, json

path = os.path.expanduser('~/.config/agent-runtimes/tokens.json')
parent = os.path.dirname(path)

errors = []

if not os.path.exists(path):
    errors.append({'type': 'token_missing', 'detail': f'Token file not found at {path}. Run \"scripts/dispatch-task --login\" to obtain a token.'})
else:
    mode = stat.S_IMODE(os.stat(path).st_mode)
    if mode != 0o600:
        errors.append({'type': 'token_perms', 'detail': f'Token file mode is {oct(mode)}; expected 0600. Run \"chmod 600 {path}\".'})

if os.path.isdir(parent):
    dmode = stat.S_IMODE(os.stat(parent).st_mode)
    if dmode != 0o700:
        errors.append({'type': 'dir_perms', 'detail': f'Token directory mode is {oct(dmode)}; expected 0700. Run \"chmod 700 {parent}\".'})

if errors:
    print(json.dumps({'type': 'validation_error', 'errors': errors}, indent=2))
    sys.exit(1)
print('ok')
"
```

If the output is not `ok`, display the error JSON and stop. Do not attempt any submissions.

### Step 3: Parse and validate .agent-tasks.json (DS-1, DS-2)

Run:
```bash
python3 -c "
import json, sys, os

with open('.agent-tasks.json') as f:
    data = json.load(f)

errors = []

# Check top-level structure
required_top = {'params', 'tasks'}
known_top = {'params', 'tasks'}
unknown_keys = set(data.keys()) - known_top
if unknown_keys:
    errors.append({'field': '<top-level>', 'reason': f'Unknown top-level key(s): {sorted(unknown_keys)}'})

# Validate params
params = data.get('params', {})
for field in ['repo_url', 'agent_repo_url', 'project_id']:
    if not params.get(field):
        errors.append({'field': f'params.{field}', 'reason': 'Required field missing or empty'})

# Check for userinfo in URLs (DS-6)
import urllib.parse
for url_field in ['repo_url', 'agent_repo_url']:
    url = params.get(url_field, '')
    try:
        p = urllib.parse.urlparse(url)
        if p.password or (p.username and '@' in url):
            errors.append({'field': f'params.{url_field}', 'reason': f'{url_field} contains userinfo. Pass credentials via SSH keys or env vars, not URL userinfo.'})
    except Exception:
        pass

# Validate tasks array
tasks = data.get('tasks', [])
if not isinstance(tasks, list) or len(tasks) == 0:
    errors.append({'field': 'tasks', 'reason': 'Must be a non-empty array'})

# Validate each task
task_ids = set()
for i, task in enumerate(tasks):
    prefix = f'tasks[{i}]'
    for field in ['id', 'name', 'prompt']:
        if not task.get(field):
            errors.append({'field': f'{prefix}.{field}', 'reason': 'Required field missing or empty'})
    
    tid = task.get('id', '')
    if tid:
        if tid in task_ids:
            errors.append({'field': f'{prefix}.id', 'reason': f'Duplicate task id: {tid!r}'})
        task_ids.add(tid)
    
    if not task.get('template') and not task.get('model'):
        errors.append({'field': f'{prefix}.template/model', 'reason': 'One of template or model is required'})
    
    if task.get('template') and task.get('model'):
        errors.append({'field': f'{prefix}.template/model', 'reason': 'Only one of template or model is allowed, not both'})
    
    if not isinstance(task.get('depends_on', []), list):
        errors.append({'field': f'{prefix}.depends_on', 'reason': 'Must be an array'})

# Validate depends_on references
for task in tasks:
    for dep in task.get('depends_on', []):
        if dep not in task_ids:
            errors.append({'field': f'tasks[{task[\"id\"]}].depends_on', 'reason': f'Unknown task id reference: {dep!r}'})

# Validate templates exist (DS-7)
templates_dir = 'task-templates'
if os.path.isdir(templates_dir):
    valid_templates = {f.replace('.yaml', '') for f in os.listdir(templates_dir) if f.endswith('.yaml')}
    for task in tasks:
        tmpl = task.get('template')
        if tmpl and tmpl not in valid_templates:
            errors.append({'field': f'tasks[{task.get(\"id\",\"?\")}].template', 'reason': f'Unknown template {tmpl!r}. Valid: {sorted(valid_templates)}'})

if errors:
    print(json.dumps({'type': 'validation_error', 'title': 'Validation failed', 'invalid-params': errors}, indent=2))
    sys.exit(1)

print('ok')
"
```

If not `ok`, display the error JSON and stop.

### Step 3b: --as-manifest translation (DS-13)

If `$ARGUMENTS` contains `--as-manifest`, run:
```bash
python3 -c "
import json, sys

RESERVED_KEYS = {'manifest_id', 'workflow_id', 'definition_version'}

with open('.agent-tasks.json') as f:
    data = json.load(f)

params = data.get('params', {})

# Check reserved keys (MN-31)
reserved_found = set(params.keys()) & RESERVED_KEYS
if reserved_found:
    for k in reserved_found:
        print(json.dumps({'type': 'validation_error', 'detail': f'params contains reserved key {k!r}. CP injects this automatically (MN-31).'}), file=sys.stderr)
    sys.exit(1)

# Build manifest
nodes = []
for task in data.get('tasks', []):
    node = {
        'node_id': task['id'],
        'prompt': task['prompt'],
        'depends_on': task.get('depends_on', []),
    }
    if task.get('template'):
        node['template'] = task['template']
    if task.get('model'):
        node['model'] = task['model']
    nodes.append(node)

manifest = {
    'definition_version': 1,
    'params': params,
    'nodes': nodes,
}

print(json.dumps(manifest, indent=2))
"
```

Print the output and stop. Do not submit any tasks.

### Step 4: Build wave structure

Run:
```bash
python3 -c "
import json, sys

with open('.agent-tasks.json') as f:
    data = json.load(f)

tasks = data['tasks']
task_map = {t['id']: t for t in tasks}

# Topological sort into waves
waves = []
resolved = set()
remaining = set(t['id'] for t in tasks)

while remaining:
    wave = []
    for tid in list(remaining):
        task = task_map[tid]
        deps = set(task.get('depends_on', []))
        if deps.issubset(resolved):
            wave.append(tid)
    
    if not wave:
        print(json.dumps({'error': 'Circular dependency detected in tasks'}), file=sys.stderr)
        sys.exit(1)
    
    waves.append(sorted(wave))
    for tid in wave:
        resolved.add(tid)
        remaining.discard(tid)

# Output wave assignments
result = {}
for wave_num, wave_ids in enumerate(waves, 1):
    for tid in wave_ids:
        result[tid] = wave_num

print(json.dumps({'waves': waves, 'wave_for_task': result}))
"
```

Store the wave structure. Present it to the user as:
```
Wave 1: <task ids>
Wave 2: <task ids> (depends on wave 1)
...
```

### Step 5: Check CP_URL

Use `CP_URL` from the environment if set. Default: `https://agents.oreillyit.nz/api`

### Step 6: Submit tasks wave by wave

For each wave in order:

**6a. Identify tasks to submit in this wave:**
- Skip tasks where `cp_task_id` is already set (non-null) — these are already submitted (DS-11)
- But first verify each already-submitted task still exists on the CP:

```bash
scripts/dispatch-task status <cp_task_id> 2>&1
```

If any returns 404: print structured error and stop:
```
Recorded cp_task_id '<id>' for task '<task_id>' returned 404 from CP. Clear cp_task_id in .agent-tasks.json to re-submit.
```

**6b. Submit each remaining task in the wave:**

For each task, build the dispatch-task command:
```bash
CP_URL=<cp_url> scripts/dispatch-task \
  --name "<task.name>" \
  --project-id "<params.project_id>" \
  --prompt "<task.prompt>" \
  [--template "<task.template>" | --model "<task.model>"] \
  --template-param "repo_url=<params.repo_url>" \
  --template-param "agent_repo_url=<params.agent_repo_url>"
```

Run the submission. Capture stdout — it contains the returned `cp_task_id` (UUID on its own line, or JSON).

**6c. Extract cp_task_id from output:**

Parse the dispatch-task output to extract the task UUID. Look for a UUID pattern: `[0-9a-f-]{36}`.

**6d. Record cp_task_id atomically (DS-3):**

After each successful submission, atomically update `.agent-tasks.json`:
```bash
python3 -c "
import json, os, tempfile

with open('.agent-tasks.json') as f:
    data = json.load(f)

for task in data['tasks']:
    if task['id'] == '<task_id>':
        task['cp_task_id'] = '<cp_task_id>'
        break

# Atomic write: tempfile + rename (DS-3)
with tempfile.NamedTemporaryFile(mode='w', dir='.', delete=False, suffix='.tmp') as tf:
    json.dump(data, tf, indent=2)
    tmpname = tf.name

os.rename(tmpname, '.agent-tasks.json')
print('recorded')
"
```

**6e. If submission fails (non-zero exit from dispatch-task):**
- Record the per-task error: `Task '<id>' submission failed: <error output>`
- Continue with the rest of the wave (DS-5)
- Set an overall failure flag (exit 1 at end)

**6f. If --wait-wave flag is set (DS-4):**

After submitting all tasks in the wave, collect all cp_task_ids for this wave and run:
```bash
CP_URL=<cp_url> scripts/wait-for-tasks <cp_task_id_1> <cp_task_id_2> ...
```

If wait-for-tasks exits non-zero (any task failed/cancelled/timed_out):
- Report the failure: show which tasks failed
- Unless `--continue-on-failure` is set, **stop here** and do not submit the next wave
- Exit 1

### Step 7: Print monitoring command (DS-8)

After all waves are submitted, print:
```
Run: CP_URL=<cp_url> scripts/agent-monitor --login --filter "project=<project_id>" --filter "age<2h"
```

### Step 8: Final exit

- If any submission failed → exit 1
- If `--wait-wave` and any wave task reached a non-success terminal state → exit 1
- Otherwise → exit 0

---

## Error output format

All errors are printed as JSON to stderr:
```json
{
  "type": "https://agent-runtimes.oreillyit.nz/errors/dispatch-validation",
  "title": "Validation failed",
  "detail": "...",
  "invalid-params": [{"name": "field", "reason": "reason"}]
}
```