fix(acp): expose session-scoped service-tier picker

ACP clients advertise edit, model and reasoning controls but omit speed,
so editors cannot select the backend's existing service tiers.

Read eligible choices from the owning gateway under its profile and
credential scope, and apply picks through the validated session fast
setter without changing profile defaults. Publish tier changes to attached
viewers and read the requested next-turn value independently of active
agent metadata. Pass the reconnecting client through snapshot projection
to avoid re-entering the connection lock.

Cover route/account eligibility, session isolation, reloads, cross-viewer
updates, older owners and reconnect projection. Update ACP documentation
and regenerate the canonical gateway contract.

acp_adapter/gateway_server.py

@@ -117,7 +117,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         self._titles = {}
         # Session -> the mode (edit approval policy) the editor was last shown (``current_mode_update``).
         self._modes = {}
-        # Session -> the mode / model / effort the editor's ``configOptions`` last showed (``config_option_update``).
+        # Session -> the settings the editor's ``configOptions`` last showed (``config_option_update``).
         self._configs = {}
         # Session -> the owner's latest usage dict (``session.usage`` / ``session.stats``).
         self._usage = {}
@@ -193,7 +193,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
                 logger.info("ACP session %s not re-attached after reconnect", session_id, exc_info=True)
                 self._snapshots.pop(session_id, None)
                 continue
-            await self._adopt(session_id, snapshot)
+            await self._adopt(session_id, snapshot, client=gateway)
             # Before this session's cards are offered again: until ``_client`` returns, ``self._gateway`` is
             # not this connection, so a card answered while the request is out would lose its answer.
             await self._request_live_reasoning(gateway, session_id)
@@ -201,7 +201,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
                 self._permission(session_id, pending)
             await self._notify_unknown(session_id)
 
-    async def _adopt(self, session_id, snapshot, resumes=None):
+    async def _adopt(self, session_id, snapshot, resumes=None, client=None):
         """Install a fresh ``session.resume`` snapshot as a session's projection: every install (create, load,
         fork, a control's refresh, reconnect, replay gap) goes through here.
 
@@ -213,7 +213,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             if resumes is not None:
                 self._adopted[session_id] = resumes
             self._changed.notify_all()
-        await self._project_mode(session_id, self._editor_id(session_id), snapshot.get("info") or {})
+        await self._project_mode(session_id, self._editor_id(session_id), snapshot.get("info") or {}, client=client)
 
     async def _request_live_reasoning(self, client, session_id):
         """Ask the owner to stream this session's reasoning (``show_reasoning``), as the CLI does.
@@ -299,7 +299,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
     async def _session_fields(self, session_id, info, editor_id):
         """Main's new/load/resume/fork reply fields (model picker, modes, provenance ``_meta``); the
         command list and context indicator follow once the reply is queued, as main scheduled them."""
-        from acp_adapter.gateway_session_meta import config_options, model_state, provenance_meta, session_modes
+        from acp_adapter.gateway_session_meta import model_state, provenance_meta, session_config_options, session_modes
         from acp_adapter.modes import MODE_TO_EDIT_APPROVAL_POLICY
         title = info.get("title")
         if isinstance(title, str) and title.strip():
@@ -314,7 +314,8 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         modes = session_modes({"edit_approval_policy": MODE_TO_EDIT_APPROVAL_POLICY[self._modes.setdefault(
             session_id, session_modes(info).current_mode_id)]})
         return {"models": models, "modes": modes, "field_meta": meta,
-                "config_options": config_options(models, info, modes.current_mode_id)}
+                "config_options": await session_config_options(
+                    self._gateway, session_id, models, info, modes.current_mode_id)}
 
     async def _after_attach(self, session_id, editor_id):
         from acp_adapter.gateway_session_meta import available_commands_update
@@ -577,11 +578,13 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
     async def set_config_option(self, config_id, session_id, value=None, **kwargs):
         """The pickers ``config_options`` advertises: the edit mode is the modes select's owner setting,
         the model ``set_session_model``'s pick, the effort the owner's ``reasoning`` mutation (``default``
-        = ``/reasoning reset``). Another id is accepted so editors do not fail on their own config writes,
+        = ``/reasoning reset``), and the tier its session-scoped speed setting.
+        Another id is accepted so editors do not fail on their own config writes,
         as main did. The reply is the full list, read back from the owner."""
         from acp.schema import SetSessionConfigOptionResponse
         from acp_adapter.gateway_session_meta import (
-            REASONING_CONFIG_ID, config_options, current_mode, model_state, reasoning_choices,
+            REASONING_CONFIG_ID, SERVICE_TIER_CONFIG_ID, SERVICE_TIER_LABELS, current_mode, model_state,
+            reasoning_choices, session_config_options,
         )
         from acp_adapter.modes import EDIT_APPROVAL_POLICY_CONFIG_ID, EDIT_APPROVAL_POLICY_TO_MODE
         editor_id, session_id = session_id, self._held_session(session_id)
@@ -598,10 +601,15 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
                 raise GatewayClientError("invalid_params")
             await self._idle_control(session_id, "reasoning", {"effort": None if value == "default" else value},
                                      "Reasoning effort")
+        elif config_id == SERVICE_TIER_CONFIG_ID:
+            if not isinstance(value, str) or value not in SERVICE_TIER_LABELS:
+                raise GatewayClientError('invalid_params')
+            self._unacked.pop(session_id, None)
+            await (await self._client()).rpc('config.set', session_id=session_id, key='fast', value=value)
         info = await (await self._client()).rpc("session.info", session_id=session_id)
         models = await asyncio.to_thread(model_state, info)
-        return SetSessionConfigOptionResponse(config_options=config_options(
-            models, info, self._modes.get(session_id) or current_mode(info)))
+        return SetSessionConfigOptionResponse(config_options=await session_config_options(
+            self._gateway, session_id, models, info, self._modes.get(session_id) or current_mode(info)))
 
     async def _set_edit_policy(self, session_id, policy):
         """The owner's ``edit_policy`` mutation; every surface attached to the session reads the
@@ -1071,7 +1079,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         await self._project_title(sid, editor_id, payload.get("title"))
         await self._project_mode(sid, editor_id, payload)
 
-    async def _project_mode(self, sid, editor_id, payload):
+    async def _project_mode(self, sid, editor_id, payload, client=None):
         """Another surface's ``edit_policy`` change (its ``session.info``) as ``current_mode_update``."""
         from acp_adapter.modes import EDIT_APPROVAL_POLICY_TO_MODE
         mode = EDIT_APPROVAL_POLICY_TO_MODE.get(payload.get("edit_approval_policy"))
@@ -1082,14 +1090,14 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
                 from acp.schema import CurrentModeUpdate  # ``acp`` does not re-export ``update_current_mode``
                 await self._conn.session_update(session_id=editor_id, update=CurrentModeUpdate(
                     session_update="current_mode_update", current_mode_id=mode))
-        await self._project_config(sid, editor_id, payload)
+        await self._project_config(sid, editor_id, payload, client=client)
 
-    async def _project_config(self, sid, editor_id, payload):
+    async def _project_config(self, sid, editor_id, payload, client=None):
         """Zed draws its pickers only from ``configOptions`` once an agent sends them (it then ignores
-        ``current_mode_update``), so a mode, model or effort change from another surface or a typed
-        ``/model`` / ``/reasoning`` resends the full list. A partial ``session.info`` keeps what it omits."""
-        from acp_adapter.gateway_session_meta import config_options, current_mode, model_state
-        fields = ("model", "provider", "reasoning_effort")
+        ``current_mode_update``), so a mode, model, effort or tier change from another surface resends
+        the full list. A partial ``session.info`` keeps what it omits."""
+        from acp_adapter.gateway_session_meta import current_mode, model_state, session_config_options
+        fields = ("model", "provider", "reasoning_effort", "service_tier")
         if not any(key in payload for key in (*fields, "edit_approval_policy")):
             return
         seen = self._configs.get(sid)
@@ -1103,10 +1111,11 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             return
         from acp.schema import ConfigOptionUpdate
         models = await asyncio.to_thread(model_state, shown)
+        options = await session_config_options(client or self._gateway, sid, models, shown, shown["mode"])
         if self._configs.get(sid) is not shown:
             return  # a newer projection (an event or ``_adopt``) replaced it during the build and sends its own
         await self._conn.session_update(session_id=editor_id, update=ConfigOptionUpdate(
-            session_update="config_option_update", config_options=config_options(models, shown, shown["mode"])))
+            session_update="config_option_update", config_options=options))
 
     async def _project_usage(self, sid, editor_id, stats):
         """The owner's usage dict: kept for the turn's ``PromptResponse.usage``, shown as ``usage_update``."""

acp_adapter/gateway_session_meta.py

@@ -70,6 +70,8 @@ def session_modes(info: dict | None = None) -> SessionModeState:
 
 
 REASONING_CONFIG_ID = "reasoning_effort"
+SERVICE_TIER_CONFIG_ID = "service_tier"
+SERVICE_TIER_LABELS = {"normal": "Normal", "fast": "Fast", "auto": "Auto", "cold": "Cold", "ultrafast": "Ultrafast"}
 
 
 def reasoning_choices() -> tuple[str, ...]:
@@ -98,6 +100,31 @@ def config_options(models: SessionModelState | None, info: dict, mode_id: str) -
     return options
 
 
+async def session_config_options(client, session_id, models, info, mode_id):
+    """Project the owner's eligible speed choices alongside the existing ACP pickers."""
+    from hermes_cli.gateway_client import GatewayRPCError
+    from hermes_cli.models import model_supports_fast_mode
+    options = config_options(models, info, mode_id)
+    if not model_supports_fast_mode(info.get('model')):
+        return options
+    try:
+        speed = await client.rpc('config.get', session_id=session_id, key='service_tier')
+    except GatewayRPCError as exc:
+        if str(exc) != 'invalid_params':
+            raise
+        return options  # Older owners do not serve the speed-choice read.
+    choices = speed.get('options') or []
+    if choices:
+        current = speed['value']
+        rows = [SessionConfigSelectOption(value=value, name=SERVICE_TIER_LABELS[value]) for value in choices]
+        if current not in choices:
+            rows.append(SessionConfigSelectOption(value=current,
+                        name=f"{SERVICE_TIER_LABELS.get(current, current)} (unavailable on this route)"))
+        options.append(SessionConfigOptionSelect(id=SERVICE_TIER_CONFIG_ID, name='Service tier', category='other',
+                                                type='select', current_value=current, options=rows))
+    return options
+
+
 def model_state(info: dict) -> SessionModelState | None:
     """The picker for the session's model (``session.info``), from the shared inventory; a single
     current-model row when nothing is listable, as main's ``_build_model_state``. Blocking."""

apps/shared/src/gateway-contract.generated.ts

@@ -5287,6 +5287,7 @@ export interface CanonicalConfigGetResult {
   mcp_rev?: string | null
   key?: string | null
   scope?: string | null
+  options?: ('normal' | 'fast' | 'auto' | 'cold' | 'ultrafast')[] | null
 }
 /** Session controls (``busy`` / ``verbose`` / ``yolo`` / ``show_reasoning`` / ``service_tier``, which need ``session_id``) and the TUI's profile preferences (``gateway/session_tui_config.py``; ``session_id`` optional). ``model`` is recognised only to be refused with ``use_session_mutation_model``: a model pick is the revision-fenced ``session.mutate`` operation, never a config write. */
 export interface CanonicalConfigSetParams {

apps/shared/src/gateway-contract.openrpc.json

@@ -8200,6 +8200,28 @@
             ],
             "default": null,
             "title": "Scope"
+          },
+          "options": {
+            "anyOf": [
+              {
+                "items": {
+                  "enum": [
+                    "normal",
+                    "fast",
+                    "auto",
+                    "cold",
+                    "ultrafast"
+                  ],
+                  "type": "string"
+                },
+                "type": "array"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "title": "Options"
           }
         },
         "title": "CanonicalConfigGetResult",

gateway/session_busy_controls.py

@@ -122,6 +122,7 @@ def service_tier_config(connection, ref, params):
     authorize(connection, ref, params, 'session:control')
     live = connection.authority.sessions[ref.session_id]
     connection.authority.runner._set_session_service_tier_override(live.route, tier or None)
+    live.event_stream.publish(ref.session_id, {'service_tier': tier or ''}, event_type='session.info')
     return {'key': 'service_tier', 'value': tier or 'normal', 'scope': 'session'}
 
 

gateway/session_config.py

@@ -98,6 +98,7 @@ async def config_get(connection, ref, params):
                                  'mcp_rev': mcp_revision(policy, cfg)},
                 'reasoning': lambda: _reasoning(agent, policy, cfg, connection, ref),
                 'fast': lambda: _fast(connection, ref, policy, agent, cfg),
+                'service_tier': lambda: _service_tier(connection, ref, policy),
                 'daybreak': lambda: {'value': (cfg.get('agent') or {}).get('daybreak') is True},
                 'skin': lambda: {'value': display.get('skin', 'default')},
                 'theme': lambda: {'value': display.get('tui_theme', 'auto')},
@@ -187,6 +188,15 @@ def _fast(connection, ref, policy, agent, cfg):
     return {'value': {'priority': 'fast', 'ultrafast': 'ultrafast'}.get(tier, 'normal')}
 
 
+def _service_tier(connection, ref, policy):
+    from agent.fast_mode import service_tier_word
+    from gateway.session_tui_config import current_tier, service_tier_options
+    if not ref.session_id:
+        raise RuntimeStoreError('invalid_params')
+    return {'value': service_tier_word(current_tier(connection, ref, policy)),
+            'options': service_tier_options(connection, ref, policy)}
+
+
 def _reasoning(agent, policy, cfg, connection=None, ref=None):
     from hermes_constants import resolve_reasoning_config
     value = getattr(agent, 'reasoning_config', None)

gateway/session_tui_config.py

@@ -230,6 +230,25 @@ def current_tier(connection, ref, policy):
     return runner._resolve_session_service_tier(source=live.source, session_key=live.route)
 
 
+def service_tier_options(connection, ref, policy, *, include_ultrafast=True):
+    """The session route's speed choices, resolved under its credential scope."""
+    from agent.codex_service_tiers import ultrafast_offered
+    from gateway.run import _profile_runtime_scope
+    from hermes_cli.models import resolve_fast_mode_overrides
+    with _profile_runtime_scope(Path(connection.authority.profile_id)):
+        agent = connection.authority.agent(ref)
+        model = getattr(agent, 'model', None) or (policy.current_model(connection.authority) if policy else None)
+        provider = getattr(agent, 'provider', None) or (policy.provider if policy else None)
+        base_url = getattr(agent, 'base_url', None) or (policy.base_url if policy else None)
+        if not model or resolve_fast_mode_overrides(model, provider=provider, base_url=base_url) is None:
+            return []
+        options = ['normal', 'fast', 'auto', 'cold']
+        if include_ultrafast and ultrafast_offered(model, provider=provider, base_url=base_url,
+                                                   access_token=getattr(agent, 'api_key', None)):
+            options.append('ultrafast')
+        return options
+
+
 async def _set_fast(connection, ref, policy, value):
     """Session-scoped /fast like main's: validated against the session's route, never config.yaml.
     The words parse through the shared tier table (``parse_exact_service_tier``, as the classic CLI's
@@ -248,23 +267,12 @@ async def _set_fast(connection, ref, policy, value):
     except ValueError:
         raise RuntimeStoreError('invalid_params') from None
     if tier in STATIC_TIERS:
-        agent = connection.authority.agent(ref)
-        model = getattr(agent, 'model', None) or (policy.current_model(connection.authority) if policy else None)
-        provider = getattr(agent, 'provider', None) or (policy.provider if policy else None)
-        base_url = getattr(agent, 'base_url', None) or (policy.base_url if policy else None)
-
-        def offered():
-            from agent.codex_service_tiers import ultrafast_offered
-            from hermes_cli.models import resolve_fast_mode_overrides
-            if not model or resolve_fast_mode_overrides(model, provider=provider, base_url=base_url,
-                                                        tier=tier if tier == 'ultrafast' else None) is None:
-                return False
-            return tier != 'ultrafast' or ultrafast_offered(model, provider=provider, base_url=base_url,
-                                                            access_token=getattr(agent, 'api_key', None))
-        if not await asyncio.to_thread(offered):
+        options = await asyncio.to_thread(service_tier_options, connection, ref, policy,
+                                          include_ultrafast=tier == 'ultrafast')
+        if service_tier_word(tier) not in options:
             raise RuntimeStoreError('fast_unavailable')
-    connection.authority.runner._set_session_service_tier_override(
-        connection.authority.sessions[ref.session_id].route, tier)
+    from gateway.session_busy_controls import service_tier_config
+    service_tier_config(connection, ref, {'session_id': ref.session_id, 'value': raw})
     return {'key': 'fast', 'value': service_tier_word(tier)}
 
 

tests/acp_adapter/test_gateway_service_tier.py

@@ -0,0 +1,122 @@
+"""Service-tier choices and selections travel through the real ACP/gateway connection."""
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+
+from acp_adapter.gateway_server import GatewayACPAgent
+from hermes_cli.gateway_client import GatewayRPCError
+from tests.acp_adapter.test_gateway_sessions import editor, viewer
+from tests.acp_adapter.test_gateway_session_metadata import _updates
+from tests.waiting import async_wait_until
+
+
+def _tier(result):
+    options = {option['id']: option for option in result['configOptions']}
+    assert 'service_tier' in options, options
+    return options['service_tier']
+
+
+@pytest.mark.platforms('linux')
+@pytest.mark.asyncio
+@pytest.mark.parametrize('daemon', [{
+    'model': {'provider': 'openai', 'default': 'gpt-6.1-sol', 'base_url': 'https://api.openai.com/v1'},
+}], indirect=True)
+async def test_tier_picker_changes_only_its_session_and_tracks_other_viewers(daemon, tmp_path, model_peer):
+    home = daemon[0]
+    original_config = (home / 'config.yaml').read_bytes()
+    async with editor(daemon, tmp_path) as acp, viewer(daemon) as ws:
+        await acp.rpc('initialize', protocolVersion=1, clientCapabilities={})
+        created = (await acp.rpc('session/new', cwd=str(tmp_path), mcpServers=[]))['result']
+        sid = created['sessionId']
+        sibling = (await acp.rpc('session/new', cwd=str(tmp_path), mcpServers=[]))['result']['sessionId']
+        option = _tier(created)
+        assert option['currentValue'] == 'normal'
+        assert {'normal', 'fast', 'ultrafast', 'auto', 'cold'} <= {row['value'] for row in option['options']}
+        for selected in ('ultrafast', 'auto', 'cold', 'normal'):
+            changed = await acp.rpc('session/set_config_option', sessionId=sid,
+                                    configId='service_tier', value=selected)
+            assert _tier(changed['result'])['currentValue'] == selected, changed
+            assert (await ws.rpc('config.get', session_id=sid, key='service_tier'))['value'] == selected
+            assert (await ws.rpc('config.get', session_id=sibling, key='service_tier'))['value'] == 'normal'
+        changed = await acp.rpc('session/set_config_option', sessionId=sid,
+                                configId='service_tier', value='ultrafast')
+        assert _tier(changed['result'])['currentValue'] == 'ultrafast'
+        invalid = await acp.rpc('session/set_config_option', sessionId=sid,
+                                configId='service_tier', value='warp')
+        assert 'error' in invalid, invalid
+        assert (await ws.rpc('config.get', session_id=sid, key='service_tier'))['value'] == 'ultrafast'
+        loaded = await acp.rpc('session/load', sessionId=sid, cwd=str(tmp_path), mcpServers=[])
+        assert _tier(loaded['result'])['currentValue'] == 'ultrafast'
+        start = len(acp.frames)
+        await ws.rpc('config.set', session_id=sid, key='fast', value='fast')
+        await async_wait_until(
+            lambda: any(_tier(update)['currentValue'] == 'fast'
+                        for update in _updates(acp, sid, 'config_option_update', start)),
+            timeout=10, what='the other viewer’s service-tier selection')
+        assert (await ws.rpc('config.get', session_id=sid, key='service_tier'))['value'] == 'fast'
+    assert (home / 'config.yaml').read_bytes() == original_config
+    assert not model_peer.requests
+
+
+@pytest.mark.platforms('linux')
+@pytest.mark.asyncio
+async def test_unsupported_route_has_no_tier_picker_and_refuses_a_direct_pick(daemon, tmp_path):
+    async with editor(daemon, tmp_path) as acp:
+        await acp.rpc('initialize', protocolVersion=1, clientCapabilities={})
+        created = (await acp.rpc('session/new', cwd=str(tmp_path), mcpServers=[]))['result']
+        assert 'service_tier' not in {option['id'] for option in created['configOptions']}
+        picked = await acp.rpc('session/set_config_option', sessionId=created['sessionId'],
+                               configId='service_tier', value='ultrafast')
+        assert 'error' in picked, picked
+
+
+@pytest.mark.asyncio
+async def test_older_owner_keeps_existing_pickers_without_a_service_tier_read():
+    from acp_adapter.gateway_session_meta import session_config_options
+
+    owner = AsyncMock()
+    owner.rpc.side_effect = GatewayRPCError('invalid_params')
+    info = {'model': 'gpt-6.1-sol', 'reasoning_effort': 'high'}
+    options = await session_config_options(owner, 'session', None, info, 'default')
+    assert {option.id for option in options} == {'edit_approval_policy', 'reasoning_effort'}
+    owner.rpc.side_effect = GatewayRPCError('permission_denied')
+    with pytest.raises(GatewayRPCError, match='permission_denied'):
+        await session_config_options(owner, 'session', None, info, 'default')
+
+
+@pytest.mark.asyncio
+async def test_an_existing_unavailable_tier_is_not_silently_displayed_as_normal():
+    from acp_adapter.gateway_session_meta import session_config_options
+
+    owner = AsyncMock()
+    owner.rpc.return_value = {'value': 'ultrafast', 'options': ['normal', 'fast']}
+    options = await session_config_options(owner, 'session', None, {'model': 'gpt-6.1-sol'}, 'default')
+    tier = next(option for option in options if option.id == 'service_tier')
+    assert tier.current_value == 'ultrafast'
+    assert next(row.name for row in tier.options if row.value == 'ultrafast') == 'Ultrafast (unavailable on this route)'
+
+
+@pytest.mark.asyncio
+async def test_reconnect_projects_speed_using_the_connection_being_attached(monkeypatch):
+    monkeypatch.setattr('acp_adapter.gateway_session_meta.model_state', lambda info: None)
+    agent = GatewayACPAgent()
+    agent._conn = AsyncMock()
+    agent._snapshots['session'] = {}
+    agent._configs['session'] = {'model': 'gpt-6.1-sol', 'mode': 'default', 'service_tier': ''}
+
+    async def rpc(method, **params):
+        if method == 'session.resume':
+            return {'info': {'model': 'gpt-6.1-sol', 'service_tier': 'ultrafast'}}
+        if method == 'config.get':
+            return {'value': 'ultrafast', 'options': ['normal', 'fast', 'ultrafast']}
+        return {}
+
+    owner = AsyncMock()
+    owner.rpc.side_effect = rpc
+    async with agent._connect_lock, asyncio.timeout(2):
+        await agent._reattach(owner)
+    updates = [call.kwargs['update'] for call in agent._conn.session_update.await_args_list
+               if call.kwargs['update'].session_update == 'config_option_update']
+    assert any(option.id == 'service_tier' and option.current_value == 'ultrafast'
+               for update in updates for option in update.config_options)

tests/gateway/test_session_service_tier_options.py

@@ -0,0 +1,42 @@
+"""The owner advertises service tiers for the selected route and account, never another session."""
+import pytest
+
+from tests.gateway.test_session_create_tier_cwd import _owner
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(('model', 'provider', 'base_url', 'entitled', 'expected'), [
+    ('gpt-6.1-sol', 'openai', 'https://api.openai.com/v1', True, {'normal', 'fast', 'auto', 'cold', 'ultrafast'}),
+    ('gpt-5.4', 'openai', 'https://api.openai.com/v1', True, {'normal', 'fast', 'auto', 'cold'}),
+    ('gpt-6.1-sol', 'openai-codex', 'https://chatgpt.com/backend-api/codex', False,
+     {'normal', 'fast', 'auto', 'cold'}),
+    ('gpt-6.1-sol', 'openai-codex', 'https://chatgpt.com/backend-api/codex', True,
+     {'normal', 'fast', 'auto', 'cold', 'ultrafast'}),
+    ('gpt-6.1-sol', 'custom', 'http://127.0.0.1:12345/v1', True, set()),
+])
+async def test_service_tier_choices_use_the_sessions_model_route_and_account(
+    tmp_path, monkeypatch, model, provider, base_url, entitled, expected,
+):
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr('agent.codex_service_tiers.codex_account_offers_tier', lambda *a, **k: entitled)
+    store, _, _, owner = _owner(tmp_path / 'owner', monkeypatch, {'model': {'default': 'fixture'}})
+    try:
+        created = await owner.dispatch({'id': 'create', 'method': 'session.create', 'params': {
+            'request_id': 'create', 'source': 'gui', 'cwd': str(tmp_path), 'toolsets': [],
+            'model': model, 'provider': provider, 'base_url': base_url,
+        }})
+        sid = created['result']['session_id']
+        reply = await owner.dispatch({'id': 'choices', 'method': 'config.get', 'params': {
+            'session_id': sid, 'key': 'service_tier',
+        }})
+        assert 'result' in reply, reply
+        assert reply['result']['value'] == 'normal'
+        assert set(reply['result']['options']) == expected
+        if 'ultrafast' not in expected:
+            rejected = await owner.dispatch({'id': 'pick', 'method': 'config.set', 'params': {
+                'session_id': sid, 'key': 'fast', 'value': 'ultrafast',
+            }})
+            assert rejected['error']['message'] == 'fast_unavailable'
+    finally:
+        await owner.close()
+        store.close_all_db_handles()

tui_gateway/contracts/canonical_projections.py

@@ -16,6 +16,7 @@ from .registry import canonical_method
 class CanonicalConfigGetResult(ConfigGetResult):
     key: str | None = None
     scope: str | None = None
+    options: list[Literal['normal', 'fast', 'auto', 'cold', 'ultrafast']] | None = None
 
 
 class CanonicalConfigSetParams(ProfileParams):

website/docs/user-guide/features/acp.md

@@ -333,6 +333,19 @@ ACP mode uses the same Hermes configuration as the CLI:
 
 Provider resolution uses Hermes' normal runtime resolver, so ACP inherits the currently configured provider and credentials. Hermes also advertises a terminal auth method (`--setup`) for first-run ACP clients; this opens Hermes' interactive model/provider setup.
 
+### Service-tier picker
+
+ACP clients that render `configOptions` can select **Service tier** for a supported
+model route: **Normal**, **Fast**, **Auto**, **Cold**, and **Ultrafast** when its
+model and account eligibility checks permit it. The gateway supplies the choices
+and validates each selection; unsupported routes do not show the picker.
+
+The selection applies to that session's next turn, not to every Hermes session,
+and does not write `config.yaml`. Reloading the editor preserves the selection
+while the same gateway session remains alive. Changes from another attached
+client update the ACP picker as well. See [the `/fast` reference](../../reference/slash-commands.md)
+for the bounded Auto and Cold modes.
+
 ## Host integration
 
 Older ACP hosts may still set this variable on the Hermes subprocess they