Skip to content

terok_agent

terok_agent

terok-agent: single-agent task runner for hardened Podman containers.

Builds agent images, launches instrumented containers, and manages the lifecycle of one AI coding agent at a time. Designed for standalone use (terok-agent run claude .) and as a library for terok orchestration.

Public API::

# Provider registry
from terok_agent import HEADLESS_PROVIDERS, HeadlessProvider, get_provider
from terok_agent import PROVIDER_NAMES, CLIOverrides
from terok_agent import apply_provider_config, build_headless_command
from terok_agent import collect_opencode_provider_env, collect_all_auto_approve_env

# Agent config preparation
from terok_agent import AgentConfigSpec, prepare_agent_config_dir, parse_md_agent

# Auth
from terok_agent import AUTH_PROVIDERS, AuthProvider, authenticate

# Instructions
from terok_agent import resolve_instructions, bundled_default_instructions

# Credential proxy
from terok_agent import ensure_proxy_routes

# Config stack
from terok_agent import ConfigStack, ConfigScope, resolve_provider_value

Internal symbols (available via submodule import for white-box tests)::

from terok_agent.headless_providers import generate_agent_wrapper, generate_all_wrappers
from terok_agent.headless_providers import OpenCodeProviderConfig, ProviderConfig, WrapperConfig
from terok_agent.config_stack import deep_merge, load_yaml_scope, load_json_scope
from terok_agent.instructions import has_custom_instructions
from terok_agent._util import podman_userns_args

AUTH_PROVIDERS = {} module-attribute

All known auth providers (agents + tools), keyed by name. Loaded from resources/agents/*.yaml.

DEFAULT_BASE_IMAGE = 'ubuntu:24.04' module-attribute

Default base OS image when none is specified.

HEADLESS_PROVIDERS = {} module-attribute

All headless agent providers, keyed by name. Loaded from resources/agents/*.yaml.

AgentConfigSpec(tasks_root, task_id, subagents, selected_agents=None, prompt=None, provider='claude', instructions=None, default_agent=None, mounts_base=None) dataclass

Groups parameters for preparing an agent-config directory.

__post_init__()

Coerce mutable sequences to tuples for true immutability.

Source code in src/terok_agent/agents.py 
410
411
412
413
414
415
def __post_init__(self) -> None:
    """Coerce mutable sequences to tuples for true immutability."""
    if isinstance(self.subagents, list):
        object.__setattr__(self, "subagents", tuple(self.subagents))
    if isinstance(self.selected_agents, list):
        object.__setattr__(self, "selected_agents", tuple(self.selected_agents))

AuthProvider(name, label, host_dir_name, container_mount, command, banner_hint, extra_run_args=tuple(), modes=('api_key',), api_key_hint='') dataclass

Describes how to authenticate one tool/agent.

name instance-attribute

Short key used in CLI and TUI dispatch (e.g. "codex").

label instance-attribute

Human-readable display name (e.g. "Codex").

host_dir_name instance-attribute

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

container_mount instance-attribute

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

command instance-attribute

Command to execute inside the container (OAuth mode only).

banner_hint instance-attribute

Provider-specific help text shown before the container runs.

extra_run_args = field(default_factory=tuple) class-attribute instance-attribute

Additional podman run arguments (e.g. port forwarding).

modes = ('api_key',) class-attribute instance-attribute

Supported auth modes: "oauth" (container), "api_key" (fast path).

api_key_hint = '' class-attribute instance-attribute

Hint shown when prompting for an API key (URL to get one).

supports_oauth property

Whether this provider supports OAuth (container-based) auth.

supports_api_key property

Whether this provider supports direct API key entry.

BuildError

Bases: RuntimeError

Raised when base-image construction cannot complete.

The CLI maps this to a user-facing error message; library callers can catch it without being terminated by SystemExit.

ImageSet(l0, l1, l1_sidecar=None) dataclass

L0 + L1 image tags produced by a build.

l0 instance-attribute

L0 base dev image tag (e.g. terok-l0:ubuntu-24.04).

l1 instance-attribute

L1 agent CLI image tag (e.g. terok-l1-cli:ubuntu-24.04).

l1_sidecar = None class-attribute instance-attribute

L1 sidecar image tag, if built (e.g. terok-l1-sidecar:ubuntu-24.04).

CommandDef(name, help='', handler=None, args=(), group='') dataclass

Definition of a terok-agent subcommand.

ConfigScope(level, source, data) dataclass

A single layer in the config stack.

ConfigStack()

Ordered collection of config scopes, lowest-priority first.

Usage::

stack = ConfigStack()
stack.push(ConfigScope("global", global_path, global_data))
stack.push(ConfigScope("project", proj_path, proj_data))
resolved = stack.resolve()

Initialise an empty config stack.

Source code in src/terok_agent/config_stack.py 
120
121
122
def __init__(self) -> None:
    """Initialise an empty config stack."""
    self._scopes: list[ConfigScope] = []

scopes property

Read-only access to the scope list (for diagnostics).

push(scope)

Append a scope (higher priority than all previous).

Source code in src/terok_agent/config_stack.py 
124
125
126
def push(self, scope: ConfigScope) -> None:
    """Append a scope (higher priority than all previous)."""
    self._scopes.append(scope)

resolve()

Deep-merge all scopes in order and return the result.

Source code in src/terok_agent/config_stack.py 
128
129
130
131
132
133
def resolve(self) -> dict:
    """Deep-merge all scopes in order and return the result."""
    result: dict = {}
    for scope in self._scopes:
        result = deep_merge(result, scope.data)
    return result

resolve_section(key)

Resolve only a single top-level section across all scopes.

Source code in src/terok_agent/config_stack.py 
135
136
137
138
139
140
141
142
def resolve_section(self, key: str) -> dict:
    """Resolve only a single top-level section across all scopes."""
    result: dict = {}
    for scope in self._scopes:
        section = scope.data.get(key)
        if isinstance(section, dict):
            result = deep_merge(result, section)
    return result

CLIOverrides(model=None, max_turns=None, timeout=None, instructions=None) dataclass

CLI flag overrides for a headless agent run.

model = None class-attribute instance-attribute

Explicit --model from CLI (takes precedence over config).

max_turns = None class-attribute instance-attribute

Explicit --max-turns from CLI.

timeout = None class-attribute instance-attribute

Explicit --timeout from CLI.

instructions = None class-attribute instance-attribute

Resolved instructions text. Delivery is provider-aware.

HeadlessProvider(name, label, binary, git_author_name, git_author_email, headless_subcommand, prompt_flag, auto_approve_env, auto_approve_flags, output_format_flags, model_flag, max_turns_flag, verbose_flag, supports_session_resume, resume_flag, continue_flag, session_file, supports_agents_json, supports_session_hook, supports_add_dir, log_format, opencode_config=None) dataclass

Describes how to run one AI agent in headless (autopilot) mode.

name instance-attribute

Short key used in CLI dispatch (e.g. "claude", "codex").

label instance-attribute

Human-readable display name (e.g. "Claude", "Codex").

binary instance-attribute

CLI binary name (e.g. "claude", "codex", "opencode").

git_author_name instance-attribute

AI identity name for Git author/committer policy application.

git_author_email instance-attribute

AI identity email for Git author/committer policy application.

headless_subcommand instance-attribute

Subcommand for headless mode (e.g. "exec" for codex, "run" for opencode).

None means the binary uses flags only (e.g. claude -p).

prompt_flag instance-attribute

Flag for passing the prompt.

"-p" for flag-based, "" for positional (after subcommand).

auto_approve_env instance-attribute

Environment variables for fully autonomous execution.

Injected into the container env by _apply_unrestricted_env() when TEROK_UNRESTRICTED=1. Read by agents regardless of launch path. Claude uses /etc/claude-code/managed-settings.json instead.

auto_approve_flags instance-attribute

CLI flags injected by the shell wrapper when TEROK_UNRESTRICTED=1.

Only for agents that lack an env var or managed config mechanism (currently Codex only). Empty for all other agents — their env vars and /etc/ config files handle permissions across all launch paths.

output_format_flags instance-attribute

Flags for structured output (e.g. ("--output-format", "stream-json")).

model_flag instance-attribute

Flag for model override ("--model", "--agent", or None).

max_turns_flag instance-attribute

Flag for maximum turns ("--max-turns" or None).

verbose_flag instance-attribute

Flag for verbose output ("--verbose" or None).

supports_session_resume instance-attribute

Whether the provider supports resuming a previous session.

resume_flag instance-attribute

Flag to resume a session (e.g. "--resume", "--session").

continue_flag instance-attribute

Flag to continue a session (e.g. "--continue").

session_file instance-attribute

Filename in /home/dev/.terok/ for stored session ID.

Providers that capture session IDs via plugin or post-run parsing set this to a filename (e.g. "opencode-session.txt"). Providers with their own hook mechanism (Claude) or no session support set this to None.

supports_agents_json instance-attribute

Whether the provider supports --agents JSON (Claude only).

supports_session_hook instance-attribute

Whether the provider supports SessionStart hooks (Claude only).

supports_add_dir instance-attribute

Whether the provider supports --add-dir "/" (Claude only).

log_format instance-attribute

Log format identifier: "claude-stream-json" or "plain".

opencode_config = None class-attribute instance-attribute

Configuration for OpenCode-based providers (Blablador, KISSKI, etc.).

When set, this provider uses OpenCode with a custom OpenAI-compatible API. The configuration includes API endpoints, model preferences, and provider-specific settings that are injected into the container environment.

uses_opencode_instructions property

Whether the provider uses OpenCode's instruction system.

CredentialProxyRoute(provider, route_prefix, upstream, auth_header='Authorization', auth_prefix='Bearer ', credential_type='api_key', credential_file='', phantom_env=dict(), oauth_phantom_env=dict(), base_url_env='', socket_path='', socket_env='', shared_config_patch=None, oauth_refresh=None) dataclass

Proxy route config parsed from a credential_proxy: YAML section.

Used to generate the routes.json that the credential proxy 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").

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.

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

Phantom env vars for API-key credentials (e.g. {"ANTHROPIC_API_KEY": true}).

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

Phantom env vars for OAuth credentials (e.g. {"CLAUDE_CODE_OAUTH_TOKEN": true}).

When the stored credential type is "oauth" and this is non-empty, these env vars are injected instead of :attr:phantom_env.

base_url_env = '' class-attribute instance-attribute

Env var to override with proxy URL (e.g. "ANTHROPIC_BASE_URL").

socket_path = '' class-attribute instance-attribute

Unix socket path for socat bridge (e.g. "/tmp/terok-claude-proxy.sock").

socket_env = '' class-attribute instance-attribute

Env var that receives :attr:socket_path (e.g. "ANTHROPIC_UNIX_SOCKET").

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}.

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.

AgentRunner(*, sandbox=None, roster=None, base_image='ubuntu:24.04')

Composes sandbox + agent config into a single container launch.

All three run methods follow the same flow:

  1. Ensure L0+L1 images exist (build if missing)
  2. Prepare agent-config directory (wrapper, instructions, prompt)
  3. Assemble environment variables and volume mounts
  4. Optionally set up gate (mirror repo, create token)
  5. Launch container via podman
Source code in src/terok_agent/runner.py 
76
77
78
79
80
81
82
83
84
85
def __init__(
    self,
    *,
    sandbox: Sandbox | None = None,
    roster: AgentRoster | None = None,
    base_image: str = "ubuntu:24.04",
) -> None:
    self._base_image = base_image
    self._sandbox: Sandbox | None = sandbox
    self._roster: AgentRoster | None = roster

sandbox property

Lazy-init sandbox facade.

roster property

Lazy-init agent roster.

run_headless(provider, repo, *, prompt, branch=None, model=None, max_turns=None, timeout=1800, gate=True, name=None, follow=False, unrestricted=True, gpu=False, hooks=None)

Launch a headless agent run. Returns container name.

The agent executes the prompt against repo (local path or git URL) and exits when done or when timeout is reached. Set follow=True to block until the agent finishes (the CLI does this by default).

Source code in src/terok_agent/runner.py 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
def run_headless(
    self,
    provider: str,
    repo: str,
    *,
    prompt: str,
    branch: str | None = None,
    model: str | None = None,
    max_turns: int | None = None,
    timeout: int = 1800,
    gate: bool = True,
    name: str | None = None,
    follow: bool = False,
    unrestricted: bool = True,
    gpu: bool = False,
    hooks: LifecycleHooks | None = None,
) -> str:
    """Launch a headless agent run. Returns container name.

    The agent executes the *prompt* against *repo* (local path or git URL)
    and exits when done or when *timeout* is reached.  Set *follow=True*
    to block until the agent finishes (the CLI does this by default).
    """
    return self._run(
        provider=provider,
        repo=repo,
        prompt=prompt,
        branch=branch,
        model=model,
        max_turns=max_turns,
        timeout=timeout,
        gate=gate,
        name=name,
        follow=follow,
        mode="headless",
        unrestricted=unrestricted,
        gpu=gpu,
        hooks=hooks,
    )

run_interactive(provider, repo, *, branch=None, gate=True, name=None, unrestricted=True, gpu=False, hooks=None)

Launch an interactive container. Returns container name.

The container stays up after init; user logs in via podman exec.

Source code in src/terok_agent/runner.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
def run_interactive(
    self,
    provider: str,
    repo: str,
    *,
    branch: str | None = None,
    gate: bool = True,
    name: str | None = None,
    unrestricted: bool = True,
    gpu: bool = False,
    hooks: LifecycleHooks | None = None,
) -> str:
    """Launch an interactive container. Returns container name.

    The container stays up after init; user logs in via ``podman exec``.
    """
    return self._run(
        provider=provider,
        repo=repo,
        branch=branch,
        gate=gate,
        name=name,
        mode="interactive",
        unrestricted=unrestricted,
        gpu=gpu,
        hooks=hooks,
    )

run_web(repo, *, port=None, branch=None, gate=True, name=None, public_url=None, unrestricted=True, gpu=False, hooks=None)

Launch a toad web container. Returns container name.

If port is None, an available port is auto-allocated.

Source code in src/terok_agent/runner.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
def run_web(
    self,
    repo: str,
    *,
    port: int | None = None,
    branch: str | None = None,
    gate: bool = True,
    name: str | None = None,
    public_url: str | None = None,
    unrestricted: bool = True,
    gpu: bool = False,
    hooks: LifecycleHooks | None = None,
) -> str:
    """Launch a toad web container. Returns container name.

    If *port* is None, an available port is auto-allocated.
    """
    if port is None:
        from terok_sandbox import find_free_port

        port = find_free_port()
    return self._run(
        provider="claude",  # toad uses claude as default
        repo=repo,
        branch=branch,
        gate=gate,
        name=name,
        mode="web",
        port=port,
        public_url=public_url,
        unrestricted=unrestricted,
        gpu=gpu,
        hooks=hooks,
    )

run_tool(tool, repo, *, tool_args=(), branch=None, gate=True, name=None, follow=True, timeout=600)

Launch a sidecar tool container. Returns container name.

Runs the named tool in a lightweight sidecar L1 image (no agent CLIs). The tool receives the real API key from the credential store — not a phantom token.

Source code in src/terok_agent/runner.py 
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
def run_tool(
    self,
    tool: str,
    repo: str,
    *,
    tool_args: tuple[str, ...] = (),
    branch: str | None = None,
    gate: bool = True,
    name: str | None = None,
    follow: bool = True,
    timeout: int = 600,
) -> str:
    """Launch a sidecar tool container. Returns container name.

    Runs the named tool in a lightweight sidecar L1 image (no agent
    CLIs).  The tool receives the real API key from the credential
    store — not a phantom token.
    """
    return self._run(
        provider=tool,
        repo=repo,
        mode="tool",
        gate=gate,
        name=name,
        follow=follow,
        timeout=timeout,
        tool_args=tool_args,
        branch=branch,
    )

resolve_provider_value(key, config, provider_name)

Extract a provider-aware config value.

Supports two forms:

  • Flat value — model: opus → same for all providers.
  • Per-provider dict — model: {claude: opus, codex: o3, _default: fast} → looks up provider_name, falls back to _default, then None.

Returns None when the key is absent or has no match for the provider.

Null override behaviour: when a per-provider dict maps a provider to null (Python None), that None is treated as "no value" and the resolver falls back to _default. This is intentional — it allows a lower-priority config layer to set a provider-specific value that a higher-priority layer can effectively unset by mapping it to null, letting the _default (or None) bubble up instead.

Source code in src/terok_agent/agent_config.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def resolve_provider_value(
    key: str,
    config: dict[str, Any],
    provider_name: str,
) -> Any | None:
    """Extract a provider-aware config value.

    Supports two forms:

    * **Flat value** — ``model: opus`` → same for all providers.
    * **Per-provider dict** — ``model: {claude: opus, codex: o3, _default: fast}``
      → looks up *provider_name*, falls back to ``_default``, then ``None``.

    Returns ``None`` when the key is absent or has no match for the provider.

    **Null override behaviour**: when a per-provider dict maps a provider to
    ``null`` (Python ``None``), that ``None`` is treated as "no value" and the
    resolver falls back to ``_default``.  This is intentional — it allows a
    lower-priority config layer to set a provider-specific value that a
    higher-priority layer can effectively *unset* by mapping it to ``null``,
    letting the ``_default`` (or ``None``) bubble up instead.
    """
    val = config.get(key)
    if val is None:
        return None
    if isinstance(val, dict):
        provider_val = val.get(provider_name)
        if provider_val is not None:
            return provider_val
        return val.get("_default")
    return val

parse_md_agent(file_path)

Parse a .md file with YAML frontmatter into an agent dict.

Expected format

name: agent-name description: ... tools: [Read, Grep] model: sonnet

System prompt body...

Source code in src/terok_agent/agents.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def parse_md_agent(file_path: str) -> dict:
    """Parse a .md file with YAML frontmatter into an agent dict.

    Expected format:
        ---
        name: agent-name
        description: ...
        tools: [Read, Grep]
        model: sonnet
        ---
        System prompt body...
    """
    path = Path(file_path)
    if not path.is_file():
        return {}
    content = path.read_text(encoding="utf-8")
    # Split YAML frontmatter from body
    if content.startswith("---"):
        parts = content.split("---", 2)
        if len(parts) >= 3:
            frontmatter = _yaml_load(parts[1]) or {}
            if not isinstance(frontmatter, dict):
                frontmatter = {}
            body = parts[2].strip()
            frontmatter["prompt"] = body
            return frontmatter
    # No frontmatter: treat entire file as prompt
    return {"prompt": content.strip()}

prepare_agent_config_dir(spec)

Create and populate the agent-config directory for a task.

Writes: - terok-agent.sh (always) — wrapper functions with git env vars - agents.json (only when provider supports it and sub-agents are non-empty) - prompt.txt (if prompt given, headless only) - instructions.md (always) — custom instructions or a neutral default - /_claude-config/settings.json — SessionStart hook (Claude only) - opencode.json entries — instructions path injected into shared OpenCode and Blablador configs

Parameters:

Name Type Description Default
spec AgentConfigSpec

All agent-config parameters bundled in an :class:AgentConfigSpec.

 required

Returns the agent_config_dir path.

Source code in src/terok_agent/agents.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
def prepare_agent_config_dir(spec: AgentConfigSpec) -> Path:
    """Create and populate the agent-config directory for a task.

    Writes:
    - terok-agent.sh (always) — wrapper functions with git env vars
    - agents.json (only when provider supports it and sub-agents are non-empty)
    - prompt.txt (if prompt given, headless only)
    - instructions.md (always) — custom instructions or a neutral default
    - <envs>/_claude-config/settings.json — SessionStart hook (Claude only)
    - opencode.json entries — ``instructions`` path injected into shared
      OpenCode and Blablador configs

    Args:
        spec: All agent-config parameters bundled in an :class:`AgentConfigSpec`.

    Returns the agent_config_dir path.
    """
    from .headless_providers import get_provider as _get_provider

    resolved = _get_provider(spec.provider, default_agent=spec.default_agent)

    task_dir = spec.tasks_root / str(spec.task_id)
    agent_config_dir = task_dir / "agent-config"
    ensure_dir(agent_config_dir)

    # Build agents JSON — only for providers that support --agents (Claude)
    has_agents = False
    if resolved.supports_agents_json and spec.subagents:
        agents_json = _subagents_to_json(spec.subagents, spec.selected_agents)
        agents_dict = json.loads(agents_json)
        if agents_dict:  # non-empty dict
            (agent_config_dir / "agents.json").write_text(agents_json, encoding="utf-8")
            has_agents = True
    elif spec.subagents or spec.selected_agents:
        import warnings

        warnings.warn(
            f"{resolved.label} does not support sub-agents (--agents); "
            f"sub-agent definitions will be ignored.",
            stacklevel=2,
        )

    # Write instructions file — always present so opencode.json `instructions`
    # references never point to a missing file.  When no custom instructions
    # are configured, a neutral default is used.
    _DEFAULT_INSTRUCTIONS = "Follow the project's coding conventions and existing patterns."

    has_instructions = bool(spec.instructions)
    instructions_text = spec.instructions or _DEFAULT_INSTRUCTIONS
    (agent_config_dir / "instructions.md").write_text(instructions_text, encoding="utf-8")

    # Inject instructions path into opencode.json configs on the host so
    # all OpenCode-based providers discover them natively (works for both
    # interactive and headless modes).
    from .headless_providers import HEADLESS_PROVIDERS

    mounts_base = spec.mounts_base
    if mounts_base is None:
        raise ValueError("mounts_base is required in AgentConfigSpec")
    _inject_opencode_instructions(mounts_base / "_opencode-config" / "opencode.json")
    for _p in HEADLESS_PROVIDERS.values():
        if _p.opencode_config is not None:
            _inject_opencode_instructions(
                mounts_base / f"_{_p.name}-config" / "opencode" / "opencode.json"
            )

    # Write shell wrapper functions for ALL providers so interactive CLI users
    # can invoke any agent (each provider gets its own shell function).
    from .headless_providers import generate_all_wrappers

    def _claude_wrapper_with_instructions(cfg: WrapperConfig) -> str:
        """Wrap _generate_claude_wrapper with the resolved has_instructions flag."""
        return _generate_claude_wrapper(
            WrapperConfig(
                has_agents=cfg.has_agents,
                has_instructions=has_instructions,
            )
        )

    wrapper = generate_all_wrappers(
        has_agents,
        claude_wrapper_fn=_claude_wrapper_with_instructions,
    )
    (agent_config_dir / "terok-agent.sh").write_text(wrapper, encoding="utf-8")

    # Write SessionStart hook — only for providers that support it (Claude)
    if resolved.supports_session_hook:
        shared_claude_dir = mounts_base / "_claude-config"
        ensure_dir_writable(shared_claude_dir, "_claude-config")
        _write_session_hook(shared_claude_dir / "settings.json")

    # Prompt (headless only)
    if spec.prompt is not None:
        (agent_config_dir / "prompt.txt").write_text(spec.prompt, encoding="utf-8")

    return agent_config_dir

authenticate(project_id, provider, *, mounts_dir, image)

Run the auth flow for provider against project_id.

Dispatches based on the provider's modes field:

  • api_key only: prompt for key, store directly (no container)
  • oauth only: launch container with vendor CLI
  • both: ask user to choose, then dispatch accordingly

Parameters:

Name Type Description Default
project_id str

Project identifier (for container naming).

 required
provider str

Auth provider name (e.g. "claude").

 required
mounts_dir Path

Base directory for shared config bind-mounts.

 required
image str

Container image to use for the auth container.

 required

Raises SystemExit if the provider name is unknown.

Source code in src/terok_agent/auth.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
def authenticate(
    project_id: str,
    provider: str,
    *,
    mounts_dir: Path,
    image: str,
) -> None:
    """Run the auth flow for *provider* against *project_id*.

    Dispatches based on the provider's ``modes`` field:

    - **api_key only**: prompt for key, store directly (no container)
    - **oauth only**: launch container with vendor CLI
    - **both**: ask user to choose, then dispatch accordingly

    Args:
        project_id: Project identifier (for container naming).
        provider: Auth provider name (e.g. ``"claude"``).
        mounts_dir: Base directory for shared config bind-mounts.
        image: Container image to use for the auth container.

    Raises ``SystemExit`` if the provider name is unknown.
    """
    info = AUTH_PROVIDERS.get(provider)
    if not info:
        available = ", ".join(AUTH_PROVIDERS)
        raise SystemExit(f"Unknown auth provider: {provider}. Available: {available}")

    if info.supports_oauth and info.supports_api_key:
        # Both modes — let the user choose
        print(f"Authenticate {info.label}:\n")
        print("  1. OAuth / interactive login (launches container)")
        print("  2. API key (paste key, no container needed)")
        print()
        choice = input("Choose [1/2]: ").strip()
        if choice == "2":
            key = _prompt_api_key(info)
            store_api_key(provider, key)
            return
        # choice == "1" or anything else → OAuth
        _run_auth_container(project_id, info, mounts_dir=mounts_dir, image=image)

    elif info.supports_api_key:
        # API key only — fast path, no container
        key = _prompt_api_key(info)
        store_api_key(provider, key)

    else:
        # OAuth only
        _run_auth_container(project_id, info, mounts_dir=mounts_dir, image=image)

store_api_key(provider, api_key, credential_set='default')

Store an API key directly in the credential DB (no container needed).

This is the non-interactive fast path for automated workflows and CI. The key is stored as {"type": "api_key", "key": "<value>"}.

Source code in src/terok_agent/auth.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
def store_api_key(
    provider: str,
    api_key: str,
    credential_set: str = "default",
) -> None:
    """Store an API key directly in the credential DB (no container needed).

    This is the non-interactive fast path for automated workflows and CI.
    The key is stored as ``{"type": "api_key", "key": "<value>"}``.
    """
    from terok_sandbox import CredentialDB, SandboxConfig

    cfg = SandboxConfig()
    db = CredentialDB(cfg.proxy_db_path)
    try:
        db.store_credential(credential_set, provider, {"type": "api_key", "key": api_key})
        print(f"API key stored for {provider} (set: {credential_set})")
    finally:
        db.close()

build_base_images(base_image=DEFAULT_BASE_IMAGE, *, rebuild=False, full_rebuild=False, build_dir=None)

Build L0 + L1 container images and return their tags.

Skips building if images already exist locally (unless rebuild or full_rebuild is set). Uses a temporary directory for the build context by default; pass build_dir to use a specific (empty or non-existent) directory instead.

Parameters:

Name Type Description Default
base_image str

Base OS image (e.g. ubuntu:24.04, nvidia/cuda:...).

 DEFAULT_BASE_IMAGE
rebuild bool

Force rebuild with cache bust (refreshes agent installs).

 False
full_rebuild bool

Force rebuild with --no-cache --pull=always.

 False
build_dir Path | None

Build context directory (must be empty or absent).

 None

Returns:

Type Description
ImageSet

class:ImageSet with the L0 and L1 image tags.

Raises:

Type Description
BuildError

If podman is missing or a build step fails.
ValueError

If build_dir exists and is non-empty.
Source code in src/terok_agent/build.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
def build_base_images(
    base_image: str = DEFAULT_BASE_IMAGE,
    *,
    rebuild: bool = False,
    full_rebuild: bool = False,
    build_dir: Path | None = None,
) -> ImageSet:
    """Build L0 + L1 container images and return their tags.

    Skips building if images already exist locally (unless *rebuild* or
    *full_rebuild* is set).  Uses a temporary directory for the build
    context by default; pass *build_dir* to use a specific (empty or
    non-existent) directory instead.

    Args:
        base_image: Base OS image (e.g. ``ubuntu:24.04``, ``nvidia/cuda:...``).
        rebuild: Force rebuild with cache bust (refreshes agent installs).
        full_rebuild: Force rebuild with ``--no-cache --pull=always``.
        build_dir: Build context directory (must be empty or absent).

    Returns:
        :class:`ImageSet` with the L0 and L1 image tags.

    Raises:
        BuildError: If podman is missing or a build step fails.
        ValueError: If *build_dir* exists and is non-empty.
    """
    # Validate arguments before any side effects (podman probe, temp dirs)
    if build_dir is not None:
        if build_dir.is_file():
            raise ValueError(f"build_dir is a file, not a directory: {build_dir}")
        if build_dir.exists() and any(build_dir.iterdir()):
            raise ValueError(f"build_dir must be empty or absent: {build_dir}")

    _check_podman()

    base_image = _normalize_base_image(base_image)
    l0_tag = l0_image_tag(base_image)
    l1_tag = l1_image_tag(base_image)

    # Skip if both images exist and no forced rebuild
    if not rebuild and not full_rebuild:
        if _image_exists(l0_tag) and _image_exists(l1_tag):
            return ImageSet(l0=l0_tag, l1=l1_tag)

    # Prepare build context in a safe directory
    import tempfile

    own_tmp = build_dir is None
    context = build_dir or Path(tempfile.mkdtemp(prefix="terok-agent-build-"))

    try:
        prepare_build_context(context)

        # Single timestamp for both render and build-arg consistency
        cache_bust = str(int(time.time()))

        # Render and write Dockerfiles into the build context
        (context / "L0.Dockerfile").write_text(render_l0(base_image))
        (context / "L1.cli.Dockerfile").write_text(render_l1(l0_tag, cache_bust=cache_bust))

        ctx = str(context)

        # Build L0 — base dev image (Ubuntu + git + SSH + init script)
        cmd_l0 = ["podman", "build", "-f", str(context / "L0.Dockerfile")]
        cmd_l0 += ["--build-arg", f"BASE_IMAGE={base_image}"]
        cmd_l0 += ["-t", l0_tag]
        if full_rebuild:
            cmd_l0 += ["--no-cache", "--pull=always"]
        cmd_l0.append(ctx)

        print("$", shlex.join(cmd_l0))
        subprocess.run(cmd_l0, check=True)

        # Build L1 — agent CLI layer (all agent installs, shell env, ACP wrappers)
        cmd_l1 = ["podman", "build", "-f", str(context / "L1.cli.Dockerfile")]
        cmd_l1 += ["--build-arg", f"BASE_IMAGE={l0_tag}"]
        cmd_l1 += ["--build-arg", f"AGENT_CACHE_BUST={cache_bust}"]
        cmd_l1 += ["-t", l1_tag]
        if full_rebuild:
            cmd_l1.append("--no-cache")
        cmd_l1.append(ctx)

        print("$", shlex.join(cmd_l1))
        subprocess.run(cmd_l1, check=True)

    except (OSError, subprocess.CalledProcessError) as e:
        raise BuildError(f"Image build failed: {e}") from e
    finally:
        if own_tmp:
            shutil.rmtree(context, ignore_errors=True)

    return ImageSet(l0=l0_tag, l1=l1_tag)

build_sidecar_image(base_image=DEFAULT_BASE_IMAGE, *, tool_name='coderabbit', rebuild=False, full_rebuild=False, build_dir=None)

Build the L1 sidecar image for a specific tool. Returns the image tag.

Ensures L0 exists first (builds it if missing), then builds the sidecar image FROM L0. The sidecar contains only the named tool — no agent CLIs, no LLMs.

Parameters:

Name Type Description Default
base_image str

Base OS image (passed through to L0 build).

 DEFAULT_BASE_IMAGE
tool_name str

Tool to install (selects Jinja2 conditional in template).

 'coderabbit'
rebuild bool

Force rebuild with cache bust.

 False
full_rebuild bool

Force rebuild with --no-cache.

 False
build_dir Path | None

Build context directory (must be empty or absent).

 None

Returns:

Type Description
str

The sidecar image tag (e.g. terok-l1-sidecar:ubuntu-24.04).

Raises:

Type Description
BuildError

If podman is missing or a build step fails.
Source code in src/terok_agent/build.py 
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
def build_sidecar_image(
    base_image: str = DEFAULT_BASE_IMAGE,
    *,
    tool_name: str = "coderabbit",
    rebuild: bool = False,
    full_rebuild: bool = False,
    build_dir: Path | None = None,
) -> str:
    """Build the L1 sidecar image for a specific tool. Returns the image tag.

    Ensures L0 exists first (builds it if missing), then builds the
    sidecar image FROM L0.  The sidecar contains only the named tool —
    no agent CLIs, no LLMs.

    Args:
        base_image: Base OS image (passed through to L0 build).
        tool_name: Tool to install (selects Jinja2 conditional in template).
        rebuild: Force rebuild with cache bust.
        full_rebuild: Force rebuild with ``--no-cache``.
        build_dir: Build context directory (must be empty or absent).

    Returns:
        The sidecar image tag (e.g. ``terok-l1-sidecar:ubuntu-24.04``).

    Raises:
        BuildError: If podman is missing or a build step fails.
    """
    _check_podman()

    base_image = _normalize_base_image(base_image)
    l0_tag = l0_image_tag(base_image)
    sidecar_tag = l1_sidecar_image_tag(base_image)

    if not rebuild and not full_rebuild and _image_exists(sidecar_tag) and _image_exists(l0_tag):
        return sidecar_tag

    # Ensure L0 exists (build if needed)
    if not _image_exists(l0_tag) or full_rebuild:
        build_base_images(base_image, rebuild=rebuild, full_rebuild=full_rebuild)

    import tempfile

    own_tmp = build_dir is None
    context = build_dir or Path(tempfile.mkdtemp(prefix="terok-agent-sidecar-"))

    try:
        prepare_build_context(context)
        cache_bust = str(int(time.time()))

        (context / "L1.sidecar.Dockerfile").write_text(
            render_l1_sidecar(l0_tag, tool_name=tool_name, cache_bust=cache_bust)
        )

        cmd = ["podman", "build", "-f", str(context / "L1.sidecar.Dockerfile")]
        cmd += ["--build-arg", f"BASE_IMAGE={l0_tag}"]
        cmd += ["--build-arg", f"TOOL_CACHE_BUST={cache_bust}"]
        cmd += ["-t", sidecar_tag]
        if full_rebuild:
            cmd.append("--no-cache")
        cmd.append(str(context))

        print("$", shlex.join(cmd))
        subprocess.run(cmd, check=True)
    except (OSError, subprocess.CalledProcessError) as e:
        raise BuildError(f"Sidecar image build failed: {e}") from e
    finally:
        if own_tmp:
            shutil.rmtree(context, ignore_errors=True)

    return sidecar_tag

l0_image_tag(base_image)

Return the L0 base dev image tag for base_image.

Source code in src/terok_agent/build.py 
95
96
97
def l0_image_tag(base_image: str) -> str:
    """Return the L0 base dev image tag for *base_image*."""
    return f"terok-l0:{_base_tag(base_image)}"

l1_image_tag(base_image)

Return the L1 agent CLI image tag for base_image.

Source code in src/terok_agent/build.py 
100
101
102
def l1_image_tag(base_image: str) -> str:
    """Return the L1 agent CLI image tag for *base_image*."""
    return f"terok-l1-cli:{_base_tag(base_image)}"

l1_sidecar_image_tag(base_image)

Return the L1 sidecar (tool-only) image tag for base_image.

Source code in src/terok_agent/build.py 
105
106
107
def l1_sidecar_image_tag(base_image: str) -> str:
    """Return the L1 sidecar (tool-only) image tag for *base_image*."""
    return f"terok-l1-sidecar:{_base_tag(base_image)}"

render_l1_sidecar(l0_image, *, tool_name='coderabbit', cache_bust='0')

Render the L1 sidecar (tool-only) Dockerfile.

The sidecar image is built FROM L0 (not L1) and installs a single tool binary — no agent CLIs, no LLMs. The tool_name selects which tool install block to activate via Jinja2 conditional.

Source code in src/terok_agent/build.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
def render_l1_sidecar(
    l0_image: str, *, tool_name: str = "coderabbit", cache_bust: str = "0"
) -> str:
    """Render the L1 sidecar (tool-only) Dockerfile.

    The sidecar image is built FROM L0 (not L1) and installs a single
    tool binary — no agent CLIs, no LLMs.  The *tool_name* selects which
    tool install block to activate via Jinja2 conditional.
    """
    return _render_template(
        "l1.sidecar.Dockerfile.template",
        {"BASE_IMAGE": l0_image, "TOOL_CACHE_BUST": cache_bust, "tool_name": tool_name},
    )

stage_scripts(dest)

Stage container helper scripts into dest.

Copies all files from terok_agent/resources/scripts/ into the given directory, replacing any existing contents. Python bytecode caches and __init__.py markers are excluded.

Source code in src/terok_agent/build.py 
159
160
161
162
163
164
165
166
167
168
169
def stage_scripts(dest: Path) -> None:
    """Stage container helper scripts into *dest*.

    Copies all files from ``terok_agent/resources/scripts/`` into the given
    directory, replacing any existing contents.  Python bytecode caches and
    ``__init__.py`` markers are excluded.
    """
    if dest.exists():
        shutil.rmtree(dest)
    _copy_package_tree("terok_agent", "resources/scripts", dest)
    _clean_packaging_artifacts(dest)

stage_tmux_config(dest)

Stage the container tmux configuration into dest.

Copies container-tmux.conf — the green-status-bar config that distinguishes container tmux sessions from host tmux.

Source code in src/terok_agent/build.py 
184
185
186
187
188
189
190
191
192
193
def stage_tmux_config(dest: Path) -> None:
    """Stage the container tmux configuration into *dest*.

    Copies ``container-tmux.conf`` — the green-status-bar config that
    distinguishes container tmux sessions from host tmux.
    """
    if dest.exists():
        shutil.rmtree(dest)
    _copy_package_tree("terok_agent", "resources/tmux", dest)
    _clean_packaging_artifacts(dest)

stage_toad_agents(dest)

Stage Toad ACP agent TOML definitions into dest.

These describe OpenCode-based agents (Blablador, KISSKI, etc.) that are injected into Toad's bundled agent directory at container build time.

Source code in src/terok_agent/build.py 
172
173
174
175
176
177
178
179
180
181
def stage_toad_agents(dest: Path) -> None:
    """Stage Toad ACP agent TOML definitions into *dest*.

    These describe OpenCode-based agents (Blablador, KISSKI, etc.) that are
    injected into Toad's bundled agent directory at container build time.
    """
    if dest.exists():
        shutil.rmtree(dest)
    _copy_package_tree("terok_agent", "resources/toad-agents", dest)
    _clean_packaging_artifacts(dest)

extract_credential(provider, base_dir)

Run the appropriate extractor for provider against base_dir.

Raises ValueError if no extractor is registered or extraction fails.

Source code in src/terok_agent/credential_extractors.py 
229
230
231
232
233
234
235
236
237
238
def extract_credential(provider: str, base_dir: Path) -> dict:
    """Run the appropriate extractor for *provider* against *base_dir*.

    Raises ``ValueError`` if no extractor is registered or extraction fails.
    """
    entry = EXTRACTORS.get(provider)
    if entry is None:
        raise ValueError(f"No credential extractor for provider {provider!r}")
    fn, *args = entry
    return fn(base_dir, *args)

agent_doctor_checks(roster, *, proxy_port=None)

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

Parameters:

Name Type Description Default
roster AgentRoster

The loaded agent roster.

 required
proxy_port int | None

Credential proxy TCP port. Required for base URL checks; if None, base URL checks are skipped.

 None

Returns:

Type Description
list[DoctorCheck]

List of :class:DoctorCheck instances ready for orchestration.
Source code in src/terok_agent/doctor.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def agent_doctor_checks(
    roster: AgentRoster,
    *,
    proxy_port: int | None = None,
) -> list[DoctorCheck]:
    """Return agent-level health checks for in-container diagnostics.

    Args:
        roster: The loaded agent roster.
        proxy_port: Credential proxy TCP port. Required for base URL checks;
            if ``None``, base URL checks are skipped.

    Returns:
        List of :class:`DoctorCheck` instances ready for orchestration.
    """
    checks: list[DoctorCheck] = [
        _make_ssh_bridge_check(),
        _make_gh_proxy_bridge_check(),
    ]
    checks.extend(_make_credential_file_checks(roster))
    checks.extend(_make_phantom_token_checks(roster))
    if proxy_port is not None:
        checks.extend(_make_base_url_checks(roster, proxy_port))
    return checks

apply_provider_config(provider, config, overrides=None)

Resolve config values for a provider with best-effort feature mapping.

CLI flag overrides take precedence over config values. When the provider lacks a feature, an analogue is used where possible (e.g. injecting max-turns guidance into the prompt), and a warning is emitted for features that have no analogue.

Parameters:

Name Type Description Default
config dict

Merged agent config dict (from :func:resolve_agent_config).

 required
overrides CLIOverrides | None

CLI flag overrides (model, max_turns, timeout, instructions).

 None
Source code in src/terok_agent/headless_providers.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def apply_provider_config(
    provider: HeadlessProvider,
    config: dict,
    overrides: CLIOverrides | None = None,
) -> ProviderConfig:
    """Resolve config values for a provider with best-effort feature mapping.

    CLI flag overrides take precedence over config values.  When the provider
    lacks a feature, an analogue is used where possible (e.g. injecting
    max-turns guidance into the prompt), and a warning is emitted for
    features that have no analogue.

    Args:
        config: Merged agent config dict (from :func:`resolve_agent_config`).
        overrides: CLI flag overrides (model, max_turns, timeout, instructions).
    """
    if overrides is None:
        overrides = CLIOverrides()
    from .agent_config import resolve_provider_value

    warnings: list[str] = []
    prompt_parts: list[str] = []

    # --- Model ---
    cfg_model = resolve_provider_value("model", config, provider.name)
    model = overrides.model or (str(cfg_model) if cfg_model is not None else None)
    if model and not provider.model_flag:
        warnings.append(
            f"{provider.label} does not support model selection; ignoring model={model!r}"
        )
        model = None

    # --- Max turns ---
    cfg_turns = resolve_provider_value("max_turns", config, provider.name)
    max_turns_raw = overrides.max_turns if overrides.max_turns is not None else cfg_turns
    max_turns: int | None = int(max_turns_raw) if max_turns_raw is not None else None
    if max_turns is not None and not provider.max_turns_flag:
        # Best-effort: inject into prompt as guidance
        prompt_parts.append(f"Important: complete this task in no more than {max_turns} steps.")
        warnings.append(
            f"{provider.label} does not support --max-turns; "
            f"added guidance to prompt instead ({max_turns} steps)"
        )
        max_turns = None

    # --- Timeout ---
    cfg_timeout = resolve_provider_value("timeout", config, provider.name)
    timeout = (
        overrides.timeout
        if overrides.timeout is not None
        else (int(cfg_timeout) if cfg_timeout is not None else 1800)
    )

    # --- Subagents (warning only — filtering is handled elsewhere) ---
    subagents = config.get("subagents")
    if subagents and not provider.supports_agents_json:
        warnings.append(
            f"{provider.label} does not support sub-agents (--agents); "
            f"sub-agent definitions will be ignored"
        )

    # --- Instructions ---
    # Claude receives instructions via --append-system-prompt in the wrapper.
    # Codex receives instructions via -c model_instructions_file=... in the wrapper.
    # OpenCode-based providers receive instructions via opencode.json `instructions`
    # array (injected by prepare_agent_config_dir).
    # Remaining providers get best-effort prompt prepending.
    instructions = overrides.instructions
    if (
        instructions
        and provider.name not in {"claude", "codex"}
        and not provider.uses_opencode_instructions
    ):
        prompt_parts.insert(0, instructions)

    return ProviderConfig(
        model=model,
        max_turns=max_turns,
        timeout=timeout,
        prompt_extra="\n".join(prompt_parts),
        warnings=tuple(warnings),
    )

build_headless_command(provider, *, timeout, model=None, max_turns=None)

Assemble the bash command string for a headless agent run.

The command assumes: - init-ssh-and-repo.sh has already set up the workspace - The prompt is in /home/dev/.terok/prompt.txt - For Claude, the claude() wrapper function is sourced via bash -l

Returns a bash command string suitable for ["bash", "-lc", cmd].

Source code in src/terok_agent/headless_providers.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def build_headless_command(
    provider: HeadlessProvider,
    *,
    timeout: int,
    model: str | None = None,
    max_turns: int | None = None,
) -> str:
    """Assemble the bash command string for a headless agent run.

    The command assumes:
    - ``init-ssh-and-repo.sh`` has already set up the workspace
    - The prompt is in ``/home/dev/.terok/prompt.txt``
    - For Claude, the ``claude()`` wrapper function is sourced via bash -l

    Returns a bash command string suitable for ``["bash", "-lc", cmd]``.
    """
    if provider.name == "claude":
        return _build_claude_command(provider, timeout=timeout, model=model, max_turns=max_turns)
    return _build_generic_command(provider, timeout=timeout, model=model, max_turns=max_turns)

collect_all_auto_approve_env()

Collect auto_approve_env from all providers into one dict.

Used by task runners to inject these env vars at the container level (not just inside shell wrappers) so that ACP-spawned agents also inherit unrestricted permissions.

Source code in src/terok_agent/headless_providers.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def collect_all_auto_approve_env() -> dict[str, str]:
    """Collect ``auto_approve_env`` from all providers into one dict.

    Used by task runners to inject these env vars at the container level
    (not just inside shell wrappers) so that ACP-spawned agents also
    inherit unrestricted permissions.
    """
    merged: dict[str, str] = {}
    for p in HEADLESS_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 environment variables for all OpenCode-based providers.

Returns a dictionary of environment variables that will be injected into containers to configure OpenCode-based providers. Each provider with opencode_config set contributes variables prefixed with TEROK_OC_{PROVIDER_NAME}_*.

Source code in src/terok_agent/headless_providers.py 
734
735
736
737
738
739
740
741
742
743
744
745
def collect_opencode_provider_env() -> dict[str, str]:
    """Collect environment variables for all OpenCode-based providers.

    Returns a dictionary of environment variables that will be injected into containers
    to configure OpenCode-based providers. Each provider with opencode_config set
    contributes variables prefixed with TEROK_OC_{PROVIDER_NAME}_*.
    """
    env: dict[str, str] = {}
    for provider in HEADLESS_PROVIDERS.values():
        if provider.opencode_config is not None:
            env.update(provider.opencode_config.to_env(provider.name))
    return env

get_provider(name, *, default_agent=None)

Resolve a provider name to a HeadlessProvider.

Resolution order
  1. Explicit name if given
  2. default_agent (from project config)
  3. "claude" (ultimate fallback)

Raises SystemExit if the resolved name is not in the registry.

Source code in src/terok_agent/headless_providers.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def get_provider(name: str | None, *, default_agent: str | None = None) -> HeadlessProvider:
    """Resolve a provider name to a ``HeadlessProvider``.

    Resolution order:
      1. Explicit *name* if given
      2. *default_agent* (from project config)
      3. ``"claude"`` (ultimate fallback)

    Raises ``SystemExit`` if the resolved name is not in the registry.
    """
    resolved = name or default_agent or "claude"
    provider = HEADLESS_PROVIDERS.get(resolved)
    if provider is None:
        valid = ", ".join(sorted(HEADLESS_PROVIDERS))
        raise SystemExit(f"Unknown headless provider {resolved!r}. Valid providers: {valid}")
    return provider

bundled_default_instructions()

Read and return the bundled default instructions from package resources.

Source code in src/terok_agent/instructions.py 
30
31
32
33
def bundled_default_instructions() -> str:
    """Read and return the bundled default instructions from package resources."""
    ref = importlib.resources.files("terok_agent.resources.instructions").joinpath("default.md")
    return ref.read_text(encoding="utf-8")

resolve_instructions(config, provider_name, project_root=None)

Resolve instructions from a merged config dict.

Supports: - Flat string: returned as-is - Per-provider dict: uses :func:resolve_provider_value, falls back to _default - List (with _inherit): splices bundled default at each _inherit sentinel - Absent/None: returns bundled default

After resolving the YAML value, appends the contents of project_root/instructions.md (if it exists and is non-empty).

Returns the final instructions text.

Source code in src/terok_agent/instructions.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def resolve_instructions(
    config: dict[str, Any],
    provider_name: str,
    project_root: Path | None = None,
) -> str:
    """Resolve instructions from a merged config dict.

    Supports:
    - Flat string: returned as-is
    - Per-provider dict: uses :func:`resolve_provider_value`, falls back to ``_default``
    - List (with ``_inherit``): splices bundled default at each ``_inherit`` sentinel
    - Absent/None: returns bundled default

    After resolving the YAML value, appends the contents of
    ``project_root/instructions.md`` (if it exists and is non-empty).

    Returns the final instructions text.
    """
    from .agent_config import resolve_provider_value

    val = config.get("instructions")
    default = bundled_default_instructions()

    if val is None:
        base = default
    elif isinstance(val, dict):
        resolved = resolve_provider_value("instructions", config, provider_name)
        if resolved is None:
            base = default
        elif isinstance(resolved, list):
            base = _splice_inherit(resolved, default)
        elif resolved == _INHERIT_SENTINEL:
            base = default
        else:
            base = str(resolved)
    elif isinstance(val, list):
        base = _splice_inherit(val, default)
    elif val == _INHERIT_SENTINEL:
        # Bare _inherit string → same as absent (use bundled default)
        base = default
    else:
        base = str(val)

    # Append standalone instructions file (purely additive)
    file_text = _read_instructions_file(project_root)
    if file_text:
        return f"{base}\n\n{file_text}" if base else file_text
    return base

mounts_dir()

Base directory for agent config bind-mounts.

Each agent/tool gets a subdirectory (e.g. _claude-config/) that is bind-mounted read-write into task containers. These directories are intentionally separated from the credentials store since they are container-exposed and subject to potential poisoning.

Source code in src/terok_agent/paths.py 
54
55
56
57
58
59
60
61
62
def mounts_dir() -> Path:
    """Base directory for agent config bind-mounts.

    Each agent/tool gets a subdirectory (e.g. ``_claude-config/``) that is
    bind-mounted read-write into task containers.  These directories are
    intentionally separated from the credentials store since they are
    container-exposed and subject to potential poisoning.
    """
    return state_root() / "mounts"

scan_leaked_credentials(mounts_base)

Return (provider, host_path) for credential files found in shared mounts.

When the credential proxy is active, real secrets should only live in the proxy's sqlite DB — not in the shared config directories that get mounted into containers. This function checks each routed provider's mount for credential files that would leak real tokens alongside phantom ones.

Files injected by :func:~terok_agent.auth._write_claude_credentials_file are recognised by their dummy accessToken marker and skipped.

Source code in src/terok_agent/proxy_commands.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def scan_leaked_credentials(mounts_base: Path) -> list[tuple[str, Path]]:
    """Return ``(provider, host_path)`` for credential files found in shared mounts.

    When the credential proxy is active, real secrets should only live in the
    proxy's sqlite DB — not in the shared config directories that get mounted
    into containers.  This function checks each routed provider's mount for
    credential files that would leak real tokens alongside phantom ones.

    Files injected by :func:`~terok_agent.auth._write_claude_credentials_file`
    are recognised by their dummy ``accessToken`` marker and skipped.
    """
    from .roster import get_roster

    roster = get_roster()
    leaked: list[tuple[str, Path]] = []
    for name, route in roster.proxy_routes.items():
        if not route.credential_file:
            continue
        auth = roster.auth_providers.get(name)
        if not auth:
            continue
        try:
            path = mounts_base / auth.host_dir_name / route.credential_file
            if (
                path.is_file()
                and path.stat().st_size > 0
                and not _is_injected_credentials_file(path)
            ):
                leaked.append((name, path))
        except (OSError, TypeError):
            continue
    return leaked

ensure_proxy_routes(cfg=None)

Generate routes.json from the YAML roster and write it to disk.

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

When cfg is None, falls back to standalone defaults.

Returns the path to the written file.

Source code in src/terok_agent/roster.py 
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
def ensure_proxy_routes(cfg: SandboxConfig | None = None) -> Path:
    """Generate ``routes.json`` from the YAML roster and write it to disk.

    The routes file is written to the path configured in
    :class:`~terok_sandbox.SandboxConfig` (typically
    ``~/.local/share/terok/proxy/routes.json``).

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

    Returns the path to the written file.
    """
    from terok_sandbox import SandboxConfig

    if cfg is None:
        cfg = SandboxConfig()
    path = cfg.proxy_routes_path
    import os
    import tempfile

    path.parent.mkdir(parents=True, exist_ok=True)
    content = get_roster().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

get_roster() cached

Return the singleton roster instance (loaded once, cached).

Source code in src/terok_agent/roster.py 
615
616
617
618
@lru_cache(maxsize=1)
def get_roster() -> AgentRoster:
    """Return the singleton roster instance (loaded once, cached)."""
    return load_roster()