# Wave Scheduler

> Dependency-aware wave scheduler for parallel agent dispatch. Reads plan files, builds a dependency DAG from ticket relationships, computes execution waves with configurable max concurrency, dispatches parallel agents per wave, and serializes dependent work. Integrates with agent_healthcheck for stall detection and cross-repo locks for conflict prevention.

- Skill: `omninode-ai/wave-scheduler` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add omninode-ai/wave-scheduler`
- Raw SKILL.md: https://api.skillmd.com/api/skills/omninode-ai/wave-scheduler/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: OmniNode-ai (https://skillmd.com/u/omninode-ai)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/omninode-ai/wave-scheduler

---


# Wave Scheduler

**Skill ID**: `onex:wave_scheduler`
**Version**: 1.0.0
**Owner**: omniclaude

---

## Backing Status (<TICKET>) — NON-RUNNABLE via canonical event-bus dispatch

**This skill has no deterministic node backing and is de-listed from event-bus / node-routed dispatch — deliberately, by design.** Its generated orchestrator shell (`node_skill_wave_scheduler_orchestrator`) is a Polly-passthrough (`maturity: stub` in its contract.yaml), and no CLI or platform-service implementation exists elsewhere. The parallel per-wave agent dispatch that consumes the computed schedule is inherently Claude-session-native.

The DAG-construction + wave-computation core (topological sort, level assignment, max-concurrency splitting) is pure and I/O-free — a strong COMPUTE-node candidate. Extracting it into a real `node_wave_scheduler_compute` is tracked in <TICKET>.

---

## Dispatch Surface

**Target**: Agent Teams

---

## Overview

Replaces ad-hoc parallel dispatch in epic-team with a deterministic, dependency-aware
wave execution model. Given a plan file with explicit `depends_on` fields, the scheduler:

1. **Parses** the plan file to extract ticket definitions and dependencies
2. **Builds** a directed acyclic graph (DAG) of ticket dependencies
3. **Validates** the DAG (no cycles, all dependencies exist)
4. **Computes** execution waves using topological sort with level grouping
5. **Dispatches** parallel agents per wave (up to max_concurrency)
6. **Monitors** agent health during execution (via agent_healthcheck)
7. **Reports** completed/failed/blocked per wave

---

## Plan File Format

The scheduler accepts plan files in YAML format with this schema:

```yaml
# plan.yaml
epic_id: EPIC-001
title: "Insights Action Plan 2026-03-28"

tickets:
  - id: TICKET-001
    title: "Implement checkpoint-based pipeline recovery"
    repo: omniclaude
    depends_on: []

  - id: TICKET-002
    title: "Encode architectural invariants"
    repo: omniclaude
    depends_on: []

  - id: TICKET-003
    title: "Build agent health-check"
    repo: omniclaude
    depends_on: [TICKET-001]

  - id: TICKET-004
    title: "Implement wave scheduler"
    repo: omniclaude
    depends_on: [TICKET-003]
```

---

## DAG Construction

```python
def build_dependency_dag(tickets: list[dict]) -> dict[str, list[str]]:
    """Build a dependency DAG from ticket definitions.

    Args:
        tickets: List of ticket dicts with 'id' and 'depends_on' fields.

    Returns:
        Adjacency list: {ticket_id: [dependent_ticket_ids]}

    Raises:
        ValueError: If a cycle is detected or a dependency references a
            ticket not in the plan.
    """
    # Build adjacency list (dependency -> dependents)
    dag = {t["id"]: [] for t in tickets}
    in_degree = {t["id"]: 0 for t in tickets}

    for ticket in tickets:
        for dep in ticket.get("depends_on", []):
            if dep not in dag:
                raise ValueError(
                    f"Ticket {ticket['id']} depends on {dep} which is not in the plan"
                )
            dag[dep].append(ticket["id"])
            in_degree[ticket["id"]] += 1

    # Cycle detection via Kahn's algorithm (see compute_waves)
    return dag, in_degree
```

---

## Wave Computation

```python
def compute_waves(
    dag: dict[str, list[str]],
    in_degree: dict[str, int],
    max_concurrency: int = 6,
) -> list[list[str]]:
    """Compute execution waves using topological sort with level grouping.

    Each wave contains tickets whose dependencies have all been satisfied.
    Waves are capped at max_concurrency tickets.

    Args:
        dag: Adjacency list from build_dependency_dag.
        in_degree: In-degree count per ticket.
        max_concurrency: Max tickets per wave.

    Returns:
        List of waves, where each wave is a list of ticket IDs.

    Raises:
        ValueError: If a cycle is detected (not all tickets can be scheduled).
    """
    from collections import deque

    # Initialize queue with tickets that have no dependencies
    queue = deque([tid for tid, deg in in_degree.items() if deg == 0])
    waves = []
    scheduled_count = 0

    while queue:
        # Take up to max_concurrency tickets for this wave
        wave = []
        next_queue = deque()

        while queue and len(wave) < max_concurrency:
            wave.append(queue.popleft())

        waves.append(wave)
        scheduled_count += len(wave)

        # Reduce in-degree for dependents of completed tickets
        for tid in wave:
            for dependent in dag.get(tid, []):
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    next_queue.append(dependent)

        queue = next_queue

    # Cycle detection
    total_tickets = len(dag)
    if scheduled_count < total_tickets:
        unscheduled = [tid for tid, deg in in_degree.items() if deg > 0]
        raise ValueError(
            f"Cycle detected: {len(unscheduled)} tickets cannot be scheduled: {unscheduled}"
        )

    return waves
```

---

## Wave Execution

```python
def execute_waves(
    waves: list[list[str]],
    tickets: dict[str, dict],
    dry_run: bool = False,
) -> dict:
    """Execute waves sequentially, tickets within each wave in parallel.

    Args:
        waves: List of waves from compute_waves.
        tickets: Ticket definitions keyed by ID.
        dry_run: Log dispatch plan without executing.

    Returns:
        Execution report with per-wave and per-ticket status.
    """
    report = {
        "status": "completed",
        "waves_completed": 0,
        "tickets_completed": 0,
        "tickets_failed": 0,
        "tickets_blocked": 0,
        "wave_results": [],
    }

    failed_tickets = set()

    for wave_idx, wave in enumerate(waves):
        log(f"Wave {wave_idx}: dispatching {len(wave)} tickets: {wave}")

        # Check if any ticket in this wave is blocked by a failed dependency
        blocked = []
        dispatchable = []
        for tid in wave:
            deps = tickets[tid].get("depends_on", [])
            if any(d in failed_tickets for d in deps):
                blocked.append(tid)
                report["tickets_blocked"] += 1
            else:
                dispatchable.append(tid)

        if blocked:
            log(f"  Blocked by failed dependencies: {blocked}")

        if dry_run:
            log(f"  [DRY RUN] Would dispatch: {dispatchable}")
            report["wave_results"].append({
                "wave": wave_idx,
                "dispatched": dispatchable,
                "blocked": blocked,
                "dry_run": True,
            })
            report["waves_completed"] += 1
            continue

        # Dispatch all tickets in this wave as parallel Task() calls
        # Each Task() invokes ticket-pipeline for the ticket
        wave_results = dispatch_parallel_tickets(dispatchable, tickets)

        # Collect results
        wave_report = {
            "wave": wave_idx,
            "dispatched": dispatchable,
            "blocked": blocked,
            "results": {},
        }

        for tid, result in wave_results.items():
            wave_report["results"][tid] = result["status"]
            if result["status"] in ("completed", "merged"):
                report["tickets_completed"] += 1
            else:
                report["tickets_failed"] += 1
                failed_tickets.add(tid)

        report["wave_results"].append(wave_report)
        report["waves_completed"] += 1

    # Determine overall status
    if report["tickets_failed"] > 0 or report["tickets_blocked"] > 0:
        report["status"] = "partial"
    if report["tickets_completed"] == 0:
        report["status"] = "failed"

    return report
```

### Parallel dispatch within a wave

```python
def dispatch_parallel_tickets(
    ticket_ids: list[str],
    tickets: dict[str, dict],
) -> dict[str, dict]:
    """Dispatch ticket-pipeline for each ticket in parallel via Task().

    All Task() calls are made in a SINGLE message for true parallelism.
    Results are collected when all tasks complete.
    """
    # Dispatch all in parallel (single message with multiple Task calls)
    tasks = {}
    for tid in ticket_ids:
        ticket = tickets[tid]
        tasks[tid] = Task(
            subagent_type="general-purpose",
            description=f"wave-scheduler: ticket-pipeline for {tid}",
            prompt=f"""Execute ticket-pipeline for {tid}: {ticket['title']}

    Invoke: Skill(skill="onex:ticket_pipeline", args="{tid}")

    Repo: {ticket['repo']}
    Execute end-to-end. Create worktree, implement, review, create PR, merge.
    Report back with: status (completed|failed|blocked), pr_url, blockers.
    """,
        )

    # Collect results
    return {tid: task.result() for tid, task in tasks.items()}
```

---

## Cross-Repo Locks

When multiple tickets in the same wave target the same repo, the scheduler prevents
conflicting edits by acquiring a per-repo lock before dispatch.

```python
REPO_LOCKS = {}  # {repo_name: lock_holder_ticket_id}

def acquire_repo_lock(repo: str, ticket_id: str) -> bool:
    """Acquire a repo lock for a ticket. Returns False if already held."""
    if repo in REPO_LOCKS and REPO_LOCKS[repo] != ticket_id:
        return False
    REPO_LOCKS[repo] = ticket_id
    return True

def release_repo_lock(repo: str, ticket_id: str) -> None:
    """Release a repo lock after ticket completes."""
    if REPO_LOCKS.get(repo) == ticket_id:
        del REPO_LOCKS[repo]
```

**Conflict resolution strategy:**
- If two tickets in the same wave target the same repo, one is deferred to the next wave
- The ticket with fewer dependencies gets priority (lower in-degree first)
- Deferred tickets are re-queued, not failed

---

## Health-Check Integration

During wave execution, the scheduler monitors dispatched agents using the
`agent_healthcheck` skill. Between polling intervals:

1. Check each active agent for stall signals (inactivity, context overflow, rate limits)
2. On stall detection: invoke agent_healthcheck recovery protocol
3. Recovery writes a checkpoint and relaunches the agent
4. If max recovery attempts (3) exceeded: mark ticket as failed, continue wave

---

## State Persistence

Wave execution state is persisted to allow resume after interruption:

```yaml
# .onex_state/wave_scheduler/{epic_id}/state.yaml
schema_version: "1.0.0"
epic_id: EPIC-001
plan_file: docs/plans/2026-03-28-insights-plan.yaml
started_at: 2026-03-28T22:00:00Z
status: in_progress
max_concurrency: 6

waves:
  - wave: 0
    tickets: [TICKET-001, TICKET-002, TICKET-005]
    status: completed
    completed_at: 2026-03-28T22:30:00Z
    results:
      TICKET-001: completed
      TICKET-002: completed
      TICKET-005: completed
  - wave: 1
    tickets: [TICKET-003]
    status: in_progress
    started_at: 2026-03-28T22:31:00Z
    results: {}
  - wave: 2
    tickets: [TICKET-004]
    status: pending
```

### Resume behavior

With `--resume`, the scheduler:
1. Reads persisted state from `.onex_state/wave_scheduler/{epic_id}/state.yaml`
2. Skips completed waves
3. Re-dispatches in-progress wave tickets that are not yet completed
4. Continues from the last incomplete wave

---

## Policy Switches

| Switch | Default | Description |
|--------|---------|-------------|
| `max_concurrency` | `6` | Maximum parallel agents per wave |
| `dispatch_timeout_minutes` | `30` | Per-agent timeout before circuit breaker |
| `max_recovery_attempts` | `3` | Max health-check recovery relaunches per ticket |
| `fail_fast` | `false` | Stop entire execution on first ticket failure |
| `defer_repo_conflicts` | `true` | Defer conflicting same-repo tickets to next wave |

---

## Example

Given a sample plan with four tickets:

```
Wave 0: [TICKET-001, TICKET-002, TICKET-005]  (no dependencies, parallel)
Wave 1: [TICKET-003]                           (depends on TICKET-001)
Wave 2: [TICKET-004]                           (depends on TICKET-003)
```

```bash
/wave-scheduler docs/plans/2026-03-28-insights-plan.yaml --max-concurrency 6
/wave-scheduler docs/plans/2026-03-28-insights-plan.yaml --dry-run
/wave-scheduler docs/plans/2026-03-28-insights-plan.yaml --resume
```

---

## Acceptance Test: Diamond Dependency Pattern (15 Tasks, 3 Repos)

This test validates correct wave computation, cross-repo locking, and failure propagation
using a realistic diamond dependency pattern across 3 repositories.

### Test Plan File

```yaml
# test-plan-diamond-15.yaml
epic_id: TEST-DIAMOND-15
title: "Diamond dependency acceptance test"

tickets:
  # Wave 0 — 5 tickets, no dependencies (roots)
  - id: T-001
    title: "Init shared types"
    repo: omnibase_core
    depends_on: []
  - id: T-002
    title: "Init API schema"
    repo: omninode_infra
    depends_on: []
  - id: T-003
    title: "Init frontend scaffold"
    repo: omnidash
    depends_on: []
  - id: T-004
    title: "Init event bus topic registration"
    repo: omnibase_core
    depends_on: []
  - id: T-005
    title: "Init DB migration framework"
    repo: omninode_infra
    depends_on: []

  # Wave 1 — 4 tickets, single dependencies
  - id: T-006
    title: "Implement core models"
    repo: omnibase_core
    depends_on: [T-001]
  - id: T-007
    title: "Implement API routes"
    repo: omninode_infra
    depends_on: [T-002]
  - id: T-008
    title: "Wire event consumers"
    repo: omnibase_core
    depends_on: [T-004]
  - id: T-009
    title: "Create DB schema"
    repo: omninode_infra
    depends_on: [T-005]

  # Wave 2 — 3 tickets, diamond top (multiple dependencies converge)
  - id: T-010
    title: "API integration with core models"
    repo: omninode_infra
    depends_on: [T-006, T-007]        # Diamond: two parents
  - id: T-011
    title: "Dashboard data layer"
    repo: omnidash
    depends_on: [T-003, T-007]        # Cross-repo diamond
  - id: T-012
    title: "Event-driven DB sync"
    repo: omninode_infra
    depends_on: [T-008, T-009]        # Same-repo diamond

  # Wave 3 — 2 tickets, deeper convergence
  - id: T-013
    title: "End-to-end API + dashboard wiring"
    repo: omnidash
    depends_on: [T-010, T-011]        # Second diamond layer
  - id: T-014
    title: "Event replay with DB state"
    repo: omnibase_core
    depends_on: [T-010, T-012]        # Cross-repo convergence

  # Wave 4 — 1 ticket, final convergence (all paths join)
  - id: T-015
    title: "Full integration verification"
    repo: omninode_infra
    depends_on: [T-013, T-014]        # Grand convergence
```

### Expected Wave Schedule

```
Wave 0: [T-001, T-002, T-003, T-004, T-005]   5 tickets (all roots)
Wave 1: [T-006, T-007, T-008, T-009]           4 tickets (single deps satisfied)
Wave 2: [T-010, T-011, T-012]                  3 tickets (diamond tops)
Wave 3: [T-013, T-014]                         2 tickets (deeper convergence)
Wave 4: [T-015]                                1 ticket  (final convergence)

Total: 15 tickets across 5 waves
Sequential estimate: 15 x avg_ticket_time
Parallel estimate: 5 x avg_ticket_time (5 waves)
Speedup: ~3x
```

### Cross-Repo Lock Behavior

In Wave 0, T-001 and T-004 both target `omnibase_core`. With `defer_repo_conflicts=true`:
- T-001 acquires the `omnibase_core` lock (fewer downstream dependents: 2 vs 2, tie-break by order)
- T-004 is deferred to a new Wave 0b
- Net effect: Wave 0 = [T-001, T-002, T-003, T-005], Wave 0b = [T-004]

With `defer_repo_conflicts=false` (default for worktree-isolated repos):
- Both T-001 and T-004 run in Wave 0 in separate worktrees
- No lock contention because each ticket gets its own worktree

### Failure Propagation

If T-007 fails in Wave 1:
- T-010 is **blocked** (depends on T-007)
- T-011 is **blocked** (depends on T-007)
- T-013 is **blocked** (depends on T-010 and T-011, both blocked)
- T-014 is **blocked** (depends on T-010, which is blocked)
- T-015 is **blocked** (depends on T-013 and T-014, both blocked)
- **Total cascade**: 1 failure blocks 5 downstream tickets
- T-006, T-008, T-009, T-012 continue unaffected

### Expected Summary Report

```yaml
status: partial
waves_completed: 5
tickets_completed: 9
tickets_failed: 1
tickets_blocked: 5
wall_clock_minutes: 150    # 5 waves x 30 min avg
sequential_estimate_minutes: 450  # 15 tickets x 30 min avg
speedup_factor: 3.0
failure_cascade:
  root_failure: T-007
  blocked_tickets: [T-010, T-011, T-013, T-014, T-015]
```

---

## See Also

- `epic-team` skill (current ad-hoc wave execution, to be replaced)
- `agent_healthcheck` skill (stall detection)
- `checkpoint` skill (checkpoint protocol)
- `ticket-pipeline` skill (per-ticket execution)
- `decompose-epic` skill (plan decomposition)