feat(providers): pluggable agent providers + Grok (xAI) backend

Add a roboco/llm/providers/ seam — an AgentProvider lifecycle ABC and a
ProviderRegistry keyed by ModelProvider — so the orchestrator can drive
agent backends other than Claude Code.

The first non-Claude backend is GrokProvider for xAI's grok-build-0.1.
xAI is OpenAI-compatible only (no Anthropic-Messages endpoint), so a Grok
agent runs an OpenAI-protocol runtime pointed at https://api.x.ai/v1
rather than the ANTHROPIC_BASE_URL injection the other providers use. It
reuses the orchestrator's existing mount/auth assembly, so it inherits the
same MCP gateway + tool-manifest wiring as every other agent by
construction, and passes its prompt via env (never an argv positional).

The change is purely additive: only GROK routes through the registry;
Anthropic / Ollama Cloud / self-hosted spawns run the existing
_spawn_container path unchanged.

Includes:
- ModelProvider.GROK (migration 038) + a seeded Grok provider row
  (migration 039) + a grok-build-0.1 catalog entry
- GET/PUT /api/providers/grok-key to store the xAI key (Fernet-encrypted,
  reusing the existing provider-key machinery)
- ClaudeCodeProvider reference adapter over the current spawn
- unit tests for the registry, GrokProvider (gateway wiring, no
  ANTHROPIC_* leak, prompt-injection safety, failure paths) and routing

The dedicated roboco-agent-grok image and the exact OpenAI-protocol CLI
invocation are the remaining piece to finalise with xAI.
This commit is contained in:
Renn F
2026-06-18 06:14:27 +02:00
parent 7209edefbe
commit a956083f9f
17 changed files with 1089 additions and 0 deletions
+6
View File
@@ -4,6 +4,12 @@ All notable changes to RoboCo are documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
### Added
- **Pluggable agent providers + a Grok (xAI) backend.** A new `roboco/llm/providers/` seam (an `AgentProvider` lifecycle ABC + a `ProviderRegistry` keyed by `ModelProvider`) lets the orchestrator drive agent backends other than Claude Code. The first is `GrokProvider` for xAI's `grok-build-0.1`: xAI is OpenAI-compatible only (no Anthropic-Messages endpoint), so a Grok agent runs an OpenAI-protocol runtime pointed at `https://api.x.ai/v1` rather than the `ANTHROPIC_BASE_URL` injection the other providers use — and it reuses the orchestrator's existing mount/auth assembly, so it gets the same MCP gateway + tool-manifest wiring as every other agent by construction (and passes its prompt via env, never an argv positional). The change is purely additive: only `GROK` routes through the registry; Anthropic / Ollama Cloud / self-hosted spawns are untouched. Includes the `grok` enum value (migration 038), a seeded Grok provider row (migration 039), a `grok-build-0.1` catalog entry, and `GET/PUT /api/providers/grok-key` to store the xAI key (Fernet-encrypted, reusing the existing provider-key machinery). The dedicated `roboco-agent-grok` image plus the exact OpenAI-protocol CLI invocation are the remaining piece to finalize together with xAI.
## [0.6.0] - 2026-06-17 ## [0.6.0] - 2026-06-17
### Added ### Added
@@ -0,0 +1,33 @@
"""Add 'grok' to the postgres modelprovider enum.
Grok (``ModelProvider.GROK`` — xAI's OpenAI-compatible grok-build-0.1) is a new
agent backend. Seeding its provider row (migration 039) and routing agents to it
requires the postgres ``modelprovider`` enum to carry the value. Mirrors the
enum-add pattern of migration 037; the row seed is split into 039 because a
newly added enum value cannot be used in the same transaction that adds it.
Revision ID: 038_modelprovider_grok
Revises: 037_agentrole_pr_reviewer
Create Date: 2026-06-18
"""
from __future__ import annotations
from alembic import op
revision = "038_modelprovider_grok"
down_revision = "037_agentrole_pr_reviewer"
branch_labels = None
depends_on = None
def upgrade() -> None:
# Unguarded (renders in offline --sql so the enum-migration-parity test
# sees it) and idempotent. PG 16 permits ADD VALUE inside a transaction.
op.execute("ALTER TYPE modelprovider ADD VALUE IF NOT EXISTS 'grok'")
def downgrade() -> None:
# Postgres does not support removing enum values without a destructive
# type recreation. Forward-only by design (see migration 037).
pass
@@ -0,0 +1,60 @@
"""Idempotently seed the Grok (xAI) provider row.
The ``modelprovider`` enum carries ``'grok'`` as of migration 038. This
migration seeds the corresponding ``provider_configs`` row so the Settings UI
can store the xAI key without an extra provisioning call.
The row starts disabled with the public xAI base URL and no key — the operator
sets the key via PUT /api/providers/grok/key, which encrypts it and enables the
provider. ON CONFLICT (name) DO NOTHING keeps this safe to re-run.
Revision ID: 039_seed_grok_provider
Revises: 038_modelprovider_grok
Create Date: 2026-06-18
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
revision = "039_seed_grok_provider"
down_revision = "038_modelprovider_grok"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.execute(
sa.text(
"""
INSERT INTO provider_configs
(id, name, type, base_url, auth_token_encrypted, enabled, created_at)
VALUES
(
gen_random_uuid(),
'Grok (xAI)',
'grok',
'https://api.x.ai/v1',
NULL,
false,
now()
)
ON CONFLICT (name) DO NOTHING
"""
)
)
def downgrade() -> None:
# Drop model_assignments pointing at the Grok row first to avoid a FK
# RESTRICT violation on provider_configs.id.
op.execute(
sa.text(
"DELETE FROM model_assignments "
"WHERE provider_config_id IN ("
" SELECT id FROM provider_configs WHERE name = 'Grok (xAI)'"
")"
)
)
op.execute(sa.text("DELETE FROM provider_configs WHERE name = 'Grok (xAI)'"))
+52
View File
@@ -15,12 +15,14 @@ from roboco.api.deps import CurrentAgentContext, DbSession, require_pm_or_above
from roboco.api.schemas.provider import ( from roboco.api.schemas.provider import (
ApplyModeRequest, ApplyModeRequest,
CatalogEntryResponse, CatalogEntryResponse,
GrokKeyStatus,
ModeResponse, ModeResponse,
OllamaKeyStatus, OllamaKeyStatus,
SelfHostedConfigRequest, SelfHostedConfigRequest,
SelfHostedConfigResponse, SelfHostedConfigResponse,
SelfHostedModelEntry, SelfHostedModelEntry,
SelfHostedTestResponse, SelfHostedTestResponse,
SetGrokKeyRequest,
SetOllamaKeyRequest, SetOllamaKeyRequest,
assignment_to_response, assignment_to_response,
) )
@@ -113,6 +115,56 @@ async def set_ollama_key(
) )
# =============================================================================
# GROK (xAI) API KEY
# =============================================================================
@router.get("/grok-key", response_model=GrokKeyStatus)
async def get_grok_key_status(
db: DbSession,
agent: CurrentAgentContext,
) -> GrokKeyStatus:
"""Return whether the Grok (xAI) key is set + enabled."""
require_pm_or_above(agent.role, "view the Grok key status")
provider_svc = get_provider_service(db)
providers = await provider_svc.list_providers(include_disabled=True)
grok = next((p for p in providers if p.type == ModelProvider.GROK), None)
if grok is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Grok provider not seeded. Run alembic upgrade head.",
)
return GrokKeyStatus(
has_key=bool(grok.auth_token_encrypted),
enabled=grok.enabled,
)
@router.put("/grok-key", response_model=GrokKeyStatus)
async def set_grok_key(
data: SetGrokKeyRequest,
db: DbSession,
agent: CurrentAgentContext,
) -> GrokKeyStatus:
"""Set or clear the Grok (xAI) API key.
Empty string → clears and disables the provider. Any other value →
Fernet-encrypts + marks enabled. Used against https://api.x.ai/v1.
"""
require_pm_or_above(agent.role, "set the Grok key")
routing = get_model_routing_service(db)
try:
provider = await routing.set_grok_api_key(data.api_key)
except NotFoundError as e:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e
await db.commit()
return GrokKeyStatus(
has_key=bool(provider.auth_token_encrypted),
enabled=provider.enabled,
)
# ============================================================================= # =============================================================================
# SELF-HOSTED (LOCAL) OLLAMA SERVER # SELF-HOSTED (LOCAL) OLLAMA SERVER
# ============================================================================= # =============================================================================
+23
View File
@@ -58,6 +58,29 @@ class SetOllamaKeyRequest(BaseModel):
api_key: str = Field(default="") api_key: str = Field(default="")
# =============================================================================
# GROK (xAI) API KEY
# =============================================================================
class GrokKeyStatus(BaseModel):
"""Whether the Grok (xAI) provider has a stored key."""
has_key: bool
enabled: bool
class SetGrokKeyRequest(BaseModel):
"""Set or clear the Grok (xAI) API key.
Pass an empty string to clear. Pass a non-empty string to save
(encrypted with Fernet) and mark the Grok provider enabled. This is the
standard xAI key used against https://api.x.ai/v1.
"""
api_key: str = Field(default="")
# ============================================================================= # =============================================================================
# SELF-HOSTED (LOCAL) OLLAMA SERVER # SELF-HOSTED (LOCAL) OLLAMA SERVER
# ============================================================================= # =============================================================================
+27
View File
@@ -0,0 +1,27 @@
"""LLM agent providers.
Each provider implements the :class:`AgentProvider` lifecycle for one backend.
The :class:`ProviderRegistry` maps :class:`~roboco.models.base.ModelProvider`
values to provider instances; the orchestrator looks a provider up at spawn time
and falls back to its built-in Claude Code spawn when none is registered.
Backends:
- :class:`ClaudeCodeProvider` Anthropic-protocol Claude Code container (default;
also serves Ollama Cloud / self-hosted via ``ANTHROPIC_BASE_URL`` injection).
- :class:`GrokProvider` xAI ``grok-build-0.1`` over the OpenAI protocol.
"""
from roboco.llm.providers.base import AgentProvider, ProviderError, SpawnResult
from roboco.llm.providers.claude_code import ClaudeCodeProvider
from roboco.llm.providers.grok import GrokProvider
from roboco.llm.providers.registry import ProviderNotRegisteredError, ProviderRegistry
__all__ = [
"AgentProvider",
"ClaudeCodeProvider",
"GrokProvider",
"ProviderError",
"ProviderNotRegisteredError",
"ProviderRegistry",
"SpawnResult",
]
+42
View File
@@ -0,0 +1,42 @@
"""Shared Docker container-lifecycle helpers for providers.
Small, dependency-free wrappers around the ``docker`` CLI so each provider's
``stop`` / ``health_check`` reads the same way. Spawn and remove stay in the
providers (spawn is backend-specific; remove delegates to the orchestrator so
its log-dump-before-remove behaviour is preserved).
"""
from __future__ import annotations
import asyncio
from roboco.llm.providers.base import ProviderError
async def stop_container(name: str, graceful: bool = True) -> None:
"""Stop a container by name/id (``docker stop`` = SIGTERM, ``kill`` = SIGKILL)."""
verb = "stop" if graceful else "kill"
proc = await asyncio.create_subprocess_exec(
"docker",
verb,
name,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.PIPE,
)
_, stderr = await proc.communicate()
if proc.returncode != 0:
raise ProviderError(f"docker {verb} {name} failed: {stderr.decode().strip()}")
async def container_running(name: str) -> bool:
"""Return True if the named container exists and is running."""
proc = await asyncio.create_subprocess_exec(
"docker",
"inspect",
"--format={{.State.Running}}",
name,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
)
stdout, _ = await proc.communicate()
return proc.returncode == 0 and stdout.decode().strip() == "true"
+92
View File
@@ -0,0 +1,92 @@
"""Abstract base class for agent lifecycle providers.
An ``AgentProvider`` encapsulates *how* an agent is spawned, stopped,
health-checked, and removed for one LLM backend. The orchestrator stays
provider-agnostic: it resolves a provider from the :class:`ProviderRegistry`
by the agent's :class:`~roboco.models.base.ModelProvider` and calls these
methods without knowing whether the agent runs as a Claude Code container, an
OpenAI-protocol subprocess, or a remote host.
This is the seam new backends plug into. ``GROK`` (xAI / grok-build-0.1) is the
first OpenAI-protocol provider; see :mod:`roboco.llm.providers.grok`.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pathlib import Path
from roboco.models.runtime import OrchestratorAgentConfig as AgentConfig
@dataclass(frozen=True)
class SpawnResult:
"""Result of a successful agent spawn.
Attributes:
instance_id: Provider-specific handle (container id, PID, ...).
agent_state: Initial state after a successful spawn.
extra: Provider-specific metadata (container name, model, url, ...).
"""
instance_id: str
agent_state: str = "active"
extra: dict[str, object] = field(default_factory=dict)
class ProviderError(Exception):
"""Raised when an agent-lifecycle operation fails inside a provider.
Attributes:
message: Human-readable description.
agent_id: The agent being operated on, if known.
cause: The original exception, if any.
"""
def __init__(
self,
message: str,
agent_id: str | None = None,
cause: BaseException | None = None,
) -> None:
super().__init__(message)
self.message = message
self.agent_id = agent_id
self.cause = cause
class AgentProvider(ABC):
"""Abstract base for an agent-lifecycle backend.
Every concrete provider implements the full lifecycle so the orchestrator
can drive any backend through one interface.
"""
@abstractmethod
async def spawn(
self,
config: AgentConfig,
initial_prompt: str | None = None,
agent_settings_path: Path | None = None,
) -> SpawnResult:
"""Spawn an agent instance and return a handle to it."""
...
@abstractmethod
async def stop(self, instance_id: str, graceful: bool = True) -> None:
"""Stop a running instance (graceful shutdown unless ``graceful=False``)."""
...
@abstractmethod
async def health_check(self, instance_id: str) -> bool:
"""Return True if the instance is still alive."""
...
@abstractmethod
async def remove(self, instance_id: str) -> None:
"""Remove the instance, releasing all resources."""
...
+92
View File
@@ -0,0 +1,92 @@
"""Claude Code provider — the default Anthropic-protocol Docker backend.
This is a thin adapter over the orchestrator's existing Docker spawn
(``_spawn_container`` / ``_remove_container``). It exists so the registry has a
first-class provider for the Claude Code runtime and so new backends
(:mod:`roboco.llm.providers.grok`) have a reference to mirror.
It deliberately *delegates* to the orchestrator rather than copying ~hundreds of
lines of mount/auth/CLI assembly. Moving that body into this class is the job of
the separate orchestrator-decomposition refactor; keeping it delegated here
means this seam adds the abstraction without destabilising the live spawn path.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Protocol
from roboco.llm.providers._docker import container_running, stop_container
from roboco.llm.providers.base import AgentProvider, ProviderError, SpawnResult
if TYPE_CHECKING:
from pathlib import Path
from roboco.models.runtime import OrchestratorAgentConfig as AgentConfig
class _ClaudeCodeHost(Protocol):
"""The slice of the orchestrator that ``ClaudeCodeProvider`` delegates to.
Typing against a Protocol (rather than importing ``AgentOrchestrator``)
keeps this module import-cycle-free and trivially mockable in tests.
"""
async def _spawn_container(
self,
config: AgentConfig,
initial_prompt: str | None = ...,
agent_settings_path: Path | None = ...,
) -> str: ...
async def _remove_container(self, container_name: str) -> None: ...
def _container_name(agent_id: str) -> str:
"""The container name the orchestrator uses for an agent."""
return f"roboco-agent-{agent_id}"
class ClaudeCodeProvider(AgentProvider):
"""Spawn agents as Claude Code Docker containers (Anthropic protocol).
Non-Anthropic models that *speak the Anthropic Messages API* (Ollama Cloud,
self-hosted) also run through this provider they are routed purely by
``ANTHROPIC_BASE_URL`` / ``ANTHROPIC_AUTH_TOKEN`` injection at spawn, which
the orchestrator already handles. Backends that speak a *different* wire
protocol (e.g. OpenAI-compatible xAI) need their own provider.
"""
def __init__(self, host: _ClaudeCodeHost) -> None:
self._host = host
async def spawn(
self,
config: AgentConfig,
initial_prompt: str | None = None,
agent_settings_path: Path | None = None,
) -> SpawnResult:
try:
container_id = await self._host._spawn_container(
config, initial_prompt, agent_settings_path
)
except Exception as exc:
# Re-wrap any spawn failure as a typed ProviderError for callers.
raise ProviderError(
f"Claude Code spawn failed: {exc}",
agent_id=config.agent_id,
cause=exc,
) from exc
return SpawnResult(
instance_id=_container_name(config.agent_id),
extra={"container_id": container_id, "model": config.model},
)
async def stop(self, instance_id: str, graceful: bool = True) -> None:
await stop_container(instance_id, graceful)
async def health_check(self, instance_id: str) -> bool:
return await container_running(instance_id)
async def remove(self, instance_id: str) -> None:
# Delegate so the orchestrator's log-dump-before-remove behaviour is kept.
await self._host._remove_container(instance_id)
+199
View File
@@ -0,0 +1,199 @@
"""Grok provider — xAI ``grok-build-0.1`` as a native OpenAI-protocol agent.
xAI's API is OpenAI-compatible *only* (``https://api.x.ai/v1``) — there is no
native Anthropic-Messages endpoint so a Grok agent cannot run through the
Claude Code path (which speaks the Anthropic Messages API via
``ANTHROPIC_BASE_URL`` injection). Instead it runs an OpenAI-protocol agent CLI
(the OpenCode pattern), pointed at ``api.x.ai/v1`` with the operator's xAI key.
Design this provider deliberately **reuses the orchestrator's proven container
assembly** (``_build_mount_args`` / ``_append_agent_auth_env`` /
``_append_git_context_env``). That means a Grok agent gets the *same* RoboCo MCP
gateway wiring as every other agent **by construction**:
* ``/app/mcp-config.json`` (the ``roboco-flow`` / ``roboco-do`` gateway) and
``/app/system-prompt.md`` are mounted by ``_build_mount_args``;
* the spawn manifest + ``ROBOCO_GATEWAY_ENABLED`` are set there too;
* the agent HMAC identity is injected by ``_append_agent_auth_env``.
An OpenAI-protocol provider that built its own bespoke spawn would have to
re-wire the MCP config itself, and skipping it leaves agents with zero gateway
verbs. Reusing the shared assembly makes the gateway wiring non-optional here.
Only two things differ from the Claude Code spawn:
1. **LLM env** ``OPENAI_BASE_URL`` / ``OPENAI_API_KEY`` (xAI) instead of the
``ANTHROPIC_*`` injection. The provider routing fields are blanked before
the shared mount step so the xAI endpoint is never mislabelled as Anthropic.
2. **Runtime** the ``roboco-agent-grok`` image, whose entrypoint launches the
OpenAI-protocol CLI from the env below (model, MCP config, system prompt,
prompt). The image + the exact CLI invocation are the one remaining piece to
finalise together with xAI ("review + help implement + test").
The initial prompt is passed via an **env var, not a positional CLI arg**, which
structurally avoids a flag-injection vector: a prompt starting with ``--`` passed
as a positional argument could otherwise be parsed as CLI options.
"""
from __future__ import annotations
import asyncio
import dataclasses
import os
from typing import TYPE_CHECKING, Protocol
from roboco.llm.providers._docker import container_running, stop_container
from roboco.llm.providers.base import AgentProvider, ProviderError, SpawnResult
if TYPE_CHECKING:
from pathlib import Path
from roboco.models.runtime import OrchestratorAgentConfig as AgentConfig
# The Grok agent image (own image, like every other agent role). Built as the
# infra follow-up — bundles the OpenAI-protocol agent CLI + an entrypoint that
# reads the env contract below. Overridable for tests / staged rollout.
_DEFAULT_GROK_IMAGE = os.environ.get(
"ROBOCO_GROK_AGENT_IMAGE", "roboco-agent-grok:latest"
)
# Default xAI endpoint when the seeded provider row carries no base_url.
_DEFAULT_XAI_BASE_URL = "https://api.x.ai/v1"
# In-container paths mounted by the orchestrator's `_build_mount_args`.
_MCP_CONFIG_IN_CONTAINER = "/app/mcp-config.json"
_SYSTEM_PROMPT_IN_CONTAINER = "/app/system-prompt.md"
# Mirrors the Claude Code tool set. The grok image entrypoint applies this to
# the OpenAI-protocol CLI via the recognised `--tools` flag (not --allowed-tools).
_DEFAULT_TOOLS = "Read,Write,Edit,Bash,Grep,Glob,TodoWrite"
def _container_name(agent_id: str) -> str:
return f"roboco-agent-{agent_id}"
class _GrokHost(Protocol):
"""The orchestrator surface GrokProvider reuses for container assembly.
Typed as a Protocol so this module never imports ``AgentOrchestrator``
(no import cycle) and is trivially mockable in tests.
"""
async def _remove_container(self, container_name: str) -> None: ...
def _resolve_host_paths(
self, config: AgentConfig, agent_settings_path: Path | None
) -> dict[str, str | None]: ...
def _build_mount_args(
self,
container_name: str,
config: AgentConfig,
hosts: dict[str, str | None],
) -> list[str]: ...
def _append_agent_auth_env(self, cmd: list[str], config: AgentConfig) -> None: ...
def _append_git_context_env(self, cmd: list[str], config: AgentConfig) -> None: ...
class GrokProvider(AgentProvider):
"""Spawn a Grok (xAI, OpenAI-protocol) agent as a gateway-wired container."""
def __init__(self, host: _GrokHost, image: str | None = None) -> None:
self._host = host
self._image = image or _DEFAULT_GROK_IMAGE
async def spawn(
self,
config: AgentConfig,
initial_prompt: str | None = None,
agent_settings_path: Path | None = None,
) -> SpawnResult:
if not config.provider_auth_token:
raise ProviderError(
"GROK spawn requires an xAI API key — set the Grok provider key "
"in Settings (PUT /api/providers/grok/key).",
agent_id=config.agent_id,
)
if not config.mcp_config_path:
raise ProviderError(
"GROK spawn requires an MCP config (gateway access).",
agent_id=config.agent_id,
)
container_name = _container_name(config.agent_id)
await self._host._remove_container(container_name)
# Reuse the orchestrator's mount/auth/git assembly so the agent gets the
# full MCP gateway + identity wiring. Blank the provider routing fields
# first: otherwise the shared builder would inject the xAI endpoint as
# ANTHROPIC_BASE_URL/ANTHROPIC_AUTH_TOKEN (wrong protocol).
mount_config = dataclasses.replace(
config, provider_base_url=None, provider_auth_token=None
)
hosts = self._host._resolve_host_paths(config, agent_settings_path)
cmd = self._host._build_mount_args(container_name, mount_config, hosts)
self._host._append_agent_auth_env(cmd, config)
self._host._append_git_context_env(cmd, config)
self._append_grok_env(cmd, config, initial_prompt)
cmd.append(self._image)
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
if proc.returncode != 0:
raise ProviderError(
f"Failed to start Grok container: {stderr.decode().strip()}",
agent_id=config.agent_id,
)
return SpawnResult(
instance_id=container_name,
extra={"container_id": stdout.decode().strip(), "model": config.model},
)
def _append_grok_env(
self, cmd: list[str], config: AgentConfig, initial_prompt: str | None
) -> None:
"""Append the OpenAI-protocol (xAI) env contract the grok image consumes.
The prompt travels as an env var, never an argv positional, so a prompt
beginning with ``--`` cannot be parsed as a CLI flag.
"""
base_url = config.provider_base_url or _DEFAULT_XAI_BASE_URL
cmd.extend(
[
# OpenAI-compatible client config (standard env the CLI reads).
"-e",
f"OPENAI_BASE_URL={base_url}",
"-e",
f"OPENAI_API_KEY={config.provider_auth_token}",
# Operational inputs for the grok image entrypoint.
"-e",
f"ROBOCO_AGENT_MODEL={config.model}",
"-e",
f"ROBOCO_MCP_CONFIG={_MCP_CONFIG_IN_CONTAINER}",
"-e",
f"ROBOCO_SYSTEM_PROMPT={_SYSTEM_PROMPT_IN_CONTAINER}",
"-e",
f"ROBOCO_AGENT_TOOLS={_DEFAULT_TOOLS}",
"-e",
f"ROBOCO_INITIAL_PROMPT={initial_prompt or ''}",
]
)
if config.claude_session_id:
# Reused as the generic agent session id so the transcript stays
# locatable at finalize, exactly as on the Claude Code path.
cmd.extend(["-e", f"ROBOCO_AGENT_SESSION_ID={config.claude_session_id}"])
async def stop(self, instance_id: str, graceful: bool = True) -> None:
await stop_container(instance_id, graceful)
async def health_check(self, instance_id: str) -> bool:
return await container_running(instance_id)
async def remove(self, instance_id: str) -> None:
await self._host._remove_container(instance_id)
+73
View File
@@ -0,0 +1,73 @@
"""Provider registry — maps :class:`ModelProvider` values to provider instances.
Usage::
registry = ProviderRegistry()
registry.register(ModelProvider.GROK, GrokProvider(...))
provider = registry.get(ModelProvider.GROK)
result = await provider.spawn(config, initial_prompt)
The orchestrator builds the registry once at startup and calls ``get()`` for any
provider that has a dedicated backend. Providers that are *not* registered fall
back to the orchestrator's built-in Claude Code container spawn — so adding a new
backend is additive and never destabilises the existing Anthropic / Ollama /
self-hosted paths.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from roboco.models.base import ModelProvider
if TYPE_CHECKING:
from roboco.llm.providers.base import AgentProvider
class ProviderNotRegisteredError(LookupError):
"""Raised when no provider is registered for a ``ModelProvider`` value."""
def __init__(self, provider_type: ModelProvider) -> None:
self.provider_type = provider_type
super().__init__(
f"No provider registered for {provider_type.value!r}. "
f"Registered: {[p.value for p in ModelProvider]}"
)
class ProviderRegistry:
"""Maps ``ModelProvider`` values to ``AgentProvider`` instances."""
def __init__(self) -> None:
self._providers: dict[ModelProvider, AgentProvider] = {}
def register(self, provider_type: ModelProvider, provider: AgentProvider) -> None:
"""Register *provider* for *provider_type* (replaces any existing)."""
self._providers[provider_type] = provider
def get(self, provider_type: ModelProvider) -> AgentProvider:
"""Return the provider for *provider_type* or raise."""
provider = self._providers.get(provider_type)
if provider is None:
raise ProviderNotRegisteredError(provider_type)
return provider
def get_or_none(self, provider_type: ModelProvider) -> AgentProvider | None:
"""Return the provider for *provider_type*, or ``None`` if unregistered.
The orchestrator uses this to decide whether a dedicated backend
exists; ``None`` means "use the built-in Claude Code spawn".
"""
return self._providers.get(provider_type)
def is_registered(self, provider_type: ModelProvider) -> bool:
"""Return True if a provider is registered for *provider_type*."""
return provider_type in self._providers
def registered_types(self) -> list[ModelProvider]:
"""Return all registered provider types."""
return list(self._providers.keys())
def unregister(self, provider_type: ModelProvider) -> None:
"""Remove a provider registration (no-op if absent)."""
self._providers.pop(provider_type, None)
+6
View File
@@ -194,6 +194,11 @@ class ModelProvider(StrEnum):
`LOCAL` is the self-hosted Ollama provider: the operator configures its `LOCAL` is the self-hosted Ollama provider: the operator configures its
base URL via PUT /api/providers/self-hosted (seeded by migration 028). base URL via PUT /api/providers/self-hosted (seeded by migration 028).
Agents assigned to LOCAL are routed to that server at spawn time. Agents assigned to LOCAL are routed to that server at spawn time.
`GROK` is xAI's OpenAI-compatible provider (grok-build-0.1 at
https://api.x.ai/v1). Unlike the others it does NOT speak the Anthropic
Messages API, so GROK agents run through a dedicated OpenAI-protocol
provider (roboco.llm.providers.grok), not ANTHROPIC_BASE_URL injection.
The xAI key is set via PUT /api/providers/grok/key.
`OPENAI` is reserved for future use. `OPENAI` is reserved for future use.
""" """
@@ -201,6 +206,7 @@ class ModelProvider(StrEnum):
OLLAMA_CLOUD = "ollama_cloud" OLLAMA_CLOUD = "ollama_cloud"
OPENAI = "openai" OPENAI = "openai"
LOCAL = "local" LOCAL = "local"
GROK = "grok"
class AssignmentScope(StrEnum): class AssignmentScope(StrEnum):
+4
View File
@@ -72,6 +72,10 @@ MODEL_CATALOG: tuple[CatalogEntry, ...] = (
CatalogEntry("glm-5.1:cloud", ModelProvider.OLLAMA_CLOUD, "GLM 5.1"), CatalogEntry("glm-5.1:cloud", ModelProvider.OLLAMA_CLOUD, "GLM 5.1"),
CatalogEntry("kimi-k2.6:cloud", ModelProvider.OLLAMA_CLOUD, "Kimi K2.6"), CatalogEntry("kimi-k2.6:cloud", ModelProvider.OLLAMA_CLOUD, "Kimi K2.6"),
CatalogEntry("minimax-m3:cloud", ModelProvider.OLLAMA_CLOUD, "Minimax M3"), CatalogEntry("minimax-m3:cloud", ModelProvider.OLLAMA_CLOUD, "Minimax M3"),
# --- Grok (xAI, OpenAI protocol) ---
# Routes to the GROK provider → GrokProvider spawn (api.x.ai/v1). The xAI
# key is set via PUT /api/providers/grok/key.
CatalogEntry("grok-build-0.1", ModelProvider.GROK, "Grok Build 0.1"),
) )
+46
View File
@@ -31,6 +31,7 @@ if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from roboco.llm.providers import AgentProvider, ProviderRegistry
from roboco.services.llm import AgentRoute from roboco.services.llm import AgentRoute
from roboco.services.task import TaskService from roboco.services.task import TaskService
import structlog import structlog
@@ -624,6 +625,12 @@ class AgentOrchestrator:
self._strategy_engine_task: asyncio.Task | None = None self._strategy_engine_task: asyncio.Task | None = None
self._external_pr_poll_task: asyncio.Task | None = None self._external_pr_poll_task: asyncio.Task | None = None
self._self_heal_task: asyncio.Task | None = None self._self_heal_task: asyncio.Task | None = None
# Provider registry: maps a ModelProvider to a dedicated AgentProvider
# backend. Only providers needing a non-Claude-Code runtime are
# registered (currently GROK, which speaks the OpenAI protocol). Agents
# on unregistered providers (Anthropic / Ollama Cloud / self-hosted) use
# the built-in _spawn_container path unchanged. Built lazily.
self._provider_registry: ProviderRegistry | None = None
# Tracks which providers have already received a CEO notification # Tracks which providers have already received a CEO notification
# during the current rate-limit episode. Cleared when the probe # during the current rate-limit episode. Cleared when the probe
# succeeds and the rate limit is lifted (tracker.clear() path). # succeeds and the rate limit is lifted (tracker.clear() path).
@@ -1954,6 +1961,36 @@ class AgentOrchestrator:
"""Return the string to pass to `claude --model`.""" """Return the string to pass to `claude --model`."""
return _resolve_agent_cli_model(config.provider_type, config.model) return _resolve_agent_cli_model(config.provider_type, config.model)
def _ensure_provider_registry(self) -> "ProviderRegistry":
"""Build (once) the registry of dedicated provider backends.
Only providers that need a runtime other than the built-in Claude Code
container are registered. Today that is GROK (xAI, OpenAI protocol).
"""
if self._provider_registry is None:
from roboco.llm.providers import GrokProvider, ProviderRegistry
from roboco.models.base import ModelProvider
registry = ProviderRegistry()
registry.register(ModelProvider.GROK, GrokProvider(self))
self._provider_registry = registry
return self._provider_registry
def _provider_for(self, provider_type: str) -> "AgentProvider | None":
"""Resolve a dedicated provider for a route's ``provider_type`` string.
Returns ``None`` for providers that use the built-in Claude Code spawn
(Anthropic / Ollama Cloud / self-hosted) or any unrecognised value the
caller then runs the existing container path unchanged.
"""
from roboco.models.base import ModelProvider
try:
model_provider = ModelProvider(provider_type)
except ValueError:
return None
return self._ensure_provider_registry().get_or_none(model_provider)
async def _spawn_container( async def _spawn_container(
self, self,
config: AgentConfig, config: AgentConfig,
@@ -1967,6 +2004,15 @@ class AgentOrchestrator:
initial_prompt: Optional initial prompt for the agent initial_prompt: Optional initial prompt for the agent
agent_settings_path: Path to per-agent Claude settings file agent_settings_path: Path to per-agent Claude settings file
""" """
# A dedicated provider backend (e.g. GROK / OpenAI protocol) handles its
# own spawn. Anthropic / Ollama Cloud / self-hosted have no dedicated
# provider registered and fall through to the Claude Code body below,
# byte-for-byte unchanged.
provider = self._provider_for(config.provider_type)
if provider is not None:
result = await provider.spawn(config, initial_prompt, agent_settings_path)
return result.instance_id
container_name = f"roboco-agent-{config.agent_id}" container_name = f"roboco-agent-{config.agent_id}"
await self._remove_container(container_name) await self._remove_container(container_name)
+21
View File
@@ -353,6 +353,27 @@ class ModelRoutingService(BaseService):
# Re-fetch for the caller. # Re-fetch for the caller.
return await self._get_seeded_provider(ModelProvider.OLLAMA_CLOUD) return await self._get_seeded_provider(ModelProvider.OLLAMA_CLOUD)
async def set_grok_api_key(self, api_key: str) -> ProviderConfigTable:
"""Set / clear the Grok (xAI) provider's API key.
Empty string clears + disables; a real key encrypts + enables.
Operates on the single pre-seeded Grok row no provider creation
happens here. The key is the standard xAI key used against
https://api.x.ai/v1.
"""
provider = await self._get_seeded_provider(ModelProvider.GROK)
provider_svc = ProviderService(self.session)
await provider_svc.update_provider(
require_uuid(provider.id),
ProviderUpdate(
auth_token=api_key if api_key else None,
clear_auth_token=not api_key,
enabled=bool(api_key),
),
)
# Re-fetch for the caller.
return await self._get_seeded_provider(ModelProvider.GROK)
async def _get_seeded_provider( async def _get_seeded_provider(
self, provider_type: ModelProvider self, provider_type: ModelProvider
) -> ProviderConfigTable: ) -> ProviderConfigTable:
+272
View File
@@ -0,0 +1,272 @@
"""Tests for the LLM agent provider seam.
Covers the ProviderRegistry, the ClaudeCodeProvider adapter, and the
GrokProvider (xAI / OpenAI protocol) especially the safety properties an
OpenAI-protocol agent provider must hold:
* the agent gets the MCP gateway wiring (reuses the orchestrator mount path);
* the xAI endpoint is injected as OPENAI_* and never mislabelled ANTHROPIC_*;
* the prompt travels via env, so a leading ``--`` cannot become a CLI flag.
"""
from __future__ import annotations
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from roboco.llm.providers import (
ClaudeCodeProvider,
GrokProvider,
ProviderError,
ProviderNotRegisteredError,
ProviderRegistry,
SpawnResult,
)
from roboco.models.base import ModelProvider
from roboco.models.runtime import OrchestratorAgentConfig
def _config(
*,
provider_type: str = "grok",
provider_base_url: str | None = "https://api.x.ai/v1",
provider_auth_token: str | None = "xai-secret-key",
mcp_config_path: Path | None = Path("/host/mcp-configs/be-dev-1.json"),
) -> OrchestratorAgentConfig:
return OrchestratorAgentConfig(
agent_id="be-dev-1",
blueprint_path=Path("/app/system-prompt.md"),
model="grok-build-0.1",
mcp_config_path=mcp_config_path,
claude_session_id="sess-1",
provider_type=provider_type,
provider_base_url=provider_base_url,
provider_auth_token=provider_auth_token,
)
class _FakeHost:
"""Implements the orchestrator surface the providers delegate to."""
def __init__(self) -> None:
self.removed: list[str] = []
self.spawn_args: tuple[object, ...] | None = None
self.mount_config: OrchestratorAgentConfig | None = None
async def _spawn_container(
self,
config: OrchestratorAgentConfig,
initial_prompt: str | None = None,
agent_settings_path: Path | None = None,
) -> str:
self.spawn_args = (config, initial_prompt, agent_settings_path)
return "container-id-abc123"
async def _remove_container(self, container_name: str) -> None:
self.removed.append(container_name)
def _resolve_host_paths(
self, config: OrchestratorAgentConfig, agent_settings_path: Path | None
) -> dict[str, str | None]:
return {
"mcp_config": str(config.mcp_config_path)
if config.mcp_config_path
else None,
"settings": str(agent_settings_path) if agent_settings_path else None,
}
def _build_mount_args(
self,
container_name: str,
config: OrchestratorAgentConfig,
hosts: dict[str, str | None],
) -> list[str]:
# Record the config the mount step saw, and MIMIC the real
# _append_provider_env so a missed blanking would leak ANTHROPIC_*.
self.mount_config = config
cmd = ["docker", "run", "-d", "--name", container_name]
mcp = hosts.get("mcp_config")
if mcp:
cmd += ["-v", f"{mcp}:/app/mcp-config.json:ro"]
if config.provider_base_url:
cmd += ["-e", f"ANTHROPIC_BASE_URL={config.provider_base_url}"]
if config.provider_auth_token:
cmd += ["-e", f"ANTHROPIC_AUTH_TOKEN={config.provider_auth_token}"]
return cmd
def _append_agent_auth_env(
self, cmd: list[str], config: OrchestratorAgentConfig
) -> None:
cmd += ["-e", f"ROBOCO_AGENT_TOKEN=hmac-{config.agent_id}"]
def _append_git_context_env(
self, cmd: list[str], config: OrchestratorAgentConfig
) -> None:
cmd += ["-e", f"ROBOCO_GIT_AGENT={config.agent_id}"]
def _proc(
returncode: int = 0, stdout: bytes = b"cid\n", stderr: bytes = b""
) -> MagicMock:
proc = MagicMock()
proc.returncode = returncode
proc.communicate = AsyncMock(return_value=(stdout, stderr))
return proc
# ---------------------------------------------------------------------------
# ProviderRegistry
# ---------------------------------------------------------------------------
def test_registry_register_and_get() -> None:
registry = ProviderRegistry()
provider = GrokProvider(_FakeHost())
registry.register(ModelProvider.GROK, provider)
assert registry.get(ModelProvider.GROK) is provider
assert registry.is_registered(ModelProvider.GROK)
assert registry.registered_types() == [ModelProvider.GROK]
def test_registry_get_unregistered_raises() -> None:
registry = ProviderRegistry()
with pytest.raises(ProviderNotRegisteredError):
registry.get(ModelProvider.GROK)
def test_registry_get_or_none_returns_none_when_absent() -> None:
registry = ProviderRegistry()
assert registry.get_or_none(ModelProvider.ANTHROPIC) is None
def test_registry_unregister() -> None:
registry = ProviderRegistry()
registry.register(ModelProvider.GROK, GrokProvider(_FakeHost()))
registry.unregister(ModelProvider.GROK)
assert not registry.is_registered(ModelProvider.GROK)
registry.unregister(ModelProvider.GROK) # idempotent
# ---------------------------------------------------------------------------
# GrokProvider
# ---------------------------------------------------------------------------
async def test_grok_spawn_requires_api_key() -> None:
provider = GrokProvider(_FakeHost())
with pytest.raises(ProviderError, match="xAI API key"):
await provider.spawn(_config(provider_auth_token=None))
async def test_grok_spawn_requires_mcp_config() -> None:
provider = GrokProvider(_FakeHost())
with pytest.raises(ProviderError, match="MCP config"):
await provider.spawn(_config(mcp_config_path=None))
async def test_grok_spawn_injects_openai_env_and_no_anthropic_leak() -> None:
host = _FakeHost()
provider = GrokProvider(host, image="roboco-agent-grok:test")
with patch(
"asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
) as exec_mock:
await provider.spawn(_config(), initial_prompt="do the work")
cmd = list(exec_mock.call_args.args)
assert "OPENAI_BASE_URL=https://api.x.ai/v1" in cmd
assert "OPENAI_API_KEY=xai-secret-key" in cmd
# The xAI endpoint must NOT be injected as an Anthropic var.
assert not any(c.startswith("ANTHROPIC_BASE_URL=") for c in cmd)
assert not any(c.startswith("ANTHROPIC_AUTH_TOKEN=") for c in cmd)
# Provider fields were blanked before the shared mount step.
assert host.mount_config is not None
assert host.mount_config.provider_base_url is None
assert host.mount_config.provider_auth_token is None
async def test_grok_spawn_wires_gateway_and_image_last() -> None:
host = _FakeHost()
provider = GrokProvider(host, image="roboco-agent-grok:test")
with patch(
"asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
) as exec_mock:
result = await provider.spawn(_config())
cmd = list(exec_mock.call_args.args)
# Gateway + operational env the grok image entrypoint consumes.
assert "ROBOCO_MCP_CONFIG=/app/mcp-config.json" in cmd
assert "ROBOCO_SYSTEM_PROMPT=/app/system-prompt.md" in cmd
assert "ROBOCO_AGENT_TOOLS=Read,Write,Edit,Bash,Grep,Glob,TodoWrite" in cmd
# Identity wiring from the shared host helpers is present.
assert "ROBOCO_AGENT_TOKEN=hmac-be-dev-1" in cmd
# The image is the final docker-run argument.
assert cmd[-1] == "roboco-agent-grok:test"
assert host.removed == ["roboco-agent-be-dev-1"]
assert result == SpawnResult(
instance_id="roboco-agent-be-dev-1",
extra={"container_id": "cid", "model": "grok-build-0.1"},
)
async def test_grok_spawn_prompt_is_injection_safe() -> None:
host = _FakeHost()
provider = GrokProvider(host)
nasty = "--model evil --session-id pwned"
with patch(
"asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
) as exec_mock:
await provider.spawn(_config(), initial_prompt=nasty)
cmd = list(exec_mock.call_args.args)
# Passed only as an env value, never as a bare argv token.
assert f"ROBOCO_INITIAL_PROMPT={nasty}" in cmd
assert nasty not in cmd
async def test_grok_spawn_defaults_base_url_when_route_blank() -> None:
provider = GrokProvider(_FakeHost())
with patch(
"asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
) as exec_mock:
await provider.spawn(_config(provider_base_url=None))
cmd = list(exec_mock.call_args.args)
assert "OPENAI_BASE_URL=https://api.x.ai/v1" in cmd
async def test_grok_spawn_raises_on_docker_failure() -> None:
provider = GrokProvider(_FakeHost())
with (
patch(
"asyncio.create_subprocess_exec",
AsyncMock(return_value=_proc(returncode=1, stderr=b"boom")),
),
pytest.raises(ProviderError, match="boom"),
):
await provider.spawn(_config())
# ---------------------------------------------------------------------------
# ClaudeCodeProvider
# ---------------------------------------------------------------------------
async def test_claude_spawn_delegates_to_host() -> None:
host = _FakeHost()
provider = ClaudeCodeProvider(host)
result = await provider.spawn(_config(provider_type="anthropic"), "prompt")
assert host.spawn_args is not None
assert result.instance_id == "roboco-agent-be-dev-1"
assert result.extra["container_id"] == "container-id-abc123"
async def test_claude_spawn_wraps_host_error() -> None:
host = _FakeHost()
host._spawn_container = AsyncMock(side_effect=RuntimeError("docker down")) # type: ignore[method-assign]
provider = ClaudeCodeProvider(host)
with pytest.raises(ProviderError, match="docker down"):
await provider.spawn(_config())
async def test_claude_remove_delegates_to_host() -> None:
host = _FakeHost()
provider = ClaudeCodeProvider(host)
await provider.remove("roboco-agent-be-dev-1")
assert host.removed == ["roboco-agent-be-dev-1"]
@@ -0,0 +1,41 @@
"""The orchestrator routes only dedicated-backend providers through the registry.
GROK gets the GrokProvider; Anthropic / Ollama Cloud / self-hosted (and any
unknown value) return None so ``_spawn_container`` runs its built-in Claude Code
path unchanged. This keeps the GROK addition purely additive.
"""
from __future__ import annotations
from unittest.mock import patch
from roboco.llm.providers import GrokProvider
from roboco.runtime.orchestrator import AgentOrchestrator
def _make_orch() -> AgentOrchestrator:
with patch.object(AgentOrchestrator, "__init__", return_value=None):
orch = AgentOrchestrator.__new__(AgentOrchestrator)
orch._provider_registry = None
return orch
def test_provider_for_grok_returns_grok_provider() -> None:
assert isinstance(_make_orch()._provider_for("grok"), GrokProvider)
def test_provider_for_anthropic_returns_none() -> None:
assert _make_orch()._provider_for("anthropic") is None
def test_provider_for_ollama_cloud_returns_none() -> None:
assert _make_orch()._provider_for("ollama_cloud") is None
def test_provider_for_unknown_value_returns_none() -> None:
assert _make_orch()._provider_for("bogus") is None
def test_provider_registry_built_once() -> None:
orch = _make_orch()
assert orch._ensure_provider_registry() is orch._ensure_provider_registry()