orchestrata_ultimate.core.capability_mesh

Reference › orchestrata_ultimate › core › capability_mesh

The single capability-to-provider resolver used by both pipeline meshes. A volk can ask for a capability such as testing instead of naming a peer, and two runners resolve it at different levels: kernel.pipeline_escalation rewrites the packet's to_agent and hops (the gRPC path), while orchestrator.pipeline_escalation inserts a stage into the remaining sequence (the gateway and UI path). The resolution step is identical in both, so it lives here once.

CAPABILITY_PROVIDERS maps the core capabilities to built-in volk aliases (for example requirements to U:Cx, design to U:Ar, testing and review to U:Qa, coding to J:Junior). resolve_capability(volks, capability) returns the built-in alias only when that volk is registered; otherwise it scans the registered volks for one whose capabilities list contains the request, which is how a plugin volk extends the mesh. It returns None when nothing provides the capability: the kernel then leaves to_agent alone, and the orchestrator escalates to a human.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

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

"""The capability -> provider resolver shared by both pipeline meshes.

A volk can ask for a CAPABILITY ("testing") instead of naming a peer, and the
mesh resolves it to whichever volk provides it. Two runners do that resolution,
at different granularities:

* ``kernel.pipeline_escalation`` routes at the PACKET level -- it rewrites
  ``packet.to_agent`` and hops, for the gRPC path;
* ``orchestrator.pipeline_escalation`` routes at the STAGE level -- it inserts
  a ``StageSpec`` into the remaining sequence, for the gateway/UI path.

Those are complements, not duplicates, and both stay. The RESOLUTION step is
identical in both, though, and lived here twice: the same function body and the
same provider table, the second copy carrying a "ported verbatim" comment. Two
copies of a routing table is how the two meshes start disagreeing about who
provides ``testing``, so it lives here once.
"""

from __future__ import annotations

from collections.abc import Mapping
from typing import Any

# Capability -> built-in provider alias. Authoritative for the CORE
# capabilities, so core routing stays stable no matter what is registered; a
# volk that declares its own ``capabilities`` list provides any ADDITIONAL
# capability, which is how a plugin volk extends the mesh without hijacking a
# core hop.
CAPABILITY_PROVIDERS: dict[str, str] = {
    "requirements": "U:Cx",  # Concierge — intake / elicitation
    "clarification": "U:Cx",
    "analysis": "U:BA",  # Analyst — user stories
    "planning": "U:PM",  # PM — scope & estimation
    "scope": "U:PM",
    "design": "U:Ar",  # Architect — system design
    "architecture": "U:Ar",
    "ux": "U:Ux",  # UX
    "implementation": "U:Cd",  # CodeDirector
    "coding": "J:Junior",  # Junior — code gen
    "testing": "U:Qa",  # QA — test gen + gates
    "review": "U:Qa",
    "compliance": "U:Cp",  # Compliance — RTM / trace
    "traceability": "U:Cp",
}


def resolve_capability(volks: Mapping[str, Any], capability: str) -> str | None:
    """Alias of a registered volk that provides *capability*, or None.

    The built-in table wins for core capabilities and is checked first, but
    only when that provider is actually registered in *volks* — an unregistered
    built-in falls through to the dynamic scan rather than returning an alias
    nobody can dispatch to.

    None means no provider is registered. The two callers do different things
    with that: the kernel leaves the packet's existing ``to_agent`` alone
    (fixed choreography preserved), the orchestrator escalates to a human
    rather than continue silently.
    """
    if not capability:
        return None
    alias = CAPABILITY_PROVIDERS.get(capability)
    if alias and alias in volks:
        return alias
    for candidate_alias, volk in volks.items():
        caps = getattr(volk, "capabilities", None)
        if caps and capability in caps:
            return candidate_alias
    return None