---
title: "Agent Waves: Implementation Plan"
description: "- [ ] Step 1: Event model extensions (wave metadata) - [ ] Step 2: HatConfig extensions (concurrency, aggregate) - [ ] Step 3: WaveTracker state machine - [ ] Step 4: Wave CLI tool (`ralph wave emit`) - [ ] Step 5: Wave worker prompt builder - [ ] Step 6: Loop runner wave execution - [ ] Step 7: Context injection for NL dispatch - [ ] Step 8: Nested wave prevention - [ ] Step 9: Diagnostics and ob"
type: skill
canonical_url: https://claudary.paisolsolutions.com/skills/plan-14
source: "Claudary"
difficulty: intermediate
author: "Claude Code Knowledge Pack"
date: 2026-07-10T11:32:01.859Z
license: CC-BY-4.0
attribution: "Agent Waves: Implementation Plan — Claudary (https://claudary.paisolsolutions.com/skills/plan-14)"
---

# Agent Waves: Implementation Plan
- [ ] Step 1: Event model extensions (wave metadata) - [ ] Step 2: HatConfig extensions (concurrency, aggregate) - [ ] Step 3: WaveTracker state machine - [ ] Step 4: Wave CLI tool (`ralph wave emit`) - [ ] Step 5: Wave worker prompt builder - [ ] Step 6: Loop runner wave execution - [ ] Step 7: Context injection for NL dispatch - [ ] Step 8: Nested wave prevention - [ ] Step 9: Diagnostics and ob

## Overview

# Agent Waves: Implementation Plan

## Checklist

- [ ] Step 1: Event model extensions (wave metadata)
- [ ] Step 2: HatConfig extensions (concurrency, aggregate)
- [ ] Step 3: WaveTracker state machine
- [ ] Step 4: Wave CLI tool (`ralph wave emit`)
- [ ] Step 5: Wave worker prompt builder
- [ ] Step 6: Loop runner wave execution
- [ ] Step 7: Context injection for NL dispatch
- [ ] Step 8: Nested wave prevention
- [ ] Step 9: Diagnostics and observability
- [ ] Step 10: Smoke tests and E2E
- [ ] Step 11: Documentation and example presets

---

## Step 1: Event Model Extensions

**Objective:** Add optional wave correlation metadata to the event system so wave events can be identified, tracked, and correlated throughout the pipeline.

**Implementation guidance:**
- Add `wave_id: Option<String>`, `wave_index: Option<u32>`, `wave_total: Option<u32>` to `Event` struct in `crates/ralph-proto/src/event.rs`
- Add builder methods: `with_wave(wave_id, index, total)`, `is_wave_event()`
- Add same fields to `EventRecord` in `crates/ralph-core/src/event_logger.rs` with `#[serde(skip_serializing_if = "Option::is_none")]`
- Update `EventReader` in `crates/ralph-core/src/event_reader.rs` to parse wave fields from JSONL using `#[serde(default)]` for backwards compatibility
- Update `EventRecord::new()` and `EventRecord::from_agent_event()` to propagate wave fields from `Event`

**Test requirements:**
- Unit: wave metadata round-trips through serialize/deserialize
- Unit: events without wave fields parse correctly (backwards compat)
- Unit: `is_wave_event()` returns correct results
- Unit: `EventReader` parses JSONL with and without wave fields

**Integration notes:** This is the foundation — every subsequent step depends on these fields existing. No behavioral changes yet; existing functionality is unaffected.

**Demo:** `cargo test -p ralph-proto` and `cargo test -p ralph-core` pass. Manually write a JSONL line with wave fields, verify `EventReader` parses it.

---

## Step 2: HatConfig Extensions

**Objective:** Add `concurrency` and `aggregate` configuration fields to hat definitions so preset authors can declare wave-capable and aggregator hats.

**Implementation guidance:**
- Add `concurrency: u32` (default 1) to `HatConfig` in `crates/ralph-core/src/config.rs`
- Add `aggregate: Option<AggregateConfig>` to `HatConfig`
- Define `AggregateConfig { mode: AggregateMode, timeout: u32 }` and `AggregateMode::WaitForAll`
- Add validation in `RalphConfig::validate()`:
  - `concurrency >= 1`
  - Error if `aggregate` set on hat with `concurrency > 1`
  - Warn if `concurrency > 1` but no downstream hat has `aggregate`
- Propagate `concurrency` to `Hat` struct in `crates/ralph-proto/src/hat.rs` if needed for runtime access

**Test requirements:**
- Unit: YAML with `concurrency: 3` and `aggregate: { mode: wait_for_all, timeout: 600 }` parses correctly
- Unit: YAML without new fields parses with defaults (concurrency=1, aggregate=None)
- Unit: validation rejects `concurrency: 0`
- Unit: validation rejects aggregate on concurrent hat
- Unit: existing preset YAML files still parse correctly

**Integration notes:** Pure config — no runtime behavior changes. Existing presets are unaffected because defaults preserve current behavior.

**Future failure modes (v2+):** v1 is hardcoded best-effort (continue-on-failure) — partial results are almost always useful, so the wave continues when instances fail and the aggregator gets structured failure metadata. `on_failure: fail_fast` is worth adding in v2 for all-or-nothing use cases (e.g., parallel builds where one failure invalidates everything). Failure thresholds and must-pass hats are better handled by aggregator instructions — the aggregator already sees which instances failed and can decide how to react. If we see patterns where people keep writing the same "abort if X failed" logic in aggregator instructions, that's the signal to promote it to config.

**Demo:** Write a test hat collection YAML with wave config, verify it parses and validates.

---

## Step 3: WaveTracker State Machine

**Objective:** Build the core state machine that tracks active waves, records results and failures, manages timeouts, and determines when aggregation gates should open.

**Implementation guidance:**
- New file: `crates/ralph-core/src/wave_tracker.rs`
- Core structs: `WaveTracker`, `WaveState`, `WaveInstance`, `InstanceStatus`, `WaveResult`, `WaveFailure`, `CompletedWave`, `WaveProgress`
- Key methods:
  - `register_wave(wave_id, events, worker_hat, timeout)` — creates new wave state
  - `record_result(wave_id, event)` → `WaveProgress` (InProgress or Complete)
  - `record_failure(wave_id, index, error, duration)` — records instance failure
  - `is_complete(wave_id)` — all results + failures == expected total
  - `check_timeouts()` → `Vec<String>` — returns timed-out wave IDs
  - `take_wave_results(wave_id)` → `CompletedWave` — consumes completed wave
  - `has_active_waves()` — any waves in progress
- Add `mod wave_tracker` to `crates/ralph-core/src/lib.rs` and export

**Test requirements:**
- Unit: register wave, record results one by one, verify Complete on last
- Unit: register wave, record some results + failure, verify completion accounting
- Unit: timeout detection with mocked time
- Unit: `take_wave_results` returns all results and failures, removes wave
- Unit: multiple concurrent waves tracked independently

**Integration notes:** Pure data structure — no I/O, no async. Can be tested entirely with synchronous unit tests. Will be integrated into the loop runner in Step 6.

**Demo:** `cargo test -p ralph-core wave_tracker` — all state transitions exercised.

---

## Step 4: Wave CLI Tool

**Objective:** Build the `ralph wave emit` command for atomic batch wave dispatch, and enhance `ralph emit` to support wave worker env vars.

**Implementation guidance:**
- New file: `crates/ralph-cli/src/wave.rs`
- Add `Wave(wave::WaveArgs)` to `Commands` enum in `crates/ralph-cli/src/main.rs`
- Wire up `wave::execute()` in the command dispatch match
- v1 only supports batch emission — no `start`/`end` subcommands (deferred to v2)
- `ralph wave emit <topic> --payloads "a" "b" "c"`:
  1. Check `RALPH_WAVE_WORKER` env var — if set, exit with error (nested wave prevention)
  2. Generate wave ID (timestamp-based hex: `w-{:08x}` from nanos mod `0xFFFF_FFFF`)
  3. Resolve events file from `.ralph/current-events` marker (falling back to `.ralph/events.jsonl`)
  4. Write N events to JSONL, each with `wave_id`, `wave_index: 0..N-1`, `wave_total: N`
  5. Print wave ID to stdout
- Enhance existing `ralph emit` (in `main.rs:emit_command`):
  - Check `RALPH_EVENTS_FILE` env var — if set, write to that file instead of default
  - Check `RALPH_WAVE_ID` + `RALPH_WAVE_INDEX` env vars — if set, auto-tag events with wave metadata
  - When no wave env vars are present, behavior is unchanged (backwards compatible)

**Test requirements:**
- Unit/CLI: `ralph wave emit topic --payloads a b c` writes 3 tagged events atomically
- Unit/CLI: `ralph emit` with `RALPH_WAVE_ID` and `RALPH_WAVE_INDEX` env vars tags events correctly
- Unit/CLI: `ralph emit` with `RALPH_EVENTS_FILE` env var writes to specified file
- Unit/CLI: `ralph emit` without wave env vars works unchanged
- Unit/CLI: `RALPH_WAVE_WORKER=1 ralph wave emit` fails with error

**Integration notes:** This is the agent-facing interface. The events file is the handoff — CLI writes tagged JSONL, loop runner reads and detects waves in Step 6. The env var approach keeps `ralph emit` backwards compatible while transparently supporting wave workers.

**Demo:** Run `ralph wave emit review.file --payloads "src/main.rs" "src/lib.rs" "src/config.rs"`. Inspect events file — three events with matching wave_id, sequential indices, wave_total=3.

---

## Step 5: Wave Worker Prompt Builder

**Objective:** Build the prompt constructor for wave worker instances — each worker gets focused context with the hat's instructions and its specific event payload.

**Implementation guidance:**
- New file: `crates/ralph-core/src/wave_prompt.rs`
- Add `mod wave_prompt` to `crates/ralph-core/src/lib.rs` and export
- `build_wave_worker_prompt(HatConfig, Event, WaveWorkerContext) -> String`
- `WaveWorkerContext` contains: `wave_id`, `wave_index`, `wave_total`, `result_topics` (from hat's `publishes`)
- Prompt sections:
  1. Hat instructions (from config)
  2. Wave context metadata (wave_id, your index, total instances)
  3. Event payload (the work item)
  4. Event writing guide (how to emit results — topic from `publishes`, env vars handle correlation transparently)
  5. Nested wave guard ("Do NOT use `ralph wave` commands")
- Keep it simple — no HATS table, no objective, no scratchpad. Workers are focused executors.
- The events file path and wave correlation metadata are communicated via env vars (Step 6), not embedded in the prompt. The prompt only includes what the agent needs to understand its task.

**Test requirements:**
- Unit: prompt includes hat instructions
- Unit: prompt includes event payload
- Unit: prompt includes wave metadata
- Unit: prompt includes nested wave prohibition
- Unit: prompt includes correct result topic from hat's `publishes`

**Integration notes:** Used by the loop runner (Step 6) when spawning wave backends. Pure string construction — no I/O.

**Demo:** Call `build_wave_worker_prompt()` with test data, inspect output string for all required sections.

---

## Step 6: Loop Runner Wave Execution

**Objective:** The core integration — detect wave events after a normal iteration, spawn concurrent backends for wave workers, collect results, merge into the main events file, and resume the normal loop. The loop runner owns the entire wave lifecycle; the event loop remains wave-agnostic.

**Implementation guidance:**
- In `crates/ralph-cli/src/loop_runner.rs`, after `process_events_from_jsonl()`:
  - New struct `DetectedWave { wave_id, target_hat: HatId, hat_config: HatConfig, events: Vec<Event>, total: u32 }`
  - New function `detect_wave_events(events, registry) -> Option<DetectedWave>` — groups events by `wave_id`, validates consistency, resolves target hat from event topic via `HatRegistry`
  - When wave detected, enter wave execution mode:
    1. Create per-worker events files (`.ralph/wave-{wave_id}-{index}.jsonl`)
    2. Register wave in `WaveTracker`
    3. Spawn concurrent backends using `tokio::sync::Semaphore` for concurrency limiting
    4. Each backend gets env vars: `RALPH_WAVE_WORKER=1`, `RALPH_WAVE_ID`, `RALPH_WAVE_INDEX`, `RALPH_EVENTS_FILE` (pointing to per-worker file)
    5. Each backend uses worker hat's backend config, prompt from `build_wave_worker_prompt()`
    6. Collect results as instances complete — read events from each per-worker file
    7. Handle failures — call `wave_tracker.record_failure()`
    8. Race against aggregate timeout (resolved from downstream aggregator hat's config)
    9. On timeout: cancel running instances (SIGTERM, then SIGKILL after 250ms)
    10. Merge all result events from per-worker files into the main events file
    11. Clean up per-worker files
    12. Increment worker hat's activation count by number of instances
    13. Accumulate costs into global `max_cost` check
  - Resume normal loop — aggregator hat sees all results as pending events on next iteration
- `WaveInstanceResult { index, status, events, cost, tokens, duration }` — returned by each instance
- The event loop never sees partial wave results. By the time it processes events on the next iteration, all results have been merged.

**Test requirements:**
- Integration: mock backend that emits result events, verify wave lifecycle end-to-end
- Integration: concurrency limiting — 5 instances, concurrency=2, verify max 2 concurrent
- Integration: timeout fires, verify partial results collected, instances terminated
- Integration: instance failure, verify wave continues, failure recorded
- Integration: activation counts incremented per instance
- Integration: cost accumulated across all instances
- Integration: per-worker event files created, read, merged, and cleaned up
- Integration: main events file not written to during wave execution

**Integration notes:** This is the largest and most complex step. It touches the main loop's execution path. Consider implementing incrementally: first get sequential wave execution working (concurrency=1), then add the semaphore-based concurrency. The aggregator gate is implicit — by merging all results at once, the event loop's existing `determine_active_hats()` naturally picks up the aggregator hat.

**Demo:** Create a test hat collection with dispatcher + worker (concurrency=2) + aggregator. Run a small wave (3 items) with a mock backend. Verify concurrent execution, result collection, and aggregator activation.

---

## Step 7: Context Injection for NL Dispatch

**Objective:** Enhance the HATS table in Ralph's prompt to include downstream hat descriptions and wave dispatch instructions, enabling natural language wave dispatch.

**Implementation guidance:**
- In `crates/ralph-core/src/hatless_ralph.rs`, in the HATS table generation (`hats_section`):
  - When the active hat has `publishes` targeting wave-capable hats (`concurrency > 1`):
    - Add "Available Downstream Hats" section with topic, name, description, concurrency
    - Add wave emission instructions (brief `ralph wave emit` usage)
  - When the active hat `publishes` target multiple different hats (scatter-gather):
    - Same enrichment, showing each target hat
- Use existing `HatInfo` and topology resolution — this already resolves `publishes` → downstream hats
- Keep it concise — a few-line table + one-liner wave instruction, not a tutorial

**Test requirements:**
- Unit: dispatcher hat with wave-capable downstream → prompt includes downstream table
- Unit: dispatcher hat with non-wave downstream (concurrency=1) → no wave context injected
- Unit: scatter-gather hat with multiple downstream hats → all listed
- Unit: prompt includes wave emission instructions
- Unit: hat with no publishes → no downstream section

**Integration notes:** Pure prompt construction — extends existing HATS table logic. No runtime behavior changes. This is what makes NL dispatch possible: the model sees what's available and decides what to fan out.

**Demo:** Build prompt for a dispatcher hat in a wave-capable collection. Inspect prompt for downstream hat table and wave instructions.

---

## Step 8: Nested Wave Prevention

**Objective:** Prevent wave workers from emitting further waves, avoiding complexity explosion in v1.

**Implementation guidance:**
- **Hard enforcement:** In `ralph wave emit` (wave.rs), check `RALPH_WAVE_WORKER` env var. If set, print error and exit with non-zero status. (This check is already partially implemented in Step 4, but verify it's in place.)
- **Soft enforcement:** In `build_wave_worker_prompt()` (Step 5), include "Do NOT use `ralph wave` commands. Nested waves are not suppo

---

Source: [Claudary](https://claudary.paisolsolutions.com/skills/plan-14) · https://claudary.paisolsolutions.com