feat(hooks): add after_build_prompt to observe built prompts

src/bub/framework.py

@@ -248,7 +248,11 @@ async def build_prompt(self, message: Envelope, session_id: str, state: dict[str
         )
         if not prompt:
             prompt = content_of(message)
-        return to_content(prompt)
+        content = to_content(prompt)
+        await self._hook_runtime.call_many(
+            "after_build_prompt", message=message, prompt=list(content), session_id=session_id, state=state
+        )
+        return content
 
     async def continue_prompt(self, prompt: list[UserContent] | None, tape: Tape, state: StreamState) -> str | None:
         """Return an optional user message for the next step of an agent loop."""

src/bub/hooks/agent.py

@@ -44,9 +44,9 @@ class AgentPlugins:
     ``before/after_llm_call`` and ``before/after_tool_call`` interception hooks,
     ``continue_prompt`` and ``provide_steering_inbox`` (agent events), and ``load_state``
     and ``provide_environment`` (turn state). ``running()`` owns the lifespan, store and
-    steering inbox. Message-pipeline hooks such as ``build_prompt``,
-    ``run_model``, ``save_state`` and outbound rendering only run in
-    ``framework.process_inbound()``.
+    steering inbox. Drained steering messages are built through ``build_prompt`` and
+    ``after_build_prompt``. Other message-pipeline hooks such as ``run_model``,
+    ``save_state`` and outbound rendering only run in ``framework.process_inbound()``.
     """
 
     def __init__(self, framework: BubFramework | None = None, *, plugins: Iterable[object] | None = None) -> None:

src/bub/hooks/specs.py

@@ -61,6 +61,18 @@ def build_prompt(self, message: Envelope, session_id: str, state: TurnState) ->
         """
         raise NotImplementedError
 
+    @hookspec
+    def after_build_prompt(
+        self, message: Envelope, prompt: list[UserContent], session_id: str, state: TurnState
+    ) -> None:
+        """Observe the prompt content built for one inbound message.
+
+        Fires for the turn's message and for each steering message drained into a
+        running turn, so plugins can relate a model-facing user message to the
+        message it came from. ``prompt`` is a copy of the built content as
+        ``list[UserContent]``, not the legacy form. Return values are ignored.
+        """
+
     @hookspec(firstresult=True)
     def run_model(self, prompt: str | list[dict], session_id: str, state: TurnState) -> str:
         """Run model for one turn and return plain text output. Should not be implemented if `run_model_stream` is implemented."""

tests/test_agent_plugins.py

@@ -23,6 +23,7 @@ def __init__(self) -> None:
         self.store = InMemoryTapeStore()
         self.inbox = InMemorySteeringInbox()
         self.continued: list[Any] = []
+        self.built: list[tuple[Any, Any]] = []
 
     @hookimpl
     def system_prompt(self, prompt: Any, state: TurnState) -> str:
@@ -49,6 +50,10 @@ async def after_llm_call(self, request: LLMCallRequest, result: LLMCallResult, s
     def continue_prompt(self, prompt: Any, tape: Any, state: Any) -> None:
         self.continued.append(prompt)
 
+    @hookimpl
+    def after_build_prompt(self, message: Any, prompt: Any) -> None:
+        self.built.append((message["content"], prompt))
+
 
 @pytest.mark.asyncio
 async def test_agent_plugins_connect_loop_hooks_to_core_agent(
@@ -86,6 +91,7 @@ async def test_agent_plugins_connect_loop_hooks_to_core_agent(
         {"role": "user", "content": "late"},
     ]
     assert plugin.continued == ["hello"]
+    assert plugin.built == [("early", ["early"]), ("late", ["late"])]
     assert plugin.inbox.message_count(state) == 0
     assert state["session_id"] == "s1"
     assert (await agent.tape("s1").read_messages())[-1] == republic.assistant("two")

tests/test_framework.py

@@ -171,6 +171,33 @@ def build_prompt(self, message, session_id, state):
     assert prompt == ["describe", republic.Image("image/png", url="https://example.test/a.png")]
 
 
+@pytest.mark.asyncio
+async def test_after_build_prompt_observes_a_copy_of_the_built_content_with_its_message() -> None:
+    framework = BubFramework()
+    observed: list[tuple[Any, list[Any], str]] = []
+
+    class FramingPlugin:
+        @hookimpl
+        def build_prompt(self, message, session_id, state):
+            return f"[framed]\n{message['content']}"
+
+    class ObserverPlugin:
+        @hookimpl
+        async def after_build_prompt(self, message, prompt, session_id):
+            observed.append((message, list(prompt), session_id))
+            prompt.clear()
+            return ["ignored"]
+
+    framework.plugin_manager.register(FramingPlugin(), name="framing")
+    framework.plugin_manager.register(ObserverPlugin(), name="observer")
+    message = {"content": "hello"}
+
+    prompt = await framework.build_prompt(message, session_id="s", state={})
+
+    assert prompt == ["[framed]\nhello"]
+    assert observed == [(message, ["[framed]\nhello"], "s")]
+
+
 @pytest.mark.asyncio
 async def test_continue_prompt_awaits_high_priority_async_hook() -> None:
     framework = BubFramework()

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

@@ -50,13 +50,14 @@ The plugin list printed by `bub hooks` follows pluggy's registration order, **no
 
 ## Turn Hooks {#turn-hooks}
 
-A single message-processing flow is called a **Turn**. The following nine hooks are triggered by `BubFramework.process_inbound()` and cover the whole process from determining the session to sending the reply.
+A single message-processing flow is called a **Turn**. The following ten hooks are triggered by `BubFramework.process_inbound()` and cover the whole process from determining the session to sending the reply.
 
 | Hook | What it does | Dispatch |
 | --- | --- | --- |
 | [`resolve_session`](#resolve_session) | Determines which session a message belongs to | First result |
 | [`load_state`](#load_state) | Restores the state needed for this Turn | Run all and merge results |
 | [`build_prompt`](#build_prompt) | Builds the prompt sent to the model | First result |
+| [`after_build_prompt`](#after_build_prompt) | Observes the prompt built for a message | Run all |
 | [`run_model`](#run_model) | Calls the model and returns the complete text | First result |
 | [`run_model_stream`](#run_model_stream) | Calls the model and returns streaming events | First result |
 | [`continue_prompt`](#continue_prompt) | Optionally appends a user message when the Agent continues running | First result |
@@ -119,6 +120,19 @@ Hooks that use the legacy prompt format (`run_model`, `run_model_stream`, `conti
 
 `to_content()` still accepts legacy strings and content-block lists, and handles the format conversion. To convert back to the legacy format, use `bub.prompt.to_legacy_prompt()`.
 
+### `after_build_prompt` {#after_build_prompt}
+
+Observes the prompt content built for one inbound message, so a plugin can relate a model-facing user message to the message it came from.
+
+```python
+def after_build_prompt(self, message: Envelope, prompt: list[UserContent], session_id: str, state: TurnState) -> None:
+```
+
+- **Dispatch**: Run all.
+- **When it fires**: After `build_prompt` for the Turn's message, and for each steering message drained into a running Turn.
+- **Prompt format**: Receives a copy of the built content as `list[UserContent]`, not the legacy format.
+- **Return value**: Ignored.
+
 ### `run_model` {#run_model}
 
 Calls the model and returns the complete text output directly.
@@ -608,7 +622,7 @@ The concept page explains who intervenes in which direction; this section covers
 | `before_tool_call` | `ToolCallStarted` | Before every tool call; may `proceed(arguments)`, `replace(result)`, or `deny(message)` |
 | `after_tool_call` | `ToolCallFinished` | Fires on success, failure, denial, and replacement |
 | `continue_prompt` | `StepFinished` | Called only when `status == "continue"`; the result becomes `next_prompt` |
-| `provide_steering_inbox` | `MessagesPending` | Drains steering messages into the next step of the current Run |
+| `provide_steering_inbox` | `MessagesPending` | Drains steering messages into the next step of the current Run; each is built through `build_prompt` and `after_build_prompt` |
 
 Model-visible tool calls, including built-in tools such as `.tape.handoff`, also pass through `ToolCallStarted` / `ToolCallFinished`.
 

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

@@ -23,6 +23,12 @@ def load_state(self, message: Envelope, session_id: str) -> TurnState:
 def build_prompt(self, message: Envelope, session_id: str, state: TurnState) -> list[UserContent] | LegacyPrompt:
 ```
 
+### `after_build_prompt` {#after_build_prompt}
+
+```python
+def after_build_prompt(self, message: Envelope, prompt: list[UserContent], session_id: str, state: TurnState) -> None:
+```
+
 ### `run_model` {#run_model}
 
 ```python
@@ -288,7 +294,7 @@ class ToolCallResult:
 | `before_tool_call` | `ToolCallStarted` | 每次工具调用前；可 `proceed(arguments)`、`replace(result)` 或 `deny(message)` |
 | `after_tool_call` | `ToolCallFinished` | 成功、失败、拒绝与替代都会触发 |
 | `continue_prompt` | `StepFinished` | 仅在 `status == "continue"` 时调用，结果写入 `next_prompt` |
-| `provide_steering_inbox` | `MessagesPending` | 排出待注入的 steering 消息，加入当前 Run 的下一步 |
+| `provide_steering_inbox` | `MessagesPending` | 排出待注入的 steering 消息，加入当前 Run 的下一步；每条消息经过 `build_prompt` 和 `after_build_prompt` 构造 |
 
 模型可见的工具调用（包括 `.tape.handoff` 这类内置工具）同样经过 `ToolCallStarted` / `ToolCallFinished`。