hermes(router-plugin): adopt HUX friendly modes at the route boundary
HUX-06: the auto-router now consults the loopback HUX service (worker trust, HMAC-derived conversation id from the persisted session) for the conversation's selected friendly mode and maps it onto the existing provider-neutral pools: fast->auto/fast, thoughtful/research->auto/deep, create->auto/balanced; private pins the local route and refuses hosted overrides. Explicit UI picks keep precedence; every HUX absence or failure falls through to the previous behaviour unchanged. 95% branch coverage; quality contract registers the new module. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BvMSXH8VH2tMWXanb8SJdf
This commit is contained in:
jenkins
parent
b3d4538b4f
commit
0755e1bbd7
@ -13,6 +13,11 @@ try:
|
||||
except ImportError: # Direct module loading in the small unit-test harness.
|
||||
from provider_status import provider_status_text
|
||||
|
||||
try:
|
||||
from .hux_mode import selection as _hux_mode_selection
|
||||
except ImportError: # Direct module loading in the small unit-test harness.
|
||||
from hux_mode import selection as _hux_mode_selection
|
||||
|
||||
|
||||
POLICY_PATH = Path("/opt/data/workspace/coordinator/route-policy.json")
|
||||
POLICY_VERSION = 2
|
||||
@ -159,7 +164,7 @@ def _explicit_ui_effort(agent: Any) -> str:
|
||||
return effort if effort in EFFORTS else ""
|
||||
|
||||
|
||||
def _boundary_selection(agent: Any) -> tuple[str, str, str]:
|
||||
def _boundary_selection(agent: Any, session_id: str = "") -> tuple[str, str, str]:
|
||||
"""Select only a public route; Switchyard selects the actual target."""
|
||||
policy = _load_policy()
|
||||
ui_route = _explicit_ui_route(agent)
|
||||
@ -171,6 +176,13 @@ def _boundary_selection(agent: Any) -> tuple[str, str, str]:
|
||||
return ui_route, effort, f"ui-{mode}"
|
||||
return ui_route, ui_effort, f"ui-{mode}"
|
||||
|
||||
# HUX-06: a friendly mode selected for this conversation shapes the next
|
||||
# request. Automatic modes stay provider-neutral pools; only private may
|
||||
# pin the local route. Any HUX absence or failure falls straight through.
|
||||
hux = _hux_mode_selection(session_id) if session_id else None
|
||||
if hux and hux[0] in ALL_REQUEST_ROUTES:
|
||||
return hux[0], ui_effort or hux[1], "hux-mode"
|
||||
|
||||
if policy["mode"] == "manual":
|
||||
manual = policy.get("manual")
|
||||
if isinstance(manual, dict):
|
||||
@ -300,7 +312,9 @@ def _route_boundary(ctx: Any, scope: str, **kwargs: Any) -> None:
|
||||
agent = kwargs.get("agent") or kwargs.get("child") or _runtime_agent(ctx)
|
||||
if agent is None:
|
||||
return
|
||||
requested_route, effort, source = _boundary_selection(agent)
|
||||
requested_route, effort, source = _boundary_selection(
|
||||
agent, str(kwargs.get("session_id") or "")
|
||||
)
|
||||
route = _resolved_route(requested_route, effort)
|
||||
_switch_agent(ctx, agent, route, effort)
|
||||
policy = _load_policy()
|
||||
|
||||
158
services/hermes/plugins/auto-router/hux_mode.py
Normal file
158
services/hermes/plugins/auto-router/hux_mode.py
Normal file
@ -0,0 +1,158 @@
|
||||
"""Adopt the conversation's selected HUX friendly mode at the route boundary.
|
||||
|
||||
HUX-06: a mode selected in the WebUI must shape the next real Switchyard
|
||||
request without ever naming a provider for automatic pools. This adapter
|
||||
reads the loopback HUX service with worker trust and maps the stored
|
||||
provider-neutral mode contract onto the public route families the
|
||||
auto-router already owns. Every failure is an explicit ``None`` so the
|
||||
existing routing behaviour is preserved bit-for-bit when HUX is absent,
|
||||
misconfigured, unreachable, or simply has no selection for the session.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import stat
|
||||
import time
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
DOMAIN = "hux.context.id.v1"
|
||||
KEY_BYTES = 32
|
||||
CACHE_SECONDS = 5.0
|
||||
TIMEOUT_SECONDS = 2.0
|
||||
MAX_RESPONSE_BYTES = 256 * 1024
|
||||
RAW_ID = re.compile(r"^[A-Za-z0-9._:@+-]{1,200}$")
|
||||
SLOT = re.compile(r"^slot-[0-9]{1,3}$")
|
||||
SUBJECT = re.compile(r"^usr_[0-9a-f]{64}$")
|
||||
LOCAL_ROUTE = "atlas/manual/local/qwen-14b"
|
||||
MANUAL_OVERRIDE = re.compile(r"^atlas/manual/(codex|claude|local)/[a-z0-9][a-z0-9/-]{0,100}$")
|
||||
# Friendly modes map onto provider-neutral automatic pools; only the explicit
|
||||
# local-only private mode may pin a route, and it pins the local model.
|
||||
MODE_ROUTES = {
|
||||
"fast": "atlas/auto/fast",
|
||||
"thoughtful": "atlas/auto/deep",
|
||||
"research": "atlas/auto/deep",
|
||||
"create": "atlas/auto/balanced",
|
||||
"private": LOCAL_ROUTE,
|
||||
}
|
||||
|
||||
_cache: dict[str, tuple[float, tuple[str, str] | None]] = {}
|
||||
|
||||
|
||||
def _read_secret(path: str, size: int | None = None, modes: frozenset[int] = frozenset({0o400, 0o600})) -> bytes | None:
|
||||
"""Read one owner-only regular file without following symlinks."""
|
||||
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
|
||||
descriptor = -1
|
||||
try:
|
||||
descriptor = os.open(Path(path), flags)
|
||||
info = os.fstat(descriptor)
|
||||
if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
|
||||
return None
|
||||
if stat.S_IMODE(info.st_mode) not in modes and info.st_uid == os.geteuid():
|
||||
return None
|
||||
value = os.read(descriptor, 4096)
|
||||
except OSError:
|
||||
return None
|
||||
finally:
|
||||
if descriptor >= 0:
|
||||
os.close(descriptor)
|
||||
if size is not None and len(value) != size:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def _derive(key: bytes, prefix: str, purpose: str, slot: str, subject: str, raw: str) -> str | None:
|
||||
"""Replicate the public ``hux.context.id.v1`` derivation exactly."""
|
||||
if not RAW_ID.fullmatch(raw or ""):
|
||||
return None
|
||||
message = "\0".join((DOMAIN, purpose, slot, subject, raw)).encode("utf-8")
|
||||
return f"{prefix}_{hmac.new(key, message, hashlib.sha256).hexdigest()[:32]}"
|
||||
|
||||
|
||||
def _fetch(url: str, headers: dict[str, str]) -> tuple[int, dict[str, Any]] | None:
|
||||
"""One bounded loopback GET; anything unexpected is None."""
|
||||
request = urllib.request.Request(url, headers=headers, method="GET")
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=TIMEOUT_SECONDS) as response: # noqa: S310 - literal loopback URL built below
|
||||
body = response.read(MAX_RESPONSE_BYTES + 1)
|
||||
if len(body) > MAX_RESPONSE_BYTES:
|
||||
return None
|
||||
return response.status, json.loads(body)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _selection_uncached(environ: dict[str, str], raw_session_id: str, fetch: Any) -> tuple[str, str] | None:
|
||||
"""Resolve ``(route, effort)`` for the session's stored mode, or None."""
|
||||
base = environ.get("HUX_BASE_URL", "")
|
||||
slot = environ.get("HUX_TENANT_SLOT", "")
|
||||
subject_file = environ.get("HUX_SUBJECT_FILE", "")
|
||||
key_file = environ.get("HUX_WORKER_KEY_FILE", "")
|
||||
context_key_file = environ.get("HUX_CONTEXT_KEY_FILE", "")
|
||||
project_source = environ.get("HUX_PROJECT_SOURCE", "profile:default")
|
||||
if not (base.startswith("http://127.0.0.1:") and SLOT.fullmatch(slot)
|
||||
and subject_file and key_file and context_key_file):
|
||||
return None
|
||||
context_key = _read_secret(context_key_file, KEY_BYTES)
|
||||
subject_raw = _read_secret(subject_file)
|
||||
worker_key_raw = _read_secret(key_file)
|
||||
if context_key is None or subject_raw is None or worker_key_raw is None:
|
||||
return None
|
||||
subject = subject_raw.decode("ascii", errors="replace").strip()
|
||||
worker_key = worker_key_raw.decode("ascii", errors="replace").strip()
|
||||
if not SUBJECT.fullmatch(subject) or not worker_key:
|
||||
return None
|
||||
conversation = _derive(context_key, "conv", "conversation", slot, subject, raw_session_id)
|
||||
project = _derive(context_key, "prj", "project", slot, subject, project_source)
|
||||
if not conversation or not project:
|
||||
return None
|
||||
result = fetch(
|
||||
f"{base}/hux/v1/projects/{project}/conversations/{conversation}/mode",
|
||||
{
|
||||
"X-Hermes-Tenant-Identity": slot,
|
||||
"X-Hux-Subject": subject,
|
||||
"X-Hux-Surface": "worker",
|
||||
"X-Hux-Trust": "worker",
|
||||
"X-Hux-Relay-Key": worker_key,
|
||||
},
|
||||
)
|
||||
if result is None or result[0] != 200 or not isinstance(result[1], dict):
|
||||
return None
|
||||
mode = result[1].get("mode")
|
||||
if not isinstance(mode, dict):
|
||||
return None
|
||||
name = mode.get("mode")
|
||||
switchyard = mode.get("switchyard")
|
||||
route = MODE_ROUTES.get(name) if isinstance(name, str) else None
|
||||
if route is None or not isinstance(switchyard, dict):
|
||||
return None
|
||||
override = switchyard.get("override_route_id")
|
||||
if isinstance(override, str) and MANUAL_OVERRIDE.fullmatch(override):
|
||||
if name == "private" and not override.startswith("atlas/manual/local/"):
|
||||
return None
|
||||
route = override
|
||||
return route, ""
|
||||
|
||||
|
||||
def selection(raw_session_id: str, environ: dict[str, str] | None = None, fetch: Any = _fetch) -> tuple[str, str] | None:
|
||||
"""Cached, fail-open mode adoption for one persisted Hermes session."""
|
||||
if not isinstance(raw_session_id, str) or not raw_session_id:
|
||||
return None
|
||||
now = time.monotonic()
|
||||
cached = _cache.get(raw_session_id)
|
||||
if cached is not None and cached[0] > now:
|
||||
return cached[1]
|
||||
try:
|
||||
value = _selection_uncached(dict(os.environ if environ is None else environ), raw_session_id, fetch)
|
||||
except Exception:
|
||||
value = None
|
||||
if len(_cache) > 512:
|
||||
_cache.clear()
|
||||
_cache[raw_session_id] = (now + CACHE_SECONDS, value)
|
||||
return value
|
||||
@ -134,6 +134,7 @@
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/__init__.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/client.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/hooks.py",
|
||||
"services/hermes/plugins/auto-router/hux_mode.py",
|
||||
"services/hermes/plugins/hux-runtime/__init__.py",
|
||||
"services/hermes/plugins/hux-runtime/context_ids.py",
|
||||
"services/hermes/plugins/hux-runtime/runtime.py",
|
||||
@ -242,6 +243,7 @@
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/__init__.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/client.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/hooks.py",
|
||||
"services/hermes/plugins/auto-router/hux_mode.py",
|
||||
"services/hermes/plugins/hux-runtime/__init__.py",
|
||||
"services/hermes/plugins/hux-runtime/context_ids.py",
|
||||
"services/hermes/plugins/hux-runtime/runtime.py",
|
||||
@ -441,6 +443,7 @@
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/__init__.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/client.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/hooks.py",
|
||||
"services/hermes/plugins/auto-router/hux_mode.py",
|
||||
"services/hermes/plugins/hux-runtime/__init__.py",
|
||||
"services/hermes/plugins/hux-runtime/context_ids.py",
|
||||
"services/hermes/plugins/hux-runtime/runtime.py",
|
||||
@ -559,6 +562,7 @@
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/__init__.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/client.py",
|
||||
"dockerfiles/hermes-worker-hux/hux_hook/hooks.py",
|
||||
"services/hermes/plugins/auto-router/hux_mode.py",
|
||||
"services/hermes/plugins/hux-runtime/__init__.py",
|
||||
"services/hermes/plugins/hux-runtime/context_ids.py",
|
||||
"services/hermes/plugins/hux-runtime/runtime.py",
|
||||
|
||||
276
testing/tests/test_hermes_hux_mode_adoption.py
Normal file
276
testing/tests/test_hermes_hux_mode_adoption.py
Normal file
@ -0,0 +1,276 @@
|
||||
"""HUX-06 adoption: the stored friendly mode shapes the next route boundary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
ROOT = Path(__file__).parents[2]
|
||||
PLUGIN = ROOT / "services/hermes/plugins/auto-router"
|
||||
sys.path.insert(0, str(PLUGIN.parent))
|
||||
|
||||
SPEC = importlib.util.spec_from_file_location("hermes_hux_mode", PLUGIN / "hux_mode.py")
|
||||
assert SPEC and SPEC.loader
|
||||
hux_mode = importlib.util.module_from_spec(SPEC)
|
||||
sys.modules[SPEC.name] = hux_mode
|
||||
SPEC.loader.exec_module(hux_mode)
|
||||
|
||||
ROUTER_SPEC = importlib.util.spec_from_file_location(
|
||||
"hermes_auto_router_hux", PLUGIN / "__init__.py"
|
||||
)
|
||||
assert ROUTER_SPEC and ROUTER_SPEC.loader
|
||||
router = importlib.util.module_from_spec(ROUTER_SPEC)
|
||||
sys.modules[ROUTER_SPEC.name] = router
|
||||
ROUTER_SPEC.loader.exec_module(router)
|
||||
|
||||
KEY = bytes(range(32))
|
||||
SUBJECT = "usr_" + "ab" * 32
|
||||
SLOT = "slot-3"
|
||||
|
||||
|
||||
def _files(tmp_path: Path) -> dict[str, str]:
|
||||
for name in ("context-key", "subject", "worker-key"):
|
||||
existing = tmp_path / name
|
||||
if existing.exists():
|
||||
existing.chmod(0o600)
|
||||
existing.unlink()
|
||||
context = tmp_path / "context-key"
|
||||
context.write_bytes(KEY)
|
||||
context.chmod(0o600)
|
||||
subject = tmp_path / "subject"
|
||||
subject.write_text(SUBJECT + "\n")
|
||||
subject.chmod(0o400)
|
||||
worker = tmp_path / "worker-key"
|
||||
worker.write_text("worker-key-value\n")
|
||||
worker.chmod(0o400)
|
||||
return {
|
||||
"HUX_BASE_URL": "http://127.0.0.1:8790",
|
||||
"HUX_TENANT_SLOT": SLOT,
|
||||
"HUX_CONTEXT_KEY_FILE": str(context),
|
||||
"HUX_SUBJECT_FILE": str(subject),
|
||||
"HUX_WORKER_KEY_FILE": str(worker),
|
||||
"HUX_PROJECT_SOURCE": "profile:default",
|
||||
}
|
||||
|
||||
|
||||
def _record(mode: str, override: str | None = None) -> dict:
|
||||
switchyard: dict = {"route_id": "atlas/auto/fast"}
|
||||
if override:
|
||||
switchyard["override_route_id"] = override
|
||||
return {"mode": {"mode": mode, "switchyard": switchyard}}
|
||||
|
||||
|
||||
def _fetch_for(record: dict, status: int = 200, seen: dict | None = None):
|
||||
def fetch(url: str, headers: dict) -> tuple[int, dict]:
|
||||
if seen is not None:
|
||||
seen["url"] = url
|
||||
seen["headers"] = headers
|
||||
seen["calls"] = seen.get("calls", 0) + 1
|
||||
return status, record
|
||||
return fetch
|
||||
|
||||
|
||||
def _selection(tmp_path, record, session="sess-1", status=200, seen=None):
|
||||
hux_mode._cache.clear()
|
||||
return hux_mode.selection(
|
||||
session, _files(tmp_path), _fetch_for(record, status, seen)
|
||||
)
|
||||
|
||||
|
||||
def test_every_friendly_mode_maps_to_a_provider_neutral_pool(tmp_path):
|
||||
assert _selection(tmp_path, _record("fast")) == ("atlas/auto/fast", "")
|
||||
assert _selection(tmp_path, _record("thoughtful")) == ("atlas/auto/deep", "")
|
||||
assert _selection(tmp_path, _record("research")) == ("atlas/auto/deep", "")
|
||||
assert _selection(tmp_path, _record("create")) == ("atlas/auto/balanced", "")
|
||||
for route in ("atlas/auto/fast", "atlas/auto/deep", "atlas/auto/balanced"):
|
||||
assert route in router.AUTO_ROUTES # both providers stay eligible
|
||||
|
||||
|
||||
def test_private_mode_is_local_only(tmp_path):
|
||||
assert _selection(tmp_path, _record("private")) == (
|
||||
"atlas/manual/local/qwen-14b", "",
|
||||
)
|
||||
# A hosted override can never escape private mode.
|
||||
assert _selection(
|
||||
tmp_path, _record("private", "atlas/manual/claude/opus")
|
||||
) is None
|
||||
assert _selection(
|
||||
tmp_path, _record("private", "atlas/manual/local/qwen-14b")
|
||||
) == ("atlas/manual/local/qwen-14b", "")
|
||||
|
||||
|
||||
def test_advanced_override_is_honoured_for_hosted_modes(tmp_path):
|
||||
assert _selection(
|
||||
tmp_path, _record("create", "atlas/manual/claude/sonnet")
|
||||
) == ("atlas/manual/claude/sonnet", "")
|
||||
# A tampered override never breaks routing: the neutral pool wins.
|
||||
assert _selection(tmp_path, _record("create", "not-a-route")) == (
|
||||
"atlas/auto/balanced", "",
|
||||
)
|
||||
|
||||
|
||||
def test_identity_and_derivation_reach_the_service_exactly(tmp_path):
|
||||
seen: dict = {}
|
||||
_selection(tmp_path, _record("fast"), seen=seen)
|
||||
conv = "conv_" + hmac.new(
|
||||
KEY,
|
||||
"\0".join(
|
||||
("hux.context.id.v1", "conversation", SLOT, SUBJECT, "sess-1")
|
||||
).encode(),
|
||||
hashlib.sha256,
|
||||
).hexdigest()[:32]
|
||||
prj = "prj_" + hmac.new(
|
||||
KEY,
|
||||
"\0".join(
|
||||
("hux.context.id.v1", "project", SLOT, SUBJECT, "profile:default")
|
||||
).encode(),
|
||||
hashlib.sha256,
|
||||
).hexdigest()[:32]
|
||||
assert seen["url"] == (
|
||||
f"http://127.0.0.1:8790/hux/v1/projects/{prj}/conversations/{conv}/mode"
|
||||
)
|
||||
assert seen["headers"]["X-Hux-Trust"] == "worker"
|
||||
assert seen["headers"]["X-Hux-Surface"] == "worker"
|
||||
assert seen["headers"]["X-Hux-Subject"] == SUBJECT
|
||||
assert seen["headers"]["X-Hux-Relay-Key"] == "worker-key-value"
|
||||
|
||||
|
||||
def test_every_failure_is_fail_open(tmp_path):
|
||||
assert hux_mode.selection("", _files(tmp_path), _fetch_for(_record("fast"))) is None
|
||||
assert _selection(tmp_path, _record("fast"), status=404) is None
|
||||
assert _selection(tmp_path, _record("unknown")) is None
|
||||
assert _selection(tmp_path, {"mode": "not-a-dict"}) is None
|
||||
env = _files(tmp_path)
|
||||
env["HUX_TENANT_SLOT"] = "slot-nope"
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", env, _fetch_for(_record("fast"))) is None
|
||||
env = _files(tmp_path)
|
||||
env["HUX_BASE_URL"] = "http://10.0.0.1:8790"
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", env, _fetch_for(_record("fast"))) is None
|
||||
env = _files(tmp_path)
|
||||
Path(env["HUX_CONTEXT_KEY_FILE"]).write_bytes(b"short")
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", env, _fetch_for(_record("fast"))) is None
|
||||
|
||||
def broken(url, headers):
|
||||
raise OSError("down")
|
||||
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", _files(tmp_path), broken) is None
|
||||
|
||||
|
||||
def test_secret_file_guards_reject_links_and_bad_shapes(tmp_path):
|
||||
env = _files(tmp_path)
|
||||
target = tmp_path / "elsewhere"
|
||||
target.write_bytes(KEY)
|
||||
link = tmp_path / "context-link"
|
||||
link.symlink_to(target)
|
||||
env["HUX_CONTEXT_KEY_FILE"] = str(link)
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", env, _fetch_for(_record("fast"))) is None
|
||||
env = _files(tmp_path)
|
||||
env["HUX_SUBJECT_FILE"] = str(tmp_path / "does-not-exist")
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", env, _fetch_for(_record("fast"))) is None
|
||||
env = _files(tmp_path)
|
||||
Path(env["HUX_SUBJECT_FILE"]).chmod(0o600)
|
||||
Path(env["HUX_SUBJECT_FILE"]).write_text("not-a-subject\n")
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", env, _fetch_for(_record("fast"))) is None
|
||||
env = _files(tmp_path)
|
||||
Path(env["HUX_WORKER_KEY_FILE"]).chmod(0o600)
|
||||
Path(env["HUX_WORKER_KEY_FILE"]).write_text("\n")
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("s", env, _fetch_for(_record("fast"))) is None
|
||||
# A malformed raw session id never reaches HMAC derivation.
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("bad session", _files(tmp_path), _fetch_for(_record("fast"))) is None
|
||||
|
||||
|
||||
def test_real_fetch_round_trip_and_refusals(tmp_path):
|
||||
import http.server
|
||||
import threading
|
||||
|
||||
record = _record("fast")
|
||||
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def do_GET(self): # noqa: N802 - stdlib naming
|
||||
body = json.dumps(record).encode()
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_message(self, *args): # noqa: D102
|
||||
return
|
||||
|
||||
server = http.server.HTTPServer(("127.0.0.1", 0), Handler)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
env = _files(tmp_path)
|
||||
env["HUX_BASE_URL"] = f"http://127.0.0.1:{server.server_port}"
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("sess-live", env) == ("atlas/auto/fast", "")
|
||||
finally:
|
||||
server.shutdown()
|
||||
thread.join(timeout=5)
|
||||
# A dead loopback port fails open through the real fetch path too.
|
||||
env = _files(tmp_path)
|
||||
env["HUX_BASE_URL"] = f"http://127.0.0.1:{server.server_port}"
|
||||
hux_mode._cache.clear()
|
||||
assert hux_mode.selection("sess-dead", env) is None
|
||||
|
||||
|
||||
def test_cache_is_bounded(tmp_path):
|
||||
env = _files(tmp_path)
|
||||
hux_mode._cache.clear()
|
||||
for index in range(514):
|
||||
hux_mode.selection(f"sess-{index}", env, _fetch_for(_record("fast")))
|
||||
assert len(hux_mode._cache) <= 513
|
||||
|
||||
|
||||
def test_selection_is_cached_per_session(tmp_path):
|
||||
seen: dict = {}
|
||||
env = _files(tmp_path)
|
||||
hux_mode._cache.clear()
|
||||
first = hux_mode.selection("sess-c", env, _fetch_for(_record("fast"), seen=seen))
|
||||
second = hux_mode.selection("sess-c", env, _fetch_for(_record("fast"), seen=seen))
|
||||
assert first == second == ("atlas/auto/fast", "")
|
||||
assert seen["calls"] == 1
|
||||
|
||||
|
||||
def test_boundary_prefers_ui_pick_and_adopts_hux_mode(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(router, "POLICY_PATH", tmp_path / "route-policy.json")
|
||||
monkeypatch.setattr(
|
||||
router, "_hux_mode_selection", lambda session: ("atlas/auto/deep", "")
|
||||
)
|
||||
agent = SimpleNamespace(
|
||||
_hermes_explicit_model_pick=False,
|
||||
_hermes_explicit_reasoning_effort="",
|
||||
_hermes_routing_priority="",
|
||||
)
|
||||
assert router._boundary_selection(agent, "sess-1") == (
|
||||
"atlas/auto/deep", "", "hux-mode",
|
||||
)
|
||||
# No session means no adoption and the existing default is untouched.
|
||||
assert router._boundary_selection(agent, "")[2] == "auto"
|
||||
# An explicit UI pick always outranks the stored mode.
|
||||
picked = SimpleNamespace(
|
||||
_hermes_explicit_model_pick=True,
|
||||
model="atlas/manual/claude/opus",
|
||||
_hermes_explicit_reasoning_effort="high",
|
||||
_hermes_routing_priority="",
|
||||
)
|
||||
assert router._boundary_selection(picked, "sess-1") == (
|
||||
"atlas/manual/claude/opus", "high", "ui-manual",
|
||||
)
|
||||
monkeypatch.setattr(router, "_hux_mode_selection", lambda session: None)
|
||||
assert router._boundary_selection(agent, "sess-1")[2] == "auto"
|
||||
Loading…
x
Reference in New Issue
Block a user