Uses of the GatewayBroker should not occasionally bypass the InferenceBroker trait

crates/harness-gateway-client/src/broker-tests.rs

@@ -8,7 +8,7 @@
 use std::num::NonZeroU32;
 use std::sync::{Arc, Mutex};
 
-use harness::InferenceBroker;
+use harness::{InferenceBroker, StreamDelta};
 use runvil::effect::Round;
 use runvil::event::ReplyOrigin;
 use runvil::ids::RoundId;
@@ -19,7 +19,6 @@ use runvil::model::{
 use serde_json::Value;
 
 use super::GatewayBroker;
-use crate::StreamDelta;
 use crate::config::{GatewayEndpoint, SecretString};
 use crate::transport::tests::{content_chunk, serve, sse_app, sse_body};
 
@@ -100,6 +99,7 @@ async fn chat_streaming_hands_each_piece_to_its_callback_in_wire_order() {
             vec![Message::user("hi")],
             Vec::new(),
             CompletionOptions::new("m"),
+            round(ReplyOrigin::Chat),
             on_piece,
         )
         .await
@@ -177,6 +177,7 @@ async fn every_round_whole_or_streamed_is_labeled_with_the_model_its_request_nam
                 vec![Message::user("hi")],
                 Vec::new(),
                 CompletionOptions::new("routed-to"),
+                round(ReplyOrigin::Chat),
                 on_piece,
             )
             .await

crates/harness-gateway-client/src/broker.rs

@@ -2,17 +2,17 @@
 //! the Gateway client, and the Gateway's model list.
 //!
 //! The Harness takes only a round's finished reply, so the broker's
-//! `InferenceBroker` round returns it whole. A Host that shows the reply
-//! as it forms runs the round through [`GatewayBroker::chat_streaming`]
-//! instead, which hands each live piece to the Host's callback as it
-//! arrives, so no fragment ever reaches the Host's recorder. Either way
+//! `chat` round returns it whole. A Host that shows the reply as it forms
+//! runs the round through [`InferenceBroker::chat_streaming`] instead,
+//! which hands each live piece to the Host's callback as it arrives, so no
+//! fragment ever reaches the Host's recorder. Either way
 //! the completion names the model the round's options sent it to, so a
 //! Host that routes a round to another model sees that model on its reply.
 
 use std::fmt;
 use std::sync::Arc;
 
-use harness::{BoxFuture, InferenceBroker};
+use harness::{BoxFuture, InferenceBroker, StreamDelta};
 use runvil::effect::Round;
 use runvil::error::Error;
 use runvil::model::{
@@ -22,7 +22,6 @@ use runvil::model::{
 use crate::catalog::fetch_model_catalog;
 use crate::config::{GatewayEndpoint, SecretString};
 use crate::transport::GatewayChat;
-use crate::wire::delta::StreamDelta;
 
 /// A model broker that sends the Harness's model rounds to the Runvil
 /// Gateway and lists the Gateway's models.
@@ -74,46 +73,6 @@ impl GatewayBroker {
             key,
         }
     }
-
-    /// Runs one model round and passes each piece of the reply to
-    /// `on_piece` as it arrives.
-    ///
-    /// The round is the same one `InferenceBroker::chat` runs. `on_piece`
-    /// receives the pieces in the order the stream delivers them. The
-    /// returned completion holds the whole reply.
-    ///
-    /// `on_piece` is called inline as each piece is read, so it must not
-    /// block.
-    pub fn chat_streaming(
-        &self,
-        _binding: ModelBinding,
-        messages: Vec<Message>,
-        tools: Vec<ToolSchema>,
-        options: CompletionOptions,
-        on_piece: Arc<dyn Fn(StreamDelta) + Send + Sync>,
-    ) -> BoxFuture<Result<Box<Completion>, Error>> {
-        self.round(messages, tools, options, move |piece| on_piece(piece))
-    }
-
-    /// One round on the client, handing each live piece to `on_piece`.
-    fn round(
-        &self,
-        messages: Vec<Message>,
-        tools: Vec<ToolSchema>,
-        options: CompletionOptions,
-        on_piece: impl Fn(StreamDelta) + Send + Sync + 'static,
-    ) -> BoxFuture<Result<Box<Completion>, Error>> {
-        let client = self.client.clone();
-        Box::pin(async move {
-            // An empty advertisement sends no `tools` field at all, the
-            // plain chat-completions shape.
-            let tools = (!tools.is_empty()).then_some(tools.as_slice());
-            client
-                .complete(&messages, tools, &options, on_piece)
-                .await
-                .map(Box::new)
-        })
-    }
 }
 
 impl InferenceBroker for GatewayBroker {
@@ -131,8 +90,47 @@ impl InferenceBroker for GatewayBroker {
         options: CompletionOptions,
         _round: Round,
     ) -> BoxFuture<Result<Box<Completion>, Error>> {
-        self.round(messages, tools, options, |_| {})
+        round(self.client.clone(), messages, tools, options, |_| {})
     }
+
+    fn chat_streaming(
+        &self,
+        _binding: ModelBinding,
+        messages: Vec<Message>,
+        tools: Vec<ToolSchema>,
+        options: CompletionOptions,
+        _round: Round,
+        on_piece: Arc<dyn Fn(StreamDelta) + Send + Sync>,
+    ) -> BoxFuture<Result<Box<Completion>, Error>> {
+        round(
+            self.client.clone(),
+            messages,
+            tools,
+            options,
+            move |piece| {
+                on_piece(piece);
+            },
+        )
+    }
+}
+
+/// One round on `client`, handing each live piece to `on_piece`.
+fn round(
+    client: GatewayChat,
+    messages: Vec<Message>,
+    tools: Vec<ToolSchema>,
+    options: CompletionOptions,
+    on_piece: impl Fn(StreamDelta) + Send + Sync + 'static,
+) -> BoxFuture<Result<Box<Completion>, Error>> {
+    Box::pin(async move {
+        // An empty advertisement sends no `tools` field at all, the plain
+        // chat-completions shape.
+        let tools = (!tools.is_empty()).then_some(tools.as_slice());
+        client
+            .complete(&messages, tools, &options, on_piece)
+            .await
+            .map(Box::new)
+    })
 }
 
 #[cfg(test)]

crates/harness-gateway-client/src/lib.rs

@@ -6,8 +6,9 @@
 //! [`GatewayChat`] under the client's own request limits, and it lists the
 //! Gateway's models through [`fetch_model_catalog`]. A Host that shows a
 //! reply as it forms runs the round through
-//! [`GatewayBroker::chat_streaming`], which hands each [`StreamDelta`] to
-//! the Host's callback as it arrives.
+//! [`InferenceBroker::chat_streaming`](harness::InferenceBroker::chat_streaming),
+//! which hands each [`StreamDelta`](harness::StreamDelta) to the Host's
+//! callback as it arrives.
 //!
 //! [`GatewayChat`] is the HTTP client that sends a `Chat` effect's round
 //! to one Gateway URL. It presents the Gateway's shared bearer key when
@@ -81,7 +82,6 @@ pub use search::GatewaySearch;
 pub use transport::GatewayChat;
 pub use wire::classify::classify_http_failure;
 pub use wire::classify::classify_stream_error;
-pub use wire::delta::StreamDelta;
 pub use wire::read::ChunkSource;
 pub use wire::read::read_body_capped;
 pub use wire::read::read_completion_stream;

crates/harness-gateway-client/src/transport.rs

@@ -17,10 +17,10 @@ use runvil::model::{Completion, CompletionOptions, Message, ToolSchema};
 use crate::config::{GatewayEndpoint, SecretString, invalid_env, missing_env};
 use crate::failure::{elapsed, transport_failure, unavailable};
 use crate::wire::classify::classify_http_failure;
-use crate::wire::delta::StreamDelta;
 use crate::wire::read::{ChunkSource, read_body_capped, read_completion_stream};
 use crate::wire::request::build_request_body;
 use crate::wire::stream::escape_controls;
+use harness::StreamDelta;
 
 /// A client that sends chat completion requests to one gateway URL.
 ///

crates/harness-gateway-client/src/transport/tests/streaming.rs

@@ -5,7 +5,7 @@ use runvil::error::{ErrorCode, FactKey};
 use runvil::model::{CompletionResult, Message};
 
 use super::*;
-use crate::StreamDelta;
+use harness::StreamDelta;
 
 #[tokio::test]
 async fn complete_sends_completion_options_and_stream_flags_on_the_wire() {

crates/harness-gateway-client/src/wire.rs

@@ -9,7 +9,6 @@
 //! `[DONE]` rule, and the one turn rule set.
 
 pub(crate) mod classify;
-pub(crate) mod delta;
 mod parse;
 pub(crate) mod read;
 pub(crate) mod request;

crates/harness-gateway-client/src/wire/read.rs

@@ -16,9 +16,9 @@ use runvil::metrics::ClientTiming;
 use runvil::model::Completion;
 use serde_json::Value;
 
-use super::delta::StreamDelta;
 use super::stream::{Applied, SseScanner, StreamAccumulator};
 use crate::failure::malformed;
+use harness::StreamDelta;
 
 /// A response body that a transport supplies one chunk at a time.
 ///

crates/harness-gateway-client/src/wire/stream.rs

@@ -27,9 +27,9 @@ use runvil::model::{Completion, RawExchange};
 use serde_json::{Map, Value};
 
 use super::classify::classify_stream_error;
-use super::delta::StreamDelta;
 use super::parse::{normalize, response_metadata};
 use crate::failure::{decoded, malformed};
+use harness::StreamDelta;
 
 /// Splits a raw SSE byte stream into `data:` payloads.
 ///

crates/harness-internal/runner/src/performers.rs

@@ -21,6 +21,9 @@
 //! design, so the loop answers it inline through the Engine's store
 //! operation.
 //!
+//! A broker may also stream a round's live pieces to a caller through
+//! [`InferenceBroker::chat_streaming`]; the effect loop never does.
+//!
 //! The runner supplies the tool performer itself: the run's snapshot of
 //! the Host's Packs, taken at preparation. The Host supplies the
 //! [`InferenceBroker`] and the [`Timer`].
@@ -66,8 +69,9 @@ pub trait InferenceBroker: Send + Sync {
     /// built from `binding`. `round` holds the round's run-wide id and the
     /// path that dispatched it.
     ///
-    /// The Harness takes only the finished reply. A broker that shows the
-    /// reply as it forms streams it on its own.
+    /// The Harness takes only the finished reply. A caller that shows the
+    /// reply as it forms runs the round through
+    /// [`chat_streaming`](InferenceBroker::chat_streaming) instead.
     fn chat(
         &self,
         binding: ModelBinding,
@@ -76,6 +80,45 @@ pub trait InferenceBroker: Send + Sync {
         options: CompletionOptions,
         round: Round,
     ) -> BoxFuture<Result<Box<Completion>, Error>>;
+
+    /// Runs the same round as [`chat`](InferenceBroker::chat) and passes
+    /// each piece of the reply to `on_piece` as it arrives.
+    ///
+    /// `on_piece` receives the pieces in the order the model sends them,
+    /// and the returned completion holds the whole reply. `on_piece` is
+    /// called inline as each piece is read, so it must not block.
+    ///
+    /// The Harness never calls this method, so no piece reaches the run's
+    /// record. The default runs `chat` and sends no pieces; a broker that
+    /// can stream overrides it.
+    fn chat_streaming(
+        &self,
+        binding: ModelBinding,
+        messages: Vec<Message>,
+        tools: Vec<ToolSchema>,
+        options: CompletionOptions,
+        round: Round,
+        on_piece: Arc<dyn Fn(StreamDelta) + Send + Sync>,
+    ) -> BoxFuture<Result<Box<Completion>, Error>> {
+        let _ = on_piece;
+        self.chat(binding, messages, tools, options, round)
+    }
+}
+
+/// One piece of a streamed model reply, delivered as it arrives.
+///
+/// [`InferenceBroker::chat_streaming`] passes each piece to its callback as
+/// soon as the broker decodes it. Answer text and reasoning arrive as
+/// separate variants, so a caller can show them differently. Tool calls
+/// appear only in the finished completion.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum StreamDelta {
+    /// A fragment of the assistant's answer text.
+    Text(String),
+    /// A fragment of the model's reasoning text. It stays separate from the
+    /// answer.
+    Reasoning(String),
 }
 
 /// The tool calls a run has in flight at once unless its performer says

crates/harness/src/lib.rs

@@ -26,6 +26,7 @@ pub use harness_runner::display_chain;
 pub use harness_runner::environment::HostSnapshot;
 pub use harness_runner::performers::BoxFuture;
 pub use harness_runner::performers::InferenceBroker;
+pub use harness_runner::performers::StreamDelta;
 pub use harness_runner::performers::Timer;
 
 pub mod pack {