reach.runtime

Agent execution runtime interfaces, CLI subprocess template drivers, Antigravity domain bridges, and trajectory telemetry.

Define the core AgentRuntime interface and shared runtime agent helpers.

FAKE_AGENT module-attribute

FAKE_AGENT = 'fake'

AgentOptions

Bases: BaseModel

Base configuration common to all agent drivers.

Source code in src/reach/runtime/__init__.py

class AgentOptions(BaseModel):
    """Base configuration common to all agent drivers."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    model: str = ""
    effort: str | None = None
    provider: str | None = None
    max_turns: int = Field(default=3, ge=1)
    early_exit: bool = True
    allowed_tools: tuple[str, ...] | None = None
    blocked_env_vars: tuple[str, ...] | None = None
    api_key: str | None = None
    use_symlinks: bool = True
    isolate_config_dir: bool = True
    auto_clean: bool = False
    json_schema: str | None = None

    #: Name of the field configuring an explicit isolation directory, if supported.
    isolation_dir_field: ClassVar[str | None] = None

    @property
    def custom_isolation_dir(self) -> Path | None:
        """Return custom isolation directory path configured on this options instance, if any."""
        if self.isolation_dir_field:
            val = getattr(self, self.isolation_dir_field, None)
            return Path(val) if val is not None else None
        return None

    @model_validator(mode="after")
    def _validate_isolation_dir_boundary(self) -> Self:
        """Validate that custom isolation directory does not point to active home or root."""
        if self.isolation_dir_field and self.custom_isolation_dir is not None:
            from reach.runtime._fs import validate_isolated_directory

            validate_isolated_directory(self.custom_isolation_dir, self.isolation_dir_field)
        return self

custom_isolation_dir property

custom_isolation_dir: Path | None

Return custom isolation directory path configured on this options instance, if any.

AgentRuntime

Bases: ABC

Define standard abstract base class for agent runtime implementations.

Source code in src/reach/runtime/__init__.py

class AgentRuntime[OptionsT: AgentOptions](ABC):
    """Define standard abstract base class for agent runtime implementations."""

    name: str
    settings: RuntimeSettings | None = None
    options: OptionsT = cast("Any", AgentOptions())
    _resident: tuple[str, ...] = ()
    is_dynamic: bool = False

    def __init__(
        self,
        settings: RuntimeSettings | None = None,
        options: OptionsT | None = None,
    ) -> None:
        """Initialize agent runtime settings and options."""
        agent_name = getattr(self, "name", "agent")
        self.settings = settings
        if options is not None:
            self.options = options
        elif settings is not None:
            resolved = resolve_options(settings)
            opt_cls = options_model(agent_name)
            if opt_cls is not None and issubclass(opt_cls, AgentOptions):
                self.options = cast(
                    "OptionsT",
                    resolved
                    if isinstance(resolved, opt_cls)
                    else opt_cls.model_validate(dict(settings.options or {})),
                )
            else:
                self.options = cast(
                    "OptionsT",
                    resolved
                    if isinstance(resolved, AgentOptions)
                    else AgentOptions.model_validate(dict(settings.options or {})),
                )
        else:
            opt_cls = options_model(agent_name)
            if opt_cls is not None and issubclass(opt_cls, AgentOptions):
                self.options = cast("OptionsT", opt_cls())
            else:
                self.options = cast("OptionsT", AgentOptions())
        self._resident = ()

    @property
    def timeout_s(self) -> int | None:
        """Return per-probe execution timeout in seconds from settings."""
        return self.settings.timeout_s if self.settings is not None else None

    @property
    def skills_subpath(self) -> str:
        """Return relative skill directory subpath from profile or runtime default."""
        profile = agent_profiles().get(self.name)

        if profile is not None and profile.skills_dir:
            return profile.skills_dir
        if subpath := getattr(self, "_skills_subpath", None):
            return str(subpath)
        return ""

    @property
    def model(self) -> str:
        """Return the identifier of the model being evaluated."""
        return self.options.model

    @model.setter
    def model(self, value: str) -> None:
        """Update the configured model identifier in options."""
        self.options = self.options.model_copy(update={"model": value})

    @property
    def effort(self) -> str | None:
        """Return the reasoning effort tier being evaluated."""
        return self.options.effort

    @property
    def provider(self) -> str | None:
        """Return the model provider identifier being evaluated."""
        return self.options.provider or getattr(self.options, "model_provider", None)

    @property
    def max_turns(self) -> int:
        """Return the maximum turns configured on the runtime options."""
        return self.options.max_turns

    @property
    def early_exit(self) -> bool:
        """Return whether early exit is enabled on the runtime options."""
        return self.options.early_exit

    @property
    def api_key(self) -> str | None:
        """Return the API key string configured on the runtime options."""
        return self.options.api_key

    @property
    def allowed_tools(self) -> tuple[str, ...] | None:
        """Return explicit allowed tools override from options or settings."""
        return (
            self.options.allowed_tools
            if self.options.allowed_tools is not None
            else getattr(self.settings, "allowed_tools", None)
        )

    @property
    def blocked_env_vars(self) -> tuple[str, ...] | None:
        """Return explicit blocked environment variables override from options or settings."""
        return resolve_blocked_env_vars(self.options, self.settings)

    @property
    def use_symlinks(self) -> bool:
        """Return whether symlink installation is enabled on the runtime options."""
        return self.options.use_symlinks

    @property
    def isolate_config_dir(self) -> bool:
        """Return whether isolated runtime config is enabled on the runtime options."""
        return self.options.isolate_config_dir

    @property
    def auto_clean(self) -> bool:
        """Return whether automatic post-probe cleanup is enabled on the runtime options."""
        return self.options.auto_clean

    @property
    def rations_catalog(self) -> bool:
        """Return True if this runtime actively rations skill listing budgets."""
        return False

    @property
    def is_cli(self) -> bool:
        """Return True if this runtime executes via a CLI subprocess."""
        return False

    @property
    def isolation_dir_field(self) -> str | None:
        """Return the options field name used for custom directory isolation, if supported."""
        return getattr(self.options, "isolation_dir_field", None)

    @property
    def custom_isolation_dir(self) -> Path | None:
        """Return the configured custom isolation directory path, or None."""
        return getattr(self.options, "custom_isolation_dir", None)

    def skills_dir(self, workdir: Path) -> Path:
        """Return standard skill directory path in the workspace."""
        return Path(workdir) / self.skills_subpath

    def skill_roots(self, workdir: Path) -> tuple[SkillRoot, ...]:
        """Return discovery directories where runtime searches for skills."""
        here = self.skills_dir(resolve_path(workdir))
        return (SkillRoot(path=here, scope="project", precedence=0),) if here.is_dir() else ()

    def fit(self, catalog: Catalog, skills: Iterable[Skill]) -> CatalogFit:  # noqa: ARG002
        """Evaluate whether a catalog fits listing budgets without materializing."""
        return CatalogFit()

    def install(self, catalog: Catalog, skills: Iterable[Skill], workdir: Path) -> Path:
        """Materialize resident skills in workspace and return workspace path."""
        target = resolve_path(workdir)
        self._validate_install(target)
        by_name = resolve_catalog_skills(catalog, skills)
        self._resident = install_skills(
            catalog,
            by_name,
            self.skills_dir(target),
            use_symlinks=self.use_symlinks,
        )
        self._post_install(target)
        return target

    def _validate_install(self, workdir: Path) -> None:
        """Validate workspace preconditions before installation."""
        del workdir

    def _post_install(self, workdir: Path) -> None:
        """Configure permissions or settings after installation."""
        del workdir

    def post_probe(self, workdir: Path) -> None:
        """Execute post-probe cleanup actions."""
        del workdir

    def clone_isolated(self) -> Self:
        """Create a thread-local isolated clone of this runtime."""
        clone = copy.copy(self)
        clone._resident = ()  # noqa: SLF001
        return clone

    def cleanup(self) -> None:  # noqa: B027
        """Release any isolated temporary resources owned by this runtime."""

    def build_env(self, workdir: Path | None = None) -> dict[str, str]:
        """Assemble process environment for agent execution."""
        del workdir
        return sanitize_subprocess_env(dict(os.environ), blocked_env_vars=self.blocked_env_vars)

    def make_tracker(self, target_skill: str | None = None) -> TrajectoryTracker:
        """Create a TrajectoryTracker configured with this runtime's turn and early-exit options."""
        return TrajectoryTracker(
            target_skill=target_skill,
            max_turns=self.options.max_turns,
            early_exit=self.options.early_exit,
        )

    @abstractmethod
    def select(
        self,
        query_text: str,
        workdir: Path,
        target_skill: str | None = None,
    ) -> SelectionOutcome:
        """Execute a single query probe and return the observed skill selection."""
        ...

    @property
    def effective_effort(self) -> str | None:
        """Return configured reasoning effort or default from model profile."""
        options = getattr(self, "options", None)
        if options is not None and (effort := getattr(options, "effort", None)):
            return None if effort.lower() in ("none", "off") else effort
        try:
            return model_profile(self.model).effort
        except (KeyError, ValueError):
            return None

    def resident_skill_paths(self, workdir: Path) -> frozenset[Path]:
        """Return set of valid filesystem directory paths for resident skills."""
        skills_dir = self.skills_dir(resolve_path(workdir))
        return frozenset(resolve_path(skills_dir / name) for name in self._resident)

    def parse_stream(
        self,
        lines: Iterable[str],
        resident: Sequence[str] = (),
        early_exit: bool = False,
    ) -> SessionSummary:
        """Parse transcript or log lines into a standardized SessionSummary.

        Default implementation records early_exit and assigns SUCCESS status
        if lines were produced. Subclasses override this to extract tool
        invocations, reasoning, and telemetry.
        """
        del resident
        line_list = list(lines)
        return SessionSummary(
            early_exit=early_exit,
            status=SessionStatus.SUCCESS if line_list else None,
        )

allowed_tools property

allowed_tools: tuple[str, ...] | None

Return explicit allowed tools override from options or settings.

api_key property

api_key: str | None

Return the API key string configured on the runtime options.

auto_clean property

auto_clean: bool

Return whether automatic post-probe cleanup is enabled on the runtime options.

blocked_env_vars property

blocked_env_vars: tuple[str, ...] | None

Return explicit blocked environment variables override from options or settings.

custom_isolation_dir property

custom_isolation_dir: Path | None

Return the configured custom isolation directory path, or None.

early_exit property

early_exit: bool

Return whether early exit is enabled on the runtime options.

effective_effort property

effective_effort: str | None

Return configured reasoning effort or default from model profile.

effort property

effort: str | None

Return the reasoning effort tier being evaluated.

is_cli property

is_cli: bool

Return True if this runtime executes via a CLI subprocess.

isolate_config_dir property

isolate_config_dir: bool

Return whether isolated runtime config is enabled on the runtime options.

isolation_dir_field property

isolation_dir_field: str | None

Return the options field name used for custom directory isolation, if supported.

max_turns property

max_turns: int

Return the maximum turns configured on the runtime options.

model property writable

model: str

Return the identifier of the model being evaluated.

provider property

provider: str | None

Return the model provider identifier being evaluated.

rations_catalog property

rations_catalog: bool

Return True if this runtime actively rations skill listing budgets.

skills_subpath property

skills_subpath: str

Return relative skill directory subpath from profile or runtime default.

timeout_s property

timeout_s: int | None

Return per-probe execution timeout in seconds from settings.

use_symlinks property

use_symlinks: bool

Return whether symlink installation is enabled on the runtime options.

__init__

__init__(
    settings: RuntimeSettings | None = None,
    options: OptionsT | None = None,
) -> None

Initialize agent runtime settings and options.

Source code in src/reach/runtime/__init__.py

def __init__(
    self,
    settings: RuntimeSettings | None = None,
    options: OptionsT | None = None,
) -> None:
    """Initialize agent runtime settings and options."""
    agent_name = getattr(self, "name", "agent")
    self.settings = settings
    if options is not None:
        self.options = options
    elif settings is not None:
        resolved = resolve_options(settings)
        opt_cls = options_model(agent_name)
        if opt_cls is not None and issubclass(opt_cls, AgentOptions):
            self.options = cast(
                "OptionsT",
                resolved
                if isinstance(resolved, opt_cls)
                else opt_cls.model_validate(dict(settings.options or {})),
            )
        else:
            self.options = cast(
                "OptionsT",
                resolved
                if isinstance(resolved, AgentOptions)
                else AgentOptions.model_validate(dict(settings.options or {})),
            )
    else:
        opt_cls = options_model(agent_name)
        if opt_cls is not None and issubclass(opt_cls, AgentOptions):
            self.options = cast("OptionsT", opt_cls())
        else:
            self.options = cast("OptionsT", AgentOptions())
    self._resident = ()

build_env

build_env(workdir: Path | None = None) -> dict[str, str]

Assemble process environment for agent execution.

Source code in src/reach/runtime/__init__.py

def build_env(self, workdir: Path | None = None) -> dict[str, str]:
    """Assemble process environment for agent execution."""
    del workdir
    return sanitize_subprocess_env(dict(os.environ), blocked_env_vars=self.blocked_env_vars)

cleanup

cleanup() -> None

Release any isolated temporary resources owned by this runtime.

Source code in src/reach/runtime/__init__.py

def cleanup(self) -> None:  # noqa: B027
    """Release any isolated temporary resources owned by this runtime."""

clone_isolated

clone_isolated() -> Self

Create a thread-local isolated clone of this runtime.

Source code in src/reach/runtime/__init__.py

def clone_isolated(self) -> Self:
    """Create a thread-local isolated clone of this runtime."""
    clone = copy.copy(self)
    clone._resident = ()  # noqa: SLF001
    return clone

fit

fit(
    catalog: Catalog, skills: Iterable[Skill]
) -> CatalogFit

Evaluate whether a catalog fits listing budgets without materializing.

Source code in src/reach/runtime/__init__.py

def fit(self, catalog: Catalog, skills: Iterable[Skill]) -> CatalogFit:  # noqa: ARG002
    """Evaluate whether a catalog fits listing budgets without materializing."""
    return CatalogFit()

install

install(
    catalog: Catalog, skills: Iterable[Skill], workdir: Path
) -> Path

Materialize resident skills in workspace and return workspace path.

Source code in src/reach/runtime/__init__.py

def install(self, catalog: Catalog, skills: Iterable[Skill], workdir: Path) -> Path:
    """Materialize resident skills in workspace and return workspace path."""
    target = resolve_path(workdir)
    self._validate_install(target)
    by_name = resolve_catalog_skills(catalog, skills)
    self._resident = install_skills(
        catalog,
        by_name,
        self.skills_dir(target),
        use_symlinks=self.use_symlinks,
    )
    self._post_install(target)
    return target

make_tracker

make_tracker(
    target_skill: str | None = None,
) -> TrajectoryTracker

Create a TrajectoryTracker configured with this runtime's turn and early-exit options.

Source code in src/reach/runtime/__init__.py

def make_tracker(self, target_skill: str | None = None) -> TrajectoryTracker:
    """Create a TrajectoryTracker configured with this runtime's turn and early-exit options."""
    return TrajectoryTracker(
        target_skill=target_skill,
        max_turns=self.options.max_turns,
        early_exit=self.options.early_exit,
    )

parse_stream

parse_stream(
    lines: Iterable[str],
    resident: Sequence[str] = (),
    early_exit: bool = False,
) -> SessionSummary

Parse transcript or log lines into a standardized SessionSummary.

Default implementation records early_exit and assigns SUCCESS status if lines were produced. Subclasses override this to extract tool invocations, reasoning, and telemetry.

Source code in src/reach/runtime/__init__.py

def parse_stream(
    self,
    lines: Iterable[str],
    resident: Sequence[str] = (),
    early_exit: bool = False,
) -> SessionSummary:
    """Parse transcript or log lines into a standardized SessionSummary.

    Default implementation records early_exit and assigns SUCCESS status
    if lines were produced. Subclasses override this to extract tool
    invocations, reasoning, and telemetry.
    """
    del resident
    line_list = list(lines)
    return SessionSummary(
        early_exit=early_exit,
        status=SessionStatus.SUCCESS if line_list else None,
    )

post_probe

post_probe(workdir: Path) -> None

Execute post-probe cleanup actions.

Source code in src/reach/runtime/__init__.py

def post_probe(self, workdir: Path) -> None:
    """Execute post-probe cleanup actions."""
    del workdir

resident_skill_paths

resident_skill_paths(workdir: Path) -> frozenset[Path]

Return set of valid filesystem directory paths for resident skills.

Source code in src/reach/runtime/__init__.py

def resident_skill_paths(self, workdir: Path) -> frozenset[Path]:
    """Return set of valid filesystem directory paths for resident skills."""
    skills_dir = self.skills_dir(resolve_path(workdir))
    return frozenset(resolve_path(skills_dir / name) for name in self._resident)

select abstractmethod

select(
    query_text: str,
    workdir: Path,
    target_skill: str | None = None,
) -> SelectionOutcome

Execute a single query probe and return the observed skill selection.

Source code in src/reach/runtime/__init__.py

@abstractmethod
def select(
    self,
    query_text: str,
    workdir: Path,
    target_skill: str | None = None,
) -> SelectionOutcome:
    """Execute a single query probe and return the observed skill selection."""
    ...

skill_roots

skill_roots(workdir: Path) -> tuple[SkillRoot, ...]

Return discovery directories where runtime searches for skills.

Source code in src/reach/runtime/__init__.py

def skill_roots(self, workdir: Path) -> tuple[SkillRoot, ...]:
    """Return discovery directories where runtime searches for skills."""
    here = self.skills_dir(resolve_path(workdir))
    return (SkillRoot(path=here, scope="project", precedence=0),) if here.is_dir() else ()

skills_dir

skills_dir(workdir: Path) -> Path

Return standard skill directory path in the workspace.

Source code in src/reach/runtime/__init__.py

def skills_dir(self, workdir: Path) -> Path:
    """Return standard skill directory path in the workspace."""
    return Path(workdir) / self.skills_subpath

AntigravityRuntime

Bases: AgentRuntime

Shared base runtime for Antigravity-ecosystem drivers (CLI and SDK).

Source code in src/reach/runtime/__init__.py

class AntigravityRuntime(AgentRuntime):
    """Shared base runtime for Antigravity-ecosystem drivers (CLI and SDK)."""

    ANTIGRAVITY_SELECTION_TOOLS: frozenset[str] = frozenset(
        {
            "view_file",
            "list_dir",
            "grep_search",
            "find_by_name",
        }
    )
    ANTIGRAVITY_BUILTIN_TOOLS: frozenset[str] = ANTIGRAVITY_SELECTION_TOOLS | frozenset(
        {
            "ask_question",
            "finish",
            "multi_replace_file_content",
            "read_url_content",
            "replace_file_content",
            "run_command",
            "search_web",
            "write_to_file",
        }
    )

    @property
    def selection_tools(self) -> frozenset[str]:
        """Return standard inspection tool identifiers permitted during skill selection."""
        return self.ANTIGRAVITY_SELECTION_TOOLS

    @property
    def effective_api_key(self) -> str | None:
        """Return configured API key or fallback to environment variables."""
        return (
            self.options.api_key
            or os.environ.get("GEMINI_API_KEY")
            or os.environ.get("GOOGLE_API_KEY")
        )

    @property
    def effective_model_provider(self) -> str | None:
        """Return configured model_provider or auto-detect 'gemini' when API keys are present."""
        return detect_model_provider(
            self.model,
            getattr(self.options, "model_provider", None),
            api_key=self.options.api_key,
        )

    @classmethod
    def selection_schema(cls, resident: Sequence[str]) -> type[SkillSelectionBase]:
        """Generate dynamic Pydantic model constraining selection to resident skills."""
        import typing

        from pydantic import create_model

        literal_type: Any = typing.cast("Any", typing.Literal)[tuple(resident)] if resident else str

        return create_model(
            "SkillSelection",
            __base__=SkillSelectionBase,
            selected_skill=(
                literal_type | None,
                Field(default=None, description="The skill to invoke, or null if no skill applies"),
            ),
            reasoning=(
                str,
                Field(
                    default="",
                    description=(
                        "Brief explanation of why this skill was selected or why null was returned"
                    ),
                ),
            ),
        )

    @classmethod
    def selection_json_schema(cls, resident: Sequence[str]) -> str:
        """Derive JSON Schema string directly from the canonical Pydantic model."""
        schema_dict = cls.selection_schema(resident).model_json_schema()
        schema_dict["required"] = ["selected_skill", "reasoning"]
        return json.dumps(schema_dict)

effective_api_key property

effective_api_key: str | None

Return configured API key or fallback to environment variables.

effective_model_provider property

effective_model_provider: str | None

Return configured model_provider or auto-detect 'gemini' when API keys are present.

selection_tools property

selection_tools: frozenset[str]

Return standard inspection tool identifiers permitted during skill selection.

selection_json_schema classmethod

selection_json_schema(resident: Sequence[str]) -> str

Derive JSON Schema string directly from the canonical Pydantic model.

Source code in src/reach/runtime/__init__.py

@classmethod
def selection_json_schema(cls, resident: Sequence[str]) -> str:
    """Derive JSON Schema string directly from the canonical Pydantic model."""
    schema_dict = cls.selection_schema(resident).model_json_schema()
    schema_dict["required"] = ["selected_skill", "reasoning"]
    return json.dumps(schema_dict)

selection_schema classmethod

selection_schema(
    resident: Sequence[str],
) -> type[SkillSelectionBase]

Generate dynamic Pydantic model constraining selection to resident skills.

Source code in src/reach/runtime/__init__.py

@classmethod
def selection_schema(cls, resident: Sequence[str]) -> type[SkillSelectionBase]:
    """Generate dynamic Pydantic model constraining selection to resident skills."""
    import typing

    from pydantic import create_model

    literal_type: Any = typing.cast("Any", typing.Literal)[tuple(resident)] if resident else str

    return create_model(
        "SkillSelection",
        __base__=SkillSelectionBase,
        selected_skill=(
            literal_type | None,
            Field(default=None, description="The skill to invoke, or null if no skill applies"),
        ),
        reasoning=(
            str,
            Field(
                default="",
                description=(
                    "Brief explanation of why this skill was selected or why null was returned"
                ),
            ),
        ),
    )

CatalogFit

Bases: BaseModel

Report catalog residency limits and potential description truncation metrics.

Source code in src/reach/runtime/__init__.py

class CatalogFit(BaseModel):
    """Report catalog residency limits and potential description truncation metrics."""

    model_config = ConfigDict(frozen=True)

    allowed: int = Field(default=0, ge=0)
    asked: int = Field(default=0, ge=0)
    unit: str = ""
    truncated: int = Field(default=0, ge=0)
    remedy: str = ""
    elided_skills: tuple[str, ...] = ()

    @property
    def rations(self) -> bool:
        """Return True if the runtime enforces space limitations on catalog listings."""
        return self.allowed > 0

    @property
    def whole(self) -> bool:
        """Return True if all skills are presented without description truncation."""
        return not self.truncated

rations property

rations: bool

Return True if the runtime enforces space limitations on catalog listings.

whole property

whole: bool

Return True if all skills are presented without description truncation.

CliAgentRuntime

Bases: AgentRuntime[CliOptionsT], ABC

Abstract base runtime for command-line interface agent drivers.

Source code in src/reach/runtime/__init__.py

class CliAgentRuntime[CliOptionsT: CliOptions](AgentRuntime[CliOptionsT], ABC):
    """Abstract base runtime for command-line interface agent drivers."""

    options: CliOptionsT
    api_key_env_var: str | None = None

    def __init__(
        self,
        settings: RuntimeSettings | None = None,
        options: CliOptionsT | None = None,
    ) -> None:
        """Initialize CLI agent runtime settings and options."""
        super().__init__(settings=settings, options=options)
        self.completion_cost_usd = 0.0
        self.completions = 0

    @property
    @override
    def is_cli(self) -> bool:
        """Return True if this runtime executes via a CLI subprocess."""
        return True

    @abstractmethod
    def build_command(self, query_text: str) -> list[str]:
        """Assemble command-line arguments for executing a probe."""
        ...

    @abstractmethod
    @override
    def parse_stream(
        self,
        lines: Iterable[str],
        resident: Sequence[str] = (),
        early_exit: bool = False,
    ) -> SessionSummary:
        """Parse CLI stdout lines into a standardized session summary."""
        ...

    def extract_skills_from_line(self, line: str) -> Sequence[str]:
        """Extract invoked skill names from an event line for early-exit detection."""
        single = self.extract_skill_from_line(line)
        return (single,) if single else ()

    def extract_skill_from_line(self, line: str) -> str | None:
        """Extract invoked skill name from an event line for early-exit detection."""
        del line
        return None

    @override
    def build_env(self, workdir: Path | None = None) -> dict[str, str]:
        """Assemble process environment with API keys and workspace overrides."""
        del workdir
        env = dict(os.environ)
        if (home_dir := getattr(self.options, "home_dir", None)) is not None:
            env["HOME"] = str(home_dir)
        if (api_key := getattr(self.options, "api_key", None)) is not None and self.api_key_env_var:
            env[self.api_key_env_var] = str(api_key)
        return sanitize_subprocess_env(env, blocked_env_vars=self.blocked_env_vars)

    def validate_outcome(
        self,
        summary: SessionSummary,
        workdir: Path,
    ) -> str | None:
        """Validate status and security isolation boundaries for a parsed session."""
        del workdir
        if summary.status is not None and summary.status != SessionStatus.SUCCESS:
            return summary.error or f"runtime error: {summary.status}"
        return check_tool_leak(summary.observed_tools, self.allowed_tools)

    isolation_dir_name: ClassVar[str | None] = None

    def effective_isolation_dir(self, workdir: Path) -> Path | None:
        """Return the effective isolated configuration directory for the given workspace."""
        if not self.options.isolate_config_dir:
            return None
        if self.custom_isolation_dir:
            return self.custom_isolation_dir
        if self.isolation_dir_name:
            return Path(workdir) / self.isolation_dir_name
        return None

    @override
    def post_probe(self, workdir: Path) -> None:
        """Clean isolated configuration directory after probe if auto_clean is enabled."""
        if not self.options.auto_clean:
            return
        iso_dir = self.effective_isolation_dir(workdir)
        if iso_dir is not None:
            safe_cleanup_isolated_dir(workdir, iso_dir)

    @override
    def select(
        self,
        query_text: str,
        workdir: Path,
        target_skill: str | None = None,
    ) -> SelectionOutcome:
        """Execute a query probe via the unified CLI subprocess template pipeline."""
        tracker = self.make_tracker(target_skill)

        def _on_line(line: str) -> bool:
            skills = self.extract_skills_from_line(line)
            return tracker.observe(skills)

        start_time = time.monotonic()
        try:
            completed, err = run_subprocess_probe(
                self.build_command(query_text),
                workdir,
                self.timeout_s,
                env=self.build_env(workdir),
                on_line=_on_line,
            )
            elapsed_ms = int((time.monotonic() - start_time) * 1000)
            if err is not None or completed is None:
                return SelectionOutcome(
                    error=err or "subprocess failed",
                    observed_catalog=self._resident,
                )

            summary = self.parse_stream(
                completed.stdout.splitlines(),
                resident=self._resident,
                early_exit=tracker.early_exit_hit,
            )
            if not summary.saw_result:
                reason = process_failure_reason(completed)
                if getattr(summary, "retries", 0) > 0:
                    reason = f"rate limit (429): {summary.retries} retries exceeded"
                return SelectionOutcome(
                    error=f"no result event: {reason}",
                    observed_catalog=self._resident,
                )

            catalog = getattr(summary, "observed_catalog", ()) or self._resident
            outcome = tracker.apply_to_outcome(
                summary.to_outcome(
                    observed_catalog=catalog,
                    fallback_model=self.model,
                    early_exit=tracker.early_exit_hit,
                ),
            )

            # Security tool leak and outcome validation
            if validation_error := self.validate_outcome(summary, workdir):
                return outcome.model_copy(update={"error": validation_error})

            duration_ms = outcome.duration_ms or elapsed_ms
            return outcome.model_copy(update={"duration_ms": duration_ms})
        except Exception as exc:  # noqa: BLE001
            return SelectionOutcome(
                error=f"unexpected runtime error: {exc}",
                observed_catalog=self._resident,
            )
        finally:
            self.post_probe(workdir)

is_cli property

is_cli: bool

Return True if this runtime executes via a CLI subprocess.

__init__

__init__(
    settings: RuntimeSettings | None = None,
    options: CliOptionsT | None = None,
) -> None

Initialize CLI agent runtime settings and options.

Source code in src/reach/runtime/__init__.py

def __init__(
    self,
    settings: RuntimeSettings | None = None,
    options: CliOptionsT | None = None,
) -> None:
    """Initialize CLI agent runtime settings and options."""
    super().__init__(settings=settings, options=options)
    self.completion_cost_usd = 0.0
    self.completions = 0

build_command abstractmethod

build_command(query_text: str) -> list[str]

Assemble command-line arguments for executing a probe.

Source code in src/reach/runtime/__init__.py

@abstractmethod
def build_command(self, query_text: str) -> list[str]:
    """Assemble command-line arguments for executing a probe."""
    ...

build_env

build_env(workdir: Path | None = None) -> dict[str, str]

Assemble process environment with API keys and workspace overrides.

Source code in src/reach/runtime/__init__.py

@override
def build_env(self, workdir: Path | None = None) -> dict[str, str]:
    """Assemble process environment with API keys and workspace overrides."""
    del workdir
    env = dict(os.environ)
    if (home_dir := getattr(self.options, "home_dir", None)) is not None:
        env["HOME"] = str(home_dir)
    if (api_key := getattr(self.options, "api_key", None)) is not None and self.api_key_env_var:
        env[self.api_key_env_var] = str(api_key)
    return sanitize_subprocess_env(env, blocked_env_vars=self.blocked_env_vars)

effective_isolation_dir

effective_isolation_dir(workdir: Path) -> Path | None

Return the effective isolated configuration directory for the given workspace.

Source code in src/reach/runtime/__init__.py

def effective_isolation_dir(self, workdir: Path) -> Path | None:
    """Return the effective isolated configuration directory for the given workspace."""
    if not self.options.isolate_config_dir:
        return None
    if self.custom_isolation_dir:
        return self.custom_isolation_dir
    if self.isolation_dir_name:
        return Path(workdir) / self.isolation_dir_name
    return None

extract_skill_from_line

extract_skill_from_line(line: str) -> str | None

Extract invoked skill name from an event line for early-exit detection.

Source code in src/reach/runtime/__init__.py

def extract_skill_from_line(self, line: str) -> str | None:
    """Extract invoked skill name from an event line for early-exit detection."""
    del line
    return None

extract_skills_from_line

extract_skills_from_line(line: str) -> Sequence[str]

Extract invoked skill names from an event line for early-exit detection.

Source code in src/reach/runtime/__init__.py

def extract_skills_from_line(self, line: str) -> Sequence[str]:
    """Extract invoked skill names from an event line for early-exit detection."""
    single = self.extract_skill_from_line(line)
    return (single,) if single else ()

parse_stream abstractmethod

parse_stream(
    lines: Iterable[str],
    resident: Sequence[str] = (),
    early_exit: bool = False,
) -> SessionSummary

Parse CLI stdout lines into a standardized session summary.

Source code in src/reach/runtime/__init__.py

@abstractmethod
@override
def parse_stream(
    self,
    lines: Iterable[str],
    resident: Sequence[str] = (),
    early_exit: bool = False,
) -> SessionSummary:
    """Parse CLI stdout lines into a standardized session summary."""
    ...

post_probe

post_probe(workdir: Path) -> None

Clean isolated configuration directory after probe if auto_clean is enabled.

Source code in src/reach/runtime/__init__.py

@override
def post_probe(self, workdir: Path) -> None:
    """Clean isolated configuration directory after probe if auto_clean is enabled."""
    if not self.options.auto_clean:
        return
    iso_dir = self.effective_isolation_dir(workdir)
    if iso_dir is not None:
        safe_cleanup_isolated_dir(workdir, iso_dir)

select

select(
    query_text: str,
    workdir: Path,
    target_skill: str | None = None,
) -> SelectionOutcome

Execute a query probe via the unified CLI subprocess template pipeline.

Source code in src/reach/runtime/__init__.py

@override
def select(
    self,
    query_text: str,
    workdir: Path,
    target_skill: str | None = None,
) -> SelectionOutcome:
    """Execute a query probe via the unified CLI subprocess template pipeline."""
    tracker = self.make_tracker(target_skill)

    def _on_line(line: str) -> bool:
        skills = self.extract_skills_from_line(line)
        return tracker.observe(skills)

    start_time = time.monotonic()
    try:
        completed, err = run_subprocess_probe(
            self.build_command(query_text),
            workdir,
            self.timeout_s,
            env=self.build_env(workdir),
            on_line=_on_line,
        )
        elapsed_ms = int((time.monotonic() - start_time) * 1000)
        if err is not None or completed is None:
            return SelectionOutcome(
                error=err or "subprocess failed",
                observed_catalog=self._resident,
            )

        summary = self.parse_stream(
            completed.stdout.splitlines(),
            resident=self._resident,
            early_exit=tracker.early_exit_hit,
        )
        if not summary.saw_result:
            reason = process_failure_reason(completed)
            if getattr(summary, "retries", 0) > 0:
                reason = f"rate limit (429): {summary.retries} retries exceeded"
            return SelectionOutcome(
                error=f"no result event: {reason}",
                observed_catalog=self._resident,
            )

        catalog = getattr(summary, "observed_catalog", ()) or self._resident
        outcome = tracker.apply_to_outcome(
            summary.to_outcome(
                observed_catalog=catalog,
                fallback_model=self.model,
                early_exit=tracker.early_exit_hit,
            ),
        )

        # Security tool leak and outcome validation
        if validation_error := self.validate_outcome(summary, workdir):
            return outcome.model_copy(update={"error": validation_error})

        duration_ms = outcome.duration_ms or elapsed_ms
        return outcome.model_copy(update={"duration_ms": duration_ms})
    except Exception as exc:  # noqa: BLE001
        return SelectionOutcome(
            error=f"unexpected runtime error: {exc}",
            observed_catalog=self._resident,
        )
    finally:
        self.post_probe(workdir)

validate_outcome

validate_outcome(
    summary: SessionSummary, workdir: Path
) -> str | None

Validate status and security isolation boundaries for a parsed session.

Source code in src/reach/runtime/__init__.py

def validate_outcome(
    self,
    summary: SessionSummary,
    workdir: Path,
) -> str | None:
    """Validate status and security isolation boundaries for a parsed session."""
    del workdir
    if summary.status is not None and summary.status != SessionStatus.SUCCESS:
        return summary.error or f"runtime error: {summary.status}"
    return check_tool_leak(summary.observed_tools, self.allowed_tools)

CliOptions

Bases: AgentOptions

Hold common configuration options for CLI subprocess-driven agent runtimes.

Source code in src/reach/runtime/__init__.py

class CliOptions(AgentOptions):
    """Hold common configuration options for CLI subprocess-driven agent runtimes."""

    executable: str = ""
    extra_args: tuple[str, ...] = ()

    def effort_args(self, flag: str = "--effort") -> list[str]:
        """Format CLI argument pair for non-empty effort setting."""
        return [flag, self.effort] if self.effort else []

    def provider_args(self, flag: str = "--provider") -> list[str]:
        """Format CLI argument pair for non-empty provider setting."""
        return [flag, self.provider] if self.provider else []

    def max_turns_args(self, flag: str = "--max-turns") -> list[str]:
        """Format CLI argument pair for non-None max_turns setting."""
        return [flag, str(self.max_turns)] if self.max_turns is not None else []

    def api_key_args(self, flag: str = "--api-key") -> list[str]:
        """Format CLI argument pair for non-empty api_key setting."""
        return [flag, self.api_key] if self.api_key else []

api_key_args

api_key_args(flag: str = '--api-key') -> list[str]

Format CLI argument pair for non-empty api_key setting.

Source code in src/reach/runtime/__init__.py

def api_key_args(self, flag: str = "--api-key") -> list[str]:
    """Format CLI argument pair for non-empty api_key setting."""
    return [flag, self.api_key] if self.api_key else []

effort_args

effort_args(flag: str = '--effort') -> list[str]

Format CLI argument pair for non-empty effort setting.

Source code in src/reach/runtime/__init__.py

def effort_args(self, flag: str = "--effort") -> list[str]:
    """Format CLI argument pair for non-empty effort setting."""
    return [flag, self.effort] if self.effort else []

max_turns_args

max_turns_args(flag: str = '--max-turns') -> list[str]

Format CLI argument pair for non-None max_turns setting.

Source code in src/reach/runtime/__init__.py

def max_turns_args(self, flag: str = "--max-turns") -> list[str]:
    """Format CLI argument pair for non-None max_turns setting."""
    return [flag, str(self.max_turns)] if self.max_turns is not None else []

provider_args

provider_args(flag: str = '--provider') -> list[str]

Format CLI argument pair for non-empty provider setting.

Source code in src/reach/runtime/__init__.py

def provider_args(self, flag: str = "--provider") -> list[str]:
    """Format CLI argument pair for non-empty provider setting."""
    return [flag, self.provider] if self.provider else []

SelectionOutcome

Bases: SessionSummary

Represent the observable result of probing an agent runtime with a query.

Source code in src/reach/runtime/__init__.py

class SelectionOutcome(SessionSummary):
    """Represent the observable result of probing an agent runtime with a query."""

    @model_validator(mode="before")
    @classmethod
    def _enforce_early_exit_invariants(cls, data: Any) -> Any:  # noqa: ANN401 (Pydantic before validator)
        """Ensure process cancellation artifacts are never reported as probe errors."""
        if isinstance(data, dict) and data.get("early_exit"):
            data = dict(data)
            err = str(data.get("error") or "")
            if err and not err.startswith(("tool leak", "residency leak")):
                data["error"] = None
        return data

SessionStatus

Bases: StrEnum

Enumerate canonical terminal execution statuses of an agent session.

Source code in src/reach/runtime/__init__.py

class SessionStatus(StrEnum):
    """Enumerate canonical terminal execution statuses of an agent session."""

    CANCELLED = "CANCELLED"
    ERROR = "ERROR"
    SUCCESS = "SUCCESS"
    TIMEOUT = "TIMEOUT"

SessionSummary

Bases: BaseModel

Hold parsed session outcomes and telemetry across CLI runtime logs.

Source code in src/reach/runtime/__init__.py

class SessionSummary(BaseModel):
    """Hold parsed session outcomes and telemetry across CLI runtime logs."""

    model_config = ConfigDict(frozen=True)

    cost_usd: float | None = None
    duration_ms: int | None = None
    prompt_tokens: int | None = None
    error: str | None = None
    invoked_skills: tuple[str, ...] = ()
    early_exit: bool = False
    turns_taken: int = Field(default=1, ge=1)
    tool_calls: tuple[ToolCallInfo, ...] = ()
    observed_tools: tuple[str, ...] = ()
    reasoning: tuple[str, ...] = ()
    observed_catalog: tuple[str, ...] = ()
    resolved_model: str = ""
    status: SessionStatus | str | None = None
    retries: int = 0

    @property
    def invoked_skill(self) -> str | None:
        """Return the first invoked skill name, or None if none was invoked."""
        return self.invoked_skills[0] if self.invoked_skills else None

    @field_validator("status", mode="before")
    @classmethod
    def _normalize_status(cls, value: object) -> SessionStatus | str | None:
        """Coerce known status variants to canonical SessionStatus enum members."""
        if value is None or isinstance(value, SessionStatus):
            return value
        if isinstance(value, str):
            normalized = value.strip()
            upper = normalized.upper()
            try:
                return SessionStatus(upper)
            except ValueError:
                return normalized
        return str(value)

    @property
    def saw_result(self) -> bool:
        """Return True if stream contained a terminal result event or early exit."""
        return self.status is not None or self.early_exit

    @model_validator(mode="before")
    @classmethod
    def _derive_observed_tools(cls, data: Any) -> Any:  # noqa: ANN401 (Pydantic before validator)
        """Populate observed tools from tool calls if not explicitly provided."""
        if isinstance(data, dict):
            data = dict(data)
            calls = data.get("tool_calls")
            if calls and not data.get("observed_tools"):
                names = {getattr(c, "name", None) or c.get("name") for c in calls if c}
                data["observed_tools"] = tuple(sorted(n for n in names if n))
        return data

    def to_outcome(
        self,
        observed_catalog: tuple[str, ...] | list[str] = (),
        fallback_model: str = "",
        early_exit: bool = False,
        turns_taken: int | None = None,
    ) -> SelectionOutcome:
        """Convert parsed session summary into a canonical SelectionOutcome."""
        effective_early_exit = early_exit or self.early_exit
        turns = turns_taken if turns_taken is not None else self.turns_taken

        catalog = tuple(observed_catalog) or self.observed_catalog
        return SelectionOutcome(
            cost_usd=self.cost_usd,
            duration_ms=self.duration_ms,
            prompt_tokens=self.prompt_tokens,
            error=self.error,
            invoked_skills=self.invoked_skills,
            early_exit=effective_early_exit,
            turns_taken=turns,
            tool_calls=self.tool_calls,
            observed_tools=self.observed_tools,
            reasoning=self.reasoning,
            resolved_model=self.resolved_model or fallback_model,
            status=self.status,
            observed_catalog=catalog,
        )

invoked_skill property

invoked_skill: str | None

Return the first invoked skill name, or None if none was invoked.

saw_result property

saw_result: bool

Return True if stream contained a terminal result event or early exit.

to_outcome

to_outcome(
    observed_catalog: tuple[str, ...] | list[str] = (),
    fallback_model: str = "",
    early_exit: bool = False,
    turns_taken: int | None = None,
) -> SelectionOutcome

Convert parsed session summary into a canonical SelectionOutcome.

Source code in src/reach/runtime/__init__.py

def to_outcome(
    self,
    observed_catalog: tuple[str, ...] | list[str] = (),
    fallback_model: str = "",
    early_exit: bool = False,
    turns_taken: int | None = None,
) -> SelectionOutcome:
    """Convert parsed session summary into a canonical SelectionOutcome."""
    effective_early_exit = early_exit or self.early_exit
    turns = turns_taken if turns_taken is not None else self.turns_taken

    catalog = tuple(observed_catalog) or self.observed_catalog
    return SelectionOutcome(
        cost_usd=self.cost_usd,
        duration_ms=self.duration_ms,
        prompt_tokens=self.prompt_tokens,
        error=self.error,
        invoked_skills=self.invoked_skills,
        early_exit=effective_early_exit,
        turns_taken=turns,
        tool_calls=self.tool_calls,
        observed_tools=self.observed_tools,
        reasoning=self.reasoning,
        resolved_model=self.resolved_model or fallback_model,
        status=self.status,
        observed_catalog=catalog,
    )

SkillRoot

Bases: BaseModel

Represent a skill discovery directory location and its resolution precedence.

Source code in src/reach/runtime/__init__.py

class SkillRoot(BaseModel):
    """Represent a skill discovery directory location and its resolution precedence."""

    model_config = ConfigDict(frozen=True)

    path: Path
    scope: str
    precedence: int = Field(default=0, ge=0)

SkillSelectionBase

Bases: BaseModel

Base model declaring structured skill selection response interface.

Source code in src/reach/runtime/__init__.py

class SkillSelectionBase(BaseModel):
    """Base model declaring structured skill selection response interface."""

    model_config = ConfigDict(extra="forbid")

    selected_skill: str | None = Field(
        default=None,
        description="The skill to invoke, or null if no skill applies",
    )
    reasoning: str = Field(
        default="",
        description="Brief explanation of why this skill was selected or why null was returned",
    )

TextGenerator

Bases: Protocol

Protocol for models or drivers capable of generating raw text completions.

Source code in src/reach/runtime/generator.py

@runtime_checkable
class TextGenerator(Protocol):
    """Protocol for models or drivers capable of generating raw text completions."""

    options: Any

    @property
    def name(self) -> str:
        """Return generator name or agent identifier."""
        ...

    @property
    def model(self) -> str:
        """Return the model identifier used for generation."""
        ...

    @model.setter
    def model(self, value: str) -> None:
        """Update configured model identifier."""
        ...

    @property
    def completions(self) -> int:
        """Return total number of completions executed."""
        ...

    @property
    def completion_cost_usd(self) -> float:
        """Return total cost accumulated across completions in USD."""
        ...

    def complete(self, prompt: str, *, schema: str | Mapping[str, Any] | None = None) -> str:
        """Generate a raw text completion for an arbitrary prompt."""
        ...

    def prompt_budget_chars(self) -> int | None:
        """Return maximum character length for prompts, or None if unbounded."""
        ...

completion_cost_usd property

completion_cost_usd: float

Return total cost accumulated across completions in USD.

completions property

completions: int

Return total number of completions executed.

model property writable

model: str

Return the model identifier used for generation.

name property

name: str

Return generator name or agent identifier.

complete

complete(
    prompt: str,
    *,
    schema: str | Mapping[str, Any] | None = None,
) -> str

Generate a raw text completion for an arbitrary prompt.

Source code in src/reach/runtime/generator.py

def complete(self, prompt: str, *, schema: str | Mapping[str, Any] | None = None) -> str:
    """Generate a raw text completion for an arbitrary prompt."""
    ...

prompt_budget_chars

prompt_budget_chars() -> int | None

Return maximum character length for prompts, or None if unbounded.

Source code in src/reach/runtime/generator.py

def prompt_budget_chars(self) -> int | None:
    """Return maximum character length for prompts, or None if unbounded."""
    ...

ToolCallInfo

Bases: BaseModel

Represent an observed tool invocation and its parameters.

Source code in src/reach/runtime/__init__.py

class ToolCallInfo(BaseModel):
    """Represent an observed tool invocation and its parameters."""

    model_config = ConfigDict(frozen=True)

    name: str = ""
    parameters: dict[str, Any] = Field(default_factory=dict)

    def __init__(
        self,
        name: str = "",
        parameters: dict[str, Any] | None = None,
        path: str | None = None,
        **data: Any,  # noqa: ANN401
    ) -> None:
        """Initialize tool invocation with optional direct path argument."""
        params = dict(parameters or data.pop("parameters", None) or {})
        if path is not None and "path" not in params and "AbsolutePath" not in params:
            params["path"] = str(path)
        super().__init__(name=name, parameters=params, **data)

    @model_validator(mode="before")
    @classmethod
    def _coerce_path(cls, data: Any) -> Any:  # noqa: ANN401
        """Populate parameters with path keyword argument when provided directly."""
        if isinstance(data, dict) and "path" in data:
            data = dict(data)
            params = dict(data.get("parameters") or {})
            val = data.pop("path")
            if val is not None and "path" not in params and "AbsolutePath" not in params:
                params["path"] = str(val)
            data["parameters"] = params
        return data

    @property
    def target_path(self) -> str | None:
        """Extract first non-empty filesystem path from recognized tool parameters."""
        from reach.runtime._fs import extract_tool_path

        return extract_tool_path(self.parameters)

    @property
    def path(self) -> str | None:
        """Return target path from parameters."""
        return self.target_path

path property

path: str | None

Return target path from parameters.

target_path property

target_path: str | None

Extract first non-empty filesystem path from recognized tool parameters.

__init__

__init__(
    name: str = "",
    parameters: dict[str, Any] | None = None,
    path: str | None = None,
    **data: Any,
) -> None

Initialize tool invocation with optional direct path argument.

Source code in src/reach/runtime/__init__.py

def __init__(
    self,
    name: str = "",
    parameters: dict[str, Any] | None = None,
    path: str | None = None,
    **data: Any,  # noqa: ANN401
) -> None:
    """Initialize tool invocation with optional direct path argument."""
    params = dict(parameters or data.pop("parameters", None) or {})
    if path is not None and "path" not in params and "AbsolutePath" not in params:
        params["path"] = str(path)
    super().__init__(name=name, parameters=params, **data)

TrajectoryTracker

Track multi-turn skill invocations and evaluate early exit conditions.

Source code in src/reach/runtime/__init__.py

class TrajectoryTracker:
    """Track multi-turn skill invocations and evaluate early exit conditions."""

    def __init__(
        self,
        target_skill: str | None = None,
        max_turns: int = 3,
        early_exit: bool = True,
    ) -> None:
        """Initialize tracker with optional target skill and turn boundaries."""
        self.target_skill = target_skill
        self.max_turns = max(max_turns, 1)
        self.early_exit = early_exit
        self.invoked_skills: list[str] = []
        self.early_exit_hit: bool = False

    def observe(self, skill: str | Sequence[str] | None) -> bool:
        """Record skill invocation(s) and return True if early-exit stop condition is met."""
        if self.early_exit and self.early_exit_hit:
            return True
        if not skill:
            return False
        items = (skill,) if isinstance(skill, str) else tuple(skill)
        for s in items:
            if not s:
                continue
            if not self.invoked_skills or self.invoked_skills[-1] != s:
                self.invoked_skills.append(s)
                if self.early_exit:
                    if self.target_skill is not None and s == self.target_skill:
                        self.early_exit_hit = True
                        return True
                    if len(self.invoked_skills) >= self.max_turns:
                        self.early_exit_hit = True
                        return True
        return False

    @property
    def turns_taken(self) -> int:
        """Return count of turns taken based on distinct recorded invocations."""
        return len(self.invoked_skills) if self.invoked_skills else 1

    def apply_to_outcome(self, outcome: SelectionOutcome) -> SelectionOutcome:
        """Normalize a SelectionOutcome through tracker early-exit and turn invariants."""
        if self.early_exit and self.early_exit_hit:
            skills = tuple(self.invoked_skills)
            early = True
        else:
            replay = TrajectoryTracker(
                target_skill=self.target_skill,
                max_turns=self.max_turns,
                early_exit=self.early_exit,
            )
            replay.observe(outcome.invoked_skills)
            skills = tuple(replay.invoked_skills[: self.max_turns])
            early = bool(self.early_exit and (replay.early_exit_hit or outcome.early_exit))
        was_truncated = len(outcome.invoked_skills) > len(skills)
        turns = min(outcome.turns_taken, len(skills) or 1) if was_truncated else outcome.turns_taken
        new_invoked_skill = skills[0] if skills else None
        return outcome.model_copy(
            update={
                "invoked_skill": new_invoked_skill,
                "invoked_skills": skills,
                "early_exit": early,
                "turns_taken": turns,
            },
        )

turns_taken property

turns_taken: int

Return count of turns taken based on distinct recorded invocations.

__init__

__init__(
    target_skill: str | None = None,
    max_turns: int = 3,
    early_exit: bool = True,
) -> None

Initialize tracker with optional target skill and turn boundaries.

Source code in src/reach/runtime/__init__.py

def __init__(
    self,
    target_skill: str | None = None,
    max_turns: int = 3,
    early_exit: bool = True,
) -> None:
    """Initialize tracker with optional target skill and turn boundaries."""
    self.target_skill = target_skill
    self.max_turns = max(max_turns, 1)
    self.early_exit = early_exit
    self.invoked_skills: list[str] = []
    self.early_exit_hit: bool = False

apply_to_outcome

apply_to_outcome(
    outcome: SelectionOutcome,
) -> SelectionOutcome

Normalize a SelectionOutcome through tracker early-exit and turn invariants.

Source code in src/reach/runtime/__init__.py

def apply_to_outcome(self, outcome: SelectionOutcome) -> SelectionOutcome:
    """Normalize a SelectionOutcome through tracker early-exit and turn invariants."""
    if self.early_exit and self.early_exit_hit:
        skills = tuple(self.invoked_skills)
        early = True
    else:
        replay = TrajectoryTracker(
            target_skill=self.target_skill,
            max_turns=self.max_turns,
            early_exit=self.early_exit,
        )
        replay.observe(outcome.invoked_skills)
        skills = tuple(replay.invoked_skills[: self.max_turns])
        early = bool(self.early_exit and (replay.early_exit_hit or outcome.early_exit))
    was_truncated = len(outcome.invoked_skills) > len(skills)
    turns = min(outcome.turns_taken, len(skills) or 1) if was_truncated else outcome.turns_taken
    new_invoked_skill = skills[0] if skills else None
    return outcome.model_copy(
        update={
            "invoked_skill": new_invoked_skill,
            "invoked_skills": skills,
            "early_exit": early,
            "turns_taken": turns,
        },
    )

observe

observe(skill: str | Sequence[str] | None) -> bool

Record skill invocation(s) and return True if early-exit stop condition is met.

Source code in src/reach/runtime/__init__.py

def observe(self, skill: str | Sequence[str] | None) -> bool:
    """Record skill invocation(s) and return True if early-exit stop condition is met."""
    if self.early_exit and self.early_exit_hit:
        return True
    if not skill:
        return False
    items = (skill,) if isinstance(skill, str) else tuple(skill)
    for s in items:
        if not s:
            continue
        if not self.invoked_skills or self.invoked_skills[-1] != s:
            self.invoked_skills.append(s)
            if self.early_exit:
                if self.target_skill is not None and s == self.target_skill:
                    self.early_exit_hit = True
                    return True
                if len(self.invoked_skills) >= self.max_turns:
                    self.early_exit_hit = True
                    return True
    return False

TwoStageRetrieverRuntime

Bases: AgentRuntime

Wrap an AgentRuntime with a Stage-1 lexical pre-filter to top-k candidate skills.

Source code in src/reach/runtime/retriever.py

class TwoStageRetrieverRuntime(AgentRuntime):
    """Wrap an AgentRuntime with a Stage-1 lexical pre-filter to top-k candidate skills."""

    is_dynamic: bool = True

    def __init__(
        self,
        inner: AgentRuntime,
        top_k: int = 8,
        scorer: TextScorer | None = None,
        *,
        cache_outcomes: bool = True,
    ) -> None:
        """Initialize the retriever runtime wrapping an underlying agent runtime."""
        self.inner = inner
        self.top_k = top_k
        self.scorer = scorer
        self.cache_outcomes = cache_outcomes
        self.name = f"retriever-{inner.name}"
        self.settings: RuntimeSettings | None = (
            getattr(inner, "settings", None) or RuntimeSettings()
        )
        self.options = inner.options
        self._all_skills: list[Skill] = []
        self._ranker: TextScorer | None = None
        self._catalog: Catalog | None = None
        self._resident: tuple[str, ...] = ()
        self._thread_local = threading.local()
        self._clone_lock = threading.Lock()
        self._worker_clones: list[AgentRuntime] = []
        self._cache_lock = threading.Lock()
        self._outcome_cache: dict[_RetrieverStage2CacheKey, SelectionOutcome] = {}
        self._scale_call_counts: dict[_RetrieverStage2CacheKey, int] = {}

    def _worker_inner(self) -> AgentRuntime:
        """Return a thread-isolated clone of the underlying inner runtime."""
        worker_rt = getattr(self._thread_local, "inner", None)
        if worker_rt is None:
            worker_rt = self.inner.clone_isolated()
            self._thread_local.inner = worker_rt
            with self._clone_lock:
                self._worker_clones.append(worker_rt)
        return worker_rt

    @override
    def cleanup(self) -> None:
        """Clean up inner runtime and all thread-local worker clones."""
        with self._clone_lock:
            clones = list(self._worker_clones)
            self._worker_clones.clear()
        for clone in clones:
            clone.cleanup()
        self.inner.cleanup()

    @property
    @override
    def model(self) -> str:
        """Return the model name configured on the underlying runtime."""
        return self.inner.model

    @model.setter
    @override
    def model(self, value: str) -> None:
        """Update the model identifier on the underlying runtime."""
        self.inner.model = value
        with self._clone_lock:
            for clone in self._worker_clones:
                clone.model = value

    @property
    @override
    def skills_subpath(self) -> str:
        """Return relative skill directory subpath from the underlying runtime."""
        return self.inner.skills_subpath

    @override
    def fit(self, catalog: Catalog, skills: Iterable[Skill]) -> CatalogFit:
        """Forward catalog fit evaluation to the underlying runtime."""
        return self.inner.fit(catalog, skills)

    @override
    def skill_roots(self, workdir: Path) -> tuple[SkillRoot, ...]:
        """Forward skill roots resolution to the underlying runtime."""
        return self.inner.skill_roots(workdir)

    @override
    def install(self, catalog: Catalog, skills: Iterable[Skill], workdir: Path) -> Path:
        """Index skills for pre-filtering, record residency, and reset per-scale call counts."""
        self._all_skills = list(skills)
        self._catalog = catalog
        self._ranker = self.scorer or Bm25Scorer.from_skills(self._all_skills)
        self._resident = catalog.skills
        with self._cache_lock:
            self._scale_call_counts.clear()
        return workdir

    @override
    def select(
        self,
        query_text: str,
        workdir: Path,
        target_skill: str | None = None,
    ) -> SelectionOutcome:
        """Pre-filter catalog to top-k skills and execute probe in an isolated slot."""
        if not self._all_skills:
            return self.inner.select(query_text, workdir, target_skill=target_skill)

        ranker = self._ranker or Bm25Scorer.from_skills(self._all_skills)
        ranked = ranker.rank_text(query_text, self._all_skills)
        top_names = tuple(name for name, _ in ranked[: self.top_k])
        if not top_names and self._all_skills:
            top_names = tuple(s.name for s in self._all_skills[: self.top_k])

        subset_skills = [s for s in self._all_skills if s.name in top_names]
        cache_key: _RetrieverStage2CacheKey | None = None
        if self.cache_outcomes:
            subset_digest = corpus_digest(subset_skills)
            base_key = _RetrieverStage2CacheKey(
                model=self.model,
                query_text=query_text,
                subset_digest=subset_digest,
                top_names=top_names,
                target_skill=target_skill,
            )
            with self._cache_lock:
                rep_idx = self._scale_call_counts.get(base_key, 0) + 1
                self._scale_call_counts[base_key] = rep_idx
                cache_key = base_key.model_copy(update={"replicate_index": rep_idx})
                cached = self._outcome_cache.get(cache_key)
            if cached is not None:
                return cached

        thread_id = probe_slot_id()
        slot_dir = probe_slot_dir(workdir)
        slot_dir.mkdir(parents=True, exist_ok=True)

        sub_catalog = Catalog(
            id=f"retrieved:{thread_id}",
            mode=CatalogMode.ALL,
            skills=top_names,
        )

        worker_rt = self._worker_inner()
        try:
            worker_rt.install(sub_catalog, subset_skills, slot_dir)
            outcome = worker_rt.select(query_text, slot_dir, target_skill=target_skill)
            final_outcome = outcome.model_copy(update={"observed_catalog": top_names})
            if self.cache_outcomes and cache_key is not None and not final_outcome.error:
                with self._cache_lock:
                    self._outcome_cache[cache_key] = final_outcome
            return final_outcome
        finally:
            if slot_dir.exists():
                shutil.rmtree(slot_dir, ignore_errors=True)

model property writable

model: str

Return the model name configured on the underlying runtime.

skills_subpath property

skills_subpath: str

Return relative skill directory subpath from the underlying runtime.

__init__

__init__(
    inner: AgentRuntime,
    top_k: int = 8,
    scorer: TextScorer | None = None,
    *,
    cache_outcomes: bool = True,
) -> None

Initialize the retriever runtime wrapping an underlying agent runtime.

Source code in src/reach/runtime/retriever.py

def __init__(
    self,
    inner: AgentRuntime,
    top_k: int = 8,
    scorer: TextScorer | None = None,
    *,
    cache_outcomes: bool = True,
) -> None:
    """Initialize the retriever runtime wrapping an underlying agent runtime."""
    self.inner = inner
    self.top_k = top_k
    self.scorer = scorer
    self.cache_outcomes = cache_outcomes
    self.name = f"retriever-{inner.name}"
    self.settings: RuntimeSettings | None = (
        getattr(inner, "settings", None) or RuntimeSettings()
    )
    self.options = inner.options
    self._all_skills: list[Skill] = []
    self._ranker: TextScorer | None = None
    self._catalog: Catalog | None = None
    self._resident: tuple[str, ...] = ()
    self._thread_local = threading.local()
    self._clone_lock = threading.Lock()
    self._worker_clones: list[AgentRuntime] = []
    self._cache_lock = threading.Lock()
    self._outcome_cache: dict[_RetrieverStage2CacheKey, SelectionOutcome] = {}
    self._scale_call_counts: dict[_RetrieverStage2CacheKey, int] = {}

cleanup

cleanup() -> None

Clean up inner runtime and all thread-local worker clones.

Source code in src/reach/runtime/retriever.py

@override
def cleanup(self) -> None:
    """Clean up inner runtime and all thread-local worker clones."""
    with self._clone_lock:
        clones = list(self._worker_clones)
        self._worker_clones.clear()
    for clone in clones:
        clone.cleanup()
    self.inner.cleanup()

fit

fit(
    catalog: Catalog, skills: Iterable[Skill]
) -> CatalogFit

Forward catalog fit evaluation to the underlying runtime.

Source code in src/reach/runtime/retriever.py

@override
def fit(self, catalog: Catalog, skills: Iterable[Skill]) -> CatalogFit:
    """Forward catalog fit evaluation to the underlying runtime."""
    return self.inner.fit(catalog, skills)

install

install(
    catalog: Catalog, skills: Iterable[Skill], workdir: Path
) -> Path

Index skills for pre-filtering, record residency, and reset per-scale call counts.

Source code in src/reach/runtime/retriever.py

@override
def install(self, catalog: Catalog, skills: Iterable[Skill], workdir: Path) -> Path:
    """Index skills for pre-filtering, record residency, and reset per-scale call counts."""
    self._all_skills = list(skills)
    self._catalog = catalog
    self._ranker = self.scorer or Bm25Scorer.from_skills(self._all_skills)
    self._resident = catalog.skills
    with self._cache_lock:
        self._scale_call_counts.clear()
    return workdir

select

select(
    query_text: str,
    workdir: Path,
    target_skill: str | None = None,
) -> SelectionOutcome

Pre-filter catalog to top-k skills and execute probe in an isolated slot.

Source code in src/reach/runtime/retriever.py

@override
def select(
    self,
    query_text: str,
    workdir: Path,
    target_skill: str | None = None,
) -> SelectionOutcome:
    """Pre-filter catalog to top-k skills and execute probe in an isolated slot."""
    if not self._all_skills:
        return self.inner.select(query_text, workdir, target_skill=target_skill)

    ranker = self._ranker or Bm25Scorer.from_skills(self._all_skills)
    ranked = ranker.rank_text(query_text, self._all_skills)
    top_names = tuple(name for name, _ in ranked[: self.top_k])
    if not top_names and self._all_skills:
        top_names = tuple(s.name for s in self._all_skills[: self.top_k])

    subset_skills = [s for s in self._all_skills if s.name in top_names]
    cache_key: _RetrieverStage2CacheKey | None = None
    if self.cache_outcomes:
        subset_digest = corpus_digest(subset_skills)
        base_key = _RetrieverStage2CacheKey(
            model=self.model,
            query_text=query_text,
            subset_digest=subset_digest,
            top_names=top_names,
            target_skill=target_skill,
        )
        with self._cache_lock:
            rep_idx = self._scale_call_counts.get(base_key, 0) + 1
            self._scale_call_counts[base_key] = rep_idx
            cache_key = base_key.model_copy(update={"replicate_index": rep_idx})
            cached = self._outcome_cache.get(cache_key)
        if cached is not None:
            return cached

    thread_id = probe_slot_id()
    slot_dir = probe_slot_dir(workdir)
    slot_dir.mkdir(parents=True, exist_ok=True)

    sub_catalog = Catalog(
        id=f"retrieved:{thread_id}",
        mode=CatalogMode.ALL,
        skills=top_names,
    )

    worker_rt = self._worker_inner()
    try:
        worker_rt.install(sub_catalog, subset_skills, slot_dir)
        outcome = worker_rt.select(query_text, slot_dir, target_skill=target_skill)
        final_outcome = outcome.model_copy(update={"observed_catalog": top_names})
        if self.cache_outcomes and cache_key is not None and not final_outcome.error:
            with self._cache_lock:
                self._outcome_cache[cache_key] = final_outcome
        return final_outcome
    finally:
        if slot_dir.exists():
            shutil.rmtree(slot_dir, ignore_errors=True)

skill_roots

skill_roots(workdir: Path) -> tuple[SkillRoot, ...]

Forward skill roots resolution to the underlying runtime.

Source code in src/reach/runtime/retriever.py

@override
def skill_roots(self, workdir: Path) -> tuple[SkillRoot, ...]:
    """Forward skill roots resolution to the underlying runtime."""
    return self.inner.skill_roots(workdir)

agent_default_model

agent_default_model(
    agent: str, config_path: Path | str | None = None
) -> str | None

Return default model identifier for the specified agent runtime.

Source code in src/reach/config.py

def agent_default_model(agent: str, config_path: Path | str | None = None) -> str | None:
    """Return default model identifier for the specified agent runtime."""
    profiles = agent_profiles(config_path)
    profile = profiles.get(agent)
    if profile is not None and profile.default_model:
        return profile.default_model
    return BUILTIN_AGENT_DEFAULT_MODELS.get(agent)

antigravity_agents

antigravity_agents(
    config_path: Path | str | None = None,
) -> tuple[str, ...]

Return tuple of supported agent runtime names that inherit from AntigravityRuntime.

Source code in src/reach/runtime/__init__.py

def antigravity_agents(config_path: Path | str | None = None) -> tuple[str, ...]:
    """Return tuple of supported agent runtime names that inherit from AntigravityRuntime."""
    return tuple(
        agent
        for agent in known_agents(config_path)
        if (cls := runtime_class(agent)) is not None and issubclass(cls, AntigravityRuntime)
    )

build_runtime

build_runtime(settings: RuntimeSettings) -> AgentRuntime

Instantiate and configure an AgentRuntime from settings.

Source code in src/reach/runtime/__init__.py

def build_runtime(settings: RuntimeSettings) -> AgentRuntime:
    """Instantiate and configure an AgentRuntime from settings."""
    if settings.agent in _AGENT_FACTORIES:
        rt = _AGENT_FACTORIES[settings.agent][0](settings)
    else:
        builtin = _load_builtin_entry(settings.agent)
        if builtin is not None:
            rt = builtin[0](settings)
        else:
            agents = ", ".join(known_agents())
            msg = f"unknown runtime agent {settings.agent!r}; expected one of {agents}"
            raise ValueError(msg)
    if getattr(rt, "settings", None) is None:
        rt.settings = settings
    return rt

build_text_generator

build_text_generator(
    model: str | None = None,
    agent: str | None = None,
    timeout_s: float | None = None,
    options: Mapping[str, Any] | None = None,
) -> BaseTextGenerator[Any]

Construct a TextGenerator instance configured for query drafting or optimization.

Source code in src/reach/runtime/generator.py

def build_text_generator(
    model: str | None = None,
    agent: str | None = None,
    timeout_s: float | None = None,
    options: Mapping[str, Any] | None = None,
) -> BaseTextGenerator[Any]:
    """Construct a TextGenerator instance configured for query drafting or optimization."""
    from reach.config import agent_default_model, default_agent
    from reach.runtime import known_agents

    target_agent = agent or default_agent()
    opts = dict(options or {})
    target_model = (
        model
        or (opts.get("model") if isinstance(opts.get("model"), str) else None)
        or agent_default_model(target_agent)
        or DEFAULT_GEMINI_MODEL
    )
    opts["model"] = target_model

    timeout_int = int(timeout_s) if timeout_s is not None else 300
    gen = _build_agent_generator(target_agent, str(target_model), timeout_int, opts)
    if gen is not None:
        return gen

    if target_agent not in known_agents():
        agents = ", ".join(known_agents())
        msg = f"unknown runtime agent {target_agent!r}; expected one of {agents}"
        raise ValueError(msg)

    msg = f"runtime agent {target_agent!r} does not support text generation"
    raise ValueError(msg)

builtin_tool_names

builtin_tool_names() -> frozenset[str]

Return canonical built-in tool primitives across all supported agent harnesses.

Source code in src/reach/runtime/__init__.py

def builtin_tool_names() -> frozenset[str]:
    """Return canonical built-in tool primitives across all supported agent harnesses."""
    from reach.runtime.claude_code import DEFAULT_DENIED_TOOLS, SKILL_TOOL_NAME

    return (
        frozenset(DEFAULT_DENIED_TOOLS)
        | {SKILL_TOOL_NAME}
        | AntigravityRuntime.ANTIGRAVITY_BUILTIN_TOOLS
        | {"load_skill"}
    )

cli_agents

cli_agents(
    config_path: Path | str | None = None,
) -> tuple[str, ...]

Return tuple of supported agent runtime names that execute via CLI subprocesses.

Source code in src/reach/runtime/__init__.py

def cli_agents(config_path: Path | str | None = None) -> tuple[str, ...]:
    """Return tuple of supported agent runtime names that execute via CLI subprocesses."""
    return tuple(
        agent
        for agent in known_agents(config_path)
        if (opt := options_model(agent)) is not None and issubclass(opt, CliOptions)
    )

find_agent_for_model

find_agent_for_model(
    model: str, config_path: Path | str | None = None
) -> str | None

Dynamically determine which agent runtime supports the given model.

Source code in src/reach/runtime/__init__.py

def find_agent_for_model(model: str, config_path: Path | str | None = None) -> str | None:
    """Dynamically determine which agent runtime supports the given model."""
    if not model or not model.strip():
        return None
    profiles = agent_profiles(config_path)
    if not profiles:
        return None
    pairs = [(name, [m.lower() for m in prof.models]) for name, prof in profiles.items()]
    return _match_agent_model(pairs, model.strip().lower())

known_agents

known_agents(
    config_path: Path | str | None = None,
) -> tuple[str, ...]

Return tuple of supported agent runtime names from configuration.

Source code in src/reach/runtime/__init__.py

def known_agents(config_path: Path | str | None = None) -> tuple[str, ...]:
    """Return tuple of supported agent runtime names from configuration."""
    profiles = agent_profiles(config_path)
    builtins = set(_BUILTIN_AGENTS.keys())
    if profiles:
        return tuple(sorted(set(profiles.keys()) | builtins | set(_AGENT_FACTORIES.keys())))
    return tuple(sorted(builtins | set(_AGENT_FACTORIES.keys())))

options_model

options_model(agent: str) -> type[BaseModel] | None

Return the options schema class corresponding to the named agent.

Source code in src/reach/runtime/__init__.py

def options_model(agent: str) -> type[BaseModel] | None:
    """Return the options schema class corresponding to the named agent."""
    if agent in _AGENT_FACTORIES:
        return _AGENT_FACTORIES[agent][1]
    builtin = _load_builtin_entry(agent)
    if builtin is not None:
        return builtin[1]
    return None

register_agent

register_agent(
    name: str,
    factory: Callable[[RuntimeSettings], AgentRuntime],
    options: type[BaseModel] | None = None,
) -> None

Register a runtime factory and optional options schema for an agent name.

Source code in src/reach/runtime/__init__.py

def register_agent(
    name: str,
    factory: Callable[[RuntimeSettings], AgentRuntime],
    options: type[BaseModel] | None = None,
) -> None:
    """Register a runtime factory and optional options schema for an agent name."""
    _AGENT_FACTORIES[name] = (factory, options)

resolve_options

resolve_options(
    settings: RuntimeSettings,
) -> BaseModel | None

Parse and validate agent-specific options dictionary against its schema.

Source code in src/reach/runtime/__init__.py

def resolve_options(settings: RuntimeSettings) -> BaseModel | None:
    """Parse and validate agent-specific options dictionary against its schema."""
    model = options_model(settings.agent)
    raw = dict(settings.options or {})
    if model is not None:
        if "max_turns" not in raw and getattr(settings, "max_turns", None) is not None:
            raw["max_turns"] = settings.max_turns
        if "early_exit" not in raw and getattr(settings, "early_exit", None) is not None:
            raw["early_exit"] = settings.early_exit
        if "allowed_tools" not in raw and getattr(settings, "allowed_tools", None) is not None:
            raw["allowed_tools"] = settings.allowed_tools
        if (
            "blocked_env_vars" not in raw
            and getattr(settings, "blocked_env_vars", None) is not None
        ):
            raw["blocked_env_vars"] = settings.blocked_env_vars
        return model.model_validate(raw)
    if raw:
        msg = f"{_no_options_reason(settings.agent)}; got {sorted(raw)}"
        raise ValueError(
            msg,
        )
    return None

runtime_class

runtime_class(agent: str) -> type[AgentRuntime] | None

Return the runtime implementation class corresponding to the named agent.

Source code in src/reach/runtime/__init__.py

def runtime_class(agent: str) -> type[AgentRuntime] | None:
    """Return the runtime implementation class corresponding to the named agent."""
    if agent in _AGENT_FACTORIES:
        factory = _AGENT_FACTORIES[agent][0]
        return factory if isinstance(factory, type) and issubclass(factory, AgentRuntime) else None
    builtin = _load_builtin_entry(agent)
    if builtin is not None:
        rt = builtin[0]
        return rt if isinstance(rt, type) and issubclass(rt, AgentRuntime) else None
    return None