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', 'EgressProjection', 'providers_config_dir'] module-attribute

AgentRoster(_agents=dict(), _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 agents, auth providers, vault routes, sidecar specs, install snippets, and help blurbs by name.

agents property

All headless agents (kind: agent only).

providers property

All vault-routed providers (LLM endpoints + tool APIs), keyed by clean name.

The endpoint axis: where requests go and how the real credential is attached. Loaded from resources/providers/*.yaml. The routes.json the sandbox vault reads is generated from these.

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

Return the names of agents, tools, and selectable LLM providers.

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.

This method raises ValueError if an include or exclude specifies a provider-only name or an unknown name. It raises TypeError if selection is a string other than "all". This check prevents iteration through the characters of a bare name such as "claude". An exclusion has no effect if the agent is not in the resolved include set.

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.

    This method raises ``ValueError`` if an include or exclude specifies a
    provider-only name or an unknown name. It raises ``TypeError`` if
    *selection* is a string other than ``"all"``. This check prevents
    iteration through the characters of a bare name such as ``"claude"``.
    An exclusion has no effect if the agent is not in the resolved include
    set.
    """
    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"}
    runtime_endpoints = {
        name
        for name, provider in self._providers.items()
        if provider.serves and name not in self._installs
    }
    endpoint_only = referenced & runtime_endpoints
    if endpoint_only:
        raise ValueError(
            f"These names are providers, not installable agents: "
            f"{sorted(endpoint_only)!r}. Add 'opencode' or 'pi' to image.agents. "
            "Then select a provider with --provider <name>."
        )
    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_agent(name, *, default_agent=None)

Resolve an agent name to an Agent.

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_agent(self, name: str | None, *, default_agent: str | None = None) -> Agent:
    """Resolve an agent name to an ``Agent``.

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

    return resolve_agent(self._agents, 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.

Emits one entry per provider — keyed by its clean name — so the vault can route to any authenticated provider, not just the one some agent binds by default. This is what lets a harness (opencode, pi) reach a provider no agent owns, and what keeps a provider routable after its shim agent is collapsed away. Empty/absent optional fields are 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.

    Emits one entry per **provider** — keyed by its clean name — so the
    vault can route to any authenticated provider, not just the one some
    agent binds by default.  This is what lets a harness (opencode, pi)
    reach a provider no agent owns, and what keeps a provider routable
    after its shim agent is collapsed away.  Empty/absent optional fields
    are stripped.
    """
    from pydantic import TypeAdapter

    routes = {
        name: _provider_route_entry(provider) for name, provider in self._providers.items()
    }
    return (
        TypeAdapter(dict[str, VaultRouteEntry])
        .dump_json(routes, indent=2, exclude_none=True)
        .decode()
    )

deny_to_vault_hosts(*, exposed_credential_providers=frozenset())

Hosts to deny directly at the egress firewall (shield security_deny, t20).

Every provider the vault relays contributes its relayed_hosts (upstream + path overrides + OAuth-refresh endpoint) — the agent must reach those only through the loopback vault. Two classes are skipped:

• shared_domain providers (gitlab.com, sonarcloud.io): the API rides an apex that also serves git push / docs, so a host-level deny would kill legitimate traffic; credential containment carries the weight instead.
• the providers bound by exposed_credential_providers — roster-entry (agent/tool) names in terok's experimental expose_oauth_token mode, where the agent holds its real credential in-container (vault bypassed) and so must reach the endpoint directly. Each entry is mapped to the provider it binds (provider_binding.default).

A pure function of the roster and exposed_credential_providers, mirroring generate_routes_json: the vault routes every provider, so every relayed host is denied.

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

def deny_to_vault_hosts(
    self, *, exposed_credential_providers: frozenset[str] = frozenset()
) -> frozenset[str]:
    """Hosts to deny directly at the egress firewall (shield ``security_deny``, t20).

    Every provider the vault relays contributes its
    [`relayed_hosts`][terok_executor.roster.types.Provider.relayed_hosts]
    (upstream + path overrides + OAuth-refresh endpoint) — the agent must
    reach those *only* through the loopback vault.  Two classes are skipped:

    - **``shared_domain`` providers** (``gitlab.com``, ``sonarcloud.io``):
      the API rides an apex that also serves ``git push`` / docs, so a
      host-level deny would kill legitimate traffic; credential containment
      carries the weight instead.
    - the providers bound by **``exposed_credential_providers``** — roster-entry
      (agent/tool) names in terok's experimental ``expose_oauth_token`` mode,
      where the agent holds its *real* credential in-container (vault bypassed)
      and so must reach the endpoint directly.  Each entry is mapped to the
      provider it binds (``provider_binding.default``).

    A pure function of the roster and *exposed_credential_providers*, mirroring
    [`generate_routes_json`][terok_executor.roster.loader.AgentRoster.generate_routes_json]:
    the vault routes every provider, so every relayed host is denied.
    """
    exposed = self._exposed_provider_names(exposed_credential_providers)
    hosts: set[str] = set()
    for name, provider in self._providers.items():
        if provider.shared_domain or name in exposed:
            continue
        hosts |= provider.relayed_hosts()
    return frozenset(hosts)

compose_egress(*, exposed_credential_providers=frozenset())

Project the roster into the shield's egress tiers.

Bundles the deny_to_vault_hosts set (t20) with the union of every provider's egress_allow (t30) into one EgressProjection; both tuples are sorted and de-duplicated for a deterministic bundle.

exposed_credential_providers — roster-entry (agent/tool) names whose real credential is exposed in-container; the providers they bind are left reachable (see deny_to_vault_hosts).

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

def compose_egress(
    self, *, exposed_credential_providers: frozenset[str] = frozenset()
) -> EgressProjection:
    """Project the roster into the shield's egress tiers.

    Bundles the [`deny_to_vault_hosts`][terok_executor.roster.loader.AgentRoster.deny_to_vault_hosts]
    set (t20) with the union of every provider's
    [`egress_allow`][terok_executor.roster.types.Provider.egress_allow]
    (t30) into one [`EgressProjection`][terok_executor.roster.types.EgressProjection];
    both tuples are sorted and de-duplicated for a deterministic bundle.

    *exposed_credential_providers* — roster-entry (agent/tool) names whose real
    credential is exposed in-container; the providers they bind are left
    reachable (see
    [`deny_to_vault_hosts`][terok_executor.roster.loader.AgentRoster.deny_to_vault_hosts]).
    """
    deny = self.deny_to_vault_hosts(exposed_credential_providers=exposed_credential_providers)
    provider_allow = {host for p in self._providers.values() for host in p.egress_allow}
    return EgressProjection(
        deny_to_vault=tuple(sorted(deny)),
        provider_allow=tuple(sorted(provider_allow)),
    )

collect_all_auto_approve_env()

Merge auto_approve.env from all agents 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 agents into one dict."""
    merged: dict[str, str] = {}
    for p in self._agents.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} (agent {p.name!r})"
                )
            merged[key] = value
    return merged

collect_opencode_provider_env()

Collect the TEROK_OC_{NAME}_* env vars for all OpenCode-driven providers.

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

def collect_opencode_provider_env(self) -> dict[str, str]:
    """Collect the ``TEROK_OC_{NAME}_*`` env vars for all OpenCode-driven 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

load() staticmethod

Load a fresh roster from the current configuration.

This method reads the configuration on each call. It does not replace the process-wide snapshot from AgentRoster.shared.

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

@staticmethod
def load() -> AgentRoster:
    """Load a fresh roster from the current configuration.

    This method reads the configuration on each call. It does not replace
    the process-wide snapshot from
    [`AgentRoster.shared`][terok_executor.roster.loader.AgentRoster.shared].
    """
    return load_roster()

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. Call AgentRoster.load when the caller needs the current configuration.

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. Call
    [`AgentRoster.load`][terok_executor.roster.loader.AgentRoster.load]
    when the caller needs the current configuration.
    """
    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.
    """
    agents = self.agents
    print("\nAvailable agents:")
    for name in sorted(self.agent_names):
        agent = agents.get(name)
        label = agent.label if agent 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)

EgressProjection(deny_to_vault=(), provider_allow=()) dataclass

Roster-derived egress policy for a task's shield bundle.

Produced by AgentRoster.compose_egress and handed (via the sandbox) to the shield's tier writer: deny_to_vault seeds the security_deny tier (t20), provider_allow the provider_allow tier (t30). Both tuples are sorted and de-duplicated so the projection is deterministic across runs.

deny_to_vault = () class-attribute instance-attribute

Relayed provider endpoints the agent may reach only through the vault, denied directly at the egress firewall (shared-domain and exposed-credential providers excluded).

provider_allow = () class-attribute instance-attribute

Extra runtime-egress hosts the providers legitimately need (egress.allow in the roster), allowed at the egress firewall.

providers_config_dir()

Return the current directory for user provider files.

The function uses the shared XDG-aware Terok configuration root. The default directory is ~/.config/terok/providers/. Terok also reads the legacy ~/.config/terok/agent/providers/ directory. Do not use the legacy directory for new files.

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

def providers_config_dir() -> Path:
    """Return the current directory for user provider files.

    The function uses the shared XDG-aware Terok configuration root. The
    default directory is ``~/.config/terok/providers/``. Terok also reads the
    legacy ``~/.config/terok/agent/providers/`` directory. Do not use the
    legacy directory for new files.
    """
    return namespace_config_dir() / _USER_PROVIDERS_DIR_NAME

__getattr__(name)

Resolve a re-exported name to .loader on first access (PEP 562).

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

def __getattr__(name: str) -> object:
    """Resolve a re-exported name to [`.loader`][terok_executor.roster.loader] on first access (PEP 562)."""
    try:
        module_path = _LAZY[name]
    except KeyError:
        raise AttributeError(f"module {__name__!r} has no attribute {name!r}") from None
    value = getattr(importlib.import_module(module_path, __name__), name)
    globals()[name] = value
    return value

__dir__()

Expose the lazy names to dir() / autocompletion.

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

def __dir__() -> list[str]:
    """Expose the lazy names to ``dir()`` / autocompletion."""
    return sorted({*globals(), *_LAZY})