Roster

roster

Loads agent and tool definitions from layered YAML config into a queryable roster.

Delegates to .loader for YAML deserialization and roster construction, and to .config_stack for generic layered config resolution.

__all__ = ['AgentRoster', 'MountDef', 'SidecarSpec', 'VaultRoute', 'load_roster'] module-attribute

AgentRoster(_providers=dict(), _auth_providers=dict(), _vault_routes=dict(), _sidecar_specs=dict(), _installs=dict(), _helps=dict(), _mounts=(), _agent_names=(), _all_names=(), _web_ingress=frozenset()) dataclass

Queryable view over the loaded set of agents and tools.

Returned by load_roster; grouped accessors expose providers, auth providers, vault routes, sidecar specs, install snippets, and help blurbs by name.

providers property

All headless agent providers (kind: agent only).

auth_providers property

All auth providers (agents + tools with auth: section).

vault_routes property

All vault routes, keyed by provider name.

sidecar_specs property

All sidecar tool specs, keyed by tool name.

agent_names property

Names of kind: agent entries (for CLI completion).

all_names property

Names of all entries (agents + tools).

installs property

All install specs, keyed by roster name (entries without one are absent).

helps property

All help blurbs, keyed by roster name (entries without one are absent).

web_ingress property

Names of entries that publish a host HTTP port (web_ingress: true).

Consumers (e.g. terok's task launcher) use this to decide whether to allocate a published port and drop a per-task auth token into the container-visible config dir.

mounts property

All shared directory mounts (auth dirs + explicit mounts: sections).

Deduplicated by host_dir — if auth and mounts define the same directory, only one entry is returned.

resolve_selection(selection)

Resolve a user-supplied selection into the full set of roster names to install.

Accepts the literal string "all" (every roster entry that has an InstallSpec) or a tuple of selection tokens. Each token is either a roster name (include) or a name prefixed with - (exclude). The pseudo-name "all" is also valid as an include token, meaning "seed from every installable entry"; this combines naturally with excludes, e.g. ("all", "-vibe") installs everything except vibe. When no include tokens are present (only excludes), the seed is the full roster.

Includes are expanded transitively via depends_on before excludes are applied, so an exclude that names a dependency of a kept agent will silently drop that dependency — likely producing a broken image, but matching the user's literal request.

Returns the names sorted alphabetically — the canonical order used for the OCI label, the tag suffix, and the in-container manifest.

Raises ValueError if a requested include or exclude name is not in the roster, or TypeError if selection is a string other than "all" (a bare name like "claude" would otherwise be iterated into characters). Excludes that name a known agent but don't appear in the resolved include set are a no-op.

Source code in src/terok_executor/roster/loader.py

def resolve_selection(self, selection: str | tuple[str, ...]) -> tuple[str, ...]:
    """Resolve a user-supplied selection into the full set of roster names to install.

    Accepts the literal string ``"all"`` (every roster entry that has an
    [`InstallSpec`][terok_executor.roster.types.InstallSpec]) or a tuple of
    selection tokens.  Each token is either a roster name (include) or a
    name prefixed with ``-`` (exclude).  The pseudo-name ``"all"`` is also
    valid as an include token, meaning "seed from every installable
    entry"; this combines naturally with excludes, e.g. ``("all",
    "-vibe")`` installs everything except vibe.  When no include tokens
    are present (only excludes), the seed is the full roster.

    Includes are expanded transitively via ``depends_on`` *before*
    excludes are applied, so an exclude that names a dependency of a
    kept agent will silently drop that dependency — likely producing a
    broken image, but matching the user's literal request.

    Returns the names sorted alphabetically — the canonical order used
    for the OCI label, the tag suffix, and the in-container manifest.

    Raises ``ValueError`` if a requested include or exclude name is not
    in the roster, or ``TypeError`` if *selection* is a string other
    than ``"all"`` (a bare name like ``"claude"`` would otherwise be
    iterated into characters).  Excludes that name a known agent but
    don't appear in the resolved include set are a no-op.
    """
    if isinstance(selection, str):
        if selection != "all":
            raise TypeError(
                f"Selection must be the literal string 'all' or a tuple of "
                f"tokens, got {selection!r}"
            )
        return tuple(sorted(self._installs))

    includes = {t for t in selection if not t.startswith("-")}
    excludes = {t[1:] for t in selection if t.startswith("-")}

    referenced = (includes | excludes) - {"all"}
    unknown = referenced - set(self._installs)
    if unknown:
        avail = ", ".join(sorted(self._installs))
        raise ValueError(f"Unknown roster entries: {sorted(unknown)!r}. Available: {avail}")

    seed = set(self._installs) if "all" in includes or not includes else includes

    resolved: set[str] = set()
    stack = list(seed)
    while stack:
        name = stack.pop()
        if name in resolved:
            continue
        resolved.add(name)
        spec = self._installs.get(name)
        if spec is None:
            continue
        for dep in spec.depends_on:
            if dep not in self._installs:
                raise ValueError(
                    f"Agent {name!r} declares depends_on {dep!r}, "
                    f"which has no install: section in the roster"
                )
            if dep not in resolved:
                stack.append(dep)
    return tuple(sorted(resolved - excludes))

get_provider(name, *, default_agent=None)

Resolve a provider name to an AgentProvider.

Falls back to default_agent, then "claude". Raises SystemExit if the resolved name is unknown.

Source code in src/terok_executor/roster/loader.py

def get_provider(self, name: str | None, *, default_agent: str | None = None) -> AgentProvider:
    """Resolve a provider name to an ``AgentProvider``.

    Falls back to *default_agent*, then ``"claude"``.
    Raises ``SystemExit`` if the resolved name is unknown.
    """
    from terok_executor.provider.providers import resolve_provider

    return resolve_provider(self._providers, name, default_agent=default_agent)

get_auth_provider(name)

Look up an auth provider by name.

Raises SystemExit if the name is unknown.

Source code in src/terok_executor/roster/loader.py

def get_auth_provider(self, name: str) -> AuthProvider:
    """Look up an auth provider by name.

    Raises ``SystemExit`` if the name is unknown.
    """
    info = self._auth_providers.get(name)
    if info is None:
        available = ", ".join(sorted(self._auth_providers))
        raise SystemExit(f"Unknown auth provider: {name!r}. Available: {available}")
    return info

get_sidecar_spec(name)

Look up a sidecar spec by tool name.

Raises SystemExit if the name has no sidecar configuration.

Source code in src/terok_executor/roster/loader.py

def get_sidecar_spec(self, name: str) -> SidecarSpec:
    """Look up a sidecar spec by tool name.

    Raises ``SystemExit`` if the name has no sidecar configuration.
    """
    spec = self._sidecar_specs.get(name)
    if spec is None:
        available = ", ".join(sorted(self._sidecar_specs)) or "(none)"
        raise SystemExit(f"No sidecar config for {name!r}. Available: {available}")
    return spec

generate_routes_json()

Generate the routes.json content for the sandbox vault server.

Returns a JSON object mapping provider name → VaultRouteEntry with empty/absent optional fields stripped.

Source code in src/terok_executor/roster/loader.py

def generate_routes_json(self) -> str:
    """Generate the ``routes.json`` content for the sandbox vault server.

    Returns a JSON object mapping provider name → [`VaultRouteEntry`][terok_executor.roster.schema.VaultRouteEntry]
    with empty/absent optional fields stripped.
    """
    from pydantic import TypeAdapter

    routes: dict[str, VaultRouteEntry] = {}
    prefix_owners: dict[str, str] = {}
    for route in self._vault_routes.values():
        existing = prefix_owners.get(route.route_prefix)
        if existing is not None:
            raise ValueError(
                f"Duplicate route prefix {route.route_prefix!r}: "
                f"providers {existing!r} and {route.provider!r}"
            )
        prefix_owners[route.route_prefix] = route.provider
        routes[route.provider] = VaultRouteEntry(
            upstream=route.upstream,
            auth_header=route.auth_header,
            auth_prefix=route.auth_prefix,
            path_upstreams=route.path_upstreams or None,
            oauth_extra_headers=route.oauth_extra_headers or None,
            oauth_refresh=route.oauth_refresh or None,
        )
    return (
        TypeAdapter(dict[str, VaultRouteEntry])
        .dump_json(routes, indent=2, exclude_none=True)
        .decode()
    )

collect_all_auto_approve_env()

Merge auto_approve.env from all providers into one dict.

Source code in src/terok_executor/roster/loader.py

def collect_all_auto_approve_env(self) -> dict[str, str]:
    """Merge ``auto_approve.env`` from all providers into one dict."""
    merged: dict[str, str] = {}
    for p in self._providers.values():
        for key, value in p.auto_approve_env.items():
            if key in merged and merged[key] != value:
                raise ValueError(
                    f"Conflicting auto_approve_env for {key!r}: "
                    f"{merged[key]!r} vs {value!r} (provider {p.name!r})"
                )
            merged[key] = value
    return merged

collect_opencode_provider_env()

Collect env vars for all OpenCode-based providers.

Source code in src/terok_executor/roster/loader.py

def collect_opencode_provider_env(self) -> dict[str, str]:
    """Collect env vars for all OpenCode-based providers."""
    env: dict[str, str] = {}
    for p in self._providers.values():
        if p.opencode_config is not None:
            env.update(p.opencode_config.to_env(p.name))
    return env

shared() staticmethod

Return the process-wide cached roster.

Loaded on first access; every subsequent call returns the same instance. Use this from anywhere that just needs the global view; tests that mutate or replace the roster should call load_roster and keep the result local.

Source code in src/terok_executor/roster/loader.py

@staticmethod
def shared() -> AgentRoster:
    """Return the process-wide cached roster.

    Loaded on first access; every subsequent call returns the same
    instance.  Use this from anywhere that just needs the global
    view; tests that mutate or replace the roster should call
    [`load_roster`][terok_executor.roster.loader.load_roster] and
    keep the result local.
    """
    return _shared_roster()

parse_selection(raw) staticmethod

Normalise a user-supplied agent selection string.

Accepts a comma-list of selection tokens or the literal "all". Each token is either an agent name ("claude") or a name prefixed with - to exclude it from the selection ("-vibe"). The pseudo-name "all" is also valid as a token, so "all,-vibe" means "everything except vibe". When the input contains only excludes ("-vibe"), the selection seeds from every installable entry — same effect as "all,-vibe".

Whitespace is stripped, empty / whitespace-only entries dropped, and case folded. Empty or all-whitespace input collapses to "all" — the same shape AgentRoster.resolve_selection expects. Unknown names are not checked here; resolve_selection does that.

Source code in src/terok_executor/roster/loader.py

@staticmethod
def parse_selection(raw: str) -> str | tuple[str, ...]:
    """Normalise a user-supplied agent selection string.

    Accepts a comma-list of selection tokens or the literal ``"all"``.
    Each token is either an agent name (``"claude"``) or a name
    prefixed with ``-`` to exclude it from the selection
    (``"-vibe"``).  The pseudo-name ``"all"`` is also valid as a
    token, so ``"all,-vibe"`` means "everything except vibe".  When
    the input contains only excludes (``"-vibe"``), the selection
    seeds from every installable entry — same effect as
    ``"all,-vibe"``.

    Whitespace is stripped, empty / whitespace-only entries dropped,
    and case folded.  Empty or all-whitespace input collapses to
    ``"all"`` — the same shape
    [`AgentRoster.resolve_selection`][terok_executor.roster.loader.AgentRoster.resolve_selection]
    expects.  Unknown names are not checked here;
    ``resolve_selection`` does that.
    """
    folded = raw.strip().lower()
    if folded == "all" or not folded:
        return "all"
    tokens = tuple(n.strip() for n in folded.split(",") if n.strip())
    return tokens or "all"

validate_selection(raw)

Reject raw with SystemExit(2) if it names roster entries we don't have.

CLI-flavoured: prints a Invalid agent selection: … line on stderr and exits. Domain callers that just want the parsed tuple should use parse_selection + resolve_selection and handle ValueError themselves.

Source code in src/terok_executor/roster/loader.py

def validate_selection(self, raw: str) -> None:
    """Reject *raw* with ``SystemExit(2)`` if it names roster entries we don't have.

    CLI-flavoured: prints a ``Invalid agent selection: …`` line on
    stderr and exits.  Domain callers that just want the parsed
    tuple should use
    [`parse_selection`][terok_executor.roster.loader.AgentRoster.parse_selection]
    + [`resolve_selection`][terok_executor.roster.loader.AgentRoster.resolve_selection]
    and handle ``ValueError`` themselves.
    """
    try:
        self.resolve_selection(self.parse_selection(raw))
    except ValueError as exc:
        print(f"Invalid agent selection: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc

prompt_selection()

Print the installed roster and read one line of executor grammar.

Empty input → "all". Non-interactive stdin (closed pipe) exits with a hint to pass the selection positionally instead.

Source code in src/terok_executor/roster/loader.py

def prompt_selection(self) -> str:
    """Print the installed roster and read one line of executor grammar.

    Empty input → ``"all"``.  Non-interactive stdin (closed pipe)
    exits with a hint to pass the selection positionally instead.
    """
    providers = self.providers
    print("\nAvailable agents:")
    for name in sorted(self.agent_names):
        provider = providers.get(name)
        label = provider.label if provider is not None else name
        print(f"  · {name}  — {label}")
    try:
        raw = input("\nType a comma list, or '-name' to exclude [all]: ").strip()
    except EOFError as exc:
        raise SystemExit(
            "No interactive stdin available.  Pass the selection positionally "
            "instead, e.g. `terok agents set all`."
        ) from exc
    return raw or "all"

ensure_vault_routes(cfg=None)

Generate routes.json from this roster and write it to disk.

The routes file is written to the path configured in SandboxConfig (typically ~/.local/share/terok/vault/routes.json).

When cfg is None, falls back to standalone defaults.

Returns the path to the written file.

Source code in src/terok_executor/roster/loader.py

def ensure_vault_routes(self, cfg: SandboxConfig | None = None) -> Path:
    """Generate ``routes.json`` from this roster and write it to disk.

    The routes file is written to the path configured in
    [`SandboxConfig`][terok_sandbox.SandboxConfig] (typically
    ``~/.local/share/terok/vault/routes.json``).

    When *cfg* is ``None``, falls back to standalone defaults.

    Returns the path to the written file.
    """
    if cfg is None:
        cfg = SandboxConfig()
    path = cfg.routes_path

    path.parent.mkdir(parents=True, exist_ok=True)
    content = self.generate_routes_json() + "\n"
    fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    tmp = Path(tmp_name)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            f.write(content)
            f.flush()
            os.fsync(f.fileno())
        tmp.replace(path)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise
    return path

doctor_checks(*, token_broker_port=None)

Return agent-level health checks for in-container diagnostics.

Delegates to terok_executor.doctor for the actual check factories; this method is the canonical entry point so consumers can discover the checks through the roster.

Parameters:

Name	Type	Description	Default
token_broker_port	int | None	Host-side vault broker TCP port. None selects socket mode; any integer selects TCP mode. Base URL checks use the port (or the in-container loopback port) to derive the expected host.	None

Source code in src/terok_executor/roster/loader.py

def doctor_checks(self, *, token_broker_port: int | None = None) -> list[DoctorCheck]:
    """Return agent-level health checks for in-container diagnostics.

    Delegates to
    [`terok_executor.doctor`][terok_executor.doctor] for the actual
    check factories; this method is the canonical entry point so
    consumers can discover the checks through the roster.

    Args:
        token_broker_port: Host-side vault broker TCP port.  ``None``
            selects socket mode; any integer selects TCP mode.  Base
            URL checks use the port (or the in-container loopback
            port) to derive the expected host.
    """
    from terok_executor.doctor import _build_agent_doctor_checks

    return _build_agent_doctor_checks(self, token_broker_port=token_broker_port)

MountDef(host_dir, container_path, label, credential_file='', provider='') dataclass

A shared directory mount derived from the agent roster.

host_dir instance-attribute

Directory name under mounts_dir() (e.g. "_codex-config").

container_path instance-attribute

Mount point inside the container (e.g. "/home/dev/.codex").

label instance-attribute

Human-readable label (e.g. "Codex config").

credential_file = '' class-attribute instance-attribute

Credential file path relative to the mount root (e.g. ".credentials.json").

Empty when the mount carries no auth artefact (e.g. opencode state dirs). Populated from the matching vault.credential_file so callers can layer a read-only shadow over the file without touching the rest of the shared mount. See terok-ai/terok#873.

provider = '' class-attribute instance-attribute

Roster entry name that contributed this mount (e.g. "claude").

Empty for explicit mounts: blocks that aren't tied to a single provider. Used by the credential-shadow path to match against ContainerEnvSpec.expose_credential_providers.

SidecarSpec(tool_name, env_map=dict()) dataclass

Sidecar container configuration parsed from a sidecar: YAML section.

Tools with sidecar specs run in a separate lightweight L1 image (no agent CLIs) and receive the real API key instead of phantom tokens.

tool_name instance-attribute

Tool identifier used to select the Jinja2 install block in the template.

env_map = field(default_factory=dict) class-attribute instance-attribute

Maps container env var names to credential dict keys.

Example: {"CODERABBIT_API_KEY": "key"} reads cred["key"] and injects it as CODERABBIT_API_KEY.

VaultRoute(provider, route_prefix, upstream, path_upstreams=dict(), oauth_extra_headers=dict(), auth_header='Authorization', auth_prefix='Bearer ', credential_type='api_key', credential_file='', token_env=dict(), base_url_env='', socket_env='', shared_config_patch=None, oauth_refresh=None, shared_domain=False) dataclass

Vault route config parsed from a vault: YAML section.

Used to generate the routes.json that the vault server reads.

provider instance-attribute

Agent/tool name (e.g. "claude").

route_prefix instance-attribute

Path prefix in the proxy (e.g. "claude" → /claude/v1/...).

upstream instance-attribute

Upstream API base URL (e.g. "https://api.anthropic.com").

path_upstreams = field(default_factory=dict) class-attribute instance-attribute

Optional request-path prefix → upstream-base overrides.

oauth_extra_headers = field(default_factory=dict) class-attribute instance-attribute

Provider-specific headers added only when forwarding OAuth credentials.

auth_header = 'Authorization' class-attribute instance-attribute

HTTP header name for the real credential.

auth_prefix = 'Bearer ' class-attribute instance-attribute

Prefix before the token value in the auth header.

credential_type = 'api_key' class-attribute instance-attribute

Type of credential: "oauth", "api_key", "oauth_token", "pat".

credential_file = '' class-attribute instance-attribute

Credential file path relative to the auth mount.

token_env = field(default_factory=dict) class-attribute instance-attribute

Phantom-token env var name, keyed by stored credential type.

The named env var carries the phantom token the agent reads in place of the real credential. Keys are credential types ("oauth", "pat", …); "_default" is the fallback for any type without an explicit entry. Most agents read one env var regardless of type ({"_default": "MISTRAL_API_KEY"}); Claude swaps the name when an OAuth token is stored ({"oauth": "CLAUDE_CODE_OAUTH_TOKEN", "_default": "ANTHROPIC_API_KEY"}).

base_url_env = '' class-attribute instance-attribute

Env var to override with the vault's HTTP URL (e.g. "ANTHROPIC_BASE_URL").

socket_env = '' class-attribute instance-attribute

Env var that receives the container-side vault socket path.

Set when the agent speaks HTTP-over-UNIX natively (e.g. Claude reads ANTHROPIC_UNIX_SOCKET). The resolved value is mode-dependent and injected centrally by the env builder.

shared_config_patch = None class-attribute instance-attribute

Optional shared config patch applied after auth (e.g. Vibe's config.toml).

oauth_refresh = None class-attribute instance-attribute

OAuth refresh config: {token_url, client_id, scope}.

shared_domain = False class-attribute instance-attribute

Whether the upstream host also serves non-API traffic.

Set on entries whose upstream host is an apex (or otherwise mixed) domain that legitimately serves docs, dashboards, git push, etc. Host-level egress denies can't separate paths, so terok's auth-protect layer skips these providers when re-applying denies after shield down — credential containment alone keeps the API safe.

Examples: gitlab.com (API + git push), sonarcloud.io (API + project pages + docs + badges).

load_roster()

Load the agent roster from bundled YAML + user overrides.

Bundled agents in resources/agents/*.yaml are loaded first, then user files in ~/.config/terok/agent/agents/*.yaml are deep-merged on top (allowing field-level overrides or entirely new agents). Each merged entry is then validated through RawAgentYaml — typos in section keys, wrong types, or unknown fields fail loud instead of silently defaulting.

Source code in src/terok_executor/roster/loader.py

def load_roster() -> AgentRoster:
    """Load the agent roster from bundled YAML + user overrides.

    Bundled agents in ``resources/agents/*.yaml`` are loaded first, then
    user files in ``~/.config/terok/agent/agents/*.yaml`` are deep-merged
    on top (allowing field-level overrides or entirely new agents).  Each
    merged entry is then validated through [`RawAgentYaml`][terok_executor.roster.schema.RawAgentYaml]
    — typos in section keys, wrong types, or unknown fields fail loud
    instead of silently defaulting.
    """
    raw = _load_bundled_agents()

    # Deep-merge user overrides on top of bundled definitions
    for name, user_data in _load_user_agents().items():
        if name in raw:
            raw[name] = deep_merge(raw[name], user_data)
        else:
            raw[name] = user_data

    providers: dict[str, AgentProvider] = {}
    auth_providers: dict[str, AuthProvider] = {}
    vault_routes: dict[str, VaultRoute] = {}
    sidecar_specs: dict[str, SidecarSpec] = {}
    installs: dict[str, InstallSpec] = {}
    helps: dict[str, HelpSpec] = {}
    agent_names: list[str] = []
    all_names: list[str] = []
    web_ingress_names: set[str] = set()

    # Collect mounts from all entries — deduplicate by host_dir
    seen_mounts: dict[str, MountDef] = {}

    for name, data in sorted(raw.items()):
        try:
            spec = RawAgentYaml.model_validate(data)
        except ValidationError as exc:
            raise ValueError(f"Agent {name!r}: invalid roster YAML\n{exc}") from exc

        label = spec.resolve_label(name)
        is_agent_kind = spec.kind not in ("tool", "runtime")

        if spec.kind != "runtime":
            all_names.append(name)
        if is_agent_kind:
            agent_names.append(name)
            providers[name] = spec.to_agent_provider(name)

        credential_file = spec.vault.credential_file if spec.vault else ""

        auth_prov: AuthProvider | None
        if spec.auth is not None:
            auth_prov = spec.auth.to_dataclass(name=name, label=label)
        elif is_agent_kind:
            auth_prov = spec.derive_opencode_auth(name)
        else:
            auth_prov = None

        if auth_prov is not None:
            auth_providers[name] = auth_prov
            if auth_prov.host_dir_name not in seen_mounts:
                seen_mounts[auth_prov.host_dir_name] = MountDef(
                    host_dir=auth_prov.host_dir_name,
                    container_path=auth_prov.container_mount,
                    label=f"{auth_prov.label} config",
                    credential_file=credential_file,
                    provider=name,
                )

        for m in spec.mounts:
            if m.host_dir not in seen_mounts:
                seen_mounts[m.host_dir] = MountDef(
                    host_dir=m.host_dir,
                    container_path=m.container_path,
                    label=m.label or name,
                )

        if spec.vault is not None:
            vault_routes[name] = spec.vault.to_dataclass(provider=name)

        if spec.sidecar is not None:
            sidecar_specs[name] = spec.sidecar.to_dataclass(default_name=name)

        if spec.install is not None:
            installs[name] = spec.install.to_dataclass()

        if spec.help is not None:
            helps[name] = spec.help.to_dataclass()

        if spec.web_ingress:
            web_ingress_names.add(name)

    return AgentRoster(
        _providers=providers,
        _auth_providers=auth_providers,
        _vault_routes=vault_routes,
        _sidecar_specs=sidecar_specs,
        _installs=installs,
        _helps=helps,
        _mounts=tuple(seen_mounts.values()),
        _agent_names=tuple(agent_names),
        _all_names=tuple(all_names),
        _web_ingress=frozenset(web_ingress_names),
    )