AI Agent Orchestration Patterns: Cordum Architecture Deep Dive

// core/workflow/models.go (excerpt)
const (
  StepTypeWorker    StepType = "worker"
  StepTypeApproval  StepType = "approval"
  StepTypeCondition StepType = "condition"
  StepTypeDelay     StepType = "delay"
  StepTypeNotify    StepType = "notify"
)

const (
  StepTypeSwitch      StepType = "switch"
  StepTypeParallel    StepType = "parallel"
  StepTypeLoop        StepType = "loop"
  StepTypeTransform   StepType = "transform"
  StepTypeStorage     StepType = "storage"
  StepTypeSubWorkflow StepType = "subworkflow"
)

// generic-dispatch types
const (
  StepTypeLLM       StepType = "llm"
  StepTypeHTTP      StepType = "http"
  StepTypeContainer StepType = "container"
  StepTypeScript    StepType = "script"
)

// core/workflow/engine_helpers.go (excerpt)
func depsSatisfied(step *Step, run *WorkflowRun, wfDef *Workflow) bool {
  if step == nil || len(step.DependsOn) == 0 {
    return true
  }

  for _, dep := range step.DependsOn {
    sr, ok := run.Steps[dep]
    if !ok || sr.Status == "" {
      return false
    }
    if sr.Status == StepStatusSucceeded {
      continue
    }
    if (sr.Status == StepStatusFailed || sr.Status == StepStatusDenied || sr.Status == StepStatusTimedOut) && wfDef != nil {
      depDef := wfDef.Steps[dep]
      if depDef != nil && depDef.OnError != "" {
        handlerSR := run.Steps[depDef.OnError]
        if handlerSR != nil && handlerSR.Status == StepStatusSucceeded {
          continue
        }
      }
    }
    return false
  }
  return true
}

// core/workflow/engine.go (excerpt)
// Persist state BEFORE dispatch for crash safety.
parentSR.Status = StepStatusRunning
parentSR.Attempts++
parentSR.JobID = jobID
run.Steps[stepID] = parentSR
if err := e.store.UpdateRun(ctx, run); err != nil {
  // revert and retry
}

packet := makeJobPacket(run.ID, req)
if err := e.bus.Publish(capsdk.SubjectSubmit, packet); err != nil {
  // revert to pending; idempotency key prevents duplicate execution
}

// core/controlplane/scheduler/strategy_least_loaded.go (excerpt)
func (s *LeastLoadedStrategy) PickSubject(req *pb.JobRequest, workers map[string]*pb.Heartbeat) (string, error) {
  topicPools := routing.Topics[req.Topic]
  eligiblePools := filterEligiblePools(topicPools, req.GetMeta().GetRequires(), routing.Pools)

  if preferredWorker := labels["preferred_worker_id"]; preferredWorker != "" {
    // used only if healthy + eligible + not overloaded
  }

  // otherwise choose least-loaded matching worker
  if subject := bus.DirectSubject(selected.WorkerId); subject != "" {
    return subject, nil
  }
  return req.Topic, nil
}