核心契约

Bub 把契约放在拥有其行为的子系统旁边。请从定义它的模块导入；顶层 bub 包只暴露常用框架入口。

内核词汇

Envelope — `bub.envelope`

from bub.envelope import Envelope, content_of, field_of, normalize_envelope, unpack_batch

type Envelope = Any

Envelope 可以是 mapping、dataclass、Pydantic model 或任意支持属性访问的对象。使用 field_of 与 content_of 读取值，避免绑定某一种传输 schema。normalize_envelope 生成可变 dict；unpack_batch 把单个 render_outbound 结果规范化为列表。

Turn — `bub.turn`

from bub.turn import TurnResult, TurnState

type TurnState = dict[str, Any]

@dataclass(frozen=True)
class TurnResult:
    session_id: str
    prompt: str | list[dict[str, Any]]
    model_output: str
    outbounds: list[Envelope] = field(default_factory=list)
    state: TurnState = field(default_factory=dict)

TurnState 是一次 turn 的所有阶段共享的可变状态。process_inbound 返回最终 TurnResult，其中包含生成 outbound envelope 时使用的 state。

模型选择 — `bub.model_selection`

from bub.model_selection import ModelChoice, ModelOptions

@dataclass(frozen=True)
class ModelChoice:
    id: str
    name: str | None = None
    description: str | None = None
    meta: dict[str, Any] | None = None

@dataclass(frozen=True)
class ModelOptions:
    models: list[ModelChoice] = field(default_factory=list)
    current_model: str | None = None

Channel 与 adapter 可向用户展示这些协议无关的选择项。BubFramework.get_model_options 合并 provide_model_options 返回的值；选择 UI 与持久化由调用方负责。

Streaming — `bub.streaming`

from bub.streaming import AsyncStreamEvents, StreamEvent, StreamState

StreamEvent.kind 可以是 text、reasoning、tool_call、tool_result、usage、error 或 final。AsyncStreamEvents 包装异步迭代器，并暴露最终 error 与 usage 状态。

Error — `bub.errors`

from bub.errors import BubError, ErrorKind

BubError 是稳定的执行错误对象。其 kind 为 ErrorKind，as_dict() 返回适合传输的 payload。

Hook 契约

Hook 声明与 marker 位于 bub.hooks；agent-loop 拦截 payload 与它们的执行语义放在一起：

from bub.hooks import BUB_HOOK_NAMESPACE, BubHookSpecs, hookimpl, hookspec
from bub.hooks.interception import (
    AgentHooks,
    LlmCallDecision,
    LlmCallRequest,
    LlmCallResult,
    ToolCall,
    ToolCallDecision,
    ToolCallResult,
)
from bub.hooks.runtime import HookRuntime

插件作者通常只需要 hookimpl 与对应 hook 使用的 payload 类型。分发和故障隔离语义见 Hook 参考。

Channel 契约

Handler 与 router — `bub.channels.contracts`

from bub.channels.contracts import ChannelRouter, MessageHandler

type MessageHandler = Callable[[Envelope], Coroutine[Any, Any, None]]

class ChannelRouter(Protocol):
    async def dispatch_output(self, message: Envelope) -> bool: ...
    def wrap_stream(
        self, message: Envelope, stream: AsyncIterable[StreamEvent]
    ) -> AsyncIterable[StreamEvent]: ...
    async def quit(self, session_id: str) -> None: ...
    async def admit_channel_message(
        self, session_id: str, message: Envelope, turn: TurnSnapshot
    ) -> AdmitDecision | None: ...

ChannelManager 实现该 protocol，并通过 BubFramework.bind_channel_router 绑定。

Admission — `bub.channels.admission`

from bub.channels.admission import AdmitDecision, SteeringInbox, TurnSnapshot

@dataclass(frozen=True)
class AdmitDecision:
    action: Literal["process", "drop", "follow_up", "steer"]
    reason: str | None = None

@dataclass(frozen=True)
class TurnSnapshot:
    session_id: str
    is_running: bool
    running_count: int
    pending_count: int
    steering_count: int = 0

admit_message 实现用这些类型选择立即处理、丢弃、follow-up 排队或 steering。SteeringInbox 是 session steering queue 的 protocol。

Channel 基类 — `bub.channels.base`

from bub.channels import Channel, Interface, Lifecycle

Channel 要求实现 start(stop_event) 与 stop()；按需实现 send、stream_events 或 admit_message。Interface 标记面向用户的 I/O；Lifecycle 标记与 channel 一同管理的后台 runtime。

`BubFramework`

连接这些契约的主要公共方法如下：

from bub import BubFramework

async def process_inbound(
    self, inbound: Envelope, stream_output: bool = False
) -> TurnResult: ...

async def get_model_options(
    self, *, session_id: str, workspace: str | Path | None = None
) -> ModelOptions: ...

async def admit_message(
    self, *, session_id: str, message: Envelope, turn: TurnSnapshot
) -> AdmitDecision | None: ...

def get_channels(self, message_handler: MessageHandler) -> dict[str, Channel]: ...
def bind_channel_router(self, router: ChannelRouter | None) -> None: ...

process_inbound 的生命周期见 Turn 流水线，channel 运行方式见 Channels。

顶层导出

bub.__init__ 刻意保持精简：BubFramework、Settings、config、ensure_config、home、hookimpl 与 tool。其余契约从上述子系统模块导入，使所有权保持明确。

核心契约

内核词汇

Envelope — bub.envelope

Turn — bub.turn

模型选择 — bub.model_selection

Streaming — bub.streaming

Error — bub.errors