feat(agent): M2 — OpenAI + Anthropic wire dialects

crates/agent-core/src/dialect/anthropic.rs

@@ -0,0 +1,246 @@
+//! Anthropic Messages wire (`/v1/messages`).
+//!
+//! The harness's internal model was chosen to be Anthropic-shaped (content blocks,
+//! `tool_use`/`tool_result`), so the request mapping is nearly an identity and the SSE decoder is a
+//! direct translation of Anthropic's `content_block_*` / `message_*` events.
+
+use serde_json::{Map, Value};
+
+use super::StreamDecoder;
+use crate::message::{StopReason, StreamEvent};
+use crate::transport::ModelRequest;
+
+/// Build the streaming request body. `system` is hoisted to a top-level field (Anthropic keeps it
+/// out of `messages`); `messages` and `tools` serialize straight from the internal model.
+pub fn build_body(req: &ModelRequest) -> Value {
+    let mut map = Map::new();
+    map.insert("model".into(), Value::String(req.model.clone()));
+    map.insert("max_tokens".into(), Value::from(req.max_tokens));
+    map.insert("stream".into(), Value::Bool(true));
+    map.insert(
+        "messages".into(),
+        serde_json::to_value(&req.messages).unwrap_or(Value::Null),
+    );
+    if let Some(system) = &req.system {
+        map.insert("system".into(), Value::String(system.clone()));
+    }
+    if !req.tools.is_empty() {
+        map.insert(
+            "tools".into(),
+            serde_json::to_value(&req.tools).unwrap_or(Value::Null),
+        );
+    }
+    Value::Object(map)
+}
+
+fn map_stop_reason(s: Option<&str>) -> StopReason {
+    match s {
+        Some("end_turn") => StopReason::EndTurn,
+        Some("tool_use") => StopReason::ToolUse,
+        Some("max_tokens") => StopReason::MaxTokens,
+        Some("stop_sequence") => StopReason::StopSequence,
+        _ => StopReason::Other,
+    }
+}
+
+/// Decodes Anthropic SSE. Tracks token counts (input from `message_start`, output from
+/// `message_delta`) and the stop reason, emitting a single `Usage` + `MessageStop` at `message_stop`.
+#[derive(Default)]
+pub struct Decoder {
+    input_tokens: u32,
+    output_tokens: u32,
+    stop_reason: StopReason,
+}
+
+impl StreamDecoder for Decoder {
+    fn push(&mut self, data: &Value) -> Vec<StreamEvent> {
+        let kind = data.get("type").and_then(Value::as_str).unwrap_or("");
+        match kind {
+            "message_start" => {
+                let usage = data.get("message").and_then(|m| m.get("usage"));
+                self.input_tokens = u32_at(usage, "input_tokens");
+                self.output_tokens = u32_at(usage, "output_tokens");
+                vec![StreamEvent::MessageStart]
+            }
+            "content_block_start" => {
+                let block = data.get("content_block");
+                if block.and_then(|b| b.get("type")).and_then(Value::as_str) == Some("tool_use") {
+                    let id = str_at(block, "id").to_string();
+                    let name = str_at(block, "name").to_string();
+                    vec![StreamEvent::ToolUseStart { id, name }]
+                } else {
+                    // A text block opening carries no event in our model — text accrues via deltas.
+                    Vec::new()
+                }
+            }
+            "content_block_delta" => {
+                let delta = data.get("delta");
+                match delta.and_then(|d| d.get("type")).and_then(Value::as_str) {
+                    Some("text_delta") => {
+                        vec![StreamEvent::TextDelta {
+                            text: str_at(delta, "text").to_string(),
+                        }]
+                    }
+                    Some("input_json_delta") => {
+                        vec![StreamEvent::InputJsonDelta {
+                            partial_json: str_at(delta, "partial_json").to_string(),
+                        }]
+                    }
+                    _ => Vec::new(),
+                }
+            }
+            "content_block_stop" => vec![StreamEvent::ContentBlockStop],
+            "message_delta" => {
+                let delta = data.get("delta");
+                self.stop_reason = map_stop_reason(
+                    delta
+                        .and_then(|d| d.get("stop_reason"))
+                        .and_then(Value::as_str),
+                );
+                let out = u32_at(data.get("usage"), "output_tokens");
+                if out > 0 {
+                    self.output_tokens = out;
+                }
+                Vec::new()
+            }
+            "message_stop" => vec![
+                StreamEvent::Usage {
+                    input_tokens: self.input_tokens,
+                    output_tokens: self.output_tokens,
+                },
+                StreamEvent::MessageStop {
+                    stop_reason: self.stop_reason,
+                },
+            ],
+            _ => Vec::new(),
+        }
+    }
+}
+
+fn str_at<'a>(v: Option<&'a Value>, key: &str) -> &'a str {
+    v.and_then(|v| v.get(key))
+        .and_then(Value::as_str)
+        .unwrap_or("")
+}
+
+fn u32_at(v: Option<&Value>, key: &str) -> u32 {
+    v.and_then(|v| v.get(key))
+        .and_then(Value::as_u64)
+        .unwrap_or(0) as u32
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::dialect::decode_sse;
+    use crate::message::{ContentBlock, Message, ToolDef};
+    use serde_json::json;
+
+    #[test]
+    fn build_body_hoists_system_and_keeps_blocks() {
+        let req = ModelRequest::new("claude-opus-4-8", vec![Message::user("hi")], 256)
+            .with_system("be brief")
+            .with_tools(vec![ToolDef {
+                name: "read".into(),
+                description: "read a file".into(),
+                input_schema: json!({ "type": "object" }),
+            }]);
+        let body = build_body(&req);
+        assert_eq!(body["model"], "claude-opus-4-8");
+        assert_eq!(body["stream"], true);
+        assert_eq!(body["system"], "be brief");
+        assert_eq!(body["messages"][0]["role"], "user");
+        assert_eq!(body["messages"][0]["content"][0]["type"], "text");
+        assert_eq!(body["tools"][0]["name"], "read");
+        assert_eq!(body["tools"][0]["input_schema"]["type"], "object");
+    }
+
+    #[test]
+    fn assistant_tool_use_round_trips_into_body() {
+        let req = ModelRequest::new(
+            "claude-opus-4-8",
+            vec![
+                Message::user("weather?"),
+                Message::assistant(vec![ContentBlock::ToolUse {
+                    id: "toolu_1".into(),
+                    name: "get_weather".into(),
+                    input: json!({ "city": "SF" }),
+                }]),
+                Message::tool_result("toolu_1", "72F", false),
+            ],
+            256,
+        );
+        let body = build_body(&req);
+        assert_eq!(body["messages"][1]["content"][0]["type"], "tool_use");
+        assert_eq!(body["messages"][1]["content"][0]["id"], "toolu_1");
+        assert_eq!(body["messages"][2]["content"][0]["type"], "tool_result");
+        assert_eq!(body["messages"][2]["content"][0]["tool_use_id"], "toolu_1");
+    }
+
+    // A recorded text + tool_use streamed response.
+    const FIXTURE: &str = r#"
+event: message_start
+data: {"type":"message_start","message":{"id":"msg_1","usage":{"input_tokens":24,"output_tokens":1}}}
+
+event: content_block_start
+data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me check."}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: content_block_start
+data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_42","name":"get_weather","input":{}}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"city\":"}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"\"SF\"}"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":1}
+
+event: message_delta
+data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":31}}
+
+event: message_stop
+data: {"type":"message_stop"}
+"#;
+
+    #[test]
+    fn decodes_text_then_tool_use_stream() {
+        let mut dec = Decoder::default();
+        let events = decode_sse(&mut dec, FIXTURE).unwrap();
+        assert_eq!(
+            events,
+            vec![
+                StreamEvent::MessageStart,
+                StreamEvent::TextDelta {
+                    text: "Let me check.".into()
+                },
+                StreamEvent::ContentBlockStop,
+                StreamEvent::ToolUseStart {
+                    id: "toolu_42".into(),
+                    name: "get_weather".into()
+                },
+                StreamEvent::InputJsonDelta {
+                    partial_json: "{\"city\":".into()
+                },
+                StreamEvent::InputJsonDelta {
+                    partial_json: "\"SF\"}".into()
+                },
+                StreamEvent::ContentBlockStop,
+                StreamEvent::Usage {
+                    input_tokens: 24,
+                    output_tokens: 31
+                },
+                StreamEvent::MessageStop {
+                    stop_reason: StopReason::ToolUse
+                },
+            ]
+        );
+    }
+}

crates/agent-core/src/dialect/mod.rs

@@ -0,0 +1,108 @@
+//! Wire dialects.
+//!
+//! The gateway relays bytes verbatim, so the harness owns the provider wire. Each [`Dialect`] knows
+//! how to (a) build a streaming request body from a [`ModelRequest`] and (b) decode that provider's
+//! SSE stream into the dialect-agnostic [`StreamEvent`] sequence the loop consumes.
+//!
+//! Selection is by model id: Claude speaks Anthropic (`/v1/messages`); everything else speaks the
+//! OpenAI wire (`/v1/chat/completions`), which is the lingua franca for the gateway's other providers.
+
+use serde_json::Value;
+
+use crate::error::{Error, Result};
+use crate::message::StreamEvent;
+use crate::transport::ModelRequest;
+
+pub mod anthropic;
+pub mod openai;
+
+/// Which provider wire a model speaks.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Dialect {
+    Anthropic,
+    OpenAi,
+}
+
+impl Dialect {
+    /// Pick the dialect for a model id. Claude → Anthropic; everything else → OpenAI wire.
+    pub fn for_model(model: &str) -> Self {
+        if model.starts_with("claude") || model.contains("anthropic") {
+            Dialect::Anthropic
+        } else {
+            Dialect::OpenAi
+        }
+    }
+
+    /// The gateway path this dialect POSTs to. Bare `/v1*` so the gateway's default-provider routing
+    /// (OpenAI on `/v1`, Anthropic on `/v1/messages`) applies; a `bai_v1` key picks the real provider.
+    pub fn endpoint_path(&self) -> &'static str {
+        match self {
+            Dialect::Anthropic => "/v1/messages",
+            Dialect::OpenAi => "/v1/chat/completions",
+        }
+    }
+
+    /// Build the JSON body for a streaming completion request.
+    pub fn build_body(&self, req: &ModelRequest) -> Value {
+        match self {
+            Dialect::Anthropic => anthropic::build_body(req),
+            Dialect::OpenAi => openai::build_body(req),
+        }
+    }
+
+    /// A fresh streaming decoder for this dialect.
+    pub fn decoder(&self) -> Box<dyn StreamDecoder> {
+        match self {
+            Dialect::Anthropic => Box::<anthropic::Decoder>::default(),
+            Dialect::OpenAi => Box::<openai::Decoder>::default(),
+        }
+    }
+}
+
+/// A stateful decoder turning a provider's SSE `data:` payloads into [`StreamEvent`]s. One per
+/// request (it carries cross-event state: token counts, the open content block, the stop reason).
+pub trait StreamDecoder: Send {
+    /// Feed one parsed SSE `data:` JSON object; return any events it produced.
+    fn push(&mut self, data: &Value) -> Vec<StreamEvent>;
+
+    /// Called once at end-of-stream. Flushes any held terminal event (OpenAI defers `MessageStop`
+    /// until the stream closes so it can land after the trailing usage chunk).
+    fn finish(&mut self) -> Vec<StreamEvent> {
+        Vec::new()
+    }
+}
+
+/// Decode a complete SSE body into events. Splits on `data:` lines, skips comments/`event:` lines
+/// and the OpenAI `[DONE]` sentinel, and flushes the decoder at the end. The streaming HTTP client
+/// reuses the same decoder incrementally; this is the buffered form used by tests.
+pub fn decode_sse(decoder: &mut dyn StreamDecoder, raw: &str) -> Result<Vec<StreamEvent>> {
+    let mut out = Vec::new();
+    for line in raw.lines() {
+        let line = line.trim_end_matches('\r');
+        let Some(payload) = line.strip_prefix("data:") else {
+            continue;
+        };
+        let payload = payload.trim();
+        if payload.is_empty() || payload == "[DONE]" {
+            continue;
+        }
+        let v: Value = serde_json::from_str(payload)
+            .map_err(|e| Error::Transport(format!("malformed SSE json: {e}")))?;
+        out.extend(decoder.push(&v));
+    }
+    out.extend(decoder.finish());
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn dialect_selection_by_model() {
+        assert_eq!(Dialect::for_model("claude-opus-4-8"), Dialect::Anthropic);
+        assert_eq!(Dialect::for_model("gpt-4o"), Dialect::OpenAi);
+        assert_eq!(Dialect::for_model("llama-3.1-70b"), Dialect::OpenAi);
+        assert_eq!(Dialect::for_model("anthropic/claude"), Dialect::Anthropic);
+    }
+}