391 lines
14 KiB
Python
391 lines
14 KiB
Python
"""Keep Hermes boundaries on the Atlas Switchyard routing authority."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
from .provider_status import provider_status_text
|
|
except ImportError: # Direct module loading in the small unit-test harness.
|
|
from provider_status import provider_status_text
|
|
|
|
|
|
POLICY_PATH = Path("/opt/data/workspace/coordinator/route-policy.json")
|
|
SWITCHYARD_PROVIDER = "atlas-switchyard"
|
|
ROUTER_PROFILE = os.environ.get("HERMES_AUTO_ROUTER_PROFILE", "").strip().lower()
|
|
if ROUTER_PROFILE not in {"chat", "triage", "agent"}:
|
|
ROUTER_PROFILE = (
|
|
"chat"
|
|
if os.environ.get("HERMES_AUTO_ROUTER_CHAT_MODE", "0") == "1"
|
|
else "agent"
|
|
)
|
|
|
|
PROFILE_ROUTE = {
|
|
"chat": "atlas/auto/fast",
|
|
"triage": "atlas/auto/deep",
|
|
"agent": "atlas/auto/maximum",
|
|
}
|
|
PRIORITY_ROUTE = {
|
|
"fast": "atlas/auto/fast",
|
|
"balanced": "atlas/auto/balanced",
|
|
"deep": "atlas/auto/deep",
|
|
"maximum": "atlas/auto/maximum",
|
|
}
|
|
AUTO_ROUTES = frozenset(PRIORITY_ROUTE.values())
|
|
MANUAL_ROUTES = frozenset(
|
|
{
|
|
"atlas/manual/codex/luna",
|
|
"atlas/manual/codex/terra",
|
|
"atlas/manual/codex/sol",
|
|
"atlas/manual/claude/haiku",
|
|
"atlas/manual/claude/sonnet",
|
|
"atlas/manual/claude/opus",
|
|
"atlas/manual/local/qwen-14b",
|
|
}
|
|
)
|
|
ALL_ROUTES = AUTO_ROUTES | MANUAL_ROUTES
|
|
EFFORTS = frozenset({"none", "minimal", "low", "medium", "high", "xhigh"})
|
|
PROVIDER_DEFAULT = {
|
|
"codex": "atlas/manual/codex/terra",
|
|
"claude": "atlas/manual/claude/sonnet",
|
|
"local": "atlas/manual/local/qwen-14b",
|
|
}
|
|
|
|
|
|
def _load_policy() -> dict[str, Any]:
|
|
"""Load the small persistent UI policy without making it an authority."""
|
|
try:
|
|
value = json.loads(POLICY_PATH.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError, TypeError):
|
|
value = {}
|
|
if not isinstance(value, dict):
|
|
value = {}
|
|
if value.get("mode") not in {"auto", "manual"}:
|
|
value["mode"] = "auto"
|
|
route = str(value.get("auto_route") or "")
|
|
if route not in AUTO_ROUTES:
|
|
value["auto_route"] = PROFILE_ROUTE[ROUTER_PROFILE]
|
|
return value
|
|
|
|
|
|
def _write_policy(value: dict[str, Any]) -> None:
|
|
"""Persist route preference atomically inside the current workspace."""
|
|
POLICY_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = POLICY_PATH.with_suffix(".json.tmp")
|
|
temporary.write_text(
|
|
json.dumps(value, indent=2, sort_keys=True) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary.replace(POLICY_PATH)
|
|
|
|
|
|
def _runtime_agent(ctx: Any) -> Any | None:
|
|
"""Return the foreground agent when the hook did not provide a child."""
|
|
cli = getattr(getattr(ctx, "_manager", None), "_cli_ref", None)
|
|
return getattr(cli, "agent", None) if cli is not None else None
|
|
|
|
|
|
def _normalise_manual_route(provider: str, model: str = "") -> str:
|
|
"""Map a provider/model override to one public Switchyard route."""
|
|
provider = provider.strip().lower()
|
|
model = model.strip().lower()
|
|
if provider not in PROVIDER_DEFAULT:
|
|
return ""
|
|
if not model:
|
|
return PROVIDER_DEFAULT[provider]
|
|
|
|
aliases = {
|
|
"codex": {
|
|
"luna": "atlas/manual/codex/luna",
|
|
"gpt-5.6-luna": "atlas/manual/codex/luna",
|
|
"terra": "atlas/manual/codex/terra",
|
|
"gpt-5.6-terra": "atlas/manual/codex/terra",
|
|
"sol": "atlas/manual/codex/sol",
|
|
"gpt-5.6-sol": "atlas/manual/codex/sol",
|
|
},
|
|
"claude": {
|
|
"haiku": "atlas/manual/claude/haiku",
|
|
"claude-haiku-4-5-20251001": "atlas/manual/claude/haiku",
|
|
"sonnet": "atlas/manual/claude/sonnet",
|
|
"claude-sonnet-5": "atlas/manual/claude/sonnet",
|
|
"opus": "atlas/manual/claude/opus",
|
|
"claude-opus-5": "atlas/manual/claude/opus",
|
|
},
|
|
"local": {
|
|
"qwen": "atlas/manual/local/qwen-14b",
|
|
"qwen-14b": "atlas/manual/local/qwen-14b",
|
|
"qwen2.5:14b-instruct-q4_0": "atlas/manual/local/qwen-14b",
|
|
},
|
|
}
|
|
return aliases[provider].get(model, "")
|
|
|
|
|
|
def _explicit_ui_route(agent: Any) -> str:
|
|
"""Return a route selected in the WebUI model picker for this request."""
|
|
if not bool(getattr(agent, "_hermes_explicit_model_pick", False)):
|
|
return ""
|
|
route = str(getattr(agent, "model", "") or "").strip()
|
|
return route if route in ALL_ROUTES else ""
|
|
|
|
|
|
def _explicit_ui_effort(agent: Any) -> str:
|
|
"""Return an exact WebUI effort override, capped by the route catalog."""
|
|
effort = str(
|
|
getattr(agent, "_hermes_explicit_reasoning_effort", "") or ""
|
|
).strip().lower()
|
|
return effort if effort in EFFORTS else ""
|
|
|
|
|
|
def _boundary_selection(agent: Any) -> tuple[str, str, str]:
|
|
"""Select only a public route; Switchyard selects the actual target."""
|
|
policy = _load_policy()
|
|
ui_route = _explicit_ui_route(agent)
|
|
ui_effort = _explicit_ui_effort(agent)
|
|
if ui_route:
|
|
mode = "auto" if ui_route in AUTO_ROUTES else "manual"
|
|
return ui_route, ui_effort, f"ui-{mode}"
|
|
|
|
if policy["mode"] == "manual":
|
|
manual = policy.get("manual")
|
|
if isinstance(manual, dict):
|
|
route = str(manual.get("route") or "")
|
|
effort = str(manual.get("effort") or "")
|
|
if route in MANUAL_ROUTES and effort in EFFORTS:
|
|
return route, ui_effort or effort, "manual"
|
|
|
|
priority = str(
|
|
getattr(agent, "_hermes_routing_priority", "") or ""
|
|
).strip().lower()
|
|
route = PRIORITY_ROUTE.get(priority)
|
|
if route:
|
|
return route, ui_effort, "ui-auto"
|
|
return str(policy["auto_route"]), ui_effort, "auto"
|
|
|
|
|
|
def _switch_agent(ctx: Any, agent: Any, route: str, effort: str) -> None:
|
|
"""Point one live Hermes agent at Switchyard and remove local failover."""
|
|
runtime_agent = _runtime_agent(ctx)
|
|
cli = (
|
|
getattr(getattr(ctx, "_manager", None), "_cli_ref", None)
|
|
if agent is runtime_agent
|
|
else None
|
|
)
|
|
|
|
if agent.provider != SWITCHYARD_PROVIDER or agent.model != route:
|
|
from hermes_cli.inventory import load_picker_context
|
|
from hermes_cli.model_switch import switch_model
|
|
|
|
picker = load_picker_context()
|
|
result = switch_model(
|
|
raw_input=route,
|
|
current_provider=agent.provider or "",
|
|
current_model=agent.model or "",
|
|
current_base_url=agent.base_url or "",
|
|
current_api_key=agent.api_key or "",
|
|
is_global=False,
|
|
explicit_provider=SWITCHYARD_PROVIDER,
|
|
user_providers=picker.user_providers,
|
|
custom_providers=picker.custom_providers,
|
|
)
|
|
if not result.success:
|
|
raise RuntimeError(result.error_message or "Switchyard model switch failed")
|
|
agent.switch_model(
|
|
new_model=result.new_model,
|
|
new_provider=result.target_provider,
|
|
api_key=result.api_key,
|
|
base_url=result.base_url,
|
|
api_mode=result.api_mode,
|
|
)
|
|
if cli is not None:
|
|
cli.model = result.new_model
|
|
cli.provider = result.target_provider
|
|
cli.requested_provider = result.target_provider
|
|
cli.api_key = result.api_key or cli.api_key
|
|
cli.base_url = result.base_url or ""
|
|
cli.api_mode = result.api_mode or cli.api_mode
|
|
cli._explicit_api_key = result.api_key
|
|
cli._explicit_base_url = result.base_url
|
|
|
|
reasoning = None
|
|
if effort:
|
|
from hermes_constants import parse_reasoning_effort
|
|
|
|
reasoning = parse_reasoning_effort(effort)
|
|
agent.reasoning_config = reasoning
|
|
if cli is not None:
|
|
cli.reasoning_config = reasoning
|
|
app = getattr(cli, "_app", None)
|
|
if app is not None:
|
|
app.invalidate()
|
|
|
|
# Cross-provider recovery belongs to Switchyard. A Hermes fallback here
|
|
# would create a second control plane and could bypass a manual constraint.
|
|
agent._fallback_chain = []
|
|
agent._fallback_index = 0
|
|
agent._fallback_activated = False
|
|
agent._fallback_model = None
|
|
agent._hermes_switchyard_route = route
|
|
|
|
if agent is runtime_agent:
|
|
try:
|
|
from agent.auxiliary_client import set_runtime_main
|
|
|
|
set_runtime_main(
|
|
agent.provider or "",
|
|
agent.model or "",
|
|
base_url=agent.base_url or "",
|
|
api_key=agent.api_key or "",
|
|
api_mode=agent.api_mode or "",
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _record_boundary(
|
|
policy: dict[str, Any], route: str, effort: str, source: str, scope: str
|
|
) -> None:
|
|
"""Record the public request route without pretending it is the target."""
|
|
record = {
|
|
"route": route,
|
|
"effort_override": effort or None,
|
|
"source": source,
|
|
"scope": scope,
|
|
"authority": "switchyard",
|
|
"updated_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
policy["last_boundary"] = record
|
|
counter = f"{scope}_boundaries_total"
|
|
policy[counter] = int(policy.get(counter) or 0) + 1
|
|
_write_policy(policy)
|
|
|
|
|
|
def _route_boundary(ctx: Any, scope: str, **kwargs: Any) -> None:
|
|
"""Route a user turn, tool continuation, or delegated child."""
|
|
agent = kwargs.get("agent") or kwargs.get("child") or _runtime_agent(ctx)
|
|
if agent is None:
|
|
return
|
|
route, effort, source = _boundary_selection(agent)
|
|
_switch_agent(ctx, agent, route, effort)
|
|
policy = _load_policy()
|
|
_record_boundary(policy, route, effort, source, scope)
|
|
|
|
if scope != "turn":
|
|
return
|
|
emit = getattr(agent, "_emit_status", None)
|
|
if callable(emit):
|
|
override = f" · effort {effort}" if effort else ""
|
|
emit(
|
|
f"{source.upper()} via Switchyard → {route}{override} · "
|
|
"target and fallback selected per boundary"
|
|
)
|
|
|
|
|
|
def _pre_turn(ctx: Any, **kwargs: Any) -> None:
|
|
"""Route a visible user turn unless Hermes is handling a slash command."""
|
|
message = str(kwargs.get("user_message") or "").strip()
|
|
if message.startswith("/"):
|
|
return
|
|
_route_boundary(ctx, "turn", **kwargs)
|
|
|
|
|
|
def _status_text(ctx: Any) -> str:
|
|
"""Describe the route contract without claiming an unseen target."""
|
|
policy = _load_policy()
|
|
agent = _runtime_agent(ctx)
|
|
runtime = "not initialized"
|
|
if agent is not None:
|
|
runtime = f"{agent.provider}/{agent.model}"
|
|
if policy["mode"] == "manual":
|
|
manual = policy.get("manual") or {}
|
|
preference = (
|
|
f"{manual.get('route', 'invalid')} at "
|
|
f"{manual.get('effort', 'default')}"
|
|
)
|
|
else:
|
|
preference = str(policy["auto_route"])
|
|
last = policy.get("last_boundary") or {}
|
|
last_text = str(last.get("route") or "none yet")
|
|
return (
|
|
"Routing authority: Switchyard\n"
|
|
f"Service posture: {ROUTER_PROFILE}\n"
|
|
f"Mode: {policy['mode'].upper()} ({preference})\n"
|
|
f"Current Hermes endpoint: {runtime}\n"
|
|
f"Last public route: {last_text}\n"
|
|
"Switchyard independently selects provider/model/effort and fallback "
|
|
"for every model-call boundary.\n"
|
|
"Commands: /route auto [fast|balanced|deep|maximum] | "
|
|
"/route manual <codex|claude|local> "
|
|
"<none|minimal|low|medium|high|xhigh> [model] | /route status"
|
|
)
|
|
|
|
|
|
def _route_command(ctx: Any, raw_args: str) -> str:
|
|
"""Persist an AUTO posture or constrained Switchyard route."""
|
|
args = raw_args.strip().split()
|
|
if not args or args[0].lower() == "status":
|
|
return _status_text(ctx)
|
|
|
|
mode = args[0].lower()
|
|
policy = _load_policy()
|
|
if mode == "auto":
|
|
posture = args[1].lower() if len(args) > 1 else ""
|
|
route = PRIORITY_ROUTE.get(posture, PROFILE_ROUTE[ROUTER_PROFILE])
|
|
if posture and posture not in PRIORITY_ROUTE:
|
|
return "AUTO posture must be fast, balanced, deep, or maximum."
|
|
policy = {"mode": "auto", "auto_route": route}
|
|
_write_policy(policy)
|
|
return "AUTO routing enabled.\n" + _status_text(ctx)
|
|
|
|
if mode != "manual" or len(args) < 3:
|
|
return (
|
|
"Usage: /route auto [fast|balanced|deep|maximum] | "
|
|
"/route manual <codex|claude|local> "
|
|
"<none|minimal|low|medium|high|xhigh> [model] | /route status"
|
|
)
|
|
provider = args[1].lower()
|
|
effort = args[2].lower()
|
|
model = args[3] if len(args) > 3 else ""
|
|
route = _normalise_manual_route(provider, model)
|
|
if not route:
|
|
return "Unknown provider/model combination for a Switchyard route."
|
|
if effort not in EFFORTS:
|
|
return "Effort must be none, minimal, low, medium, high, or xhigh."
|
|
policy = {
|
|
"mode": "manual",
|
|
"auto_route": PROFILE_ROUTE[ROUTER_PROFILE],
|
|
"manual": {"route": route, "effort": effort},
|
|
}
|
|
_write_policy(policy)
|
|
return "Manual Switchyard constraint enabled.\n" + _status_text(ctx)
|
|
|
|
|
|
def register(ctx: Any) -> None:
|
|
"""Register thin boundary hooks; Switchyard owns every routing decision."""
|
|
ctx.register_hook("pre_turn_route", lambda **kwargs: _pre_turn(ctx, **kwargs))
|
|
ctx.register_hook(
|
|
"pre_internal_route",
|
|
lambda **kwargs: _route_boundary(ctx, "internal", **kwargs),
|
|
)
|
|
try:
|
|
ctx.register_hook(
|
|
"pre_subagent_route",
|
|
lambda **kwargs: _route_boundary(ctx, "subagent", **kwargs),
|
|
)
|
|
except ValueError:
|
|
pass
|
|
ctx.register_command(
|
|
"route",
|
|
lambda raw_args: _route_command(ctx, raw_args),
|
|
description="Show or constrain Switchyard AUTO routing",
|
|
args_hint="auto [posture]|status|manual provider effort [model]",
|
|
)
|
|
ctx.register_command(
|
|
"providers",
|
|
lambda _raw_args: provider_status_text(),
|
|
description="Show Codex, Claude, local, and Switchyard status",
|
|
)
|