From 3de6fb2833461ee2198b262dea080a269d1f3a34 Mon Sep 17 00:00:00 2001 From: David Rauschenbach Date: Tue, 3 Nov 2026 12:06:38 -0800 Subject: [PATCH] Uses of the GatewayBroker should not occasionally bypass the InferenceBroker trait --- .../src/broker-tests.rs | 5 +- crates/harness-gateway-client/src/broker.rs | 92 ++++++------ crates/harness-gateway-client/src/lib.rs | 6 +- .../harness-gateway-client/src/transport.rs | 2 +- .../src/transport/tests/streaming.rs | 2 +- crates/harness-gateway-client/src/wire.rs | 1 - .../harness-gateway-client/src/wire/delta.rs | 20 --- .../harness-gateway-client/src/wire/read.rs | 2 +- .../harness-gateway-client/src/wire/stream.rs | 2 +- .../harness-internal/runner/src/performers.rs | 47 +++++- crates/harness/src/lib.rs | 1 + crates/harness/tests/suite/broker.rs | 57 +++++++- .../server/src/agents/broker-tests.rs | 4 +- crates/workshop/server/src/agents/broker.rs | 57 ++++---- ...26-11-03-1-widen-inference-broker-trait.md | 134 ++++++++++++++++++ 15 files changed, 322 insertions(+), 110 deletions(-) delete mode 100644 crates/harness-gateway-client/src/wire/delta.rs create mode 100644 vibe/2026-11/2026-11-03-1-widen-inference-broker-trait.md diff --git a/crates/harness-gateway-client/src/broker-tests.rs b/crates/harness-gateway-client/src/broker-tests.rs index acb2539ab..cc5eb0e77 100644 --- a/crates/harness-gateway-client/src/broker-tests.rs +++ b/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 diff --git a/crates/harness-gateway-client/src/broker.rs b/crates/harness-gateway-client/src/broker.rs index 2d2fb1e78..d998e1117 100644 --- a/crates/harness-gateway-client/src/broker.rs +++ b/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, - tools: Vec, - options: CompletionOptions, - on_piece: Arc, - ) -> BoxFuture, 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, - tools: Vec, - options: CompletionOptions, - on_piece: impl Fn(StreamDelta) + Send + Sync + 'static, - ) -> BoxFuture, 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, Error>> { - self.round(messages, tools, options, |_| {}) + round(self.client.clone(), messages, tools, options, |_| {}) } + + fn chat_streaming( + &self, + _binding: ModelBinding, + messages: Vec, + tools: Vec, + options: CompletionOptions, + _round: Round, + on_piece: Arc, + ) -> BoxFuture, 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, + tools: Vec, + options: CompletionOptions, + on_piece: impl Fn(StreamDelta) + Send + Sync + 'static, +) -> BoxFuture, 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)] diff --git a/crates/harness-gateway-client/src/lib.rs b/crates/harness-gateway-client/src/lib.rs index 2d6bcdfda..b86377eeb 100644 --- a/crates/harness-gateway-client/src/lib.rs +++ b/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; diff --git a/crates/harness-gateway-client/src/transport.rs b/crates/harness-gateway-client/src/transport.rs index 7b68cfc10..d7824c85f 100644 --- a/crates/harness-gateway-client/src/transport.rs +++ b/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. /// diff --git a/crates/harness-gateway-client/src/transport/tests/streaming.rs b/crates/harness-gateway-client/src/transport/tests/streaming.rs index 0534ed9a7..65918d82e 100644 --- a/crates/harness-gateway-client/src/transport/tests/streaming.rs +++ b/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() { diff --git a/crates/harness-gateway-client/src/wire.rs b/crates/harness-gateway-client/src/wire.rs index f7aeaf35f..ae0ab5a90 100644 --- a/crates/harness-gateway-client/src/wire.rs +++ b/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; diff --git a/crates/harness-gateway-client/src/wire/delta.rs b/crates/harness-gateway-client/src/wire/delta.rs deleted file mode 100644 index 5c9586503..000000000 --- a/crates/harness-gateway-client/src/wire/delta.rs +++ /dev/null @@ -1,20 +0,0 @@ -//! The live pieces of a streamed reply, as the read loop hands them out. - -/// One piece of a streamed model reply, delivered as it arrives. -/// -/// The client passes each piece to the caller's callback as soon as it -/// decodes the piece from the stream. Answer text and reasoning arrive as -/// separate variants, so a caller can show them differently. -/// -/// The client buffers tool-call fragments until the whole batch of tool -/// calls is complete and validated. They 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), -} diff --git a/crates/harness-gateway-client/src/wire/read.rs b/crates/harness-gateway-client/src/wire/read.rs index 3f44627b3..6866adb3b 100644 --- a/crates/harness-gateway-client/src/wire/read.rs +++ b/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. /// diff --git a/crates/harness-gateway-client/src/wire/stream.rs b/crates/harness-gateway-client/src/wire/stream.rs index 392d70ab6..52c44d723 100644 --- a/crates/harness-gateway-client/src/wire/stream.rs +++ b/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. /// diff --git a/crates/harness-internal/runner/src/performers.rs b/crates/harness-internal/runner/src/performers.rs index 0b874fabe..bfb3c3a17 100644 --- a/crates/harness-internal/runner/src/performers.rs +++ b/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, 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, + tools: Vec, + options: CompletionOptions, + round: Round, + on_piece: Arc, + ) -> BoxFuture, 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 diff --git a/crates/harness/src/lib.rs b/crates/harness/src/lib.rs index 9f91fc71e..29cc7c150 100644 --- a/crates/harness/src/lib.rs +++ b/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 { diff --git a/crates/harness/tests/suite/broker.rs b/crates/harness/tests/suite/broker.rs index 85562ea2d..4703e0392 100644 --- a/crates/harness/tests/suite/broker.rs +++ b/crates/harness/tests/suite/broker.rs @@ -1,6 +1,8 @@ //! The inference broker a Host hands `Harness::new`: a section's own //! round reaches it with origin `Chat`, and a nested `models.infer` round -//! with origin `Infer`, each under its run-wide round id. +//! with origin `Infer`, each under its run-wide round id. A broker that +//! does not stream answers `chat_streaming` with its `chat` round and +//! hands the callback no piece. use std::num::NonZeroU32; use std::sync::{Arc, Mutex, PoisonError}; @@ -8,14 +10,14 @@ use std::sync::{Arc, Mutex, PoisonError}; use harness::pack::{HostContext, HostServices, PackRegistry}; use harness::record::{MemoryRecorder, RunOutcome}; use harness::vfs::VfsRef; -use harness::{BoxFuture, Harness, HostSnapshot, InferenceBroker, RunRequest}; +use harness::{BoxFuture, Harness, HostSnapshot, InferenceBroker, RunRequest, StreamDelta}; use runvil::effect::Round; use runvil::error::Error; use runvil::event::ReplyOrigin; use runvil::ids::RoundId; use runvil::model::{ Completion, CompletionOptions, CompletionResult, Message, ModelBinding, ModelCatalog, ModelId, - ModelInfo, ThinkingMode, ToolSchema, + ModelInfo, ModelInvocation, ThinkingMode, ToolSchema, }; use crate::support::Clock; @@ -122,3 +124,52 @@ async fn a_sections_round_reaches_the_broker_as_chat_and_a_nested_infer_round_as "each round reaches the broker numbered in dispatch order, under the path that sent it" ); } + +#[tokio::test] +async fn a_broker_that_does_not_stream_answers_chat_streaming_with_its_chat_round() { + let broker = ScriptedBroker { + catalog: ModelCatalog::new([]).expect("an empty catalog is valid"), + rounds: Mutex::new(Vec::new()), + }; + let pieces: Arc>> = Arc::default(); + let sink = Arc::clone(&pieces); + let round = Round::new(RoundId::new(3), ReplyOrigin::Chat); + let binding = ModelBinding::new( + "writer", + "the round's model", + ModelId::gateway(MODEL).expect("a literal model name is valid"), + ModelInvocation::new(), + NonZeroU32::new(131_072).expect("131072 is non-zero"), + ); + + let completion = broker + .chat_streaming( + binding, + vec![Message::user("hi")], + Vec::new(), + CompletionOptions::new(MODEL), + round, + Arc::new(move |piece| { + sink.lock() + .unwrap_or_else(PoisonError::into_inner) + .push(piece); + }), + ) + .await + .expect("the scripted round completes"); + + assert_eq!( + completion.result(), + &CompletionResult::Text("re: hi".to_owned()), + "the default runs the broker's own chat round" + ); + assert_eq!( + *broker.rounds.lock().expect("the round log is healthy"), + [round], + "the chat round ran once, under the round it was handed" + ); + assert!( + pieces.lock().expect("the piece log is healthy").is_empty(), + "a broker that does not stream hands the callback no piece" + ); +} diff --git a/crates/workshop/server/src/agents/broker-tests.rs b/crates/workshop/server/src/agents/broker-tests.rs index 513f876e8..b0ad33c74 100644 --- a/crates/workshop/server/src/agents/broker-tests.rs +++ b/crates/workshop/server/src/agents/broker-tests.rs @@ -16,8 +16,7 @@ use axum::Router; use axum::http::{HeaderMap, StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; -use harness::InferenceBroker; -use harness_gateway_client::StreamDelta; +use harness::{InferenceBroker, StreamDelta}; use runvil::effect::Round; use runvil::error::{Error, ErrorCode}; use runvil::event::ReplyOrigin; @@ -140,6 +139,7 @@ async fn streamed_round(broker: &WorkshopBroker) -> (Result, Vec< vec![Message::user("hi")], Vec::new(), CompletionOptions::new("m"), + Round::new(RoundId::new(0), ReplyOrigin::Chat), Arc::new(move |piece| { sink.lock() .unwrap_or_else(PoisonError::into_inner) diff --git a/crates/workshop/server/src/agents/broker.rs b/crates/workshop/server/src/agents/broker.rs index 7ab7dded9..0658cc1ce 100644 --- a/crates/workshop/server/src/agents/broker.rs +++ b/crates/workshop/server/src/agents/broker.rs @@ -18,8 +18,8 @@ use std::num::NonZeroU32; use std::sync::{Arc, Mutex, PoisonError}; -use harness::{BoxFuture, InferenceBroker}; -use harness_gateway_client::{GatewayBroker, StreamDelta}; +use harness::{BoxFuture, InferenceBroker, StreamDelta}; +use harness_gateway_client::GatewayBroker; use runvil::effect::Round; use runvil::error::{Error, ErrorCode}; use runvil::event::ReplyOrigin; @@ -42,9 +42,12 @@ pub(super) struct WorkshopBroker { registry: Registry, /// The broker built for the last generation served under, shared with /// the model lists still waiting. - cached: Arc>>, + cached: Arc>>, } +/// A gateway generation beside the broker built for it. +type GenerationBroker = (u64, Arc); + impl WorkshopBroker { /// Builds the broker over the server's subsystem registry. pub(super) fn new(registry: Registry) -> Self { @@ -57,7 +60,7 @@ impl WorkshopBroker { /// The Gateway broker for the current generation, built on the first /// call under it. No usable gateway fails as `Unavailable` with the /// code's fixed phrase, so no URL or key reaches the message. - fn current(&self) -> Result { + fn current(&self) -> Result, Error> { let unavailable = ErrorCode::Unavailable; let gateway = usable_gateway(&self.registry) .ok_or_else(|| completion_error(unavailable, phrase(unavailable)))?; @@ -66,10 +69,11 @@ impl WorkshopBroker { if let Some((generation, broker)) = cached.as_ref() && *generation == gateway.generation { - return Ok(broker.clone()); + return Ok(Arc::clone(broker)); } - let broker = GatewayBroker::new(gateway.endpoint, gateway.key); - *cached = Some((gateway.generation, broker.clone())); + let broker: Arc = + Arc::new(GatewayBroker::new(gateway.endpoint, gateway.key)); + *cached = Some((gateway.generation, Arc::clone(&broker))); Ok(broker) } @@ -115,49 +119,50 @@ impl WorkshopBroker { } options.with_model(pick) } +} - /// Runs one round through the current generation's Gateway broker, - /// sent to the dropdown's current pick, handing `on_piece` each live - /// piece of the reply as it arrives. - fn chat_streaming( +impl InferenceBroker for WorkshopBroker { + fn models(&self) -> BoxFuture> { + let broker = self.clone(); + Box::pin(async move { + broker.chat_model_ready().await; + broker.current()?.models().await + }) + } + + fn chat( &self, binding: ModelBinding, messages: Vec, tools: Vec, options: CompletionOptions, - on_piece: Arc, + round: Round, ) -> BoxFuture, Error>> { match self.current() { Ok(broker) => { let options = self.routed(&binding, options); - broker.chat_streaming(binding, messages, tools, options, on_piece) + broker.chat(binding, messages, tools, options, round) } Err(error) => Box::pin(async move { Err(error) }), } } -} -impl InferenceBroker for WorkshopBroker { - fn models(&self) -> BoxFuture> { - let broker = self.clone(); - Box::pin(async move { - broker.chat_model_ready().await; - broker.current()?.models().await - }) - } - - fn chat( + /// Runs the round on the current generation's Gateway broker, sent to + /// the dropdown's current pick, handing `on_piece` each live piece of + /// the reply as it arrives. + fn chat_streaming( &self, binding: ModelBinding, messages: Vec, tools: Vec, options: CompletionOptions, round: Round, + on_piece: Arc, ) -> BoxFuture, Error>> { match self.current() { Ok(broker) => { let options = self.routed(&binding, options); - broker.chat(binding, messages, tools, options, round) + broker.chat_streaming(binding, messages, tools, options, round, on_piece) } Err(error) => Box::pin(async move { Err(error) }), } @@ -229,7 +234,7 @@ impl InferenceBroker for RunBroker { conversation.publish_delta(round.id, kind, content); }); self.inner - .chat_streaming(binding, messages, tools, options, on_piece) + .chat_streaming(binding, messages, tools, options, round, on_piece) } } diff --git a/vibe/2026-11/2026-11-03-1-widen-inference-broker-trait.md b/vibe/2026-11/2026-11-03-1-widen-inference-broker-trait.md new file mode 100644 index 000000000..ccac237f5 --- /dev/null +++ b/vibe/2026-11/2026-11-03-1-widen-inference-broker-trait.md @@ -0,0 +1,134 @@ +--- +name: Widen the InferenceBroker trait +overview: "Code reaches GatewayBroker's streaming round through an inherent method instead of the InferenceBroker trait. Widen the trait so it is the broker's whole public interface, and reduce GatewayBroker's inherent impl to the `new` factory." +todos: + - id: move-stream-delta + content: "Move `StreamDelta` from `harness-gateway-client` to the Harness, beside `InferenceBroker`, and re-export it from `harness`" + status: completed + - id: widen-trait + content: "Add a streaming round to `InferenceBroker` as a default method that runs `chat` and sends no pieces" + status: completed + - id: gateway-impl-factory-only + content: "Reduce `impl GatewayBroker` to `new`; implement the streaming round in `impl InferenceBroker for GatewayBroker`" + status: completed + - id: callers-through-trait + content: "Move Workshop's brokers and `broker-tests.rs` onto the trait method; delete `WorkshopBroker::chat_streaming`" + status: completed +isProject: false +--- + +# Widen the InferenceBroker trait + +## Functional Specification + +### Problem + +`harness::InferenceBroker` exists so that a broker's users depend on one +interface, not on a concrete broker. `GatewayBroker` +(`crates/harness-gateway-client/src/broker.rs`) breaks that. Besides its +`InferenceBroker` impl, its inherent `impl GatewayBroker` block exposes a +second round path: + +- `chat_streaming`, a public inherent method. It runs the same round as + `InferenceBroker::chat` and also hands each `StreamDelta` to a callback. +- `round`, the private helper that `chat` and `chat_streaming` share + +The streaming round can only be reached by naming `GatewayBroker`: + +- `WorkshopBroker` (`crates/workshop/server/src/agents/broker.rs`) holds a + concrete `GatewayBroker` and calls its `chat_streaming`. It then repeats + the pattern with its own inherent `WorkshopBroker::chat_streaming`, which + `RunBroker::chat` calls. +- `crates/harness-gateway-client/src/broker-tests.rs` calls + `GatewayBroker::chat_streaming` directly. + +So the trait isn't the whole public interface. Anything that wants +streaming has to bypass it and name the concrete type, and every wrapper +broker has to copy the bypass. The trait doc even says so: "A broker that +shows the reply as it forms streams it on its own." + +### Requirements + +1. `InferenceBroker` is the broker's whole public interface. Everything a + caller can do with a broker, including a streaming round, is a trait + method. +2. The inherent `impl GatewayBroker` block holds only the `new` factory. + `chat_streaming` and `round` leave it. The round logic lives in + `impl InferenceBroker for GatewayBroker` or in a private free function. + The `fmt::Debug` impl stays, because it is a trait impl. +3. No code outside `harness-gateway-client` calls a `GatewayBroker` method + other than `new`. Workshop holds the broker as an `InferenceBroker` and + streams through the trait. +4. No other `InferenceBroker` implementor keeps an inherent copy of a trait + method. `WorkshopBroker::chat_streaming` is removed, and + `RunBroker::chat` calls the trait method on its inner broker. +5. `broker-tests.rs` calls only `GatewayBroker::new` and trait methods. + The behaviors it tests now stay tested: + - pieces reach the callback in wire order + - `chat` returns the whole reply + - each round is sent to the model its options name + - `models` lists the Gateway's catalog + +### Design + +These are the smallest changes to existing facilities that meet the +requirements. + +- **Piece type.** Move `StreamDelta` (`Text`, `Reasoning`, non-exhaustive) + from `harness-gateway-client/src/wire/delta.rs` to the crate that defines + `InferenceBroker` (`harness-runner`, `performers.rs`), and re-export it + from `harness`. The Gateway client's read loop produces this type, and + the Gateway client stops exporting its own copy. +- **Trait method.** Add one method to `InferenceBroker`, with the same + arguments as `chat` plus the callback: + + ```rust + fn chat_streaming( + &self, + binding: ModelBinding, + messages: Vec, + tools: Vec, + options: CompletionOptions, + round: Round, + on_piece: Arc, + ) -> BoxFuture, Error>> { + self.chat(binding, messages, tools, options, round) + } + ``` + + The default runs `chat` and sends no pieces. That leaves the trait's many + test and fixture implementors unchanged. A broker that can stream + overrides it. +- **Harness behavior.** It doesn't change. The Harness still calls only + `chat`, still takes only the finished reply, and records no fragment. + Update the trait doc: replace "streams it on its own" with a pointer to + `chat_streaming`. +- **Round parameter.** The trait method takes `round`, unlike today's + inherent method. A wrapper such as `RunBroker` therefore has the round's + id and origin on every path. + +### Non-goals + +- Making the Harness stream, or making it record or forward pieces. +- Changing the wire client (`GatewayChat`), the catalog fetch, or the + request limits. +- Changing what Workshop shows, or how it stamps pieces with a round id. + +## Verification + +- `impl GatewayBroker { ... }` contains exactly one function, `new`. +- No crate outside `harness-gateway-client` names a `GatewayBroker` method + other than `new`. `WorkshopBroker` has no inherent `chat_streaming`. +- The docs in `harness-gateway-client/src/lib.rs` and `broker.rs` that link + `GatewayBroker::chat_streaming` link `InferenceBroker::chat_streaming` + instead. +- The new trait method and the moved `StreamDelta` appear in the Harness's + surface. If either reaches the `runvil` facade, regenerate + `crates/runvil/public-api.txt`. +- These pass: + - the full suite and the workshop suite + - clippy, both runs + - fmt + - docs + - `cargo test -p build-xtask` + - `node --test crates/workshop/ui/test/docs-claims.mjs`