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

name, description, user_invocable, allowed-tools

name	description	user_invocable	allowed-tools
dispatch	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.	true	Read, Write, Bash(python3 *), Bash(scripts/dispatch-task *), Bash(scripts/wait-for-tasks *), Bash(ls *), Bash(date *), Bash(cat *)

/dispatch Skill

dispatch

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:

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:

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:

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:

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:

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:

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:

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:

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:

{
  "type": "https://agent-runtimes.oreillyit.nz/errors/dispatch-validation",
  "title": "Validation failed",
  "detail": "...",
  "invalid-params": [{"name": "field", "reason": "reason"}]
}