cervellaswarm-task-orchestration

Deterministic task classification, routing, and validation for multi-agent systems.

Zero LLM calls. File-based state. Git-friendly audit trail.

Task Description ──> Classifier ──> Router ──> Workers
                     (rule-based)   (deterministic)  │
                                                     ▼
                                              Output Validator
                                              (score 0-100)

Why This Exists

Every multi-agent framework routes tasks using LLM calls. That means:

• Non-deterministic results (same input, different routing)
• API costs for every routing decision
• Latency before any real work starts
• No way to unit-test routing logic

This package takes a different approach: rule-based classification with zero LLM calls. The classifier uses keyword scoring, file count estimation, and pattern matching to produce a deterministic complexity score. Same input always produces the same routing decision.

Installation

pip install cervellaswarm-task-orchestration

Requirements: Python 3.10+ | Dependencies: None (zero external deps!)

Quick Start

Classify a Task

from cervellaswarm_task_orchestration import classify_task, TaskComplexity

result = classify_task("refactor the authentication module across 5 files")
print(result.complexity)       # TaskComplexity.CRITICAL
print(result.should_architect)  # True
print(result.confidence)        # 1.0
print(result.triggers)          # ['refactor', 'files>5 (5)']

Route to Architect or Workers

from cervellaswarm_task_orchestration import route_task

decision = route_task("fix typo in README")
print(decision.use_architect)      # False
print(decision.suggested_workers)  # [WorkerType.DOCS]

decision = route_task("redesign the entire API layer")
print(decision.use_architect)      # True
print(decision.reason)             # "Task critical: Detailed planning required"

Validate an Architect Plan

from cervellaswarm_task_orchestration import validate_plan

result = validate_plan(plan_markdown)
print(result.is_valid)   # True/False
print(result.score)      # 0.0 - 10.0
print(result.errors)     # ['Missing section: ## Phase 2: Design']
print(result.warnings)   # ['Plan too short (< 500 chars)']

Manage Task State (File-Based)

from cervellaswarm_task_orchestration import (
    create_task, mark_ready, mark_working, mark_done, get_task_status
)

# Create a task (writes .swarm/tasks/TASK_001.md)
create_task("TASK_001", "backend-worker", "Implement user auth endpoint", risk_level=2)

# State machine: created -> ready -> working -> done
mark_ready("TASK_001")
mark_working("TASK_001")   # ATOMIC: only one worker can claim!
mark_done("TASK_001")

print(get_task_status("TASK_001"))  # "done"

Validate Worker Output

from cervellaswarm_task_orchestration import validate_output
from pathlib import Path

result = validate_output(Path(".swarm/tasks/TASK_001_output.md"))
print(result.valid)         # True/False
print(result.score)         # 0-100
print(result.retry_needed)  # True if score < 50
print(result.errors)        # ['Error markers found: Traceback, ERROR:']

CLI Tools

# Classify task complexity
cervella-classify "refactor the database layer"

# Route task to architect or workers
cervella-route "add pagination to the API" --json

# Validate an architect plan
cervella-validate-plan .swarm/plans/PLAN_001.md

# Validate worker output
cervella-validate-output --file output.md

# Manage tasks
cervella-task create TASK_001 backend-worker "Implement auth" --risk 2
cervella-task list
cervella-task status TASK_001

# Unified CLI
cervella-orchestrate classify "migrate to PostgreSQL"
cervella-orchestrate route "fix typo in docs"
cervella-orchestrate task list

Architecture

Task Classifier (task_classifier.py)

Rule-based complexity scoring with zero LLM calls:

Complexity levels: SIMPLE < MEDIUM < COMPLEX < CRITICAL

Architect Flow (architect_flow.py)

Routing + plan validation + fallback logic:

classify_task() ──> should_architect?
                    │
              ┌─────┴─────┐
              No           Yes
              │            │
         Direct to     Architect
         Workers       Planning
              │            │
              │      validate_plan()
              │            │
              │      ┌─────┴─────┐
              │      Valid       Invalid
              │      │           │
              │      Approve     Reject (max 2x)
              │                  │
              │            FALLBACK_TO_WORKER
              ▼                  │
         Output Validator  <─────┘

Task Manager (task_manager.py)

File-based state machine with atomic operations:

created ──> ready ──> working ──> done
                        │
                  ATOMIC CLAIM
                 (exclusive create)

• Race condition protection via open(file, 'x') exclusive create
• Path traversal prevention with strict ID validation
• All state visible in the filesystem (git-friendly audit trail)

Output Validator (output_validator.py)

Reflection pattern with cumulative scoring:

Score < 50 triggers retry_needed = True

Comparison with Other Frameworks

Honest note: CrewAI and AutoGen have larger ecosystems and multi-LLM support. LangGraph offers more flexible graph-based workflows. This package focuses on deterministic, testable orchestration with zero external dependencies.

Customization

Custom Tasks Directory

from cervellaswarm_task_orchestration.task_manager import create_task

create_task("TASK_001", "worker", "My task", tasks_dir="my_project/.tasks")

Plan Validation Sections

The plan validator checks for these required sections:

• ## Metadata (with Task ID, Complexity, Files Affected)
• ## Phase 1: Understanding
• ## Phase 2: Design
• ## Phase 3: Review
• ## Phase 4: Final Plan (with ### Success Criteria)

Worker Type Routing

The router maps keywords to worker types:

API Reference

task_classifier

• classify_task(description, estimated_files?, has_breaking_changes?, force_architect?) -> ClassificationResult
• should_use_architect(description) -> bool
• estimate_files_affected(description) -> int
• calculate_keyword_score(description) -> tuple[float, list[str]]
• is_simple_task(description) -> bool
• has_multifile_pattern(description) -> bool

architect_flow

• route_task(description, task_id?, force_architect?, force_direct?) -> RoutingDecision
• validate_plan(content) -> PlanValidationResult
• validate_plan_file(path) -> PlanValidationResult
• create_session(task_id, description) -> ArchitectSession
• approve_plan(session, approved_by?) -> ArchitectSession
• handle_plan_rejection(session, reason) -> tuple[ArchitectSession, str]
• should_fallback(session) -> bool
• create_fallback_instruction(session) -> str
• save_session_state(session, output_dir?) -> Path

task_manager

• create_task(task_id, agent, description, risk_level?, tasks_dir?) -> str
• list_tasks(tasks_dir?) -> list[dict]
• mark_ready(task_id, tasks_dir?) -> bool
• mark_working(task_id, tasks_dir?) -> bool (atomic!)
• mark_done(task_id, tasks_dir?) -> bool
• ack_received(task_id, tasks_dir?) -> bool
• ack_understood(task_id, tasks_dir?) -> bool
• get_task_status(task_id, tasks_dir?) -> str
• get_ack_status(task_id, tasks_dir?) -> str
• cleanup_task(task_id, tasks_dir?) -> bool
• validate_task_id(task_id) -> bool

output_validator

• validate_output(output_file, logs_dir?) -> ValidationResult
• find_last_output(tasks_dir?) -> Path | None
• find_task_output(task_id, tasks_dir?) -> Path | None

Development

git clone https://github.com/rafapra3008/cervellaswarm.git
cd cervellaswarm/packages/task-orchestration
pip install -e ".[dev]"
pytest

License

Apache-2.0 - see LICENSE for details.

Part of CervellaSwarm - Build AI agent teams that remember.