coordinator.agents

coordinator.agents

Agent persona templates for subagent prompts.

Each persona is a plain string injected into the Task tool prompt when the coordinator spawns a subagent. Keep these short and directive -- the subagent's context window should be spent on the actual task, not on lengthy system prompts.

find_agents_file

find_agents_file() -> Path | None

Resolve the agents.yaml path: AGENTS_FILE env > .cursor/ > .claude/ > None.

Source code in coordinator/agents.py

def find_agents_file() -> Path | None:
    """Resolve the agents.yaml path: AGENTS_FILE env > .cursor/ > .claude/ > None."""
    env = os.environ.get("AGENTS_FILE")
    if env:
        p = Path(env)
        return p if p.exists() else None
    for candidate in (".cursor/agents.yaml", ".claude/agents.yaml"):
        p = Path(candidate)
        if p.exists():
            return p
    return None

load_local_agents

load_local_agents(agents_file: Path | None = None) -> dict[str, dict]

Parse agents.yaml into project-local override maps (no global state).

Discovery order: AGENTS_FILE env var > .cursor/agents.yaml > .claude/agents.yaml.

Returns a dict with six sub-maps keyed by agent name

{"personas": {...}, "descriptions": {...}, "roles": {...}, "models": {...}, "passes": {...}, "schemas": {...}}

Each agent entry in the YAML may define persona, description, role (validated against VALID_ROLES), model (default model slug), passes (int >= 1, clamped), schema (default extraction-schema name for this agent's tasks), and extends (the name of a base agent this entry inherits unspecified fields from -- see apply_inheritance). Malformed values are warned about and skipped rather than raising, mirroring the coordinator's tolerant loading.

Source code in coordinator/agents.py

def load_local_agents(agents_file: Path | None = None) -> dict[str, dict]:
    """Parse agents.yaml into project-local override maps (no global state).

    Discovery order: AGENTS_FILE env var > .cursor/agents.yaml > .claude/agents.yaml.

    Returns a dict with six sub-maps keyed by agent name:
        {"personas": {...}, "descriptions": {...}, "roles": {...},
         "models": {...}, "passes": {...}, "schemas": {...}}

    Each agent entry in the YAML may define ``persona``, ``description``,
    ``role`` (validated against VALID_ROLES), ``model`` (default model slug),
    ``passes`` (int >= 1, clamped), ``schema`` (default extraction-schema
    name for this agent's tasks), and ``extends`` (the name of a base agent
    this entry inherits unspecified fields from -- see ``apply_inheritance``).
    Malformed values are warned about and skipped rather than raising,
    mirroring the coordinator's tolerant loading.
    """
    local: dict[str, dict] = {
        "personas": {},
        "descriptions": {},
        "roles": {},
        "models": {},
        "passes": {},
        "schemas": {},
        "native_subagents": {},
        "extends": {},
    }
    if agents_file is None:
        agents_file = find_agents_file()
    if agents_file is None or not agents_file.exists():
        return local
    try:
        import yaml
        data = yaml.safe_load(agents_file.read_text(encoding="utf-8")) or {}
    except Exception as e:
        logger.warning("Failed to load %s: %s", agents_file, e)
        return local

    for name, agent in data.items():
        if name in RESERVED_AGENT_KEYS:
            continue
        if not isinstance(agent, dict):
            continue
        extends = agent.get("extends")
        if extends is not None:
            local["extends"][name] = str(extends)
        if "persona" in agent:
            local["personas"][name] = agent["persona"]
        if "description" in agent:
            local["descriptions"][name] = agent["description"]
        model = agent.get("model")
        if model is not None:
            local["models"][name] = str(model)
        schema = agent.get("schema")
        if schema is not None:
            local["schemas"][name] = str(schema)
        native = agent.get("native_subagent")
        if native is not None:
            local["native_subagents"][name] = str(native)
        passes = agent.get("passes")
        if passes is not None:
            try:
                n = int(passes)
            except (TypeError, ValueError):
                logger.warning(
                    "Agent %r has non-integer passes %r; ignoring "
                    "(defaulting to %d).",
                    name, passes, DEFAULT_PASSES,
                )
            else:
                if n < 1:
                    logger.warning(
                        "Agent %r has passes %d < 1; clamping to 1.", name, n,
                    )
                    n = 1
                local["passes"][name] = n
        role = agent.get("role")
        if role is not None:
            if role in VALID_ROLES:
                local["roles"][name] = role
            else:
                logger.warning(
                    "Agent %r has invalid role %r (valid: %s); "
                    "treating as 'checker'.",
                    name, role, ", ".join(sorted(VALID_ROLES)),
                )
        elif name not in AGENT_ROLES and name not in local["extends"]:
            logger.warning(
                "Agent %r has no 'role' field; treating as 'checker' "
                "(must run downstream of an implementer).",
                name,
            )
    return local

load_local_config

load_local_config(agents_file: Path | None = None) -> dict

Parse top-level coordinator settings (reserved keys) from agents.yaml.

Returns {"rigor": str|None, "max_extensions": int|None}. These keys configure the coordinator itself and are never treated as agent definitions (see RESERVED_AGENT_KEYS). Malformed values are warned about and ignored.

Source code in coordinator/agents.py

def load_local_config(agents_file: Path | None = None) -> dict:
    """Parse top-level coordinator settings (reserved keys) from agents.yaml.

    Returns ``{"rigor": str|None, "max_extensions": int|None}``. These keys
    configure the coordinator itself and are never treated as agent definitions
    (see RESERVED_AGENT_KEYS). Malformed values are warned about and ignored.
    """
    config: dict = {"rigor": None, "max_extensions": None}
    if agents_file is None:
        agents_file = find_agents_file()
    if agents_file is None or not agents_file.exists():
        return config
    try:
        import yaml
        data = yaml.safe_load(agents_file.read_text(encoding="utf-8")) or {}
    except Exception as e:
        logger.warning("Failed to load %s: %s", agents_file, e)
        return config
    if not isinstance(data, dict):
        return config

    rigor = data.get("rigor")
    if rigor is not None:
        r = str(rigor).lower()
        if r in RIGOR_PROFILES:
            config["rigor"] = r
        else:
            logger.warning(
                "Invalid rigor %r (valid: %s); ignoring.",
                rigor, ", ".join(sorted(RIGOR_PROFILES)),
            )

    max_ext = data.get("max_extensions")
    if max_ext is not None:
        try:
            n = int(max_ext)
        except (TypeError, ValueError):
            logger.warning(
                "Non-integer max_extensions %r; ignoring.", max_ext,
            )
        else:
            if n < 0:
                logger.warning("Negative max_extensions %d; clamping to 0.", n)
                n = 0
            config["max_extensions"] = n
    return config

resolve_rigor

resolve_rigor(rigor: str | None = None, max_extensions: int | None = None, *, config: dict | None = None) -> dict

Resolve the effective rigor settings for a run.

Returns {"rigor": <name>, "max_extensions": int, "passes": {agent: n}}.

Precedence

• rigor name: explicit rigor arg > config rigor > DEFAULT_RIGOR
• max_extensions: explicit max_extensions arg > (when rigor is explicitly given) that profile's cap > config max_extensions > the resolved profile's cap
• passes: the resolved rigor profile's passes map, used as defaults that sit below agents.yaml per-agent passes and task-level passes

Source code in coordinator/agents.py

def resolve_rigor(
    rigor: str | None = None,
    max_extensions: int | None = None,
    *,
    config: dict | None = None,
) -> dict:
    """Resolve the effective rigor settings for a run.

    Returns ``{"rigor": <name>, "max_extensions": int, "passes": {agent: n}}``.

    Precedence:
      - rigor name: explicit ``rigor`` arg > config ``rigor`` > DEFAULT_RIGOR
      - max_extensions: explicit ``max_extensions`` arg > (when ``rigor`` is
        explicitly given) that profile's cap > config ``max_extensions`` >
        the resolved profile's cap
      - passes: the resolved rigor profile's passes map, used as defaults that
        sit below agents.yaml per-agent passes and task-level passes
    """
    config = config or {}
    name = rigor or config.get("rigor") or DEFAULT_RIGOR
    if name not in RIGOR_PROFILES:
        name = DEFAULT_RIGOR
    profile = RIGOR_PROFILES[name]
    if max_extensions is not None:
        effective_max = max_extensions
    elif rigor:
        # An explicit per-run rigor selection carries its own cap, which beats
        # the persistent agents.yaml max_extensions default.
        effective_max = profile["max_extensions"]
    elif config.get("max_extensions") is not None:
        effective_max = config["max_extensions"]
    else:
        effective_max = profile["max_extensions"]
    return {
        "rigor": name,
        "max_extensions": int(effective_max),
        "passes": dict(profile.get("passes", {})),
    }

apply_inheritance

apply_inheritance(local: dict[str, dict], contributed: dict[str, dict] | None = None) -> dict[str, dict]

Fill unspecified fields of extends: agents from their base agent.

Mutates local in place (and returns it). For each agent that declares extends: <base>, any attribute it did not set explicitly is copied from the base's effective value. The base may be a coordinator built-in, a capability-contributed agent, or another project-local agent, and chains are followed (A extends B extends engineer). Unknown bases and cycles are warned about and skipped, leaving whatever the agent did set explicitly.

contributed is the {name: definition} map (loaded lazily when omitted) so a base that is a contributed agent resolves correctly.

Source code in coordinator/agents.py

def apply_inheritance(
    local: dict[str, dict],
    contributed: dict[str, dict] | None = None,
) -> dict[str, dict]:
    """Fill unspecified fields of ``extends:`` agents from their base agent.

    Mutates ``local`` in place (and returns it). For each agent that declares
    ``extends: <base>``, any attribute it did not set explicitly is copied from
    the base's effective value. The base may be a coordinator built-in, a
    capability-contributed agent, or another project-local agent, and chains are
    followed (A extends B extends engineer). Unknown bases and cycles are warned
    about and skipped, leaving whatever the agent did set explicitly.

    ``contributed`` is the ``{name: definition}`` map (loaded lazily when
    omitted) so a base that is a contributed agent resolves correctly.
    """
    extends: dict[str, str] = local.get("extends", {})
    if not extends:
        return local

    contrib = contributed_agent_maps(contributed)

    known = (
        set(_BUILTIN_ATTR_DEFAULTS["roles"])
        | set(PERSONAS)
        | set(AGENT_DESCRIPTIONS)
        | set(AGENT_MODELS)
        | set(AGENT_PASSES)
        | set(contrib["names"])  # type: ignore[arg-type]
    )
    for sub in _INHERITABLE_ATTRS:
        known |= set(local.get(sub, {}))
    known |= set(extends)

    def resolve_attr(attr: str, name: str, seen: frozenset[str]):
        """Effective (value, found) for ``name``'s ``attr``, following extends."""
        if name in local.get(attr, {}):
            return local[attr][name], True
        if name in contrib.get(attr, {}):
            return contrib[attr][name], True
        if name in _BUILTIN_ATTR_DEFAULTS[attr]:
            return _BUILTIN_ATTR_DEFAULTS[attr][name], True
        base = extends.get(name)
        if base is None or name in seen:
            return None, False
        return resolve_attr(attr, base, seen | {name})

    for name, base in extends.items():
        if base not in known:
            logger.warning(
                "Agent %r extends unknown agent %r; skipping inheritance.",
                name, base,
            )
            continue
        if _detect_extends_cycle(name, extends):
            logger.warning(
                "Agent %r has a cyclic extends chain; skipping inheritance.",
                name,
            )
            continue
        for attr in _INHERITABLE_ATTRS:
            if name in local.get(attr, {}):
                continue  # explicit value wins
            value, found = resolve_attr(attr, base, frozenset({name}))
            if found and value is not None:
                local.setdefault(attr, {})[name] = value
    return local

contributed_agent_maps

contributed_agent_maps(contributed: dict[str, dict] | None = None) -> dict[str, dict]

Split a contributed-agent map into per-attribute sub-maps.

contributed is the {name: definition} map from coordinator.contrib.contributed_agents() (loaded lazily when omitted). Returns {"personas", "descriptions", "roles", "models", "schemas", "names"} so callers can merge capability contributions the same way they merge project-local overrides.

Source code in coordinator/agents.py

def contributed_agent_maps(
    contributed: dict[str, dict] | None = None,
) -> dict[str, dict]:
    """Split a contributed-agent map into per-attribute sub-maps.

    ``contributed`` is the ``{name: definition}`` map from
    ``coordinator.contrib.contributed_agents()`` (loaded lazily when omitted).
    Returns ``{"personas", "descriptions", "roles", "models", "schemas",
    "names"}`` so callers can merge capability contributions the same way they
    merge project-local overrides.
    """
    if contributed is None:
        from coordinator.contrib import contributed_agents

        contributed = contributed_agents()

    maps: dict[str, dict] = {
        "personas": {},
        "descriptions": {},
        "roles": {},
        "models": {},
        "schemas": {},
        "native_subagents": {},
    }
    names: set[str] = set()
    for name, definition in (contributed or {}).items():
        if not isinstance(definition, dict):
            continue
        names.add(name)
        if definition.get("persona"):
            maps["personas"][name] = definition["persona"]
        if definition.get("description"):
            maps["descriptions"][name] = definition["description"]
        if definition.get("role"):
            maps["roles"][name] = definition["role"]
        if definition.get("model"):
            maps["models"][name] = str(definition["model"])
        if definition.get("schema"):
            maps["schemas"][name] = str(definition["schema"])
        if definition.get("native_subagent"):
            maps["native_subagents"][name] = str(definition["native_subagent"])
    maps["names"] = names  # type: ignore[assignment]
    return maps

build_agent_registry

build_agent_registry(local: dict[str, dict] | None = None, contributed: dict[str, dict] | None = None) -> list[dict]

Return the merged roster of built-in + contributed + project-local agents.

Precedence (lowest to highest): coordinator core built-ins < agents contributed by an enabled capability (e.g. the zettelkasten trio) < project-local overrides (agents.yaml). The result is a list (sorted by name) of dicts, one per agent: {"name", "role", "description", "model" (or None), "passes", "has_persona", "builtin"}

Used by the dashboard's workflow builder and the coordinator's list_agents so both surface the same agents. Pass local / contributed to reuse already-loaded maps; otherwise they are loaded from disk / discovered.

Source code in coordinator/agents.py

def build_agent_registry(
    local: dict[str, dict] | None = None,
    contributed: dict[str, dict] | None = None,
) -> list[dict]:
    """Return the merged roster of built-in + contributed + project-local agents.

    Precedence (lowest to highest): coordinator core built-ins < agents
    contributed by an enabled capability (e.g. the zettelkasten trio) <
    project-local overrides (``agents.yaml``). The result is a list (sorted by
    name) of dicts, one per agent:
        {"name", "role", "description", "model" (or None), "passes",
         "has_persona", "builtin"}

    Used by the dashboard's workflow builder and the coordinator's ``list_agents``
    so both surface the same agents. Pass ``local`` / ``contributed`` to reuse
    already-loaded maps; otherwise they are loaded from disk / discovered.
    """
    if local is None:
        local = load_local_agents()
    contrib = contributed_agent_maps(contributed)
    contrib_names = contrib["names"]  # type: ignore[index]
    apply_inheritance(local, contributed)

    descriptions = {**AGENT_DESCRIPTIONS, **contrib["descriptions"], **local["descriptions"]}
    roles = {**AGENT_ROLES, **contrib["roles"], **local["roles"]}
    models = {**AGENT_MODELS, **contrib["models"], **local["models"]}
    passes = {**AGENT_PASSES, **local["passes"]}
    native = {
        **NATIVE_SUBAGENTS,
        **contrib["native_subagents"],
        **local["native_subagents"],
    }
    persona_names = set(PERSONAS) | set(contrib["personas"]) | set(local["personas"])

    names = (
        set(descriptions)
        | set(roles)
        | set(models)
        | set(passes)
        | persona_names
    )

    registry: list[dict] = []
    for name in sorted(names):
        registry.append({
            "name": name,
            "role": roles.get(name, "checker"),
            "description": descriptions.get(name, ""),
            "model": models.get(name),
            "passes": passes.get(name, DEFAULT_PASSES),
            "native_subagent": native.get(name),
            "has_persona": name in persona_names,
            "builtin": (
                name in AGENT_ROLES or name in PERSONAS or name in contrib_names
            ),
        })
    return registry