refactor(providers)!: data-only provider configuration and stream structs

CHANGELOG.md

@@ -165,6 +165,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `StreamingToolCallDelta::provider_metadata`, read from the delta that
   starts a call. `TrackerError` converts into
   `AiMuxError::InvalidResponseData`.
+- `StreamingToolCallTracker::with_generate_id` and `SigV4Fetch::with_clock`
+  take plain function pointers (`IdGenerator = fn() -> String`,
+  `fn() -> DateTime<Utc>`) instead of closures.
+- Gains `Transformer`, `TransformStreamController` and `pipe_through`, the
+  Rust form of the WHATWG `TransformStream` the AI SDK models build `doStream`
+  with (`start` / `transform` / `flush`; `enqueue`, `enqueue_error`, `error`,
+  `terminate`).
 
 **Rust (aimux-providers)**
 
@@ -183,6 +190,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `generate_id` instead of `tool-call` / `tool-call-N`, and
   `DeltaToolCall.index` is required again, so a chunk without it fails to
   parse.
+- Provider configuration is data, not closures. Removed: the
+  `transform_request_body` setting of every provider and of
+  `PresetSettings` (use the provider's providerOptions namespace for
+  vendor fields), and `OpenAICompatibleProviderSettings::metadata_extractor`
+  with the `MetadataExtractor` / `StreamMetadataExtractor` traits (nothing
+  implemented them). Changed: `OpenAICompatibleProviderSettings::supported_urls`
+  is a `SupportedUrls` value and `convert_usage` a `ConvertUsage` enum
+  (`OpenAICompatible`, `Alibaba`, `MoonshotAI`; registry rows select it with
+  `profile.convert_usage`); the `generate_id` settings of Anthropic, Google,
+  Cohere and Mistral take an `IdGenerator` (`fn() -> String`) instead of
+  `Arc<dyn Fn() -> String>`. User-supplied credentials (`Resolvable`) and the
+  `fetch` transport are unchanged. Requests are unchanged for callers that
+  did not set a removed field.
+- Every language-model `doStream` (OpenAI chat and Responses, OpenAI-compatible
+  chat and completion, Groq, DeepSeek, Mistral, Anthropic, Bedrock, Cohere,
+  Google, Vertex, xAI, Hugging Face, Open Responses) is a named stream struct
+  driven by `pipe_through`; emitted parts are unchanged. After an Anthropic
+  `message_stop` or a Google/Vertex/xAI/Hugging Face error event, the rest of
+  the response body is read and ignored before `Finish`, as the AI SDK's
+  `flush` does, instead of dropping the connection early.
 
 **Rust (provider factory, RFC-0036)**
 

ROADMAP.md

@@ -31,7 +31,7 @@ aimux 是 **provider 接入与治理的运行时**：
 ├─ E 轨 内部清理（#164 已合入，改为 master 上独立 PR）+ 文档三层重组
 │
 ├── 新增（RFC-0036）：L0 native passthrough | ops 协议 schema | stdio CLI | 性能门禁 P
-├── 独立项：#185 ToolInput 类型化 + tracker 对齐上游并移入 provider-utils（#204）；调整：见 [docs/aisdk-architecture-alignment.md §0.7](docs/aisdk-architecture-alignment.md#07-与-roadmap--rfc-0036-既有承诺的关系)
+├── 独立项：#185 由 RFC-0036 栈取代：tracker 对齐上游并移入 provider-utils（原 #204，已并入 integration），Raw / Parsed 由 provider 侧 `RawToolCall` 与 core 侧 `ToolCall` 两套类型区分；调整：见 [docs/aisdk-architecture-alignment.md §0.7](docs/aisdk-architecture-alignment.md#07-与-roadmap--rfc-0036-既有承诺的关系)
 ├── 独立项：#170 registry 维护自动化 → #171 triage（前半：27 个 base_url 分歧）
 └── 小项（随时插空）：#181 session 后续 | #180 cache probe §10
 
@@ -50,7 +50,7 @@ aimux 是 **provider 接入与治理的运行时**：
 | **P** | 性能回归门禁：单请求开销、流式吞吐、RSS 增长 |
 | A2–A5 | 删 `ProviderName` 枚举及各语言副本；归档 `provider-inventory/`；aimux-web 类型副本；Flutter example 桌面脚手架 |
 | E1 | 内部清理 + 文档三层重组 |
-| #185 | `ToolInput{Raw,Parsed}`；tracker 不删，改为对齐 `@ai-sdk/provider-utils` 并移入 aimux-provider-utils（#204）；调整：见 [docs/aisdk-architecture-alignment.md §0.7](docs/aisdk-architecture-alignment.md#07-与-roadmap--rfc-0036-既有承诺的关系) |
+| #185 | 由 RFC-0036 栈取代：tracker 不删，对齐 `@ai-sdk/provider-utils` 并移入 aimux-provider-utils（原 #204，已并入 integration）；`ToolInput{Raw,Parsed}` 不再引入，由 `RawToolCall` / `ToolCall` 两套类型区分；调整：见 [docs/aisdk-architecture-alignment.md §0.7](docs/aisdk-architecture-alignment.md#07-与-roadmap--rfc-0036-既有承诺的关系) |
 | #170 / #171 前半 | sync/probe 脚本 + scheduled Action；27 个 base_url 分歧 triage |
 | **B1 + C2** | 解锁项：protocol 列 + `from_resolved`；40 个 FFI 构造器转发 shim（B1 形式由 descriptor 数据源 + `create_xxx` 工厂替代，C2 shim 已取消；调整：见 [docs/aisdk-architecture-alignment.md §0.7](docs/aisdk-architecture-alignment.md#07-与-roadmap--rfc-0036-既有承诺的关系)） |
 | **ops 协议 schema** | op 表、消息与错误信封、二进制帧、版本协商——先以文档 + 测试落地（RFC-0036 §4） |
@@ -116,7 +116,7 @@ aimux 是 **provider 接入与治理的运行时**：
 1. **RFC-0032 / 0033 入库**：B1 与 #174 的前提。
 2. **S1 + S2**：一个小 PR——改 profile 名、staticlib job 换 profile、全产物体积阈值。
 3. **A2–A5**：四个独立 PR；#166 的 A1 checkbox 补勾（#169 已合入）。
-4. **#185**：先删 tracker，再换 `ToolInput`。
+4. **#185**：由 RFC-0036 栈取代（tracker 保留并对齐上游；Raw / Parsed 由 `RawToolCall` / `ToolCall` 区分），待维护者确认后关闭。
 5. **B1 + C2**：解锁项。
 
 ## 5. 版本节奏预估

aimux-provider-utils/src/generate_id.rs

@@ -1,6 +1,10 @@
 use std::hash::{BuildHasher, RandomState};
 use std::sync::atomic::{AtomicU64, Ordering};
 
+/// A function that returns a fresh id (the AI SDK's `IdGenerator`). A plain
+/// function pointer: it captures no state and can sit in a `const` config.
+pub type IdGenerator = fn() -> String;
+
 /// Generate a non-cryptographic, 16-character alphanumeric ID, like `generateId`.
 #[must_use]
 pub fn generate_id() -> String {

aimux-provider-utils/src/lib.rs

@@ -30,6 +30,7 @@ pub mod response_handler;
 pub mod sigv4_fetch;
 pub mod streaming_tool_call_argument_state;
 pub mod streaming_tool_call_tracker;
+pub mod transform_stream;
 pub mod url;
 /// WebSocket client for realtime provider APIs (RFC-0028). Empty unless the
 /// `ws` feature is enabled.
@@ -43,7 +44,7 @@ pub use fetch::{
     Fetch, FetchError, FetchFunction, FetchRequest, FetchResponse, PinnedFetch, RequestCredentials,
     ReqwestFetch, default_fetch,
 };
-pub use generate_id::generate_id;
+pub use generate_id::{IdGenerator, generate_id};
 pub use get_from_api::get_from_api;
 pub use headers::{
     HeaderMapOpt, HeadersFn, combine_headers, normalize_headers, with_user_agent_suffix,
@@ -71,4 +72,5 @@ pub use streaming_tool_call_argument_state::{
 pub use streaming_tool_call_tracker::{
     StreamingToolCallDelta, StreamingToolCallTracker, TrackerError, TypeValidation,
 };
+pub use transform_stream::{TransformStreamController, Transformer, pipe_through};
 pub use url::{validate_base_url, without_trailing_slash, without_trailing_slash_opt};

aimux-provider-utils/src/sigv4_fetch.rs

@@ -13,7 +13,6 @@
 //! Reference: <https://docs.aws.amazon.com/IAM/latest/UserGuide/create-signed-request.html>
 
 use std::fmt;
-use std::sync::Arc;
 
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
@@ -279,7 +278,9 @@ fn hmac_sha256(key: &[u8], data: &[u8]) -> Vec<u8> {
     mac.finalize().into_bytes().to_vec()
 }
 
-type Clock = Arc<dyn Fn() -> DateTime<Utc> + Send + Sync>;
+/// The signing clock: a plain function, so a fixed time in tests is a named
+/// function or a non-capturing closure.
+type Clock = fn() -> DateTime<Utc>;
 
 /// A [`Fetch`] decorator that signs every request with SigV4 and forwards it
 /// to the wrapped transport. A request marked [`RequestCredentials::Omit`] is
@@ -311,14 +312,14 @@ impl SigV4Fetch {
             inner,
             credentials,
             service: service.into(),
-            clock: Arc::new(Utc::now),
+            clock: Utc::now,
         }
     }
 
     /// Replace the signing clock (deterministic signatures in tests).
     #[must_use]
-    pub fn with_clock(mut self, clock: impl Fn() -> DateTime<Utc> + Send + Sync + 'static) -> Self {
-        self.clock = Arc::new(clock);
+    pub fn with_clock(mut self, clock: fn() -> DateTime<Utc>) -> Self {
+        self.clock = clock;
         self
     }
 }
@@ -376,7 +377,7 @@ impl Fetch for SigV4Fetch {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Mutex;
+    use std::sync::{Arc, Mutex};
 
     use bytes::Bytes;
     use chrono::TimeZone;

aimux-provider-utils/src/streaming_tool_call_tracker.rs

@@ -34,6 +34,7 @@ use aimux_core::stream_part::StreamPart;
 use aimux_core::tool::RawToolCall;
 use aimux_core::types::ProviderMetadata;
 
+use crate::generate_id::IdGenerator;
 use crate::streaming_tool_call_argument_state::{
     StreamingToolCallArgumentState, starts_with_structured_value,
 };
@@ -110,7 +111,7 @@ enum Resolution {
 /// Tracks the tool calls of one streamed response. See the module docs.
 pub struct StreamingToolCallTracker {
     calls: Vec<ToolCall>,
-    generate_id: Box<dyn Fn() -> String + Send + Sync>,
+    generate_id: IdGenerator,
     type_validation: TypeValidation,
 }
 
@@ -127,16 +128,16 @@ impl StreamingToolCallTracker {
     pub fn new() -> Self {
         Self {
             calls: Vec::new(),
-            generate_id: Box::new(|| FALLBACK_TOOL_CALL_ID.to_string()),
+            generate_id: || FALLBACK_TOOL_CALL_ID.to_string(),
             type_validation: TypeValidation::None,
         }
     }
 
     /// Set the generator for calls without a usable wire id. Blank or
     /// repeated outputs are turned into unique ids.
     #[must_use]
-    pub fn with_generate_id(mut self, f: impl Fn() -> String + Send + Sync + 'static) -> Self {
-        self.generate_id = Box::new(f);
+    pub fn with_generate_id(mut self, generate_id: IdGenerator) -> Self {
+        self.generate_id = generate_id;
         self
     }
 

aimux-provider-utils/src/transform_stream.rs

@@ -0,0 +1,182 @@
+//! The Rust form of the WHATWG `TransformStream` that the AI SDK's models
+//! build their `doStream` output with (`response.pipeThrough(new
+//! TransformStream({ start, transform, flush }))`).
+//!
+//! The state the TypeScript closures capture lives in a [`Transformer`]
+//! value instead, so a model's stream logic is a plain struct with methods
+//! that can be driven and tested without any I/O.
+
+use futures::{Stream, StreamExt};
+
+use aimux_core::error::AiMuxError;
+
+/// What a transformer writes to: the `TransformStreamDefaultController`.
+pub struct TransformStreamController<O> {
+    queue: Vec<Result<O, AiMuxError>>,
+    closed: bool,
+}
+
+impl<O> TransformStreamController<O> {
+    fn new() -> Self {
+        Self {
+            queue: Vec::new(),
+            closed: false,
+        }
+    }
+
+    /// Emit one value downstream.
+    pub fn enqueue(&mut self, value: O) {
+        if !self.closed {
+            self.queue.push(Ok(value));
+        }
+    }
+
+    /// Emit `error` and keep the stream open: the next chunk is transformed
+    /// as usual. For errors a consumer may skip (a malformed event inside an
+    /// otherwise valid stream).
+    pub fn enqueue_error(&mut self, error: AiMuxError) {
+        if !self.closed {
+            self.queue.push(Err(error));
+        }
+    }
+
+    /// Emit `error` and end the stream: no further chunk is transformed and
+    /// `flush` is not called.
+    pub fn error(&mut self, error: AiMuxError) {
+        if !self.closed {
+            self.queue.push(Err(error));
+            self.closed = true;
+        }
+    }
+
+    /// End the stream without an error: no further chunk is transformed and
+    /// `flush` is not called.
+    pub fn terminate(&mut self) {
+        self.closed = true;
+    }
+}
+
+/// The `{ start, transform, flush }` object of a `TransformStream`.
+pub trait Transformer: Send + 'static {
+    type Input: Send + 'static;
+    type Output: Send + 'static;
+
+    /// Called once, before the first chunk.
+    fn start(&mut self, controller: &mut TransformStreamController<Self::Output>) {
+        let _ = controller;
+    }
+
+    /// Called for every input chunk.
+    fn transform(
+        &mut self,
+        chunk: Self::Input,
+        controller: &mut TransformStreamController<Self::Output>,
+    );
+
+    /// Called once when the input ends, unless the stream was ended earlier
+    /// with [`TransformStreamController::error`] or
+    /// [`TransformStreamController::terminate`].
+    fn flush(self, controller: &mut TransformStreamController<Self::Output>);
+}
+
+/// `input.pipeThrough(new TransformStream(transformer))`.
+pub fn pipe_through<S, T>(
+    input: S,
+    mut transformer: T,
+) -> impl Stream<Item = Result<T::Output, AiMuxError>> + Send
+where
+    S: Stream<Item = T::Input> + Send + 'static,
+    T: Transformer,
+{
+    async_stream::stream! {
+        let mut controller = TransformStreamController::new();
+        transformer.start(&mut controller);
+        for item in std::mem::take(&mut controller.queue) {
+            yield item;
+        }
+        if controller.closed {
+            return;
+        }
+        let mut input = Box::pin(input);
+        while let Some(chunk) = input.next().await {
+            transformer.transform(chunk, &mut controller);
+            for item in std::mem::take(&mut controller.queue) {
+                yield item;
+            }
+            if controller.closed {
+                return;
+            }
+        }
+        transformer.flush(&mut controller);
+        for item in std::mem::take(&mut controller.queue) {
+            yield item;
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Doubles each number, reports a skippable error for zero, stops with an
+    /// error on a negative one, and emits the running total on flush.
+    #[derive(Default)]
+    struct Doubler {
+        total: i32,
+    }
+
+    impl Transformer for Doubler {
+        type Input = i32;
+        type Output = i32;
+
+        fn start(&mut self, controller: &mut TransformStreamController<i32>) {
+            controller.enqueue(0);
+        }
+
+        fn transform(&mut self, chunk: i32, controller: &mut TransformStreamController<i32>) {
+            if chunk == 0 {
+                controller.enqueue_error(AiMuxError::InvalidResponseData("zero".into()));
+                return;
+            }
+            if chunk < 0 {
+                controller.error(AiMuxError::InvalidResponseData("negative".into()));
+                return;
+            }
+            self.total += chunk;
+            controller.enqueue(chunk * 2);
+        }
+
+        fn flush(self, controller: &mut TransformStreamController<i32>) {
+            controller.enqueue(self.total);
+        }
+    }
+
+    async fn run(input: Vec<i32>) -> Vec<Result<i32, String>> {
+        pipe_through(futures::stream::iter(input), Doubler::default())
+            .map(|item| item.map_err(|error| error.to_string()))
+            .collect()
+            .await
+    }
+
+    #[tokio::test]
+    async fn start_transform_then_flush_in_order() {
+        assert_eq!(run(vec![1, 2]).await, vec![Ok(0), Ok(2), Ok(4), Ok(3)]);
+    }
+
+    #[tokio::test]
+    async fn enqueue_error_keeps_transforming_and_flushes() {
+        let out = run(vec![1, 0, 2]).await;
+        assert_eq!(out.len(), 5);
+        assert_eq!(out[..2], [Ok(0), Ok(2)]);
+        assert!(out[2].is_err());
+        assert_eq!(out[3..], [Ok(4), Ok(3)]);
+    }
+
+    #[tokio::test]
+    async fn error_ends_the_stream_without_flush() {
+        let out = run(vec![1, -1, 5]).await;
+        assert_eq!(out.len(), 3);
+        assert_eq!(out[..2], [Ok(0), Ok(2)]);
+        assert!(out[2].is_err());
+    }
+}