Agent Waves: Design Document

Overview

Agent Waves introduce intra-loop parallelism to Ralph's orchestration loop. Today, Ralph executes one hat per iteration, sequentially. Waves allow a dispatcher hat to fan out work to multiple concurrent backend instances, collect results, and aggregate them — all within a single orchestration run.

This is a general-purpose parallel execution primitive. Use cases include deep research (parallel topic exploration), multi-perspective analysis, parallel code review, scatter-gather for any domain, and multi-agent debate patterns.

Waves are built on three primitives inspired by Enterprise Integration Patterns:

1. Wave-aware event emission — events tagged with correlation metadata
2. Concurrent hat execution — the loop runner spawns multiple backends in parallel
3. Aggregator gate — a hat that buffers results and activates only when all correlated results arrive

Source: https://github.com/mikeyobrien/ralph-orchestrator/issues/210

Why not just spawn subagents?

An agent could spawn N backends in a single step and collect results — no new infrastructure needed. Waves add value in two specific ways:

1. Wall-clock time. This is the primary motivation. 5 file reviews at 2 minutes each: sequential is 10 minutes, concurrent is 2. For multi-hat presets, sequential execution is the dominant bottleneck. Waves eliminate it.

2. Fresh context for synthesis. When N workers each produce substantial output, an in-context approach forces the dispatching agent to hold all results in one context window. Waves route results to a dedicated aggregator hat that activates in a fresh iteration — purpose-built instructions, no context pressure from the dispatch phase.

The concurrent execution is the real value. The event plumbing (per-worker files, env vars, correlation metadata) is what makes it work correctly within Ralph's existing architecture.

---

Architectural Impact

Today, next_hat() always returns "ralph" in multi-hat mode. Custom hats never get their own backend process — they are personas that Ralph wears during coordination. Wave workers are the first case where hats execute directly with their own backend process, outside Ralph's coordination context.

This is a deliberate, bounded exception to the Hatless Ralph model:

• Why it's safe: Wave workers have no coordination role. They receive a single task payload, execute with their hat's instructions, emit a result event, and exit. They cannot emit waves (hard-blocked via env var), have no access to Ralph's HATS table, scratchpad, or objective, and cannot influence hat selection.
• What's preserved: Ralph still owns all coordination — hat selection, event routing, aggregation, and loop control. The loop runner manages the wave lifecycle entirely; the event loop remains wave-agnostic.
• Bounded scope: Workers are structurally isolated. Each gets a per-worker events file, a fresh backend process, and env vars that identify it as a wave worker. The loop runner collects results and merges them back into the main event stream only after the wave completes.

---

Detailed Requirements

Core Architecture (from requirements clarification)

v1 Scope

Included:

• Wave CLI tool (ralph wave emit — atomic batch emission)
• Event correlation metadata (wave_id, wave_index, wave_total)
• Concurrent backend spawning in loop runner (respecting concurrency limit)
• aggregate.mode: wait_for_all with configurable timeout (default 300s)
• Context injection (downstream hat descriptions in prompt for NL dispatch)
• Best-effort failure handling with structured failure metadata
• Per-instance activation and cost accounting
• Per-worker events files (merged by loop runner after collection)
• Worker env var injection (RALPH_WAVE_WORKER, RALPH_WAVE_ID, RALPH_WAVE_INDEX, RALPH_EVENTS_FILE)
• Shared workspace (no filesystem isolation)
• No nested waves

Deferred to v2+:

• ralph wave start/ralph wave end (incremental wave emission)
• Nested waves
• Additional aggregation modes (first_n, quorum, external_event)
• Configurable failure modes (on_failure: fail_fast)
• Wave-level cost limits
• Dedicated aggregator backends
• Multi-round debate optimizations

---

Architecture Overview

Normal Iteration

graph TD
    S1[Hat Selection] --> E1[Ralph executes iteration<br/>wearing hat persona]
    E1 --> P1[Process events from JSONL]
    P1 --> N1[Next hat selected<br/>based on pending events]
    N1 --> S1

Wave Iteration

graph TD
    S2[Hat Selection] --> E2[Ralph executes iteration<br/>wearing dispatcher persona]
    E2 --> P2[Process events from JSONL<br/>wave events detected]
    P2 --> Spawn[Enter wave execution mode]

    Spawn --> W1[Worker 1<br/>own backend]
    Spawn --> W2[Worker 2<br/>own backend]
    Spawn --> W3[Worker 3<br/>own backend]
    Spawn -.->|queued · concurrency=3| W4[Worker 4]
    Spawn -.-> W5[Worker 5]

    W1 --> Collect[Collect results + failures]
    W2 --> Collect
    W3 --> Collect
    W4 --> Collect
    W5 --> Collect

    Collect --> Merge[Loop runner merges results<br/>into main events file]
    Merge --> Agg[Ralph activates as<br/>aggregator persona]

    Agg --> Next[Resume normal iteration loop]

Wave Lifecycle

sequenceDiagram
    participant R as Ralph (Dispatcher)
    participant LR as Loop Runner
    participant W1 as Worker 1
    participant W2 as Worker 2
    participant W3 as Worker 3
    participant RA as Ralph (Aggregator)

    R->>LR: Emit wave events (wave_id=w-abc, total=3)
    Note over R: Iteration ends normally

    LR->>LR: Detect wave events, enter wave execution mode

    par Concurrent execution (concurrency limit)
        LR->>W1: Spawn backend with hat instructions + payload[0]
        LR->>W2: Spawn backend with hat instructions + payload[1]
        LR->>W3: Spawn backend with hat instructions + payload[2]
    end

    Note over W1,W3: Each worker writes to its own events file
    W1->>LR: Completes (result in wave-w-abc-0.jsonl)
    W3->>LR: Completes (result in wave-w-abc-2.jsonl)
    W2->>LR: Completes (result in wave-w-abc-1.jsonl)

    LR->>LR: All 3/3 complete, merge results into main events file
    LR->>RA: Normal iteration — all results appear as pending events
    RA->>LR: Aggregated output event

Component Interaction

graph TB
    subgraph "Config Layer"
        HC[HatConfig<br/>+ concurrency<br/>+ aggregate]
    end

    subgraph "CLI Layer"
        WBE[ralph wave emit]
        RE[ralph emit]
    end

    subgraph "Event Layer"
        EM[Event Model<br/>+ wave_id<br/>+ wave_index<br/>+ wave_total]
        ER[EventReader]
        EL[EventLogger]
    end

    subgraph "Orchestration Layer"
        LR[Loop Runner]
        WT[WaveTracker]
        EV[EventLoop]
    end

    subgraph "Execution Layer"
        BP[Backend Pool<br/>concurrent spawning]
        WPB[Wave Worker<br/>Prompt Builder]
    end

    subgraph "Prompt Layer"
        HR[HatlessRalph<br/>+ context injection]
    end

    HC --> LR
    WBE --> EM
    RE --> EM
    EM --> ER
    ER --> LR
    LR --> WT
    WT --> BP
    BP --> WPB
    WPB --> BP
    HR --> LR
    LR --> EV
    EL --> EM

---

Components and Interfaces

1. Event Model Extensions

File: crates/ralph-proto/src/event.rs

Add optional wave metadata to the Event struct:

pub struct Event {
    pub topic: Topic,
    pub payload: String,
    pub source: Option,
    pub target: Option,
    // New wave fields
    pub wave_id: Option,
    pub wave_index: Option<u32>,
    pub wave_total: Option<u32>,
}

Builder methods:

impl Event {
    pub fn with_wave(mut self, wave_id: String, index: u32, total: u32) -> Self {
        self.wave_id = Some(wave_id);
        self.wave_index = Some(index);
        self.wave_total = Some(total);
        self
    }

    pub fn is_wave_event(&self) -> bool {
        self.wave_id.is_some()
    }
}

File: crates/ralph-core/src/event_logger.rs

Extend EventRecord with optional wave fields:

pub struct EventRecord {
    // ... existing fields ...
    #[serde(skip_serializing_if = "Option::is_none")]
    pub wave_id: Option,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub wave_index: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub wave_total: Option<u32>,
}

File: crates/ralph-core/src/event_reader.rs

Update the JSONL deserializer to parse wave fields. Use #[serde(default)] so existing events without wave fields parse correctly. The existing deserialize_flexible_payload function handles string/object/null payloads — no changes needed there, but the Event struct in event_reader.rs (distinct from ralph-proto's Event) must gain the optional wave fields.

2. HatConfig Extensions

File: crates/ralph-core/src/config.rs

pub struct HatConfig {
    // ... existing fields ...
    /// Maximum concurrent instances when processing wave events.
    /// Default: 1 (sequential, current behavior).
    #[serde(default = "default_concurrency")]
    pub concurrency: u32,
    /// Aggregation configuration. When set, this hat buffers incoming
    /// wave-correlated events and only activates once all results arrive.
    #[serde(default)]
    pub aggregate: Option,
}

fn default_concurrency() -> u32 { 1 }

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AggregateConfig {
    /// Aggregation mode. v1 only supports `wait_for_all`.
    pub mode: AggregateMode,
    /// Timeout in seconds. Aggregator activates with partial results
    /// if not all wave results arrive within this duration.
    /// Default: 300 seconds.
    #[serde(default = "default_aggregate_timeout")]
    pub timeout: u32,
}

fn default_aggregate_timeout() -> u32 { 300 }

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AggregateMode {
    WaitForAll,
}

Validation (in RalphConfig::validate()):

• concurrency must be >= 1
• If aggregate is set, mode must be wait_for_all
• Warn if concurrency > 1 but no downstream hat has aggregate configured (likely misconfiguration)
• Error if aggregate is set on a hat that also has concurrency > 1 (an aggregator shouldn't be a concurrent worker)

3. WaveTracker

New file: crates/ralph-core/src/wave_tracker.rs

Central state machine for tracking active waves.

pub struct WaveTracker {
    active_waves: HashMap<String, WaveState>,
}

pub struct WaveState {
    pub wave_id: String,
    pub expected_total: u32,
    pub source_hat: HatId,           // dispatcher hat
    pub worker_hat: HatId,           // hat that processes wave events
    pub result_topic: Option, // topic workers publish to
    pub dispatched: Vec,
    pub results: Vec,
    pub failures: Vec,
    pub started_at: Instant,
    pub timeout: Duration,
}

pub struct WaveInstance {
    pub index: u32,
    pub event: Event,              // the original wave event
    pub status: InstanceStatus,
}

pub enum InstanceStatus {
    Queued,
    Running,
    Completed,
    Failed(String),                // error message
    TimedOut,
}

pub struct WaveResult {
    pub index: u32,
    pub event: Event,              // the result event from worker
}

pub struct WaveFailure {
    pub index: u32,
    pub error: String,
    pub duration: Duration,
}

impl WaveTracker {
    pub fn new() -> Self;

    /// Register a new wave from detected wave events.
    pub fn register_wave(&mut self, wave_id: String, events: Vec,
                         worker_hat: HatId, timeout: Duration) -> &WaveState;

    /// Record a result event for a wave.
    pub fn record_result(&mut self, wave_id: &str, event: Event) -> WaveProgress;

    /// Record a failure for a wave instance.
    pub fn record_failure(&mut self, wave_id: &str, index: u32,
                          error: String, duration: Duration);

    /// Check if a wave is complete (all results or timeout).
    pub fn is_complete(&self, wave_id: &str) -> bool;

    /// Check for timed-out waves. Returns wave IDs that have timed out.
    pub fn check_timeouts(&mut self) -> Vec;

    /// Get all results and failures for a completed wave.
    pub fn take_wave_results(&mut self, wave_id: &str) -> Option;

    /// Check if any wave is currently active.
    pub fn has_active_waves(&self) -> bool;
}

pub struct CompletedWave {
    pub wave_id: String,
    pub results: Vec,
    pub failures: Vec,
    pub timed_out: bool,
    pub duration: Duration,
}

pub enum WaveProgress {
    /// More results expected.
    InProgress { received: u32, expected: u32 },
    /// All results received, wave complete.
    Complete,
}

4. Wave CLI Tool

New file: crates/ralph-cli/src/wave.rs

Top-level command (like ralph emit):

#[derive(Parser, Debug)]
pub struct WaveArgs {
    #[command(subcommand)]
    pub command: WaveCommands,
}

#[derive(Subcommand, Debug)]
pub enum WaveCommands {
    /// Batch emit: generate wave ID, emit N events atomically.
    Emit(WaveBatchEmitArgs),
}

#[derive(Parser, Debug)]
pub struct WaveBatchEmitArgs {
    /// Event topic for all wave events.
    pub topic: String,
    /// Payloads for each wave event.
    #[arg(long, num_args = 1..)]
    pub payloads: Vec,
}

ralph wave emit <topic> --payloads "a" "b" "c": Atomic batch emission — no state file needed:

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 `0xFFF