External/roboco
1
0
Fork 0
mirror of https://github.com/rennf93/roboco.git synced 2026-08-03 07:23:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(grok): route interactive intake/secretary to opencode-serve images

Browse Source 
Wire the GROK interactive path the in-place way (matching how the interactive
roles already choose ANTHROPIC_* per route), so a GROK route launches the
Grok-native opencode-serve image instead of the Claude SDK-driver image:

- _spawn_intake_container / _spawn_secretary_container pick the
  grok-prompter / grok-secretary image (ensuring the base→grok→interactive
  build chain) when the route is GROK, and stamp provider_type on the spec +
  AgentConfig so finalize routes usage to the opencode store.
- _build_intake_run_cmd / _build_secretary_run_cmd inject OPENAI_* + the
  opencode store mount + system-prompt env for GROK via a shared
  _append_interactive_provider_env, keeping ANTHROPIC_* for every other
  provider. The intake's minimal mounts (no gateway MCP) are preserved, so
  Grok intake matches the Claude intake's tool surface (the spec).
- Add a per-agent opencode store mount to the interactive host paths so
  interactive Grok usage/cost is captured like the one-shot path.

Removes the interim Phase-0 routing guard (the real path supersedes it) and
retires the unused AgentProvider.spawn_interactive/InteractiveSpawnSpec seam —
the interactive roles have a bespoke assembly that the one-shot provider
surface doesn't fit, so the fork lives in their own builders.

UNVERIFIED-LIVE: end-to-end intake/secretary chat on Grok needs the stack up +
opencode serve confirmed against grok-build-0.1.
This commit is contained in:
Renn F
2026-06-18 12:32:41 +02:00
parent 3753fbccc4
commit 0d76ebf586
7 changed files with 211 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
-2
View File
@@ -170,8 +170,6 @@ select = [
# signature for keyword-argument compatibility (mypy override check), but the
# stub bodies are empty — ARG002 would require renaming them, which breaks mypy.
"tests/unit/services/test_optimal_grounding.py" = ["ARG002"]
# Minimal AgentProvider stub exercising the interactive default — same reason.
"tests/unit/llm/test_provider_base.py" = ["ARG002"]
# =============================================================================
# MyPy Configuration
roboco/llm/providers/base.py
+2 -42
View File
@@ -38,27 +38,6 @@ class SpawnResult:
extra: dict[str, object] = field(default_factory=dict)
@dataclass(frozen=True)
class InteractiveSpawnSpec:
"""Inputs to spawn a long-lived **interactive** agent (intake / secretary).
Interactive roles run a held-open chat session the human types into, rather
than a one-shot task. This bundles everything a provider's
:meth:`AgentProvider.spawn_interactive` needs so it never imports
orchestrator internals: the resolved :class:`AgentConfig` (carries the
provider routing creds, model, mcp config, agent id), the per-session id the
relay/SSE keys on, the role-specific interactive image, and — for the
secretary, whose directive tools authenticate to the API — the HMAC token.
"""
config: AgentConfig
image: str
session_id: str
role: str # "prompter" | "secretary"
agent_token: str | None = None
agent_settings_path: Path | None = None
class ProviderError(Exception):
"""Raised when an agent-lifecycle operation fails inside a provider.
@@ -83,18 +62,10 @@ class ProviderError(Exception):
class AgentProvider(ABC):
"""Abstract base for an agent-lifecycle backend.
Every concrete provider implements the full one-shot lifecycle so the
orchestrator can drive any backend through one interface. Interactive
(held-open chat) spawns are opt-in: a provider that serves the human-facing
intake/secretary roles sets ``supports_interactive = True`` and overrides
:meth:`spawn_interactive`. One-shot-only providers inherit the default,
which declines — so adding the interactive surface breaks no existing
provider.
Every concrete provider implements the full lifecycle so the orchestrator
can drive any backend through one interface.
"""
#: Whether this backend can spawn long-lived interactive (chat) agents.
supports_interactive: bool = False
@abstractmethod
async def spawn(
self,
@@ -105,17 +76,6 @@ class AgentProvider(ABC):
"""Spawn an agent instance and return a handle to it."""
...
async def spawn_interactive(self, spec: InteractiveSpawnSpec) -> SpawnResult:
"""Spawn a long-lived interactive agent (intake / secretary).
Non-abstract: the default declines so one-shot-only providers need no
change. Providers that set ``supports_interactive = True`` override this.
"""
raise ProviderError(
f"{type(self).__name__} does not support interactive spawns",
agent_id=spec.config.agent_id,
)
@abstractmethod
async def stop(self, instance_id: str, graceful: bool = True) -> None:
"""Stop a running instance (graceful shutdown unless ``graceful=False``)."""
roboco/runtime/orchestrator.py
+116 -14
View File
@@ -169,6 +169,16 @@ DATA_HOST_PATH = os.environ.get("ROBOCO_HOST_DATA_DIR", "")
# Claude transcript is read from the mounted ~/.claude). Override for local runs.
OPENCODE_DATA_DIR = os.environ.get("ROBOCO_OPENCODE_DATA_DIR", "/data/opencode")
# Interactive Grok images (opencode-serve drivers) — selected for the intake /
# secretary roles when their route resolves to GROK, instead of the Claude
# prompter/secretary images. Their dockerfiles build FROM roboco-agent-grok.
GROK_PROMPTER_IMAGE = "roboco-agent-grok-prompter"
GROK_SECRETARY_IMAGE = "roboco-agent-grok-secretary"
_GROK_INTERACTIVE_DOCKERFILES = {
GROK_PROMPTER_IMAGE: "agent-grok-prompter.Dockerfile",
GROK_SECRETARY_IMAGE: "agent-grok-secretary.Dockerfile",
}
# =============================================================================
# ORCHESTRATOR
@@ -199,6 +209,8 @@ class _IntakeRunSpec:
api_url: str
provider_base_url: str | None
provider_auth_token: str | None
provider_type: str = "anthropic"
model: str = ""
@dataclass
@@ -220,6 +232,8 @@ class _SecretaryRunSpec:
agent_token: str
provider_base_url: str | None
provider_auth_token: str | None
provider_type: str = "anthropic"
model: str = ""
def _read_project_slug(task: dict[str, Any]) -> str | None:
@@ -826,6 +840,30 @@ class AgentOrchestrator:
build_context,
)
async def _ensure_grok_interactive_image(self, image: str) -> None:
"""Ensure a Grok interactive image and its base→runtime chain exist.
The grok-prompter / grok-secretary images build FROM roboco-agent-grok,
which builds FROM the agent base, so the whole chain must be present
before a local build of the interactive image can succeed (on the
registry path each is already pulled and this just verifies presence).
"""
if PROJECT_HOST_PATH:
build_context = PROJECT_HOST_PATH
docker_dir = f"{PROJECT_HOST_PATH}/docker"
else:
build_context = str(self.project_root)
docker_dir = str(self.project_root / "docker")
chain = [
(AGENT_BASE_IMAGE, "agent-base.Dockerfile"),
("roboco-agent-grok", "agent-grok.Dockerfile"),
(image, _GROK_INTERACTIVE_DOCKERFILES[image]),
]
for img, dockerfile in chain:
await self._ensure_image_present(
img, f"{docker_dir}/{dockerfile}", build_context
)
async def _ensure_image_present(
self, bare_image: str, dockerfile_path: str, build_context: str
) -> None:
@@ -2914,6 +2952,8 @@ class AgentOrchestrator:
if INTAKE_AGENT_ID in self._instances:
await self.stop_agent(INTAKE_AGENT_ID, graceful=False)
from roboco.models.base import ModelProvider
cwd, cloned = await self._clone_intake_scope(project_slug, product_id)
prompt_path = self._generate_composed_prompt(INTAKE_AGENT_ID)
@@ -2927,14 +2967,21 @@ class AgentOrchestrator:
else f"http://127.0.0.1:{settings.port}"
)
await self._ensure_agent_image(INTAKE_AGENT_ID)
# GROK runs the interactive driver on its own opencode-serve image; every
# other provider uses the Claude SDK-driver prompter image.
is_grok = route.provider_type == ModelProvider.GROK
image = GROK_PROMPTER_IMAGE if is_grok else get_agent_image(INTAKE_AGENT_ID)
if is_grok:
await self._ensure_grok_interactive_image(image)
else:
await self._ensure_agent_image(INTAKE_AGENT_ID)
container_name = f"roboco-agent-{INTAKE_AGENT_ID}"
await self._remove_container(container_name)
cmd = self._build_intake_run_cmd(
_IntakeRunSpec(
container_name=container_name,
image=get_agent_image(INTAKE_AGENT_ID),
image=image,
hosts=self._resolve_intake_host_paths(),
session_id=session_id,
cwd=cwd,
@@ -2942,6 +2989,8 @@ class AgentOrchestrator:
api_url=api_url,
provider_base_url=route.base_url,
provider_auth_token=route.auth_token,
provider_type=route.provider_type.value,
model=route.model_name,
)
)
container_id = await self._run_container_cmd(cmd)
@@ -2951,6 +3000,7 @@ class AgentOrchestrator:
blueprint_path=prompt_path,
model=route.model_name,
git_context=None,
provider_type=route.provider_type.value,
)
instance = AgentInstance(
agent_id=INTAKE_AGENT_ID,
@@ -3057,6 +3107,7 @@ class AgentOrchestrator:
"""
from roboco.agents_config import issue_agent_token
from roboco.foundation.identity import AGENTS
from roboco.models.base import ModelProvider
if SECRETARY_AGENT_ID in self._instances:
await self.stop_agent(SECRETARY_AGENT_ID, graceful=False)
@@ -3072,7 +3123,12 @@ class AgentOrchestrator:
else f"http://127.0.0.1:{settings.port}"
)
await self._ensure_agent_image(SECRETARY_AGENT_ID)
is_grok = route.provider_type == ModelProvider.GROK
image = GROK_SECRETARY_IMAGE if is_grok else get_agent_image(SECRETARY_AGENT_ID)
if is_grok:
await self._ensure_grok_interactive_image(image)
else:
await self._ensure_agent_image(SECRETARY_AGENT_ID)
container_name = f"roboco-agent-{SECRETARY_AGENT_ID}"
await self._remove_container(container_name)
@@ -3080,7 +3136,7 @@ class AgentOrchestrator:
cmd = self._build_secretary_run_cmd(
_SecretaryRunSpec(
container_name=container_name,
image=get_agent_image(SECRETARY_AGENT_ID),
image=image,
hosts=self._resolve_secretary_host_paths(),
session_id=session_id,
cwd="/app",
@@ -3090,6 +3146,8 @@ class AgentOrchestrator:
agent_token=issue_agent_token(agent_uuid, "secretary", ""),
provider_base_url=route.base_url,
provider_auth_token=route.auth_token,
provider_type=route.provider_type.value,
model=route.model_name,
)
)
container_id = await self._run_container_cmd(cmd)
@@ -3099,6 +3157,7 @@ class AgentOrchestrator:
blueprint_path=prompt_path,
model=route.model_name,
git_context=None,
provider_type=route.provider_type.value,
)
instance = AgentInstance(
agent_id=SECRETARY_AGENT_ID,
@@ -3145,6 +3204,7 @@ class AgentOrchestrator:
"prompt": (
f"{DATA_HOST_PATH}/prompts-generated/{SECRETARY_AGENT_ID}-prompt.md"
),
"opencode": f"{DATA_HOST_PATH}/opencode/{SECRETARY_AGENT_ID}",
}
return {
"claude": CLAUDE_AUTH_HOST_PATH,
@@ -3153,6 +3213,9 @@ class AgentOrchestrator:
/ "roboco-prompts"
/ f"{SECRETARY_AGENT_ID}-prompt.md"
),
"opencode": str(
Path(tempfile.gettempdir()) / "roboco-opencode" / SECRETARY_AGENT_ID
),
}
@staticmethod
@@ -3190,10 +3253,7 @@ class AgentOrchestrator:
f"CLAUDE_CODE_SUBAGENT_MODEL={spec.cli_model}",
]
)
if spec.provider_base_url:
cmd.extend(["-e", f"ANTHROPIC_BASE_URL={spec.provider_base_url}"])
if spec.provider_auth_token:
cmd.extend(["-e", f"ANTHROPIC_AUTH_TOKEN={spec.provider_auth_token}"])
AgentOrchestrator._append_interactive_provider_env(cmd, spec)
cmd.append(spec.image)
return cmd
@@ -3261,6 +3321,7 @@ class AgentOrchestrator:
f"{DATA_HOST_PATH}/prompts-generated/{INTAKE_AGENT_ID}-prompt.md"
),
"workspaces": f"{DATA_HOST_PATH}/workspaces",
"opencode": f"{DATA_HOST_PATH}/opencode/{INTAKE_AGENT_ID}",
}
return {
"claude": CLAUDE_AUTH_HOST_PATH,
@@ -3270,8 +3331,52 @@ class AgentOrchestrator:
/ f"{INTAKE_AGENT_ID}-prompt.md"
),
"workspaces": str(Path(settings.workspaces_root)),
"opencode": str(
Path(tempfile.gettempdir()) / "roboco-opencode" / INTAKE_AGENT_ID
),
}
@staticmethod
def _append_interactive_provider_env(
cmd: list[str], spec: "_IntakeRunSpec | _SecretaryRunSpec"
) -> None:
"""Inject the per-provider LLM env for an interactive container.
GROK runs natively on opencode: ``OPENAI_*`` (xAI) + ``ROBOCO_AGENT_MODEL``
+ the mounted system prompt the driver renders ``opencode.json`` from,
plus the per-agent opencode store mount so finalize can read usage back.
Every other provider uses the Claude path's ``ANTHROPIC_*`` injection (or
the mounted ``~/.claude`` default when the route carries no creds).
"""
from roboco.models.base import ModelProvider
base_url = spec.provider_base_url
auth_token = spec.provider_auth_token
if spec.provider_type == ModelProvider.GROK.value:
opencode_host = spec.hosts.get("opencode")
if opencode_host:
# opencode persists usage to opencode.db under its data dir; the
# per-agent host mount lets the finalizer read it (same in-
# container path as the one-shot Grok store).
cmd.extend(["-v", f"{opencode_host}:/home/agent/.local/share/opencode"])
cmd.extend(
[
"-e",
f"OPENAI_BASE_URL={base_url or 'https://api.x.ai/v1'}",
"-e",
f"OPENAI_API_KEY={auth_token or ''}",
"-e",
f"ROBOCO_AGENT_MODEL={spec.model}",
"-e",
"ROBOCO_SYSTEM_PROMPT=/app/system-prompt.md",
]
)
return
if base_url:
cmd.extend(["-e", f"ANTHROPIC_BASE_URL={base_url}"])
if auth_token:
cmd.extend(["-e", f"ANTHROPIC_AUTH_TOKEN={auth_token}"])
@staticmethod
def _build_intake_run_cmd(spec: _IntakeRunSpec) -> list[str]:
"""Compose the `docker run` argv for the persistent intake container.
@@ -3312,12 +3417,9 @@ class AgentOrchestrator:
f"CLAUDE_CODE_SUBAGENT_MODEL={spec.cli_model}",
]
)
# Non-Anthropic providers need explicit endpoint/token; the Anthropic
# default uses the mounted ~/.claude login (same as every agent).
if spec.provider_base_url:
cmd.extend(["-e", f"ANTHROPIC_BASE_URL={spec.provider_base_url}"])
if spec.provider_auth_token:
cmd.extend(["-e", f"ANTHROPIC_AUTH_TOKEN={spec.provider_auth_token}"])
# GROK runs opencode (OPENAI_* + opencode store); other providers use the
# ANTHROPIC_* injection or the mounted ~/.claude default.
AgentOrchestrator._append_interactive_provider_env(cmd, spec)
cmd.append(spec.image)
return cmd
roboco/services/llm.py
+1 -32
View File
@@ -116,12 +116,6 @@ class _ResolvedAssignment:
model_name: str
# The two human-facing interactive roles (held-open chat sessions). Kept as a
# literal here — not imported from runtime.orchestrator — to avoid a service →
# orchestrator import cycle; the slugs are stable seed identities.
_INTERACTIVE_AGENT_SLUGS = frozenset({"intake-1", "secretary-1"})
class ModelRoutingService(BaseService):
"""Resolves per-agent routes from `model_assignments` + legacy fallback."""
@@ -140,34 +134,9 @@ class ModelRoutingService(BaseService):
if resolved is not None and resolved.provider.enabled:
route = await self._route_from_resolved(resolved, agent_slug)
if route is not None:
return self._guard_interactive(route, agent_slug, role)
return route
return self._legacy_route(role)
def _guard_interactive(
self, route: AgentRoute, agent_slug: str, role: str
) -> AgentRoute:
"""Keep the human-facing interactive roles off GROK for now.
intake (prompter) and secretary run a held-open chat the human types
into, driven by the Claude Agent SDK. GROK has no interactive runtime
yet, and the interactive spawn injects the route's creds as
``ANTHROPIC_*`` against ``api.x.ai/v1`` — the wrong protocol — yielding a
silent, empty reply. Until the opencode interactive driver lands,
downgrade a GROK route for these slugs to the Anthropic default. (The
one-shot delivery roles route to GROK unchanged.)
"""
if (
route.provider_type == ModelProvider.GROK
and agent_slug in _INTERACTIVE_AGENT_SLUGS
):
self.log.warning(
"GROK route for an interactive role downgraded to Anthropic "
"(no Grok interactive runtime yet)",
agent_slug=agent_slug,
)
return self._legacy_route(role)
return route
async def _resolve_assignment(
self, agent_slug: str, role: str
) -> _ResolvedAssignment | None:
tests/unit/llm/test_provider_base.py
-55
View File
@@ -1,55 +0,0 @@
"""AgentProvider interactive scaffolding: opt-in, declines by default."""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from roboco.llm.providers.base import (
AgentProvider,
InteractiveSpawnSpec,
ProviderError,
SpawnResult,
)
if TYPE_CHECKING:
from pathlib import Path
from roboco.models.runtime import OrchestratorAgentConfig as AgentConfig
class _OneShotOnly(AgentProvider):
"""A minimal provider that implements only the one-shot lifecycle."""
async def spawn(
self,
config: AgentConfig,
initial_prompt: str | None = None,
agent_settings_path: Path | None = None,
) -> SpawnResult:
return SpawnResult(instance_id="x")
async def stop(self, instance_id: str, graceful: bool = True) -> None: ...
async def health_check(self, instance_id: str) -> bool:
return True
async def remove(self, instance_id: str) -> None: ...
def _spec() -> InteractiveSpawnSpec:
cfg = type("C", (), {"agent_id": "intake-1"})()
return InteractiveSpawnSpec(
config=cfg, image="img", session_id="s1", role="prompter"
)
def test_provider_declines_interactive_by_default() -> None:
assert _OneShotOnly.supports_interactive is False
@pytest.mark.asyncio
async def test_spawn_interactive_default_raises() -> None:
# A one-shot-only provider declines interactive spawns rather than crashing.
with pytest.raises(ProviderError):
await _OneShotOnly().spawn_interactive(_spec())
tests/unit/runtime/test_interactive_grok_spawn.py
+92
View File
@@ -0,0 +1,92 @@
"""Interactive intake/secretary builders fork a GROK route onto opencode.
A GROK route swaps the Claude SDK-driver image for the opencode-serve image and
the ANTHROPIC_* env for OPENAI_* + the opencode store mount; every other
provider keeps the Claude path's ANTHROPIC_* behaviour.
"""
from __future__ import annotations
from roboco.runtime.orchestrator import (
GROK_PROMPTER_IMAGE,
GROK_SECRETARY_IMAGE,
AgentOrchestrator,
_IntakeRunSpec,
_SecretaryRunSpec,
)
_HOSTS = {
"claude": "/h/.claude",
"prompt": "/h/p.md",
"workspaces": "/h/ws",
"opencode": "/h/oc/intake-1",
}
def _intake_spec(
provider_type: str, *, base_url: str | None, token: str | None
) -> _IntakeRunSpec:
return _IntakeRunSpec(
container_name="roboco-agent-intake-1",
image=GROK_PROMPTER_IMAGE
if provider_type == "grok"
else "roboco-agent-prompter",
hosts=_HOSTS,
session_id="sess-1",
cwd="/data/workspace",
cli_model="grok-build-0.1",
api_url="http://roboco-orchestrator:8000",
provider_base_url=base_url,
provider_auth_token=token,
provider_type=provider_type,
model="grok-build-0.1",
)
def test_intake_grok_uses_openai_env_and_opencode_mount() -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(
_intake_spec("grok", base_url="https://api.x.ai/v1", token="xai-key")
)
assert "OPENAI_BASE_URL=https://api.x.ai/v1" in cmd
assert "OPENAI_API_KEY=xai-key" in cmd
assert "ROBOCO_AGENT_MODEL=grok-build-0.1" in cmd
assert "ROBOCO_SYSTEM_PROMPT=/app/system-prompt.md" in cmd
assert "/h/oc/intake-1:/home/agent/.local/share/opencode" in cmd
assert cmd[-1] == GROK_PROMPTER_IMAGE
# The xAI endpoint is never mislabelled as Anthropic.
assert not any(c.startswith("ANTHROPIC_") for c in cmd)
def test_intake_anthropic_keeps_anthropic_env() -> None:
cmd = AgentOrchestrator._build_intake_run_cmd(
_intake_spec("anthropic", base_url="https://api.anthropic.com", token="sk-ant")
)
assert "ANTHROPIC_BASE_URL=https://api.anthropic.com" in cmd
assert "ANTHROPIC_AUTH_TOKEN=sk-ant" in cmd
assert not any(c.startswith("OPENAI_") for c in cmd)
assert cmd[-1] == "roboco-agent-prompter"
def test_secretary_grok_uses_openai_env_and_grok_image() -> None:
spec = _SecretaryRunSpec(
container_name="roboco-agent-secretary-1",
image=GROK_SECRETARY_IMAGE,
hosts={"claude": "/h/.claude", "prompt": "/h/p.md", "opencode": "/h/oc/sec-1"},
session_id="sess-2",
cwd="/app",
cli_model="grok-build-0.1",
api_url="http://roboco-orchestrator:8000",
agent_uuid="uuid-sec",
agent_token="hmac-secretary",
provider_base_url="https://api.x.ai/v1",
provider_auth_token="xai-key",
provider_type="grok",
model="grok-build-0.1",
)
cmd = AgentOrchestrator._build_secretary_run_cmd(spec)
assert "OPENAI_API_KEY=xai-key" in cmd
assert "/h/oc/sec-1:/home/agent/.local/share/opencode" in cmd
# The HMAC identity the directive tools authenticate with survives.
assert "ROBOCO_AGENT_TOKEN=hmac-secretary" in cmd
assert cmd[-1] == GROK_SECRETARY_IMAGE
assert not any(c.startswith("ANTHROPIC_") for c in cmd)
tests/unit/services/test_llm_routing_interactive_guard.py
-60
View File
@@ -1,60 +0,0 @@
"""The interactive-role routing guard keeps intake/secretary off GROK.
GROK has no interactive runtime yet, and the interactive spawn would inject the
route's creds as ANTHROPIC_* against api.x.ai/v1 (wrong protocol → silent empty
reply). Until the opencode interactive driver lands, a GROK route for those
slugs is downgraded to the Anthropic default. The one-shot delivery roles route
to GROK unchanged. DB-free: exercises `_guard_interactive` directly.
"""
from __future__ import annotations
import structlog
from roboco.models.base import ModelProvider
from roboco.services.llm import AgentRoute, ModelRoutingService
def _svc() -> ModelRoutingService:
svc = ModelRoutingService.__new__(ModelRoutingService)
svc.log = structlog.get_logger()
return svc
def _grok_route() -> AgentRoute:
return AgentRoute(
provider_id=None,
provider_type=ModelProvider.GROK,
base_url="https://api.x.ai/v1",
auth_token="xai-key",
model_name="grok-build-0.1",
)
def test_grok_interactive_role_downgrades_to_anthropic() -> None:
svc = _svc()
for slug in ("intake-1", "secretary-1"):
route = svc._guard_interactive(_grok_route(), slug, "prompter")
assert route.provider_type == ModelProvider.ANTHROPIC
# No xAI creds leak onto the Claude SDK path.
assert route.base_url is None
assert route.auth_token is None
def test_grok_delivery_role_is_left_on_grok() -> None:
svc = _svc()
route = svc._guard_interactive(_grok_route(), "be-dev-1", "developer")
assert route.provider_type == ModelProvider.GROK
assert route.base_url == "https://api.x.ai/v1"
def test_non_grok_interactive_route_is_unchanged() -> None:
svc = _svc()
anthropic = AgentRoute(
provider_id=None,
provider_type=ModelProvider.ANTHROPIC,
base_url=None,
auth_token=None,
model_name="claude-opus-4-6",
)
route = svc._guard_interactive(anthropic, "intake-1", "prompter")
assert route is anthropic