"""Fresh local policy is frozen and scoped, not process launcher state.""" import os from concurrent.futures import ThreadPoolExecutor import pytest def test_policy_selects_surface_and_isolates_cwd(tmp_path): from gateway.session_policy import build_policy, policy_scope from agent.runtime_cwd import resolve_agent_cwd from tools.terminal_scope import terminal_env from hermes_state_runtime import RuntimeStoreError cfg = {'platform_toolsets': {'cli': ['terminal']}} before = dict(os.environ) policies = [] for source in ('cli', 'tui', 'gui'): cwd = tmp_path / source cwd.mkdir() policies.append(build_policy({'source': source, 'cwd': str(cwd), 'model': source}, cfg)) assert policies[2].platform == 'desktop' assert 'desktop_ui' in policies[2].toolsets assert all('desktop_ui' not in p.toolsets for p in policies[:2]) # Platform-gated toolsets (catalog) ride on the Desktop surface only, like the native TUI factory. assert 'catalog' in policies[2].toolsets and all('catalog' not in p.toolsets for p in policies[:2]) cfg['platform_toolsets']['cli'].clear() assert 'terminal' in policies[0].toolsets def run(policy): with policy_scope(policy): assert str(resolve_agent_cwd()) == policy.cwd assert terminal_env('TERMINAL_CWD') == policy.cwd with ThreadPoolExecutor(max_workers=3) as pool: list(pool.map(run, policies)) assert dict(os.environ) == before for params in ({'source': 'cron'}, {'toolsets': ['not-a-toolset']}, {'cwd': '.'}, {'provider': 12}, {'skills': 'x'}, {'skills': []}, {'pass_session_id': 1}, {'toolsets': ['desktop_ui'], 'source': 'cli'}): with pytest.raises(RuntimeStoreError, match='invalid_params'): build_policy(params, {}) @pytest.mark.asyncio async def test_launch_options_reach_the_owner_frozen_and_hooks_consented(tmp_path, monkeypatch): """`-s/--skills`, `--checkpoints`, `--pass-session-id`, `--accept-hooks` (and an exported HERMES_ACCEPT_HOOKS=1, which the TUI forwards) were refused by session.create, so no session could be created with them. They are this session's creation policy, applied by the owner.""" import json from types import SimpleNamespace from gateway import run from gateway.config import GatewayConfig from gateway.session import SessionStore from gateway.session_authority import initialize_session_authority from gateway.session_controls import AuthorityConnection from gateway.session_policy import restore_policy from hermes_state_local import local_receipt from agent import shell_hooks from hermes_cli.plugins import get_plugin_manager home = tmp_path / 'home' (home / 'skills' / 'launch-fixture').mkdir(parents=True) (home / 'skills' / 'launch-fixture' / 'SKILL.md').write_text( '---\nname: launch-fixture\ndescription: fixture\n---\n\nLAUNCH_SKILL_BODY\n') monkeypatch.setenv('HERMES_HOME', str(home)) monkeypatch.delenv('HERMES_ACCEPT_HOOKS', raising=False) hook = tmp_path / 'hook.sh' hook.write_text('#!/bin/sh\nexit 0\n') cfg = {'platform_toolsets': {'cli': []}, 'hooks': {'pre_tool_call': [{'command': str(hook)}]}} monkeypatch.setattr(run, '_load_gateway_config', lambda: cfg) monkeypatch.setattr(run, '_resolve_gateway_model', lambda config: 'fixture') monkeypatch.setattr(get_plugin_manager(), '_hooks', {}) shell_hooks.reset_for_tests() store = SessionStore(tmp_path / 'sessions', GatewayConfig()) runner = SimpleNamespace(session_store=store, _session_db=store._db, adapters={}, _draining=False, _cached_agent_for=lambda route: None) runner._adapter_for_source = lambda source: runner.adapters.get(source.platform) authority = await initialize_session_authority(runner, profile_id=str(home), instance_id='first') connection = AuthorityConnection(authority, object(), {'user_id': 'human'}) async def create(request_id, **params): reply = await connection.dispatch({'id': 1, 'method': 'session.create', 'params': { 'request_id': request_id, 'source': 'tui', 'cwd': str(tmp_path), **params}}) return reply.get('result', {}).get('session_id') or reply['error']['message'] launch = dict(skills=['launch-fixture', 'missing-typo'], checkpoints=True, pass_session_id=True, accept_hooks=True) sid = await create('launch', **launch) def frozen(session_id): return restore_policy(local_receipt(authority.db, session_id)['policy']) policy = frozen(sid) assert 'LAUNCH_SKILL_BODY' in policy.skills_prompt and policy.pass_session_id and policy.checkpoints_enabled assert str(hook) in (home / shell_hooks.ALLOWLIST_FILENAME).read_text() assert len(get_plugin_manager()._hooks['pre_tool_call']) == 1 # Retrying the request id resumes the frozen route; nothing re-renders or re-registers. assert await create('launch', **launch) == sid assert len(get_plugin_manager()._hooks['pre_tool_call']) == 1 plain = frozen(await create('plain')) assert plain.skills_prompt is None and not plain.pass_session_id and not plain.checkpoints_enabled # Never silently dropped: nothing resolving, bad shapes, or profile reads on a bypass launch refuse. for params in ({'skills': ['missing-typo']}, {'skills': 'launch-fixture'}, {'checkpoints': 'yes'}, {'model': 'm', 'safe_mode': True, 'skills': ['launch-fixture']}): assert await create('refused-' + json.dumps(params), **params) in {'unknown_skill', 'invalid_params'}, params # An exported HERMES_ACCEPT_HOOKS=1 must not brick a --safe-mode TUI: no profile hooks, no refusal. assert (await create('safe', model='m', safe_mode=True, accept_hooks=True)).startswith('local-') def test_launch_policy_reaches_real_turn_runner(tmp_path): import json from pathlib import Path import subprocess import sys repo = Path(__file__).resolve().parents[2] home, state = tmp_path / 'home', tmp_path / 'state' home.mkdir() state.mkdir() from tests.gateway.fixtures.local_recovery_probe import child_env env = child_env() env.update(HOME=str(home), USERPROFILE=str(home), HERMES_HOME=str(state), PYTHONPATH=str(repo)) result = subprocess.run([sys.executable, str(Path(__file__).parent / 'fixtures' / 'session_policy_peer.py')], cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=130, check=False) assert result.returncode == 0, result.stdout + '\n' + result.stderr receipt = json.loads((state / 'policy-receipt.json').read_text()) assert receipt['cwd_effects'] and receipt['same_agents'] and receipt['no_spill'] print(json.dumps(receipt)) def test_launch_options_are_frozen_and_validated(tmp_path): from gateway.session_policy import build_policy from hermes_constants import parse_reasoning_effort from hermes_state_runtime import RuntimeStoreError cfg = {'agent': {'max_turns': 8, 'reasoning_effort': 'low'}} params = dict(cwd=str(tmp_path), provider='custom', base_url='http://127.0.0.1:1234/v1', model='fixture', reasoning='high', max_turns=3, ignore_rules=True) policy = build_policy(params, cfg) assert policy.provider == 'custom' and policy.base_url == params['base_url'] assert policy.ignore_rules and policy.max_turns == 3 assert policy.reasoning_config == parse_reasoning_effort('high') cfg['agent']['reasoning_effort'] = 'none' assert policy.reasoning_config == parse_reasoning_effort('high') assert build_policy(dict(cwd=str(tmp_path), ignore_rules=False), cfg).ignore_rules is False for bad in ({'max_turns': True}, {'max_turns': 'garbage'}, {'max_turns': 2.5}, {'reasoning': 'garbage'}, {'ignore_rules': 'false'}, {'base_url': 'http://user:secret@localhost/v1'}): with pytest.raises(RuntimeStoreError, match='invalid_params'): build_policy(dict(cwd=str(tmp_path), **bad), cfg) def test_provider_override_drops_the_previous_providers_route(tmp_path): """A provider pick is a whole-route change: folding it must not keep the profile's endpoint/wire-protocol/credential pointers, else the new provider's key is sent to the old host (#138843). An explicit base_url is applied after the drop, and a same-provider re-pick keeps what it carried.""" from gateway.session_policy import build_policy cfg = {'model': {'default': 'deepseek-chat', 'provider': 'deepseek', 'base_url': 'https://api.deepseek.com/v1', 'api_mode': 'chat_completions', 'key_env': 'HERMES_CUSTOM_QWENAI_API_KEY'}} policy = build_policy(dict(cwd=str(tmp_path), model='mimo-v2.6-flash', provider='xiaomi'), cfg) assert policy.config()['model'] == {'default': 'deepseek-chat', 'provider': 'xiaomi'} # A provider change plus an explicit URL still drops the replaced provider's wire protocol # and credential pointer — only the pinned endpoint comes back. explicit = build_policy(dict(cwd=str(tmp_path), model='m', provider='xiaomi', base_url='https://proxy.example/v1'), cfg) assert explicit.config()['model'] == {'default': 'deepseek-chat', 'provider': 'xiaomi', 'base_url': 'https://proxy.example/v1'} same = build_policy(dict(cwd=str(tmp_path), model='deepseek-r2', provider='deepseek'), cfg) assert same.config()['model']['base_url'] == 'https://api.deepseek.com/v1' # A profile without a recorded provider still leaves its endpoint behind for the pick to replace. unrecorded = build_policy(dict(cwd=str(tmp_path), provider='xiaomi'), {'model': {'default': 'm', 'base_url': 'https://api.deepseek.com/v1', 'api_key': 'sk-inline'}}) assert unrecorded.config()['model'] == {'default': 'm', 'provider': 'xiaomi'} def test_provider_change_drops_the_old_providers_context_window(tmp_path): """The profile's context_length was sized for its own model: a pick to another provider must not pin that window on the new model, while a same-route alias pick keeps it.""" from gateway.session_policy import build_policy cfg = {'model': {'default': 'qwen', 'provider': 'custom', 'base_url': 'http://localhost:1234/v1', 'context_length': 8192}} moved = build_policy(dict(cwd=str(tmp_path), model='claude-opus-4.6', provider='anthropic'), cfg) assert 'context_length' not in moved.config()['model'] alias = build_policy(dict(cwd=str(tmp_path), model='qwen', provider='vllm'), cfg) assert alias.config()['model']['context_length'] == 8192 def test_provider_pick_over_auto_template_sends_its_key_to_its_own_endpoint(tmp_path, monkeypatch): """#139018: the install template (`provider: auto` + OpenRouter base_url) plus a Desktop Google AI Studio pick resolved provider=gemini against openrouter.ai, so the Google key went to OpenRouter (401). The frozen route must resolve to Google's own endpoint.""" from gateway.session_policy import build_policy from hermes_cli.runtime_provider import frozen_runtime_config, resolve_runtime_provider monkeypatch.setenv('GOOGLE_API_KEY', 'sk-google') monkeypatch.setenv('OPENROUTER_API_KEY', 'sk-openrouter') cfg = {'model': {'default': 'anthropic/claude-opus-4.6', 'provider': 'auto', 'base_url': 'https://openrouter.ai/api/v1'}} policy = build_policy(dict(cwd=str(tmp_path), source='gui', model='gemini-flash-latest', provider='gemini'), cfg) with frozen_runtime_config(policy.config()): runtime = resolve_runtime_provider(requested='gemini', target_model='gemini-flash-latest') assert (runtime['provider'], runtime['api_key']) == ('gemini', 'sk-google') assert 'openrouter.ai' not in runtime['base_url'] assert runtime['base_url'].startswith('https://generativelanguage.googleapis.com/') def test_provider_alias_pick_keeps_the_profiles_endpoint(tmp_path, monkeypatch): """`--provider vllm` over `provider: custom` + a local base_url is the same provider under an alias: the pick must keep the configured endpoint and key, not fail with no endpoint.""" from gateway.session_policy import build_policy from hermes_cli.runtime_provider import frozen_runtime_config, resolve_runtime_provider for var in ('CUSTOM_BASE_URL', 'OPENAI_BASE_URL'): monkeypatch.delenv(var, raising=False) monkeypatch.setenv('LOCAL_LLM_KEY', 'sk-local') cfg = {'model': {'default': 'qwen', 'provider': 'custom', 'base_url': 'http://localhost:1234/v1', 'key_env': 'LOCAL_LLM_KEY'}} policy = build_policy(dict(cwd=str(tmp_path), model='qwen', provider='vllm'), cfg) with frozen_runtime_config(policy.config()): runtime = resolve_runtime_provider(requested='vllm', target_model='qwen') assert (runtime['base_url'], runtime['api_key']) == ('http://localhost:1234/v1', 'sk-local') def test_custom_pick_over_unrecorded_or_auto_profile_keeps_its_endpoint(tmp_path, monkeypatch): """A profile block with a base_url and no provider (or `provider: auto`) already routes to `custom` at that URL, so a `custom` pick is the same route: it must keep the endpoint and key pointer instead of dropping them and failing with no endpoint configured.""" from gateway.session_policy import build_policy from hermes_cli.runtime_provider import frozen_runtime_config, resolve_runtime_provider for var in ('CUSTOM_BASE_URL', 'OPENAI_BASE_URL'): monkeypatch.delenv(var, raising=False) monkeypatch.setenv('LOCAL_KEY', 'sk-local') for held in ({}, {'provider': 'auto'}): cfg = {'model': {'default': 'qwen', 'base_url': 'http://localhost:1234/v1', 'key_env': 'LOCAL_KEY', **held}} policy = build_policy(dict(cwd=str(tmp_path), model='qwen', provider='custom'), cfg) assert policy.config()['model']['key_env'] == 'LOCAL_KEY', held with frozen_runtime_config(policy.config()): runtime = resolve_runtime_provider(requested='custom', target_model='qwen') assert (runtime['provider'], runtime['base_url'], runtime['api_key']) == \ ('custom', 'http://localhost:1234/v1', 'sk-local'), held def test_provider_alias_spelling_and_auto_pick_keep_the_profiles_route(tmp_path, monkeypatch): """`deepinfra` and `deep-infra` are one provider, and an explicit `auto` pick (the thin-client sentinel) means "the profile's route": neither drops the configured proxy and wire mode.""" from gateway.session_policy import build_policy from hermes_cli.runtime_provider import frozen_runtime_config, resolve_runtime_provider for var in ('CUSTOM_BASE_URL', 'OPENAI_BASE_URL'): monkeypatch.delenv(var, raising=False) monkeypatch.setenv('DEEPINFRA_API_KEY', 'sk-deepinfra') for held, picked in (('deepinfra', 'deep-infra'), ('deep-infra', 'deepinfra')): cfg = {'model': {'default': 'm', 'provider': held, 'base_url': 'http://localhost:4321/v1', 'api_mode': 'chat_completions'}} model = build_policy(dict(cwd=str(tmp_path), model='m', provider=picked), cfg).config()['model'] assert (model['base_url'], model['api_mode']) == ('http://localhost:4321/v1', 'chat_completions'), picked monkeypatch.setenv('LOCAL_KEY', 'sk-local') cfg = {'model': {'default': 'qwen', 'provider': 'custom', 'base_url': 'http://localhost:1234/v1', 'key_env': 'LOCAL_KEY'}} monkeypatch.delenv('DEEPINFRA_API_KEY') policy = build_policy(dict(cwd=str(tmp_path), model='qwen', provider='auto'), cfg) assert policy.config()['model']['provider'] == 'custom' with frozen_runtime_config(policy.config()): runtime = resolve_runtime_provider(target_model='qwen') assert (runtime['provider'], runtime['base_url'], runtime['api_key']) == \ ('custom', 'http://localhost:1234/v1', 'sk-local') def test_bare_pick_of_a_shadowed_custom_entry_keeps_the_profiles_route(tmp_path, monkeypatch): """#140107: an entry named like a canonical built-in (`providers: deepseek`, or the legacy list) is shadowed by the built-in in raw provider resolution, so a Desktop pick of its row arrives as the entry's bare key. Freezing that spelling pins the session to the built-in — a different provider than the row the user picked — and drops the entry's endpoint and key pointer with it. The entry's own alias set is what says the pick is the same route.""" from gateway.session_policy import build_policy for var in ('DEEPSEEK_API_KEY', 'OPENAI_API_KEY'): monkeypatch.delenv(var, raising=False) new_style = {'providers': {'deepseek': {'name': 'deepseek', 'api': 'https://api.deepseek.com', 'key_env': 'DEEPSEEK_ENTRY_KEY', 'transport': 'chat_completions', 'default_model': 'deepseek-flash'}}} legacy = {'custom_providers': [{'name': 'deepseek', 'base_url': 'https://api.deepseek.com', 'key_env': 'DEEPSEEK_ENTRY_KEY', 'model': 'deepseek-flash'}]} for extra in (new_style, legacy): cfg = {**extra, 'model': {'default': 'deepseek-flash', 'provider': 'custom:deepseek', 'base_url': 'https://api.deepseek.com', 'key_env': 'DEEPSEEK_ENTRY_KEY'}} model = build_policy(dict(cwd=str(tmp_path), model='deepseek-flash', provider='deepseek'), cfg).config()['model'] assert model['provider'] == 'custom:deepseek', extra assert model['base_url'] == 'https://api.deepseek.com', extra assert model['key_env'] == 'DEEPSEEK_ENTRY_KEY', extra def test_unlimited_max_turns_spellings_create_an_uncapped_session(tmp_path): """`--max-turns 0` / -1 / none / unlimited meant "no cap" before the gateway cutover: session.create freezes them as the unlimited budget, and `hermes chat` sends 0 instead of dropping it as falsy.""" import argparse import sys from gateway.session_policy import build_policy from hermes_cli.gateway_chat import _launch_flags, _requested_policy cfg = {'agent': {'max_turns': 8}} for spelling in (0, -1, 'none', 'unlimited', '0'): assert build_policy(dict(cwd=str(tmp_path), max_turns=spelling), cfg).max_turns == sys.maxsize, spelling assert build_policy(dict(cwd=str(tmp_path), max_turns='12'), cfg).max_turns == 12 for zero in (0, -1): args = argparse.Namespace(max_turns=zero, model=None, ignore_rules=False, yolo=False) assert _launch_flags(args) == _requested_policy(args) == {'max_turns': zero} def test_explicit_key_is_private_and_missing_after_restart_fails_closed(tmp_path): import json from dataclasses import asdict from types import SimpleNamespace from gateway.session_policy import build_policy, bind_launch_key, launch_key from hermes_state_runtime import RuntimeStoreError authority = SimpleNamespace(instance_id='owned', profile_id='profile', epoch=1) sibling = SimpleNamespace(instance_id='sibling', profile_id='profile', epoch=1) raw = 'UNIQUE-PRIVATE-LAUNCH-KEY' before = dict(os.environ) params = dict(cwd=str(tmp_path), model='fixture', api_key=raw) private = {} policy = build_policy(params, {'model': {'api_key': raw}}, private_secrets=private) policy = bind_launch_key(authority, 'session-a', policy, raw, config_secrets=private) assert policy.config(authority)['model']['api_key'] == raw assert raw not in json.dumps(asdict(policy)) assert launch_key(authority, policy) == raw retry_private = {} retry = build_policy(params, {'model': {'api_key': raw}}, private_secrets=retry_private) assert bind_launch_key(authority, 'session-a', retry, raw, config_secrets=retry_private) == policy for other in (sibling, SimpleNamespace(instance_id='owned', profile_id='profile', epoch=2)): with pytest.raises(RuntimeStoreError, match='launch_credentials_unavailable'): launch_key(other, policy) with pytest.raises(RuntimeStoreError, match='launch_credentials_unavailable'): policy.config(other) with pytest.raises(RuntimeStoreError, match='admission_conflict'): bind_launch_key(authority, 'session-a', build_policy(params, {}), 'different') assert dict(os.environ) == before def test_ordinary_daemon_cli_launch_policy(tmp_path): import subprocess import sys import json from pathlib import Path root = Path(__file__).resolve().parents[2] result = subprocess.run([sys.executable, '-c', 'import json,sys; from pathlib import Path; ' 'from tests.gateway.fixtures.cli_launch_policy_probe import probe; ' 'print(json.dumps(probe(Path(sys.argv[1]))))', str(tmp_path)], cwd=root, stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=180, check=False) assert result.returncode == 0, result.stdout + result.stderr print(json.dumps(json.loads(result.stdout.splitlines()[-1]))) def test_null_config_sections_read_as_absent(tmp_path): # A fresh install's config.yaml carries a bare ``gateway:`` (YAML null); the managed-worker gate chains # ``.get('gateway', {}).get(...)`` on the frozen policy and must read the default, not crash the turn. from gateway.session_policy import build_policy, LocalSessionPolicy from dataclasses import replace import json cfg = {'gateway': None, 'display': None, 'platform_toolsets': {'cli': ['terminal']}} policy = build_policy({'source': 'cli', 'cwd': str(tmp_path)}, cfg) assert policy.config().get('gateway', {}).get('managed_workers') is None legacy = replace(policy, config_json=json.dumps(cfg)) # persisted before normalization assert isinstance(legacy, LocalSessionPolicy) assert legacy.config().get('display', {}).get('busy_input_mode', 'interrupt') == 'interrupt' def test_frozen_route_keeps_its_endpoint_after_live_config_edit(tmp_path, monkeypatch): """R2-M2: the frozen policy's config-derived endpoint is the one its frozen credential belongs to; a later ``model.base_url`` edit must not redirect that credential.""" import json from types import SimpleNamespace from gateway.session_policy import build_policy, bind_launch_key from gateway.run_turn_prepare import GatewayTurnPrepareMixin from hermes_cli.config_effective import load_user_config_effective home = tmp_path / 'home' home.mkdir() monkeypatch.setenv('HERMES_HOME', str(home)) def write(url, key): (home / 'config.yaml').write_text(json.dumps( {'model': {'provider': 'custom', 'base_url': url, 'api_key': key, 'default': 'm'}})) write('http://127.0.0.1:1/v1', 'sk-frozen-endpoint-one') private = {} policy = build_policy({'cwd': str(tmp_path), 'model': 'm'}, load_user_config_effective(home / 'config.yaml'), private_secrets=private) authority = SimpleNamespace(instance_id='i', epoch=1, profile_id='p', db=None) policy = bind_launch_key(authority, 'sid', policy, None, config_secrets=private) write('http://127.0.0.1:2/v1', 'sk-new-endpoint-two') monkeypatch.setattr('gateway.session_policy.policy_for_source', lambda runner, source: policy) runner = SimpleNamespace(session_authority=authority) _, runtime = GatewayTurnPrepareMixin._resolve_session_agent_runtime(runner, source=SimpleNamespace()) assert (runtime['base_url'], runtime['api_key']) == ('http://127.0.0.1:1/v1', 'sk-frozen-endpoint-one') def test_frozen_bare_custom_route_keeps_its_endpoint_pool_and_launch_key_wins(tmp_path, monkeypatch): """The frozen ``model.api_key`` is config, not a launch key: a URL-matched credential pool still serves the frozen endpoint (refresh/rotation on 401/429), while a bound launch key wins (R2-M1).""" import json from types import SimpleNamespace from gateway.session_policy import build_policy, bind_launch_key from gateway.run_turn_prepare import GatewayTurnPrepareMixin from hermes_cli.config_effective import load_user_config_effective home = tmp_path / 'home' home.mkdir() monkeypatch.setenv('HERMES_HOME', str(home)) url = 'http://127.0.0.1:1/v1' (home / 'config.yaml').write_text(json.dumps( {'model': {'provider': 'custom', 'base_url': url, 'api_key': 'sk-frozen-config', 'default': 'm'}})) pool = object() monkeypatch.setattr('hermes_cli.runtime_provider._try_resolve_from_custom_pool', lambda base_url, *a, **k: { 'provider': 'custom', 'api_mode': 'chat_completions', 'base_url': base_url, 'api_key': 'sk-pooled', 'source': 'pool:custom', 'credential_pool': pool} if base_url.rstrip('/') == url else None) authority = SimpleNamespace(instance_id='i', epoch=1, profile_id='p', db=None) runner = SimpleNamespace(session_authority=authority) runtimes = [] for sid, launch in (('config-keyed', None), ('launch-keyed', 'sk-launch-explicit')): private = {} params = {'cwd': str(tmp_path), 'model': 'm'} | ({'api_key': launch} if launch else {}) policy = build_policy(params, load_user_config_effective(home / 'config.yaml'), private_secrets=private) policy = bind_launch_key(authority, sid, policy, launch, config_secrets=private) monkeypatch.setattr('gateway.session_policy.policy_for_source', lambda runner, source, p=policy: p) runtimes.append(GatewayTurnPrepareMixin._resolve_session_agent_runtime(runner, source=SimpleNamespace())[1]) assert (runtimes[0]['base_url'], runtimes[0]['credential_pool']) == (url, pool) assert (runtimes[1]['api_key'], runtimes[1]['credential_pool']) == ('sk-launch-explicit', None) def _frozen_fixture_routes(tmp_path, monkeypatch, *launches): """Freeze ``custom_providers: Fixture`` at endpoint A with a pool-only key in auth.json, freeze one route per launch ``base_url``, then edit the LIVE entry to endpoint B; return each route's runtime.""" import json from types import SimpleNamespace from gateway.session_policy import build_policy, bind_launch_key from gateway.run_turn_prepare import GatewayTurnPrepareMixin from hermes_cli.config_effective import load_user_config_effective home = tmp_path / 'home' home.mkdir() monkeypatch.setenv('HERMES_HOME', str(home)) def write(url): (home / 'config.yaml').write_text(json.dumps({'model': {'provider': 'fixture', 'default': 'm'}, 'custom_providers': [{'name': 'Fixture', 'base_url': url}]})) write('https://a.fixture.example/v1') (home / 'auth.json').write_text(json.dumps({'version': 1, 'providers': {}, 'credential_pool': {'custom:fixture': [ {'id': 'k1', 'label': 'a', 'auth_type': 'api_key', 'priority': 0, 'source': 'manual', 'access_token': 'sk-pool-key-for-endpoint-a'}]}})) authority = SimpleNamespace(instance_id='i', epoch=1, profile_id='p', db=None) policies = [] for n, url in enumerate(launches): private = {} params = {'cwd': str(tmp_path), 'model': 'm', 'provider': 'fixture'} | ({'base_url': url} if url else {}) policy = build_policy(params, load_user_config_effective(home / 'config.yaml'), private_secrets=private) policies.append(bind_launch_key(authority, f's{n}', policy, None, config_secrets=private)) write('https://b.fixture.example/v1') runner = SimpleNamespace(session_authority=authority) runtimes = [] for policy in policies: monkeypatch.setattr('gateway.session_policy.policy_for_source', lambda runner, source, p=policy: p) runtimes.append(GatewayTurnPrepareMixin._resolve_session_agent_runtime(runner, source=SimpleNamespace())[1]) return runtimes def test_frozen_named_route_keeps_its_pool_after_live_endpoint_edit(tmp_path, monkeypatch): """A frozen named route stays on endpoint A after the live entry moves to B; A's pool key goes with it (the pool-key lookup reads the FROZEN entry, not live config) instead of no-key-required.""" (runtime,) = _frozen_fixture_routes(tmp_path, monkeypatch, None) assert (runtime['base_url'], runtime['api_key']) == ('https://a.fixture.example/v1', 'sk-pool-key-for-endpoint-a') assert runtime['credential_pool'] is not None def test_frozen_launch_url_override_never_borrows_the_named_pool(tmp_path, monkeypatch): """R2-M1 under a frozen route: a launch ``base_url`` B for the provider frozen at A never borrows A's pool key, even after the live entry is edited to B.""" (runtime,) = _frozen_fixture_routes(tmp_path, monkeypatch, 'https://b.fixture.example/v1') assert runtime['base_url'] == 'https://b.fixture.example/v1' assert (runtime['api_key'], runtime['credential_pool']) == ('no-key-required', None) def test_lazy_info_reports_the_free_tier_pinned_model(tmp_path, monkeypatch): """A not-yet-built session with no launch model/provider runs on ``nous/welcome`` on the free tier (the agent build pins it), so ``session.info`` says so; an explicit launch model stands.""" from dataclasses import replace from types import SimpleNamespace from gateway.session_local import _lazy_model from gateway.session_policy import build_policy from hermes_cli import anon_auth authority = SimpleNamespace(runner=None, profile_id='fixture') monkeypatch.setattr(anon_auth, 'free_tier_route', lambda: True) configured = replace(build_policy({'cwd': str(tmp_path)}, {'model': {'default': 'cfg-model'}}), model='cfg-model') assert _lazy_model(authority, configured) == anon_auth.GUEST_MODEL assert _lazy_model(authority, build_policy({'cwd': str(tmp_path), 'model': 'pick'}, {})) == 'pick' monkeypatch.setattr(anon_auth, 'free_tier_route', lambda: False) assert _lazy_model(authority, configured) == 'cfg-model' def test_run_budget_is_a_frozen_launch_option(tmp_path): """`hermes chat --run-budget N` (main: AIAgent(run_budget_seconds=N)): frozen as the session's agent.run_budget_seconds; the profile's own value applies without the flag; junk is refused.""" from gateway.session_policy import build_policy from hermes_state_runtime import RuntimeStoreError cfg = {'agent': {'run_budget_seconds': 300}} assert build_policy(dict(cwd=str(tmp_path), run_budget=120), cfg).run_budget_seconds == 120.0 assert build_policy(dict(cwd=str(tmp_path)), cfg).run_budget_seconds == 300.0 assert build_policy(dict(cwd=str(tmp_path)), {}).run_budget_seconds is None for bad in (0, -5, True, 'soon'): with pytest.raises(RuntimeStoreError, match='invalid_params'): build_policy(dict(cwd=str(tmp_path), run_budget=bad), cfg) def test_run_budget_passes_the_session_create_contract(): """The canonical dispatcher refuses undeclared session.create keys before the policy sees them.""" from tui_gateway.contracts.registry import CANONICAL_METHODS, canonical_param_problems from gateway.session_policy import CREATE_FIELDS contract = CANONICAL_METHODS['session.create'] assert canonical_param_problems(contract, {'request_id': 'r', 'source': 'cli', 'run_budget': 120.0}) == [] declared = set(contract.params.model_fields) assert CREATE_FIELDS <= declared | {'request_id'}, CREATE_FIELDS - declared def test_verbose_is_a_frozen_launch_option(tmp_path): """`hermes chat -v` (main: AIAgent(quiet_mode=False)): the session agent's status lines are not quieted.""" from gateway.session_policy import build_policy from hermes_state_runtime import RuntimeStoreError assert build_policy(dict(cwd=str(tmp_path), verbose=True), {}).verbose is True assert build_policy(dict(cwd=str(tmp_path)), {}).verbose is False with pytest.raises(RuntimeStoreError, match='invalid_params'): build_policy(dict(cwd=str(tmp_path), verbose='yes'), {})