Loader

loader

Loads agent and tool definitions from YAML and assembles them into a queryable roster.

Loads per-agent definition files from bundled package resources and optional user extensions, validates them through the strict schema (typo-rejecting Pydantic models), and projects each entry onto the runtime types dataclasses.

Directory layout::

resources/agents/claude.yaml      (bundled, shipped in wheel)
resources/agents/codex.yaml
...
~/.config/terok/agent/agents/      (user overrides / additions)

ROSTER_VERSION = 1 module-attribute

Schema version of the agent-roster YAML format.

Bundled agent YAMLs and user override files declare a top-level roster_version: 1 that matches this constant. A file with no roster_version is treated as version 1 (forward-compat for existing user overrides written before the marker existed). A file declaring a future version is still loaded but the loader logs a warning — the host and container may be on incompatible contracts. Bumped only on breaking changes to the roster schema, never per release.

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)

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),
    )