orchestrata_ultimate.orchestrator.pipeline_router

Reference › orchestrata_ultimate › orchestrator › pipeline_router

The planner that turns a kind of work into an ordered list of volk stages. The classification side of the intent pipeline (intent_classifier, intent_vocabulary) decides the WorkType of a human request; this module decides which volks handle it and in what order. It is used by pipeline_runner, pipeline_escalation and verification_gate, and by the cheetahclaws integration and the lexicon scripts. It imports nothing internal but the traceability decorator.

Two enums carry the vocabulary. Capability names an atomic ability (intake, analysis, scoping, architecture, implementation, testing, traceability, documentation, ui_design, parallel), and WorkType names the kind of work detected during intake (feature, refactor, test, documentation, analysis, fix, review, spec, dependency, performance). Module-level tables hold the design: CAPABILITY_AGENT maps each capability to a volk alias, WORK_TYPE_CAPABILITIES lists the capabilities each work type needs in order, CAPABILITY_SUBJECTS gives the CXP subject used between two consecutive capabilities, CAPABILITY_TIMEOUTS the per-stage timeout, and AGENT_CLASS_MAP the dotted class path for each alias, loaded lazily into AGENT_CLASS.

PipelineRouter.route accepts a WorkType or its string value (an unknown value raises ValueError) and returns a list of StageSpec records, each holding the volk alias, capability, input and output subjects, timeout and the artifact types the stage is expected to produce. When has_ui is set it inserts the UI design stage before architecture or implementation if the sequence lacks one. get_agent_class resolves an alias to its class.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

Capability (enum)

Bases: StrEnum · Line 20

Atomic capability provided by an agent.

WorkType (enum)

Bases: StrEnum · Line 38

Work type detected by Concierge during intake.

StageSpec (dataclass)

Decorators: traces(spec='003', story='US-004', reqs=['REQ-003-US-004-010']), dataclass · Line 236

Specification for one pipeline stage, resolved from capabilities.

PipelineRouter

Decorators: traces(spec='003', story='US-004', reqs=['REQ-003-US-004-011']) · Line 248

Maps work types to agent stage sequences.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/orchestrator/pipeline_router.py

"""Pipeline router — maps work types to agent stage sequences.

Each work type requires specific capabilities. The router resolves
capabilities → agents and builds the stage list for PipelineRunner.

Capabilities are the atomic units; agents provide one or more capabilities.
Compliance (U:Cp) always runs last for traceability.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import StrEnum

from orchestrata_ultimate.core.base.traces import traces

# ── Capabilities ──────────────────────────────────────────────────────


class Capability(StrEnum):
    """Atomic capability provided by an agent."""

    INTAKE = "intake"  # Requirements elicitation (Concierge)
    ANALYSIS = "analysis"  # Requirements analysis, decomposition (Analyst)
    SCOPING = "scoping"  # Effort estimation, warnings (PM)
    ARCHITECTURE = "architecture"  # System design, file tree, API spec (Architect)
    IMPLEMENTATION = "implementation"  # Code generation, file creation (CodeDirector)
    TESTING = "testing"  # Test generation, code review, QA (QA)
    TRACEABILITY = "traceability"  # RTM, compliance, doc traces (Compliance)
    DOCUMENTATION = "documentation"  # Doc generation, spec writing (Documentor)
    UI_DESIGN = "ui_design"  # UX design artifacts (UX)
    PARALLEL = "parallel"  # Swarm parallel generation (Junior/Swarm)


# ── Work Types ────────────────────────────────────────────────────────


class WorkType(StrEnum):
    """Work type detected by Concierge during intake."""

    FEATURE = "feature"
    REFACTOR = "refactor"
    TEST = "test"
    DOCUMENTATION = "documentation"
    ANALYSIS = "analysis"
    FIX = "fix"
    REVIEW = "review"
    SPEC = "spec"
    DEPENDENCY = "dependency"
    PERFORMANCE = "performance"


# ── Capability → Agent ─────────────────────────────────────────────────


CAPABILITY_AGENT: dict[Capability, str] = {
    Capability.INTAKE: "U:Cx",
    Capability.ANALYSIS: "U:BA",
    Capability.SCOPING: "U:PM",
    Capability.ARCHITECTURE: "U:Ar",
    Capability.IMPLEMENTATION: "U:Cd",
    Capability.TESTING: "U:Qa",
    Capability.TRACEABILITY: "U:Cp",
    Capability.DOCUMENTATION: "U:Do",
    Capability.UI_DESIGN: "U:Ux",
    Capability.PARALLEL: "J:Junior",
}

# Agent → Agent class
AGENT_CLASS: dict[str, type] = {}  # Populated lazily

AGENT_CLASS_MAP: dict[str, str] = {
    "U:Cx": "orchestrata_ultimate.agents.volks.concierge.volk.ConciergeVolk",
    "U:BA": "orchestrata_ultimate.agents.volks.analyst.agent.AnalystAgent",
    "U:PM": "orchestrata_ultimate.agents.volks.pm.volk.PMVolk",
    "U:Ar": "orchestrata_ultimate.agents.volks.architect.volk.ArchitectVolk",
    "U:Cd": "orchestrata_ultimate.agents.volks.codedirector.volk.CodeDirectorVolk",
    "U:Qa": "orchestrata_ultimate.agents.volks.qa.volk.QaVolk",
    "U:Cp": "orchestrata_ultimate.agents.volks.compliance.volk.ComplianceVolk",
    "U:Do": "orchestrata_ultimate.agents.volks.documentor.volk.DocumentorVolk",
    "U:Ux": "orchestrata_ultimate.agents.volks.ux.volk.UXVolk",
    # U:SD and J:Junior are dispatchable kernel-side (kernel/pipeline.py
    # registers all 11 built-in volks in a flat alias dict, so `to_agent=`
    # addressing already reaches them) but were missing here, so
    # get_agent_class() raised ValueError for anyone resolving them through
    # this router. Added so both resolve; NOT wired into any
    # WORK_TYPE_CAPABILITIES sequence — see the module-level note below
    # ("Reachability note") for why that part of the brief was not guessed.
    "U:SD": "orchestrata_ultimate.agents.volks.senior_dev.agent.SeniorDevAgent",
    "J:Junior": "orchestrata_ultimate.agents.volks.junior.volk.JuniorVolk",
}


# ── Work Type → Capability Sequence ───────────────────────────────────

# Reachability note (2026-08-19): Capability.PARALLEL (→ J:Junior) appears in
# NO sequence below, and there is no Capability member at all for U:SD's
# oscillation-recovery role. Both were deliberately left unrouted here rather
# than guessed in:
#   - No spec (specs/027-junior-volk, specs/012-pm-volk, specs/000-orchestrata-
#     cli) defines a WorkType/Capability position for either volk. Their
#     established, spec-grounded invocation is point-to-point CXP addressing —
#     U:Ar⇄J:Junior (design_review) and U:Cd⇄J:Junior (code_generate) per
#     SPEC-027 US-001/US-002, and goal-runner→U:SD→U:Cx (subject
#     "prompt_refinement") per senior_dev/agent.py's own docstring — not a
#     forward linear pipeline stage.
#   - Inserting Capability.PARALLEL between ARCHITECTURE and IMPLEMENTATION in
#     any sequence empties the existing (ARCHITECTURE, IMPLEMENTATION) =
#     "code_implementation" CAPABILITY_SUBJECTS pairing for that WorkType
#     (route()'s subject lookup is by adjacent-capability pair), silently
#     breaking today's handoff unless new subject entries are added — and the
#     only subjects J:Junior's process() actually answers are "design_review"/
#     "code_generate", not "code_implementation", so a same-named substitute
#     would produce a stage nothing downstream can service.
#   - Appending either capability at the end of any sequence violates the
#     "Compliance (U:Cp) always runs last" invariant stated at the top of this
#     module and asserted by test_route_all_end_with_compliance.
# Routed anyway, this is a real design fork (a wrong choice silently changes
# production routing for every request of that WorkType) — reported to the
# orchestrator rather than picked here per this repo's wiring discipline.
WORK_TYPE_CAPABILITIES: dict[WorkType, list[Capability]] = {
    WorkType.FEATURE: [
        Capability.INTAKE,
        Capability.ANALYSIS,
        Capability.SCOPING,
        Capability.ARCHITECTURE,
        Capability.IMPLEMENTATION,
        Capability.TESTING,
        Capability.TRACEABILITY,
    ],
    WorkType.REFACTOR: [
        Capability.ARCHITECTURE,
        Capability.IMPLEMENTATION,
        Capability.TESTING,
        Capability.TRACEABILITY,
    ],
    WorkType.TEST: [
        Capability.IMPLEMENTATION,
        Capability.TESTING,
        Capability.TRACEABILITY,
    ],
    WorkType.DOCUMENTATION: [
        Capability.INTAKE,
        Capability.DOCUMENTATION,
        Capability.TRACEABILITY,
    ],
    WorkType.ANALYSIS: [
        Capability.INTAKE,
        Capability.ANALYSIS,
        Capability.TRACEABILITY,
    ],
    WorkType.FIX: [
        Capability.TESTING,
        Capability.IMPLEMENTATION,
        Capability.TESTING,
        Capability.TRACEABILITY,
    ],
    WorkType.REVIEW: [
        Capability.TESTING,
        Capability.TRACEABILITY,
    ],
    WorkType.SPEC: [
        Capability.INTAKE,
        Capability.ANALYSIS,
        Capability.DOCUMENTATION,
        Capability.TRACEABILITY,
    ],
    WorkType.DEPENDENCY: [
        Capability.ARCHITECTURE,
        Capability.IMPLEMENTATION,
        Capability.TESTING,
        Capability.TRACEABILITY,
    ],
    WorkType.PERFORMANCE: [
        Capability.TESTING,
        Capability.ARCHITECTURE,
        Capability.IMPLEMENTATION,
        Capability.TESTING,
        Capability.TRACEABILITY,
    ],
}

# Subject routing: which CXP subject connects each capability pair
CAPABILITY_SUBJECTS: dict[tuple[Capability, Capability], str] = {
    (Capability.INTAKE, Capability.ANALYSIS): "analyze_requirements",
    (Capability.INTAKE, Capability.DOCUMENTATION): "generate_docs",
    # Every value here must be a subject the TARGET volk's process() actually
    # answers — the same defect class as the F18 "documentation_complete" fix
    # below. Measured 2026-08-29 by probing process() directly: U:PM, U:Ar,
    # U:Cd, U:Qa and U:Ux each returned None for the subject this table sent,
    # before any LLM call, so feature/refactor/performance routes died at
    # their second stage while the suite stayed green. The replacements are
    # taken from each volk's own vocabulary, not invented:
    #   U:PM  process() -> implement | scope_review | progress_update
    #   U:Ar  process() -> implement | scope_validated | ux_complete | ...
    #   U:Cd  process() -> implement | implementation_complete | fix_issues
    #   U:Qa  process() -> review_and_test | review_edit
    #   U:Ux  process() -> ux_design
    # Guarded now by test_every_routed_subject_is_accepted_by_its_volk.
    (Capability.ANALYSIS, Capability.SCOPING): "scope_review",
    (Capability.ANALYSIS, Capability.TRACEABILITY): "compliance_check",
    (Capability.ANALYSIS, Capability.DOCUMENTATION): "generate_docs",
    (Capability.SCOPING, Capability.ARCHITECTURE): "scope_validated",
    (Capability.ARCHITECTURE, Capability.IMPLEMENTATION): "implement",
    (Capability.IMPLEMENTATION, Capability.TESTING): "review_and_test",
    (Capability.IMPLEMENTATION, Capability.TRACEABILITY): "compliance_check",
    (Capability.TESTING, Capability.IMPLEMENTATION): "fix_issues",
    (Capability.TESTING, Capability.TRACEABILITY): "compliance_check",
    # "compliance_check" like every other →TRACEABILITY pair: ComplianceVolk
    # handles exactly that one subject (dogfood F18 — the previous
    # "documentation_complete" mapping made Cp return None and fail the
    # documentation route silently on every run).
    (Capability.DOCUMENTATION, Capability.TRACEABILITY): "compliance_check",
}

# Approximate timeouts per capability (seconds)
CAPABILITY_TIMEOUTS: dict[Capability, int] = {
    Capability.INTAKE: 300,
    Capability.ANALYSIS: 300,
    Capability.SCOPING: 120,
    Capability.ARCHITECTURE: 300,
    Capability.IMPLEMENTATION: 600,
    Capability.TESTING: 600,
    Capability.TRACEABILITY: 120,
    Capability.DOCUMENTATION: 300,
    Capability.UI_DESIGN: 180,
    Capability.PARALLEL: 300,
}


# ── Router ────────────────────────────────────────────────────────────


@traces(spec="003", story="US-004", reqs=["REQ-003-US-004-010"])
@dataclass
class StageSpec:
    """Specification for one pipeline stage, resolved from capabilities."""

    alias: str
    capability: Capability
    input_subject: str = ""
    output_subject: str = ""
    timeout_seconds: int = 300
    artifact_types: list[str] = field(default_factory=list)


@traces(spec="003", story="US-004", reqs=["REQ-003-US-004-011"])
class PipelineRouter:
    """Maps work types to agent stage sequences."""

    def __init__(self) -> None:
        self._cap_artifacts: dict[Capability, list[str]] = {
            Capability.INTAKE: ["requirements", "prolog_facts", "task_summary"],
            Capability.ANALYSIS: [
                "user_stories",
                "requirements_analysis",
                "project_meta",
            ],
            Capability.SCOPING: ["scope_estimate", "warnings"],
            Capability.ARCHITECTURE: [
                "architecture_design",
                "file_tree",
                "api_spec",
                "class_diagram",
            ],
            Capability.IMPLEMENTATION: ["source_code", "implementation_plan"],
            Capability.TESTING: ["test_code", "qa_report"],
            Capability.TRACEABILITY: ["rtm", "compliance_report"],
            Capability.DOCUMENTATION: ["doc_output"],
            Capability.UI_DESIGN: ["ux_design"],
            Capability.PARALLEL: ["source_code"],
        }

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-012"])
    def route(self, work_type: WorkType | str, has_ui: bool = False) -> list[StageSpec]:
        """Build stage sequence for a work type.

        Args:
            work_type: WorkType enum or string value.
            has_ui: Whether UI keywords were detected (adds U:Ux stage).

        Returns:
            Ordered list of StageSpec for PipelineRunner.

        Raises:
            ValueError: If work_type is unknown.
        """
        if isinstance(work_type, str):
            work_type = WorkType(work_type)

        capabilities = WORK_TYPE_CAPABILITIES.get(work_type)
        if capabilities is None:
            raise ValueError(f"Unknown work type: {work_type}")

        stages: list[StageSpec] = []
        prev_cap: Capability | None = None

        for cap in capabilities:
            agent_alias = CAPABILITY_AGENT.get(cap)
            if not agent_alias:
                continue

            # Determine input/output subjects
            input_subject = ""
            output_subject = ""
            if prev_cap:
                input_subject = CAPABILITY_SUBJECTS.get((prev_cap, cap), "")
            if len(capabilities) > capabilities.index(cap) + 1:
                next_cap = capabilities[capabilities.index(cap) + 1]
                output_subject = CAPABILITY_SUBJECTS.get((cap, next_cap), "")

            stages.append(
                StageSpec(
                    alias=agent_alias,
                    capability=cap,
                    input_subject=input_subject,
                    output_subject=output_subject,
                    timeout_seconds=CAPABILITY_TIMEOUTS.get(cap, 300),
                    artifact_types=self._cap_artifacts.get(cap, []),
                )
            )

            prev_cap = cap

        # Inject UX stage if UI keywords detected and not already included
        if has_ui and not any(s.capability == Capability.UI_DESIGN for s in stages):
            # Add UX before architecture if present, otherwise before implementation
            ux_alias = CAPABILITY_AGENT[Capability.UI_DESIGN]
            insert_idx = next(
                (
                    i
                    for i, s in enumerate(stages)
                    if s.capability
                    in (Capability.ARCHITECTURE, Capability.IMPLEMENTATION)
                ),
                len(stages),
            )
            stages.insert(
                insert_idx,
                StageSpec(
                    alias=ux_alias,
                    capability=Capability.UI_DESIGN,
                    # "ux_design", not "design_ux": UXVolk.process() answers
                    # exactly one subject and it is spelled the other way
                    # round. The transposed string returned None before any
                    # LLM call, so every ui_design route died here.
                    input_subject="ux_design",
                    output_subject="",
                    timeout_seconds=CAPABILITY_TIMEOUTS[Capability.UI_DESIGN],
                    artifact_types=self._cap_artifacts[Capability.UI_DESIGN],
                ),
            )

        return stages

    @traces(spec="003", story="US-004", reqs=["REQ-003-US-004-013"])
    def get_agent_class(self, alias: str) -> type:
        """Resolve agent alias to class (lazy import)."""
        import importlib

        class_path = AGENT_CLASS_MAP.get(alias)
        if not class_path:
            raise ValueError(f"No class mapping for agent: {alias}")
        module_path, class_name = class_path.rsplit(".", 1)
        module = importlib.import_module(module_path)
        agent_class: type = getattr(module, class_name)
        return agent_class