Add /dispatch skill (F114): wave-based CP task submission

Implements DS-1..DS-13 from spec/dispatch-skill.md. Reads
.agent-tasks.json, validates structure, submits tasks to the control
plane in dependency-ordered waves, and records cp_task_ids atomically
after each submission. Supports --wait-wave, --continue-on-failure, and
--as-manifest flags.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

skills/dispatch/SKILL.md

@@ -0,0 +1,372 @@
+---
+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"}]
+}
+```