refactor: introduce StreamOptions bag for provider stream parameters

Replace scattered thinking/signal/on_payload/on_response parameters
with a single StreamOptions object that flows transparently from
Agent through the loop to Provider.stream(). This mirrors pi's
SimpleStreamOptions pattern — adding new stream options no longer
requires changing signatures at every layer.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: xfgong

cubepi/__init__.py

@@ -16,6 +16,7 @@
     Model,
     Provider,
     StreamEvent,
+    StreamOptions,
     TextContent,
     ThinkingBudgets,
     ThinkingLevel,
@@ -38,6 +39,7 @@
     "Model",
     "Provider",
     "StreamEvent",
+    "StreamOptions",
     "TextContent",
     "ThinkingBudgets",
     "ThinkingLevel",

cubepi/agent/agent.py

@@ -19,7 +19,10 @@
     AssistantMessage,
     Message,
     Model,
+    OnPayloadCallback,
+    OnResponseCallback,
     Provider,
+    StreamOptions,
     TextContent,
     ThinkingLevel,
     Usage,
@@ -116,6 +119,8 @@ def __init__(
         before_tool_call: Callable | None = None,
         after_tool_call: Callable | None = None,
         should_stop_after_turn: Callable | None = None,
+        on_payload: OnPayloadCallback | None = None,
+        on_response: OnResponseCallback | None = None,
         steering_mode: str = "one-at-a-time",
         follow_up_mode: str = "one-at-a-time",
         tool_execution: str = "parallel",
@@ -135,6 +140,8 @@ def __init__(
         self.before_tool_call = before_tool_call
         self.after_tool_call = after_tool_call
         self.should_stop_after_turn = should_stop_after_turn
+        self.on_payload = on_payload
+        self.on_response = on_response
         self.tool_execution = tool_execution
         self.checkpointer = checkpointer
         self.thread_id = thread_id
@@ -222,6 +229,14 @@ async def resume(self) -> None:
 
         await self._run_continuation()
 
+    def _build_stream_options(self, signal: asyncio.Event) -> StreamOptions:
+        return StreamOptions(
+            thinking=self._state.thinking,
+            signal=signal,
+            on_payload=self.on_payload,
+            on_response=self.on_response,
+        )
+
     async def _run_prompt(self, messages: list[Any]) -> None:
         await self._run_with_lifecycle(
             lambda signal: run_agent_loop(
@@ -236,9 +251,8 @@ async def _run_prompt(self, messages: list[Any]) -> None:
                 should_stop_after_turn=self.should_stop_after_turn,
                 get_steering_messages=self._make_async_drain(self._steering_queue),
                 get_follow_up_messages=self._make_async_drain(self._follow_up_queue),
-                thinking=self._state.thinking,
+                stream_options=self._build_stream_options(signal),
                 tool_execution=self.tool_execution,
-                signal=signal,
                 emit=lambda e: self._process_event(e),
             )
         )
@@ -256,9 +270,8 @@ async def _run_continuation(self) -> None:
                 should_stop_after_turn=self.should_stop_after_turn,
                 get_steering_messages=self._make_async_drain(self._steering_queue),
                 get_follow_up_messages=self._make_async_drain(self._follow_up_queue),
-                thinking=self._state.thinking,
+                stream_options=self._build_stream_options(signal),
                 tool_execution=self.tool_execution,
-                signal=signal,
                 emit=lambda e: self._process_event(e),
             )
         )

cubepi/agent/loop.py

@@ -19,7 +19,7 @@
     AssistantMessage,
     Model,
     Provider,
-    ThinkingLevel,
+    StreamOptions,
     ToolCall,
     ToolResultMessage,
 )
@@ -45,9 +45,8 @@ async def run_agent_loop(
     should_stop_after_turn: Callable | None = None,
     get_steering_messages: Callable | None = None,
     get_follow_up_messages: Callable | None = None,
-    thinking: ThinkingLevel = "off",
+    stream_options: StreamOptions | None = None,
     tool_execution: str = "parallel",
-    signal: asyncio.Event | None = None,
     system_prompt: str | None = None,
 ) -> list[Any]:
     new_messages: list[Any] = list(prompts)
@@ -77,9 +76,8 @@ async def run_agent_loop(
         should_stop_after_turn=should_stop_after_turn,
         get_steering_messages=get_steering_messages,
         get_follow_up_messages=get_follow_up_messages,
-        thinking=thinking,
+        stream_options=stream_options,
         tool_execution=tool_execution,
-        signal=signal,
         emit=emit,
     )
     return new_messages
@@ -98,9 +96,8 @@ async def run_agent_loop_continue(
     should_stop_after_turn: Callable | None = None,
     get_steering_messages: Callable | None = None,
     get_follow_up_messages: Callable | None = None,
-    thinking: ThinkingLevel = "off",
+    stream_options: StreamOptions | None = None,
     tool_execution: str = "parallel",
-    signal: asyncio.Event | None = None,
     system_prompt: str | None = None,
 ) -> list[Any]:
     if not context.messages:
@@ -130,9 +127,8 @@ async def run_agent_loop_continue(
         should_stop_after_turn=should_stop_after_turn,
         get_steering_messages=get_steering_messages,
         get_follow_up_messages=get_follow_up_messages,
-        thinking=thinking,
+        stream_options=stream_options,
         tool_execution=tool_execution,
-        signal=signal,
         emit=emit,
     )
     return new_messages
@@ -151,11 +147,11 @@ async def _run_loop(
     should_stop_after_turn: Callable | None,
     get_steering_messages: Callable | None,
     get_follow_up_messages: Callable | None,
-    thinking: ThinkingLevel,
+    stream_options: StreamOptions | None,
     tool_execution: str,
-    signal: asyncio.Event | None,
     emit: Callable,
 ) -> None:
+    opts = stream_options or StreamOptions()
     first_turn = True
 
     while True:
@@ -173,8 +169,7 @@ async def _run_loop(
                 model,
                 convert_to_llm,
                 transform_context,
-                thinking,
-                signal,
+                opts,
                 emit,
             )
             new_messages.append(message)
@@ -195,7 +190,7 @@ async def _run_loop(
                     tool_execution=tool_execution,
                     before_tool_call=before_tool_call,
                     after_tool_call=after_tool_call,
-                    signal=signal,
+                    signal=opts.signal,
                     emit=emit,
                 )
                 tool_results = batch.messages
@@ -251,13 +246,12 @@ async def _stream_assistant_response(
     model: Model,
     convert_to_llm: Callable,
     transform_context: Callable | None,
-    thinking: ThinkingLevel,
-    signal: asyncio.Event | None,
+    options: StreamOptions,
     emit: Callable,
 ) -> AssistantMessage:
     messages = context.messages
     if transform_context:
-        messages = await transform_context(messages, signal=signal)
+        messages = await transform_context(messages, signal=options.signal)
 
     llm_messages = convert_to_llm(messages)
     if asyncio.iscoroutine(llm_messages):
@@ -272,8 +266,7 @@ async def _stream_assistant_response(
         llm_messages,
         system_prompt=context.system_prompt,
         tools=tools_defs,
-        thinking=thinking,
-        signal=signal,
+        options=options,
     )
 
     partial_message: AssistantMessage | None = None

cubepi/providers/__init__.py

@@ -11,6 +11,7 @@
     Provider,
     ProviderResponse,
     StreamEvent,
+    StreamOptions,
     TextContent,
     ThinkingBudgets,
     ThinkingContent,
@@ -70,6 +71,7 @@ def get_openai_responses_provider():
     "Provider",
     "ProviderResponse",
     "StreamEvent",
+    "StreamOptions",
     "TextContent",
     "ThinkingBudgets",
     "ThinkingContent",

cubepi/providers/anthropic.py

@@ -10,14 +10,11 @@
     Message,
     MessageStream,
     Model,
-    OnPayloadCallback,
-    OnResponseCallback,
     ProviderResponse,
     StreamEvent,
+    StreamOptions,
     TextContent,
-    ThinkingBudgets,
     ThinkingContent,
-    ThinkingLevel,
     ToolCall,
     ToolDefinition,
     ToolResultMessage,
@@ -51,14 +48,11 @@ async def stream(
         *,
         system_prompt: str = "",
         tools: list[ToolDefinition] | None = None,
-        thinking: ThinkingLevel = "off",
-        thinking_budgets: ThinkingBudgets | None = None,
-        signal: asyncio.Event | None = None,
-        on_payload: OnPayloadCallback | None = None,
-        on_response: OnResponseCallback | None = None,
+        options: StreamOptions | None = None,
     ) -> MessageStream:
+        opts = options or StreamOptions()
         ms = MessageStream()
-        thinking = clamp_thinking_level(model, thinking)
+        thinking = clamp_thinking_level(model, opts.thinking)
 
         cache_control = self._get_cache_control()
         api_messages = [self._convert_message(m) for m in messages]
@@ -69,7 +63,7 @@ async def stream(
             base_max_tokens=model.max_tokens,
             model_max_tokens=model.context_window,
             reasoning_level=thinking,
-            custom_budgets=thinking_budgets,
+            custom_budgets=opts.thinking_budgets,
         )
 
         kwargs: dict[str, Any] = {
@@ -99,14 +93,14 @@ async def stream(
         async def _produce() -> None:
             try:
                 nonlocal kwargs
-                kwargs = await _invoke_on_payload(on_payload, kwargs, model)
+                kwargs = await _invoke_on_payload(opts.on_payload, kwargs, model)
 
                 async with self._client.messages.stream(**kwargs) as stream:
                     # Invoke on_response with HTTP metadata if available
                     http_response = getattr(stream, "response", None)
                     if http_response is not None:
                         await _invoke_on_response(
-                            on_response,
+                            opts.on_response,
                             ProviderResponse(
                                 status=http_response.status_code,
                                 headers=dict(http_response.headers),
@@ -122,7 +116,7 @@ async def _produce() -> None:
                     )
 
                     async for event in stream:
-                        if signal and signal.is_set():
+                        if opts.signal and opts.signal.is_set():
                             aborted = partial.model_copy(
                                 update={
                                     "stop_reason": "aborted",

cubepi/providers/base.py

@@ -13,7 +13,7 @@
     runtime_checkable,
 )
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 ThinkingLevel = Literal["off", "minimal", "low", "medium", "high", "xhigh"]
 
@@ -210,6 +210,18 @@ class ProviderResponse:
 """Optional callback invoked after an HTTP response is received."""
 
 
+class StreamOptions(BaseModel):
+    """Options bag for Provider.stream(), transparent to the agent loop."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    thinking: ThinkingLevel = "off"
+    thinking_budgets: ThinkingBudgets | None = None
+    signal: asyncio.Event | None = None
+    on_payload: OnPayloadCallback | None = None
+    on_response: OnResponseCallback | None = None
+
+
 async def _invoke_on_payload(
     callback: OnPayloadCallback | None,
     payload: dict,
@@ -246,9 +258,5 @@ async def stream(
         *,
         system_prompt: str = "",
         tools: list[ToolDefinition] | None = None,
-        thinking: ThinkingLevel = "off",
-        thinking_budgets: ThinkingBudgets | None = None,
-        signal: asyncio.Event | None = None,
-        on_payload: OnPayloadCallback | None = None,
-        on_response: OnResponseCallback | None = None,
+        options: StreamOptions | None = None,
     ) -> MessageStream: ...

cubepi/providers/faux.py

@@ -13,10 +13,9 @@
     MessageStream,
     Model,
     StreamEvent,
+    StreamOptions,
     TextContent,
-    ThinkingBudgets,
     ThinkingContent,
-    ThinkingLevel,
     ToolCall,
     ToolDefinition,
     Usage,
@@ -223,12 +222,9 @@ async def stream(
         *,
         system_prompt: str = "",
         tools: list[ToolDefinition] | None = None,
-        thinking: ThinkingLevel = "off",
-        thinking_budgets: ThinkingBudgets | None = None,
-        signal: asyncio.Event | None = None,
-        on_payload: Any = None,
-        on_response: Any = None,
+        options: StreamOptions | None = None,
     ) -> MessageStream:
+        opts = options or StreamOptions()
         ms = MessageStream()
         self.call_count += 1
 
@@ -273,7 +269,7 @@ async def _produce() -> None:
                 )
                 resolved = resolved.model_copy(update={"usage": cache_usage})
 
-                await self._stream_with_deltas(ms, resolved, signal)
+                await self._stream_with_deltas(ms, resolved, opts.signal)
             except BaseException as exc:
                 error_msg = AssistantMessage(
                     content=[],