Wave Scheduler

Wave Scheduler

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

Backing Status () — 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 .

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:

# 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

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

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

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

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.

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:

# .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

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)

/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

# 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

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)