Agents

agents

Prepares agent config directories with wrappers, instructions, and sub-agent definitions.

Parses .md frontmatter for sub-agent definitions, converts them to Claude's --agents JSON format, and generates the terok-executor.sh wrapper that sets up git identity and CLI flags inside task containers.

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.

tasks_root instance-attribute

task_id instance-attribute

subagents instance-attribute

selected_agents = None class-attribute instance-attribute

prompt = None class-attribute instance-attribute

provider = 'claude' class-attribute instance-attribute

instructions = None class-attribute instance-attribute

default_agent = None class-attribute instance-attribute

mounts_base = None class-attribute instance-attribute

__post_init__()

Coerce mutable sequences to tuples for true immutability.

Defensive against callers that build the spec from json.loads / yaml.load output where the runtime types are list instead of tuple. Mypy sees the static annotations and reports the isinstance(..., list) branches as unreachable; the runtime coercion remains correct.

Source code in src/terok_executor/provider/agents.py

def __post_init__(self) -> None:
    """Coerce mutable sequences to tuples for true immutability.

    Defensive against callers that build the spec from
    ``json.loads`` / ``yaml.load`` output where the runtime types are
    ``list`` instead of ``tuple``.  Mypy sees the static annotations
    and reports the ``isinstance(..., list)`` branches as unreachable;
    the runtime coercion remains correct.
    """
    if isinstance(self.subagents, list):  # type: ignore[unreachable]
        object.__setattr__(self, "subagents", tuple(self.subagents))  # type: ignore[unreachable]
    if isinstance(self.selected_agents, list):
        object.__setattr__(self, "selected_agents", tuple(self.selected_agents))  # type: ignore[unreachable]

prepare_agent_config_dir(spec)

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

Writes: - terok-executor.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 AgentConfigSpec.	required

Returns the agent_config_dir path.

Source code in src/terok_executor/provider/agents.py

def prepare_agent_config_dir(spec: AgentConfigSpec) -> Path:
    """Create and populate the agent-config directory for a task.

    Writes:
    - terok-executor.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 [`AgentConfigSpec`][terok_executor.provider.agents.AgentConfigSpec].

    Returns the agent_config_dir path.
    """
    from .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."

    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).
    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 AGENT_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 .wrappers import generate_all_wrappers

    wrapper = generate_all_wrappers(has_agents)
    (agent_config_dir / "terok-executor.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

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_executor/provider/agents.py

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()}