feat: expose runtime model options

src/bub/__init__.py

@@ -12,12 +12,15 @@
 from bub.configure import Settings, config, ensure_config
 from bub.framework import DEFAULT_HOME, BubFramework
 from bub.hookspecs import hookimpl
+from bub.runtime_options import RuntimeChoice, RuntimeOptions
 from bub.tools import tool
 from bub.turn_admission import AdmitDecision, TurnSnapshot
 
 __all__ = [
     "AdmitDecision",
     "BubFramework",
+    "RuntimeChoice",
+    "RuntimeOptions",
     "Settings",
     "TurnSnapshot",
     "config",

src/bub/builtin/hook_impl.py

@@ -9,14 +9,15 @@
 from bub import inquirer as bub_inquirer
 from bub.builtin.agent import Agent
 from bub.builtin.context import default_tape_context
-from bub.builtin.settings import DEFAULT_MODEL
+from bub.builtin.settings import DEFAULT_MODEL, load_settings
 from bub.builtin.steering import InMemorySteeringInbox
 from bub.channels.base import Channel
 from bub.channels.message import ChannelMessage, MediaItem
 from bub.envelope import content_of, field_of
 from bub.framework import BubFramework
 from bub.hookspecs import hookimpl
 from bub.runtime import AsyncStreamEvents
+from bub.runtime_options import RuntimeChoice, RuntimeOptions
 from bub.tape import TapeContext, TapeStore
 from bub.turn_admission import AdmitDecision, TurnSnapshot
 from bub.types import Envelope, MessageHandler, State, SteeringInboxProtocol
@@ -119,6 +120,12 @@ def _default_enabled_channels(current_value: object, available_channels: list[st
             return selected
         return available_channels
 
+    @staticmethod
+    def _configured_models() -> list[str]:
+        settings = load_settings()
+        models = [settings.model, *(settings.fallback_models or [])]
+        return list(dict.fromkeys(model for model in models if model))
+
     @hookimpl
     def resolve_session(self, message: ChannelMessage) -> str:
         session_id = field_of(message, "session_id")
@@ -141,6 +148,8 @@ async def load_state(self, message: ChannelMessage, session_id: str) -> State:
         # fresh/unknown session never inherits another session's model.
         if model := await self._recover_session_model(session_id):
             state["model"] = model
+        if model := field_of(message, "context", {}).get("model"):
+            state["model"] = model
         if thread_id := field_of(message, "context", {}).get("thread_id"):
             state["_runtime_thread_id"] = thread_id
         return state
@@ -252,6 +261,22 @@ def onboard_config(self, current_config: dict[str, object]) -> dict[str, object]
             config["api_base"] = api_base
         return config
 
+    @hookimpl
+    def provide_runtime_options(
+        self,
+        session_id: str,
+        workspace: Path | None = None,
+    ) -> RuntimeOptions | None:
+        del session_id, workspace
+        models = self._configured_models()
+        if not models:
+            return None
+
+        return RuntimeOptions(
+            models=[RuntimeChoice(id=model, name=model) for model in models],
+            current_model=models[0],
+        )
+
     def _read_agents_file(self, state: State) -> str:
         workspace = state.get("_runtime_workspace", str(Path.cwd()))
         prompt_path = Path(workspace) / AGENTS_FILE_NAME

src/bub/framework.py

@@ -18,6 +18,7 @@
 from bub.hook_runtime import _SKIP_VALUE, HookRuntime
 from bub.hookspecs import BUB_HOOK_NAMESPACE, BubHookSpecs
 from bub.runtime import BubError, ErrorKind
+from bub.runtime_options import RuntimeOptions
 from bub.tape import AsyncTapeStore, TapeContext, TapeStore
 from bub.turn_admission import AdmitDecision, TurnSnapshot
 from bub.types import Envelope, MessageHandler, OutboundChannelRouter, State, SteeringInboxProtocol, TurnResult
@@ -238,6 +239,38 @@ async def admit_message(self, *, session_id: str, message: Envelope, turn: TurnS
             return decision
         raise TypeError("hook.admit_message must return AdmitDecision or None")
 
+    async def get_runtime_options(
+        self,
+        *,
+        session_id: str,
+        workspace: str | Path | None = None,
+    ) -> RuntimeOptions:
+        """Collect protocol-neutral runtime choices for one session."""
+
+        resolved_workspace = self._resolve_workspace(workspace)
+        results = await self._hook_runtime.call_many(
+            "provide_runtime_options",
+            session_id=session_id,
+            workspace=resolved_workspace,
+        )
+
+        merged = RuntimeOptions()
+        for result in results:
+            if result is None:
+                continue
+            if not isinstance(result, RuntimeOptions):
+                raise TypeError("hook.provide_runtime_options must return RuntimeOptions or None")
+            merged = RuntimeOptions(
+                models=[*merged.models, *result.models],
+                current_model=merged.current_model or result.current_model,
+            )
+        return merged
+
+    def _resolve_workspace(self, workspace: str | Path | None) -> Path:
+        if workspace is None:
+            return self.workspace
+        return Path(workspace).expanduser().resolve()
+
     async def steer_message(
         self,
         *,

src/bub/hookspecs.py

@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 import pluggy
 
 from bub.runtime import AsyncStreamEvents
+from bub.runtime_options import RuntimeOptions
 from bub.tape import AsyncTapeStore, TapeContext, TapeStore
 from bub.turn_admission import AdmitDecision, TurnSnapshot
 from bub.types import Envelope, MessageHandler, State, SteeringInboxProtocol
@@ -91,6 +93,14 @@ def register_cli_commands(self, app: Any) -> None:
     def onboard_config(self, current_config: dict[str, Any]) -> dict[str, Any] | None:
         """Collect a plugin config fragment for the interactive onboarding command."""
 
+    @hookspec
+    def provide_runtime_options(
+        self,
+        session_id: str,
+        workspace: Path | None,
+    ) -> RuntimeOptions | None:
+        """Provide protocol-neutral runtime choices for a session."""
+
     @hookspec
     def on_error(self, stage: str, error: Exception, message: Envelope | None) -> None:
         """Observe framework errors from any stage."""

src/bub/runtime_options.py

@@ -0,0 +1,24 @@
+"""Protocol-neutral runtime option types."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class RuntimeChoice:
+    """One selectable runtime value."""
+
+    id: str
+    name: str | None = None
+    description: str | None = None
+    meta: dict[str, Any] | None = None
+
+
+@dataclass(frozen=True)
+class RuntimeOptions:
+    """Runtime choices that a channel or adapter may present to a user."""
+
+    models: list[RuntimeChoice] = field(default_factory=list)
+    current_model: str | None = None

tests/test_builtin_hook_impl.py

@@ -69,8 +69,8 @@ def _raise_value_error() -> None:
     raise ValueError("boom")
 
 
-def _build_impl(tmp_path: Path) -> tuple[BubFramework, BuiltinImpl, FakeAgent]:
-    framework = BubFramework()
+def _build_impl(tmp_path: Path, config_file: Path | None = None) -> tuple[BubFramework, BuiltinImpl, FakeAgent]:
+    framework = BubFramework(config_file=config_file) if config_file is not None else BubFramework()
     impl = BuiltinImpl(framework)
     agent = FakeAgent(tmp_path)
     impl._agent = agent
@@ -221,6 +221,25 @@ async def test_run_model_stream_delegates_to_agent(tmp_path: Path) -> None:
     assert agent.run_calls == []
 
 
+@pytest.mark.asyncio
+async def test_runtime_model_override_is_passed_to_agent(tmp_path: Path) -> None:
+    _, impl, agent = _build_impl(tmp_path)
+    message = ChannelMessage(
+        session_id="session",
+        channel="cli",
+        chat_id="room",
+        content="hello",
+        context={"model": "anthropic:claude-sonnet-4-5"},
+    )
+
+    state = await impl.load_state(message=message, session_id="session")
+    stream = await impl.run_model_stream(prompt="prompt", session_id="session", state=state)
+    events = [event async for event in stream]
+
+    assert [(event.kind, event.data) for event in events] == [("text", {"delta": "agent-output"})]
+    assert agent.run_stream_calls == [("session", "prompt", state, "anthropic:claude-sonnet-4-5")]
+
+
 @pytest.mark.asyncio
 async def test_run_model_stream_forwards_state_model_override(tmp_path: Path) -> None:
     """state['model'] must be forwarded as the per-call model override."""
@@ -243,6 +262,27 @@ async def test_run_model_stream_passes_none_when_state_has_no_model(tmp_path: Pa
     assert agent.run_stream_calls[-1][3] is None
 
 
+def test_builtin_provides_model_runtime_options(tmp_path: Path, load_config) -> None:
+    with pytest.MonkeyPatch.context() as monkeypatch:
+        monkeypatch.delenv("BUB_MODEL", raising=False)
+        monkeypatch.delenv("BUB_FALLBACK_MODELS", raising=False)
+        config_file = load_config(
+            """
+model: openai:gpt-5
+fallback_models:
+  - anthropic:claude-sonnet-4-5
+  - openai:gpt-5
+""".strip()
+        )
+        _, impl, _ = _build_impl(tmp_path, config_file=config_file)
+
+        options = impl.provide_runtime_options(session_id="session")
+
+        assert options is not None
+        assert options.current_model == "openai:gpt-5"
+        assert [item.id for item in options.models] == ["openai:gpt-5", "anthropic:claude-sonnet-4-5"]
+
+
 def test_system_prompt_appends_workspace_agents_file(tmp_path: Path) -> None:
     _, impl, _ = _build_impl(tmp_path)
     (tmp_path / AGENTS_FILE_NAME).write_text("local rules", encoding="utf-8")

tests/test_framework.py

@@ -20,6 +20,7 @@
 from bub.framework import BubFramework
 from bub.hookspecs import hookimpl
 from bub.runtime import AsyncStreamEvents, StreamEvent, StreamState
+from bub.runtime_options import RuntimeChoice, RuntimeOptions
 from bub.turn_admission import AdmitDecision, TurnSnapshot
 
 
@@ -312,6 +313,43 @@ def admit_message(self, session_id, message, turn):
     assert decision == AdmitDecision("follow_up", reason="busy")
 
 
+@pytest.mark.asyncio
+async def test_get_runtime_options_collects_models_by_priority(tmp_path: Path) -> None:
+    framework = BubFramework()
+
+    class LowPriorityPlugin:
+        @hookimpl
+        def provide_runtime_options(self, session_id, workspace):
+            assert session_id == "session"
+            assert workspace == tmp_path.resolve()
+            return RuntimeOptions(
+                models=[RuntimeChoice(id="low", name="Low")],
+                current_model="low",
+            )
+
+    class HighPriorityPlugin:
+        @hookimpl
+        def provide_runtime_options(self, session_id, workspace):
+            assert session_id == "session"
+            assert workspace == tmp_path.resolve()
+            return RuntimeOptions(
+                models=[RuntimeChoice(id="high", name="High"), RuntimeChoice(id="mid", name="Mid")],
+                current_model="high",
+            )
+
+    framework._plugin_manager.register(LowPriorityPlugin(), name="low")
+    framework._plugin_manager.register(HighPriorityPlugin(), name="high")
+
+    options = await framework.get_runtime_options(session_id="session", workspace=tmp_path)
+
+    assert [(choice.id, choice.name) for choice in options.models] == [
+        ("high", "High"),
+        ("mid", "Mid"),
+        ("low", "Low"),
+    ]
+    assert options.current_model == "high"
+
+
 @pytest.mark.asyncio
 async def test_process_inbound_streams_when_requested() -> None:  # noqa: C901
     framework = BubFramework()

website/src/content/docs/docs/reference/hooks.mdx

@@ -25,6 +25,7 @@ For the *why* and *how* of each stage see [Turn pipeline](/docs/concepts/turn-pi
 | `dispatch_outbound` | broadcast | `(message: Envelope) -> bool` | sent flag | `process_inbound` per outbound | Each outbound is fanned out to every impl. |
 | `register_cli_commands` | sync-only consumer | `(app: typer.Typer) -> None` | none | `BubFramework.create_cli_app` (`call_many_sync`) | Bootstrap only; async impls log a warning and are skipped. |
 | `onboard_config` | sync-only consumer (custom merge) | `(current_config: dict) -> dict \| None` | config fragment | `BubFramework.collect_onboard_config` | Iterated by priority; each fragment is merged via `configure.merge`. Non-dict returns raise `TypeError`. |
+| `provide_runtime_options` | broadcast | `(session_id: str, workspace: Path \| None) -> RuntimeOptions \| None` | runtime choices | `BubFramework.get_runtime_options` | Model choices are appended in hook priority order. Selection state is owned by the caller or adapter. |
 | `on_error` | observer | `(stage: str, error: Exception, message: Envelope \| None) -> None` | none | `HookRuntime.notify_error` / `notify_error_sync` | Failures inside an `on_error` impl are caught and logged so other observers still run. |
 | `system_prompt` | broadcast (joined) | `(prompt, state) -> str` | prompt fragment | `BubFramework.get_system_prompt` (`call_many_sync`) | Results are reversed and joined with `\n\n`; truthy fragments only. |
 | `provide_tape_store` | firstresult | `() -> TapeStore \| AsyncTapeStore` | tape store | `BubFramework.running()` | Resolved once when the runtime scope opens; sync/async iterators are entered as context managers. |

website/src/content/docs/docs/reference/index.mdx

@@ -10,4 +10,4 @@ This page indexes the four reference tables for Bub's public surface. Each page
 | [Hooks](/docs/reference/hooks/) | Every spec in `BubHookSpecs` with kind, parameters, return type, invocation site. |
 | [CLI](/docs/reference/cli/) | Every `bub` command and subcommand, with options, defaults, and behavior notes. |
 | [Settings](/docs/reference/settings/) | All `BUB_*` environment variables, `pydantic-settings` classes, and `~/.bub/config.yml` keys. |
-| [Types](/docs/reference/types/) | Public types from `bub`: `Envelope`, `State`, `TurnResult`, `Channel`, `OutboundChannelRouter`, `BubFramework`. |
+| [Types](/docs/reference/types/) | Public types from `bub`: `Envelope`, `State`, `TurnResult`, runtime options, `Channel`, `OutboundChannelRouter`, `BubFramework`. |

website/src/content/docs/docs/reference/types.mdx

@@ -95,6 +95,28 @@ class TurnResult:
 
 Returned by `BubFramework.process_inbound`. `prompt` is the resolved prompt. The source annotation is currently `str`, but a `build_prompt` hook may return multimodal content parts and the runtime preserves that list. `outbounds` is the flattened result of every `render_outbound` impl.
 
+## Runtime Option Types
+
+These dataclasses live in `src/bub/runtime_options.py` and are exported from `bub`. They describe protocol-neutral runtime choices that a channel or adapter may present to a user.
+
+```python
+@dataclass(frozen=True)
+class RuntimeChoice:
+    id: str
+    name: str | None = None
+    description: str | None = None
+    meta: dict[str, Any] | None = None
+```
+
+```python
+@dataclass(frozen=True)
+class RuntimeOptions:
+    models: list[RuntimeChoice] = field(default_factory=list)
+    current_model: str | None = None
+```
+
+`id` is the stable value an adapter may use when applying a runtime choice. Bub does not own UI state for these choices; callers store selections and apply them through their own integration.
+
 ## Turn Admission Types
 
 These types are exported from `bub` and defined in `src/bub/turn_admission.py`.
@@ -189,6 +211,9 @@ class BubFramework:
     def get_tape_store(self) -> TapeStore | AsyncTapeStore | None: ...
     def get_system_prompt(self, prompt: str | list[dict], state: dict[str, Any]) -> str: ...
     def hook_report(self) -> dict[str, list[str]]: ...
+    async def get_runtime_options(
+        self, *, session_id: str, workspace: str | Path | None = None
+    ) -> RuntimeOptions: ...
     async def admit_message(
         self, *, session_id: str, message: Envelope, turn: TurnSnapshot
     ) -> AdmitDecision | None: ...
@@ -212,6 +237,7 @@ class BubFramework:
 | `get_tape_store()` | Return the tape store entered by `running()`, or `None` outside the scope. |
 | `get_system_prompt(prompt, state)` | Run `system_prompt` impls (sync), reverse, and join non-empty results with `\n\n`. |
 | `hook_report()` | Map hook name → discovered adapter names. Backs `bub hooks`; read the hook reference before treating this order as runtime precedence. |
+| `get_runtime_options(...)` | Collect protocol-neutral runtime choices from hooks, appending model choices in hook priority order. |
 | `admit_message(...)` | Call the `admit_message` hook and return the selected decision. Used by `ChannelManager`. |
 | `running()` | Async context manager; resolves `provide_tape_store` once and binds the resulting store for the duration. |
 | `bind_outbound_router(router)` | Attach (or detach with `None`) the `OutboundChannelRouter`. The `ChannelManager` calls this on start/stop. |
@@ -228,6 +254,8 @@ From `src/bub/__init__.py`:
 | --- | --- | --- |
 | `BubFramework` | class | Framework runtime (above). |
 | `AdmitDecision` | dataclass | Decision returned by `admit_message`. |
+| `RuntimeChoice` | dataclass | One selectable runtime value. |
+| `RuntimeOptions` | dataclass | Available runtime choices for models. |
 | `Settings` | class | Base class for plugin settings (re-exported from `bub.configure`). |
 | `TurnSnapshot` | dataclass | Snapshot passed to `admit_message`. |
 | `config` | decorator | `@config(name="...")` registers a settings class for YAML/env validation. |

website/src/content/docs/zh-cn/docs/reference/hooks.mdx

@@ -25,6 +25,7 @@ description: BubHookSpecs 中每个钩子的类型、签名、返回值与调用
 | `dispatch_outbound` | broadcast | `(message: Envelope) -> bool` | sent flag | `process_inbound` per outbound | 每个 outbound 都会广播给所有实现。 |
 | `register_cli_commands` | sync-only consumer | `(app: typer.Typer) -> None` | none | `BubFramework.create_cli_app` (`call_many_sync`) | 仅用于启动期;async 实现会被跳过并产生告警。 |
 | `onboard_config` | sync-only consumer (custom merge) | `(current_config: dict) -> dict \| None` | config fragment | `BubFramework.collect_onboard_config` | 按优先级遍历;每个返回值通过 `configure.merge` 合并;非 dict 返回会抛出 `TypeError`。 |
+| `provide_runtime_options` | broadcast | `(session_id: str, workspace: Path \| None) -> RuntimeOptions \| None` | runtime choices | `BubFramework.get_runtime_options` | 模型选项会按 hook 优先级顺序追加。选择状态由调用方或 adapter 自己持有。 |
 | `on_error` | observer | `(stage: str, error: Exception, message: Envelope \| None) -> None` | none | `HookRuntime.notify_error` / `notify_error_sync` | `on_error` 实现内部抛出的异常会被吞掉并写日志,确保其他观察者继续运行。 |
 | `system_prompt` | broadcast (joined) | `(prompt, state) -> str` | prompt fragment | `BubFramework.get_system_prompt` (`call_many_sync`) | 结果先反转再用 `\n\n` 拼接,只保留真值片段。 |
 | `provide_tape_store` | firstresult | `() -> TapeStore \| AsyncTapeStore` | tape store | `BubFramework.running()` | 仅在 runtime 作用域开启时解析一次;返回同步或异步迭代器时会被作为 context manager 进入。 |

website/src/content/docs/zh-cn/docs/reference/index.mdx

@@ -10,4 +10,4 @@ description: Bub 的钩子、CLI 命令、配置项与公共类型查找表。
 | [Hooks](/zh-cn/docs/reference/hooks/) | `BubHookSpecs` 中每个钩子的类型、参数、返回值与调用位置。 |
 | [CLI](/zh-cn/docs/reference/cli/) | 每个 `bub` 命令与子命令的选项、默认值与行为说明。 |
 | [Settings](/zh-cn/docs/reference/settings/) | 全部 `BUB_*` 环境变量、`pydantic-settings` 配置类与 `~/.bub/config.yml` 字段。 |
-| [Types](/zh-cn/docs/reference/types/) | `bub` 暴露的公共类型:`Envelope`、`State`、`TurnResult`、`Channel`、`OutboundChannelRouter`、`BubFramework`。 |
+| [Types](/zh-cn/docs/reference/types/) | `bub` 暴露的公共类型:`Envelope`、`State`、`TurnResult`、runtime options、`Channel`、`OutboundChannelRouter`、`BubFramework`。 |