OrionScheduler

// runLoop owns all mutable state.
// Workers communicate via typed channels only.
select {
case req := <-s.ingestChan:
    s.handleIngest(req)
case taskID := <-s.completeChan:
    s.handleComplete(taskID)
case taskID := <-s.failChan:
    s.handleFailure(taskID)
case taskID := <-s.retryChan:
    s.handleRetryEnqueue(taskID)
}

func (w *WAL) append(entry WalEntry) error {
    data, _ := json.Marshal(entry)
    data = append(data, '\n')
    w.mu.Lock()
    defer w.mu.Unlock()
    w.file.Write(data)
    // Strict write-ahead durability boundary.
    return w.file.Sync()
}

func (s *Scheduler) propagateFailure(rootID string) {
    queue := []string{rootID}
    visited := make(map[string]bool)
    for len(queue) > 0 {
        current := queue[0]; queue = queue[1:]
        if visited[current] { continue }
        visited[current] = true
        task.Cancelled.Store(1) // abort in-flight workers
        s.wal.AppendCascadeFail(current)
        queue = append(queue, s.dependents[current]...)
    }
}

// WAL entry: cascade vs organic failure.
type WalEntry struct {
    Type      string `json:"type"`
    TaskID    string `json:"task_id,omitempty"`
    IsCascade bool   `json:"is_cascade,omitempty"`
}

// Replay: different paths for different failure origins.
if req.isCascade {
    task.Status = models.StatusFailed
    task.Cancelled.Store(1)
    return // never re-enqueue
}
// Organic: apply normal retry logic.
task.RetryCount++

// emitEvent never blocks the run loop.
func (s *Scheduler) emitEvent(event models.TaskEvent) {
    select {
    case s.EventChan <- event:
    default:
        // Buffer full — count the drop, move on.
        s.droppedEvents.Add(1)
    }
}