coordinator.contrib

coordinator.contrib

Capability contribution discovery for the coordinator.

A capability (e.g. the zettelkasten grounded-extraction bundle) can contribute built-in agents and reusable schema definitions into the coordinator WITHOUT the coordinator importing it directly. This keeps the coordinator core generic: it only knows how to discover and merge contributions, not what any particular capability does.

Capabilities register via the angelo.coordinator.capabilities entry-point group. Each entry point resolves to a zero-argument callable returning a contribution dict::

{
    "name": str,                       # capability id (e.g. "zettelkasten")
    "enabled": bool,                   # decided at runtime by the capability
    "agents": {                        # built-in agents contributed
        name: {
            "persona": str,
            "role": "planner"|"implementer"|"checker"|"meta",
            "description": str,
            "model": str | None,       # optional default model
            "schema": str | None,      # optional default schema name
        },
        ...
    },
    "schemas": {                       # optional schema provider
        "expand": Callable[[str], dict | None],  # name -> expanded object
        "list": Callable[[], list[str]],         # available schema names
    },
}

Only contributions whose enabled is true are surfaced. A disabled capability contributes nothing: its agents do not appear in agents(action="list"/"get") and its schemas cannot be resolved. This is the mechanism behind the spec's "conditional shipping": the zettelkasten capability reports enabled=False unless the project opted into it (ANGELO_ZETTELKASTEN env flag), so a coordinator-only project never sees the trio or the schema attribute.

Discovery is cached for the process lifetime. Tests can call reset_cache() to force a reload after mutating the environment.

reset_cache

reset_cache() -> None

Drop the discovery cache so the next call re-discovers capabilities.

Source code in coordinator/contrib.py

def reset_cache() -> None:
    """Drop the discovery cache so the next call re-discovers capabilities."""
    global _cache
    _cache = None

enabled_capabilities

enabled_capabilities() -> list[dict]

Return the cached list of enabled capability contributions.

Source code in coordinator/contrib.py

def enabled_capabilities() -> list[dict]:
    """Return the cached list of enabled capability contributions."""
    global _cache
    if _cache is None:
        _cache = _discover()
    return _cache

capability_enabled

capability_enabled(name: str) -> bool

True when a capability with name is installed and enabled.

Source code in coordinator/contrib.py

def capability_enabled(name: str) -> bool:
    """True when a capability with ``name`` is installed and enabled."""
    return any(c.get("name") == name for c in enabled_capabilities())

contributed_agents

contributed_agents() -> dict[str, dict]

Merged {agent_name: definition} from all enabled capabilities.

Each definition carries the keys a capability supplied: persona, role, description, model (optional), schema (optional default schema name). When two enabled capabilities define the same agent name the first discovered wins (stable, capability-order-dependent).

Source code in coordinator/contrib.py

def contributed_agents() -> dict[str, dict]:
    """Merged ``{agent_name: definition}`` from all enabled capabilities.

    Each definition carries the keys a capability supplied: ``persona``,
    ``role``, ``description``, ``model`` (optional), ``schema`` (optional default
    schema name). When two enabled capabilities define the same agent name the
    first discovered wins (stable, capability-order-dependent).
    """
    merged: dict[str, dict] = {}
    for cap in enabled_capabilities():
        agents = cap.get("agents")
        if not isinstance(agents, dict):
            continue
        for agent_name, definition in agents.items():
            if agent_name not in merged and isinstance(definition, dict):
                merged[agent_name] = definition
    return merged

agent_default_schemas

agent_default_schemas() -> dict[str, str]

Per-agent default schema names declared by contributed agents.

Source code in coordinator/contrib.py

def agent_default_schemas() -> dict[str, str]:
    """Per-agent default schema names declared by contributed agents."""
    defaults: dict[str, str] = {}
    for agent_name, definition in contributed_agents().items():
        schema = definition.get("schema")
        if schema:
            defaults[agent_name] = str(schema)
    return defaults

schemas_available

schemas_available() -> bool

True when at least one enabled capability provides a schema registry.

Source code in coordinator/contrib.py

def schemas_available() -> bool:
    """True when at least one enabled capability provides a schema registry."""
    for cap in enabled_capabilities():
        schemas = cap.get("schemas")
        if isinstance(schemas, dict) and callable(schemas.get("expand")):
            return True
    return False

expand_schema

expand_schema(name: str) -> dict | None

Resolve a schema name to its fully-expanded object, or None.

Searches each enabled capability's schema provider in discovery order and returns the first match. Returns None when no enabled capability knows the name (the caller turns that into a validation error).

Source code in coordinator/contrib.py

def expand_schema(name: str) -> dict | None:
    """Resolve a schema ``name`` to its fully-expanded object, or ``None``.

    Searches each enabled capability's schema provider in discovery order and
    returns the first match. Returns ``None`` when no enabled capability knows
    the name (the caller turns that into a validation error).
    """
    if not name:
        return None
    for cap in enabled_capabilities():
        schemas = cap.get("schemas")
        if not isinstance(schemas, dict):
            continue
        expand = schemas.get("expand")
        if not callable(expand):
            continue
        try:
            result = expand(name)
        except Exception as e:
            logger.warning("Schema expander for %r raised on %r: %s",
                           cap.get("name"), name, e)
            continue
        if result:
            return result
    return None

list_schemas

list_schemas() -> list[str]

Names of all schemas exposed by enabled capabilities (sorted, deduped).

Source code in coordinator/contrib.py

def list_schemas() -> list[str]:
    """Names of all schemas exposed by enabled capabilities (sorted, deduped)."""
    names: set[str] = set()
    for cap in enabled_capabilities():
        schemas = cap.get("schemas")
        if not isinstance(schemas, dict):
            continue
        lister = schemas.get("list")
        if not callable(lister):
            continue
        try:
            result = lister()
        except Exception as e:  # pragma: no cover - defensive
            logger.warning("Schema lister for %r raised: %s", cap.get("name"), e)
            continue
        if result:
            names.update(str(n) for n in result)
    return sorted(names)