headless_providers

headless_providers

Headless (autopilot) provider registry for multi-agent support.

Defines a frozen dataclass per provider and a registry dict, following the same pattern as AuthProvider in security/auth.py. Dispatch functions resolve the active provider, build the headless CLI command, and generate the per-provider shell wrapper.

Instruction delivery ~~~~~~~~~~~~~~~~~~~~ Custom instructions are delivered via a provider-specific channel:

• Claude: --append-system-prompt flag (injected by the wrapper).
• Codex: model_instructions_file config (-c flag in the wrapper).
• OpenCode / Blablador / KISSKI: "instructions" array in opencode.json pointing to /home/dev/.terok/instructions.md (injected on the host by :func:~terok.lib.instrumentation.agents._inject_opencode_instructions).
• Other providers (Copilot, Vibe, …): best-effort prompt prepending via prompt_extra in :class:ProviderConfig.

The instructions file is always written (with a neutral default when no custom text is configured) so that config-file references never dangle.

HEADLESS_PROVIDERS = {} module-attribute

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

OpenCodeProviderConfig(display_name, base_url, preferred_model, fallback_model, env_var_prefix, config_dir, auth_key_url) dataclass

Immutable descriptor for an OpenCode-based provider wrapper.

display_name instance-attribute

Human-readable display name (e.g., 'Helmholtz Blablador').

base_url instance-attribute

Base URL for the OpenAI-compatible API (e.g., 'https://api.helmholtz-blablador.fz-juelich.de/v1').

preferred_model instance-attribute

Preferred model ID (e.g., 'alias-huge').

fallback_model instance-attribute

Fallback model ID if preferred is unavailable (e.g., 'alias-code').

env_var_prefix instance-attribute

Environment variable prefix for API key (e.g., 'BLABLADOR' → BLABLADOR_API_KEY).

config_dir instance-attribute

Configuration directory name (e.g., '.blablador').

auth_key_url instance-attribute

URL where users can obtain API keys for documentation.

to_env(name)

Return env vars for container injection, keyed by TEROK_OC_{NAME}_*.

Source code in src/terok_agent/headless_providers.py

def to_env(self, name: str) -> dict[str, str]:
    """Return env vars for container injection, keyed by TEROK_OC_{NAME}_*."""
    prefix = f"TEROK_OC_{name.upper()}_"
    return {
        f"{prefix}BASE_URL": self.base_url,
        f"{prefix}PREFERRED_MODEL": self.preferred_model,
        f"{prefix}FALLBACK_MODEL": self.fallback_model,
        f"{prefix}DISPLAY_NAME": self.display_name,
        f"{prefix}ENV_VAR_PREFIX": self.env_var_prefix,
        f"{prefix}CONFIG_DIR": self.config_dir,
    }

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.

ProviderConfig(model, max_turns, timeout, prompt_extra, warnings) dataclass

Resolved per-run config for a headless provider.

Produced by :func:apply_provider_config after best-effort feature mapping.

model instance-attribute

Model override for providers that support it, else None.

max_turns instance-attribute

Max turns for providers that support it, else None.

timeout instance-attribute

Effective timeout in seconds.

prompt_extra instance-attribute

Extra text to append to the prompt (best-effort feature analogues).

warnings instance-attribute

Warnings about unsupported features (for user display).

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.

WrapperConfig(has_agents, has_instructions=False) dataclass

Groups parameters for generating the Claude shell wrapper.

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

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

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

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

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

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

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)

generate_agent_wrapper(provider, has_agents, *, claude_wrapper_fn=None)

Generate the shell wrapper function content for a single provider.

For Claude, uses claude_wrapper_fn (which should be agents._generate_claude_wrapper) to produce the full wrapper with --add-dir /, --agents, and session resume support. The function is passed in by the caller to avoid a circular import between this module and agents.

For other providers, produces a simpler wrapper that sets git env vars and delegates to the binary. Instructions are delivered via opencode.json (OpenCode/Blablador), model_instructions_file (Codex), or --append-system-prompt (Claude) — not via the wrapper.

Parameters:

Name	Type	Description	Default
claude_wrapper_fn	Callable[[WrapperConfig], str] | None	(cfg: WrapperConfig) -> str. Required when provider.name == "claude".	None

See also :func:generate_all_wrappers which produces wrappers for every registered provider in one file.

Source code in src/terok_agent/headless_providers.py

def generate_agent_wrapper(
    provider: HeadlessProvider,
    has_agents: bool,
    *,
    claude_wrapper_fn: Callable[[WrapperConfig], str] | None = None,
) -> str:
    """Generate the shell wrapper function content for a single provider.

    For Claude, uses *claude_wrapper_fn* (which should be
    ``agents._generate_claude_wrapper``) to produce the full wrapper with
    ``--add-dir /``, ``--agents``, and session resume support.  The function is passed in by the caller to
    avoid a circular import between this module and ``agents``.

    For other providers, produces a simpler wrapper that sets git env vars
    and delegates to the binary.  Instructions are delivered via
    ``opencode.json`` (OpenCode/Blablador), ``model_instructions_file``
    (Codex), or ``--append-system-prompt`` (Claude) — not via the wrapper.

    Args:
        claude_wrapper_fn: ``(cfg: WrapperConfig) -> str``.
            Required when ``provider.name == "claude"``.

    See also :func:`generate_all_wrappers` which produces wrappers for every
    registered provider in one file.
    """
    if provider.name == "claude":
        if claude_wrapper_fn is None:
            raise ValueError("claude_wrapper_fn is required for Claude provider")
        return claude_wrapper_fn(WrapperConfig(has_agents=has_agents))

    return _generate_generic_wrapper(provider)

generate_all_wrappers(has_agents, *, claude_wrapper_fn=None)

Generate shell wrappers for all registered providers in one file.

The output file contains a shell function per provider (claude(), codex(), vibe(), etc.), each with correct git env vars, timeout support, and session resume logic. This allows interactive CLI users to invoke any agent regardless of which provider was configured as default.

A shared _terok_resume_or_fresh helper is emitted at the top of the file for stale-session fallback (see :data:_RESUME_FALLBACK_FN).

Parameters:

Name	Type	Description	Default
claude_wrapper_fn	Callable[[WrapperConfig], str] | None	Required — produces the Claude wrapper.	None

Source code in src/terok_agent/headless_providers.py

def generate_all_wrappers(
    has_agents: bool,
    *,
    claude_wrapper_fn: Callable[[WrapperConfig], str] | None = None,
) -> str:
    """Generate shell wrappers for **all** registered providers in one file.

    The output file contains a shell function per provider (``claude()``,
    ``codex()``, ``vibe()``, etc.), each with correct git env vars, timeout
    support, and session resume logic.  This allows interactive CLI users to
    invoke any agent regardless of which provider was configured as default.

    A shared ``_terok_resume_or_fresh`` helper is emitted at the top of the
    file for stale-session fallback (see :data:`_RESUME_FALLBACK_FN`).

    Args:
        claude_wrapper_fn: Required — produces the Claude wrapper.
    """
    sections: list[str] = [_RESUME_FALLBACK_FN]
    for provider in HEADLESS_PROVIDERS.values():
        section = generate_agent_wrapper(
            provider,
            has_agents,
            claude_wrapper_fn=claude_wrapper_fn,
        )
        sections.append(section)
    return "\n".join(sections)

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

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