From 0d4b43dd5e0ace884e3be9761e1b77797451dce5 Mon Sep 17 00:00:00 2001 From: Paul O'Reilly Date: Tue, 28 Apr 2026 13:45:11 +1200 Subject: [PATCH] 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 --- skills/dispatch/SKILL.md | 372 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 372 insertions(+) create mode 100644 skills/dispatch/SKILL.md diff --git a/skills/dispatch/SKILL.md b/skills/dispatch/SKILL.md new file mode 100644 index 0000000..beeda4c --- /dev/null +++ b/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 + +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: +```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': '', '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: +Wave 2: (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 2>&1 +``` + +If any returns 404: print structured error and stop: +``` +Recorded cp_task_id '' for task '' 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= scripts/dispatch-task \ + --name "" \ + --project-id "" \ + --prompt "" \ + [--template "" | --model ""] \ + --template-param "repo_url=" \ + --template-param "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['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 '' submission failed: ` +- 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= scripts/wait-for-tasks ... +``` + +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= scripts/agent-monitor --login --filter "project=" --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"}] +} +```