diff --git a/crates/daemon/src/router.rs b/crates/daemon/src/router.rs index 628bc92e..f8d81f6a 100644 --- a/crates/daemon/src/router.rs +++ b/crates/daemon/src/router.rs @@ -24,7 +24,7 @@ pub mod oauth; pub mod proxy; pub mod translate; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; @@ -102,6 +102,16 @@ pub fn provider_dialect(provider: &str) -> Option { } } +/// Whether a target refuses an assistant turn that does not carry back the +/// reasoning it produced for that turn (spec 0181). +/// +/// Measured, not documented: DeepSeek's thinking mode rejects a replayed +/// tool-calling turn whose `reasoning_content` is missing, while accepting +/// the same turn when it is present — including when it is empty. +pub fn provider_echoes_reasoning(provider: &str) -> bool { + matches!(provider.to_ascii_lowercase().as_str(), "deepseek") +} + /// How a route's target consumes a requested reasoning effort (spec 0160). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum EffortSupport { @@ -400,6 +410,9 @@ pub struct ArmedRoute { pub client_dialect: Dialect, /// Whether and how the target honors a requested reasoning effort. pub effort: EffortSupport, + /// The target requires each assistant turn to carry back the reasoning + /// it produced for that turn (spec 0181). + pub reasoning_echo: bool, /// Pin-chosen effort applied when this arm is the session's durable /// pin (spec 0165). Catalog-resolved request-scoped arms leave this /// `None` so the harness request body remains authoritative. @@ -434,6 +447,9 @@ impl ArmedRoute { || self.system_prefix.is_some() || !self.extra_headers.is_empty() || !self.drop_params.is_empty() + // Carrying reasoning back is a rewrite of the request body, + // which byte-forwarding never performs (spec 0181). + || self.reasoning_echo } } @@ -456,6 +472,9 @@ pub struct SessionRouting { /// request. role_models: HashSet, route: RwLock>, + /// Reasoning the session's target has produced, kept so a later request + /// can hand each assistant turn its own back (spec 0181). + reasoning: RwLock, /// Bumped on every route change. Open pass-through tunnels compare it /// against the value they started with to notice they are stale: a /// tunnel decides tunnel-vs-intercept once, at CONNECT, and a harness @@ -468,11 +487,64 @@ pub struct SessionRouting { observed_tx: tokio::sync::mpsc::UnboundedSender, } +/// Reasoning a target produced for one of its own tool-calling turns, +/// keyed by the tool-call id that turn carried (spec 0181). +/// +/// Bounded rather than complete: an entry that has aged out degrades to an +/// empty echo, which the target accepts, instead of growing a session's +/// memory with every turn it has ever taken. +#[derive(Default)] +pub struct ReasoningMemo { + entries: VecDeque<(String, String)>, + bytes: usize, +} + +impl ReasoningMemo { + /// Reasoning kept per session before the oldest turns are forgotten. + const MAX_BYTES: usize = 1 << 20; + + fn remember(&mut self, id: String, reasoning: String) { + if id.is_empty() || self.entries.iter().any(|(known, _)| *known == id) { + return; + } + self.bytes += id.len() + reasoning.len(); + self.entries.push_back((id, reasoning)); + while self.bytes > Self::MAX_BYTES { + let Some((id, reasoning)) = self.entries.pop_front() else { + break; + }; + self.bytes -= id.len() + reasoning.len(); + } + } + + fn recall(&self, id: &str) -> Option { + self.entries + .iter() + .find(|(known, _)| known == id) + .map(|(_, reasoning)| reasoning.clone()) + } +} + impl SessionRouting { pub fn armed_route(&self) -> Option { self.route.read().unwrap().clone() } + /// Record the reasoning a target produced for the turn that made these + /// tool calls. Each call id resolves to it, since the harness may + /// replay the turn's calls in any order. + pub fn remember_reasoning(&self, tool_call_ids: &[String], reasoning: &str) { + let mut memo = self.reasoning.write().unwrap(); + for id in tool_call_ids { + memo.remember(id.clone(), reasoning.to_string()); + } + } + + /// The reasoning that accompanied `tool_call_id`, if still remembered. + pub fn recall_reasoning(&self, tool_call_id: &str) -> Option { + self.reasoning.read().unwrap().recall(tool_call_id) + } + /// Whether a request's model names an internal seat the harness chose /// for itself rather than the model the session runs on (spec 0166). pub fn is_role_model(&self, model: &str) -> bool { @@ -796,6 +868,7 @@ impl Router { catalog_enabled: AtomicBool::new(catalog_enabled), role_models, route: RwLock::new(None), + reasoning: RwLock::new(ReasoningMemo::default()), route_epoch: std::sync::atomic::AtomicU64::new(0), observed: AtomicBool::new(false), observed_tx: self.observed_tx.clone(), @@ -1023,6 +1096,7 @@ impl Router { target_dialect: provider.dialect(), client_dialect: routing.dialect, effort: oauth::effort_support(provider, &model), + reasoning_echo: false, pin_effort: None, client: reqwest::Client::new(), }) @@ -1148,6 +1222,7 @@ impl Router { target_dialect, client_dialect: routing.dialect, effort: profile_effort_support(&profile.provider, &resolved_model), + reasoning_echo: provider_echoes_reasoning(&profile.provider), model: resolved_model, pin_effort: None, client: reqwest::Client::new(), @@ -2019,6 +2094,72 @@ mod tests { ); } + /// A session remembers the reasoning its target produced, and forgets + /// the oldest of it rather than growing without bound. A forgotten + /// turn is a miss, which the caller answers with an empty echo — never + /// with another turn's reasoning (spec 0181). + #[test] + fn a_session_remembers_reasoning_per_tool_call_and_then_forgets_it() { + let mut memo = ReasoningMemo::default(); + memo.remember("call_1".into(), "first".into()); + memo.remember("call_2".into(), "second".into()); + memo.remember("call_1".into(), "a later turn must not overwrite".into()); + assert_eq!(memo.recall("call_1").as_deref(), Some("first")); + assert_eq!(memo.recall("call_2").as_deref(), Some("second")); + assert_eq!(memo.recall("call_unknown"), None); + + memo.remember(String::new(), "an id-less turn is not addressable".into()); + assert_eq!(memo.recall(""), None); + + let bulk = "x".repeat(64 * 1024); + for i in 0..24 { + memo.remember(format!("call_bulk_{i}"), bulk.clone()); + } + assert_eq!( + memo.recall("call_1"), + None, + "the oldest reasoning is dropped once the budget is spent" + ); + assert_eq!(memo.recall("call_bulk_23").as_deref(), Some(bulk.as_str())); + assert!(memo.bytes <= ReasoningMemo::MAX_BYTES); + } + + /// DeepSeek refuses a replayed tool-calling turn whose reasoning is + /// missing, so its arms must rebuild the request even when the harness + /// already speaks the target's dialect (spec 0181). + #[test] + fn a_deepseek_arm_carries_reasoning_and_always_rebuilds() { + assert!(provider_echoes_reasoning("deepseek")); + assert!(provider_echoes_reasoning("DeepSeek")); + assert!(!provider_echoes_reasoning("openai")); + assert!(!provider_echoes_reasoning("anthropic")); + + let mut route = ArmedRoute { + name: "deepseek".into(), + endpoint: "https://api.deepseek.com/v1/chat/completions".into(), + base_url: "https://api.deepseek.com/v1".into(), + model: "deepseek-v4-flash".into(), + api_key: "sk-test".into(), + auth: TargetAuth::Bearer, + system_prefix: None, + extra_headers: Vec::new(), + drop_params: &[], + target_dialect: Dialect::OpenAiChat, + client_dialect: Dialect::OpenAiChat, + effort: EffortSupport::DeepSeek, + reasoning_echo: provider_echoes_reasoning("deepseek"), + pin_effort: None, + client: reqwest::Client::new(), + }; + assert!(!route.translates(), "same dialect on both ends"); + assert!( + route.needs_rebuild(), + "byte-forwarding would ship the turn without its reasoning" + ); + route.reasoning_echo = false; + assert!(!route.needs_rebuild()); + } + /// The built-in DeepSeek target (spec 0179) reaches the router as an /// ordinary profile, so it must be selectable from an Anthropic harness /// and carry every model the shared catalog lists for the provider — @@ -2061,6 +2202,10 @@ mod tests { .unwrap(); assert_eq!(armed.model, "deepseek-v4-pro"); let ctx = r.sessions.read().unwrap()["s1"].clone(); + assert!( + ctx.armed_route().unwrap().reasoning_echo, + "a DeepSeek arm must carry reasoning back (spec 0181)" + ); assert!( ctx.armed_route().unwrap().translates(), "a chat-completions target from an anthropic harness must translate" diff --git a/crates/daemon/src/router/proxy.rs b/crates/daemon/src/router/proxy.rs index 2444ca40..50fc4b04 100644 --- a/crates/daemon/src/router/proxy.rs +++ b/crates/daemon/src/router/proxy.rs @@ -664,7 +664,7 @@ where { ctx.mark_observed(); let streaming = wants_stream(&body); - return match forward_translated(body, &route, client_dialect).await { + return match forward_translated(body, &route, client_dialect, &ctx).await { Ok(forwarded) => { write_translated_response( stream, @@ -673,6 +673,7 @@ where client_dialect, streaming, &forwarded.context, + &ctx, ) .await } @@ -1004,10 +1005,14 @@ async fn forward_translated( body: Vec, route: &ArmedRoute, client_dialect: Dialect, + ctx: &SessionRouting, ) -> Result { let source: serde_json::Value = serde_json::from_slice(&body).context("parse intercepted request body")?; let mut canon = translate::parse_request(client_dialect, &source); + if route.reasoning_echo { + restore_reasoning(&mut canon, |id| ctx.recall_reasoning(id)); + } // Durable pin effort overrides the harness body on pin-routed turns // (spec 0165). Catalog-resolved arms leave pin_effort empty so the // request body remains the authority. @@ -1096,6 +1101,45 @@ async fn forward_translated( }) } +/// Give each replayed assistant turn back the reasoning the target +/// produced for it (spec 0181). +/// +/// A harness that speaks a dialect without a reasoning field cannot carry +/// it, so the proxy remembers it instead — keyed by the turn's tool-call +/// ids, which the harness does replay. A turn whose reasoning is no longer +/// remembered gets an empty one: the target accepts that, and an empty +/// echo is the honest statement that we no longer have it. Nothing is +/// invented. +fn restore_reasoning( + canon: &mut translate::CanonRequest, + recall: impl Fn(&str) -> Option, +) { + for message in &mut canon.messages { + if message.role != translate::CanonRole::Assistant { + continue; + } + // The harness carried it itself — leave its own account alone. + if message + .blocks + .iter() + .any(|block| matches!(block, translate::CanonBlock::Thinking(_))) + { + continue; + } + let called = message.blocks.iter().find_map(|block| match block { + translate::CanonBlock::ToolUse { id, .. } => Some(id.clone()), + _ => None, + }); + // Only tool-calling turns are refused without it; a plain answer + // needs no echo and gains nothing from an empty one. + let Some(id) = called else { continue }; + let reasoning = recall(&id).unwrap_or_default(); + message + .blocks + .insert(0, translate::CanonBlock::Thinking(reasoning)); + } +} + /// Map Codex's effort vocabulary onto the K3 scale. fn kimi_effort(effort: &str) -> &'static str { match effort { @@ -1146,12 +1190,15 @@ async fn write_translated_response( client_dialect: Dialect, streaming: bool, context: &translate::TranslationContext, + ctx: &SessionRouting, ) -> Result<()> where S: tokio::io::AsyncWrite + Unpin, { use futures::StreamExt; + let mut capture = ReasoningCapture::new(route.reasoning_echo); + let status = response.status(); if !status.is_success() { let bytes = tokio::time::timeout( @@ -1200,14 +1247,16 @@ where let body = translate::error_body(client_dialect, message).to_string(); return write_simple(stream, 502, &body).await; } - let body = translate::encode_response_with_context( - client_dialect, - route.target_dialect, - &parsed, - &route.model, - context, - ) - .to_string(); + let events = + translate::decode_full_response_with_context(route.target_dialect, &parsed, context); + for event in &events { + capture.observe(event); + } + if let Some((ids, reasoning)) = capture.take() { + ctx.remember_reasoning(&ids, &reasoning); + } + let body = + translate::encode_full_response(client_dialect, &events, &route.model).to_string(); return write_simple(stream, 200, &body).await; } @@ -1257,6 +1306,7 @@ where route.target_dialect, data.trim(), context, + &mut capture, ) .await? { @@ -1282,6 +1332,7 @@ where route.target_dialect, data.trim(), context, + &mut capture, ) .await? { @@ -1307,6 +1358,11 @@ where failed = true; } if !failed { + // Only a turn that arrived whole is worth remembering: a truncated + // one is not the reasoning the target will expect back. + if let Some((ids, reasoning)) = capture.take() { + ctx.remember_reasoning(&ids, &reasoning); + } let tail = encoder.finish(); write_chunk(stream, tail.as_bytes()).await?; } @@ -1316,6 +1372,53 @@ where Ok(()) } +/// The reasoning and tool-call ids of one response, held until the turn +/// completes and then recorded against the session (spec 0181). +#[derive(Default)] +struct ReasoningCapture { + armed: bool, + reasoning: String, + tool_call_ids: Vec, +} + +impl ReasoningCapture { + fn new(armed: bool) -> Self { + Self { + armed, + ..Self::default() + } + } + + fn observe(&mut self, event: &translate::CanonEvent) { + if !self.armed { + return; + } + match event { + translate::CanonEvent::ThinkingDelta(delta) => self.reasoning.push_str(delta), + translate::CanonEvent::ToolStart { id, .. } if !id.is_empty() => { + if !self.tool_call_ids.iter().any(|known| known == id) { + self.tool_call_ids.push(id.clone()); + } + } + _ => {} + } + } + + /// The completed turn, if it is one worth remembering. A turn with no + /// tool calls is never replayed as one, and a turn with no reasoning + /// has nothing to hand back that the empty default does not already + /// cover. + fn take(&mut self) -> Option<(Vec, String)> { + if self.reasoning.is_empty() || self.tool_call_ids.is_empty() { + return None; + } + Some(( + std::mem::take(&mut self.tool_call_ids), + std::mem::take(&mut self.reasoning), + )) + } +} + enum SseOutcome { Continue, Frame { terminal: bool }, @@ -1328,6 +1431,7 @@ async fn process_target_sse_data( dialect: Dialect, data: &str, context: &translate::TranslationContext, + capture: &mut ReasoningCapture, ) -> Result where S: tokio::io::AsyncWrite + Unpin, @@ -1363,6 +1467,7 @@ where && matches!(event, translate::CanonEvent::Usage { .. })) }); for event in events { + capture.observe(&event); let out = encoder.push(&event); if !out.is_empty() { write_chunk(stream, out.as_bytes()).await?; @@ -1568,6 +1673,120 @@ mod tests { /// The client's own credential must never ride along to a different /// vendor's endpoint. + /// A turn whose reasoning is still remembered gets it back verbatim; + /// one that aged out gets an empty reasoning rather than none, because + /// a thinking target refuses the turn outright when the field is + /// missing and nothing may be invented in its place (spec 0181). + #[test] + fn a_replayed_tool_turn_gets_its_reasoning_back() { + let mut canon = translate::CanonRequest { + messages: vec![ + translate::CanonMessage { + role: translate::CanonRole::Assistant, + blocks: vec![translate::CanonBlock::ToolUse { + id: "call_known".into(), + name: "ls".into(), + input: serde_json::json!({}), + }], + }, + translate::CanonMessage { + role: translate::CanonRole::Assistant, + blocks: vec![translate::CanonBlock::ToolUse { + id: "call_forgotten".into(), + name: "ls".into(), + input: serde_json::json!({}), + }], + }, + translate::CanonMessage { + role: translate::CanonRole::Assistant, + blocks: vec![translate::CanonBlock::Text("done".into())], + }, + ], + ..Default::default() + }; + restore_reasoning(&mut canon, |id| { + (id == "call_known").then(|| "the root listing first".to_string()) + }); + assert_eq!( + canon.messages[0].blocks[0], + translate::CanonBlock::Thinking("the root listing first".into()) + ); + assert_eq!( + canon.messages[1].blocks[0], + translate::CanonBlock::Thinking(String::new()) + ); + assert!( + !canon.messages[2] + .blocks + .iter() + .any(|b| matches!(b, translate::CanonBlock::Thinking(_))), + "a turn that called no tool is not refused without reasoning" + ); + } + + /// A harness that carries reasoning itself is the authority on its own + /// turn — the proxy's memory must not overwrite it. + #[test] + fn a_turn_that_carries_its_own_reasoning_is_left_alone() { + let mut canon = translate::CanonRequest { + messages: vec![translate::CanonMessage { + role: translate::CanonRole::Assistant, + blocks: vec![ + translate::CanonBlock::Thinking("what the harness kept".into()), + translate::CanonBlock::ToolUse { + id: "call_known".into(), + name: "ls".into(), + input: serde_json::json!({}), + }, + ], + }], + ..Default::default() + }; + restore_reasoning(&mut canon, |_| Some("what the proxy kept".to_string())); + assert_eq!( + canon.messages[0].blocks[0], + translate::CanonBlock::Thinking("what the harness kept".into()) + ); + assert_eq!(canon.messages[0].blocks.len(), 2); + } + + #[test] + fn a_capture_keeps_only_a_reasoned_tool_turn() { + let mut capture = ReasoningCapture::new(true); + capture.observe(&translate::CanonEvent::ThinkingDelta("weigh".into())); + capture.observe(&translate::CanonEvent::ThinkingDelta("ing".into())); + capture.observe(&translate::CanonEvent::ToolStart { + index: 0, + id: "call_1".into(), + name: "ls".into(), + }); + capture.observe(&translate::CanonEvent::ToolStart { + index: 1, + id: "call_2".into(), + name: "grep".into(), + }); + let (ids, reasoning) = capture.take().expect("a reasoned tool turn is kept"); + assert_eq!(ids, vec!["call_1".to_string(), "call_2".to_string()]); + assert_eq!(reasoning, "weighing"); + assert!(capture.take().is_none(), "a turn is recorded once"); + + // Text-only turns are never replayed as tool calls, and a target + // that reasoned nothing has nothing to hand back. + let mut text_only = ReasoningCapture::new(true); + text_only.observe(&translate::CanonEvent::ThinkingDelta("weighing".into())); + assert!(text_only.take().is_none()); + + // A target the flag was never raised for costs nothing to stream. + let mut disarmed = ReasoningCapture::new(false); + disarmed.observe(&translate::CanonEvent::ThinkingDelta("weighing".into())); + disarmed.observe(&translate::CanonEvent::ToolStart { + index: 0, + id: "call_1".into(), + name: "ls".into(), + }); + assert!(disarmed.take().is_none()); + } + /// Build a session whose routable host is loopback, so the drain path /// can be exercised without DNS. fn drainable_ctx(dir: &tempfile::TempDir) -> Arc { @@ -1589,6 +1808,7 @@ mod tests { catalog_enabled: std::sync::atomic::AtomicBool::new(false), role_models: std::collections::HashSet::new(), route: std::sync::RwLock::new(None), + reasoning: std::sync::RwLock::new(Default::default()), route_epoch: std::sync::atomic::AtomicU64::new(0), observed: std::sync::atomic::AtomicBool::new(false), observed_tx: tx, @@ -1795,6 +2015,7 @@ mod tests { Dialect::GoogleGemini, payload, &translate::TranslationContext::default(), + &mut ReasoningCapture::default(), ) .await .unwrap(); diff --git a/crates/daemon/src/router/translate.rs b/crates/daemon/src/router/translate.rs index 2d2478f3..32e02545 100644 --- a/crates/daemon/src/router/translate.rs +++ b/crates/daemon/src/router/translate.rs @@ -127,6 +127,10 @@ pub enum CanonEvent { id: String, }, TextDelta(String), + /// Model-private reasoning as the target streamed it. Not re-encoded + /// into any client dialect — it exists so the proxy can remember what + /// a target reasoned and hand it back on the next request (spec 0181). + ThinkingDelta(String), ToolStart { index: usize, id: String, @@ -464,27 +468,42 @@ impl ClientEncoder { } } -/// Non-streaming response body in the client's dialect. -pub fn encode_response_with_context( - dialect: Dialect, +/// Decode a whole non-streaming response into canonical events. +pub fn decode_full_response_with_context( target: Dialect, body: &Value, - model: &str, context: &TranslationContext, -) -> Value { - let events = match target { +) -> Vec { + match target { Dialect::AnthropicMessages => anthropic::decode_full_response(body), Dialect::GoogleGemini => google::decode_full_response(body, context), Dialect::OpenAiResponses => responses::decode_full_response(body), Dialect::OpenAiChat => openai_chat::decode_full_response(body), - }; + } +} + +/// Re-encode decoded events as a non-streaming body in `dialect`. +pub fn encode_full_response(dialect: Dialect, events: &[CanonEvent], model: &str) -> Value { match dialect { - Dialect::OpenAiChat => openai_chat::encode_full(&events, model), - Dialect::OpenAiResponses => responses::encode_full(&events, model), - _ => anthropic::encode_full(&events, model), + Dialect::OpenAiChat => openai_chat::encode_full(events, model), + Dialect::OpenAiResponses => responses::encode_full(events, model), + _ => anthropic::encode_full(events, model), } } +/// Non-streaming response body in the client's dialect. +#[cfg_attr(not(test), allow(dead_code))] +pub fn encode_response_with_context( + dialect: Dialect, + target: Dialect, + body: &Value, + model: &str, + context: &TranslationContext, +) -> Value { + let events = decode_full_response_with_context(target, body, context); + encode_full_response(dialect, &events, model) +} + /// Rough token estimate for endpoints a target cannot answer. /// /// Deliberately approximate and deliberately not silent: refusing would diff --git a/crates/daemon/src/router/translate/anthropic.rs b/crates/daemon/src/router/translate/anthropic.rs index a940ea9b..8fb5cbb4 100644 --- a/crates/daemon/src/router/translate/anthropic.rs +++ b/crates/daemon/src/router/translate/anthropic.rs @@ -452,6 +452,7 @@ pub fn encode_full(events: &[CanonEvent], model: &str) -> Value { for event in events { match event { CanonEvent::TextDelta(t) => text.push_str(t), + CanonEvent::ThinkingDelta(_) => {} CanonEvent::ToolStart { id, name, .. } => { tools.push((id.clone(), name.clone(), String::new())) } @@ -564,6 +565,7 @@ impl StreamEncoder { }), )); } + CanonEvent::ThinkingDelta(_) => {} CanonEvent::ToolStart { index, id, name } => { if !self.tools.contains(index) { out.push_str(&self.close_open()); diff --git a/crates/daemon/src/router/translate/openai_chat.rs b/crates/daemon/src/router/translate/openai_chat.rs index 3c4e1fed..6c88e3d9 100644 --- a/crates/daemon/src/router/translate/openai_chat.rs +++ b/crates/daemon/src/router/translate/openai_chat.rs @@ -29,6 +29,12 @@ pub fn parse_request(body: &Value) -> CanonRequest { continue; } let mut blocks = Vec::new(); + // Reasoning-model dialects carry the model's own thinking beside + // the content. It leads the turn so it stays ahead of the text it + // produced when the blocks are emitted again. + if let Some(reasoning) = message.get("reasoning_content").and_then(Value::as_str) { + blocks.push(CanonBlock::Thinking(reasoning.to_string())); + } match message.get("content") { Some(Value::String(t)) if !t.is_empty() => blocks.push(CanonBlock::Text(t.clone())), Some(Value::Array(parts)) => { @@ -191,6 +197,7 @@ pub fn emit_request(req: &CanonRequest, model: &str) -> Value { // conversation otherwise. let mut parts: Vec = Vec::new(); let mut calls: Vec = Vec::new(); + let mut reasoning: Option = None; for block in &message.blocks { match block { CanonBlock::Text(t) => parts.push(json!({"type":"text","text":t})), @@ -204,7 +211,12 @@ pub fn emit_request(req: &CanonRequest, model: &str) -> Value { CanonBlock::ToolResult { id, text, .. } => { messages.push(json!({"role":"tool","tool_call_id":id,"content":text})); } - CanonBlock::Thinking(_) => {} + // A thinking target rejects a turn whose reasoning it + // issued and cannot see again, so it is carried rather + // than dropped (spec 0181). + CanonBlock::Thinking(t) => { + reasoning.get_or_insert_with(String::new).push_str(t); + } } } if parts.is_empty() && calls.is_empty() { @@ -234,6 +246,9 @@ pub fn emit_request(req: &CanonRequest, model: &str) -> Value { if !calls.is_empty() { m.insert("tool_calls".into(), json!(calls)); } + if let Some(reasoning) = reasoning.filter(|_| message.role == CanonRole::Assistant) { + m.insert("reasoning_content".into(), json!(reasoning)); + } messages.push(Value::Object(m)); } @@ -305,6 +320,13 @@ pub fn decode_event(data: &Value) -> Vec { return out; }; if let Some(delta) = choice.get("delta") { + if let Some(reasoning) = delta + .get("reasoning_content") + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + { + out.push(CanonEvent::ThinkingDelta(reasoning.to_string())); + } if let Some(text) = delta .get("content") .and_then(Value::as_str) @@ -368,6 +390,13 @@ pub fn decode_full_response(body: &Value) -> Vec { .and_then(Value::as_array) .and_then(|c| c.first()); let message = choice.and_then(|c| c.get("message")); + if let Some(reasoning) = message + .and_then(|m| m.get("reasoning_content")) + .and_then(Value::as_str) + .filter(|s| !s.is_empty()) + { + events.push(CanonEvent::ThinkingDelta(reasoning.to_string())); + } if let Some(text) = message .and_then(|m| m.get("content")) .and_then(Value::as_str) @@ -490,6 +519,9 @@ impl StreamEncoder { CanonEvent::TextDelta(text) => { out.push_str(&self.chunk(json!({"content":text}), Value::Null, Value::Null)); } + // Reasoning is remembered by the proxy, not replayed to the + // harness: no client dialect asked for it (spec 0181). + CanonEvent::ThinkingDelta(_) => {} CanonEvent::ToolStart { index, id, name } => { if self.tool_args.contains_key(index) { return out; @@ -585,6 +617,7 @@ pub fn encode_full(events: &[CanonEvent], model: &str) -> Value { CanonEvent::Start { id: event_id } if !event_id.is_empty() => id = event_id.clone(), CanonEvent::Start { .. } => {} CanonEvent::TextDelta(delta) => text.push_str(delta), + CanonEvent::ThinkingDelta(_) => {} CanonEvent::ToolStart { index, id, name } => { tools .entry(*index) @@ -681,6 +714,104 @@ mod tests { assert_eq!(messages[1]["tool_call_id"], "t1"); } + /// A thinking target refuses a replayed tool-calling turn whose + /// reasoning it cannot see again, so the turn has to carry it (spec + /// 0181) — including the empty reasoning that stands for "no longer + /// remembered". + #[test] + fn an_assistant_turn_carries_its_reasoning_back() { + let req = CanonRequest { + messages: vec![ + CanonMessage { + role: CanonRole::Assistant, + blocks: vec![ + CanonBlock::Thinking("the repo root is the place to look".into()), + CanonBlock::Text("Looking.".into()), + CanonBlock::ToolUse { + id: "call_1".into(), + name: "ls".into(), + input: json!({"p": "/"}), + }, + ], + }, + CanonMessage { + role: CanonRole::Assistant, + blocks: vec![ + CanonBlock::Thinking(String::new()), + CanonBlock::ToolUse { + id: "call_2".into(), + name: "ls".into(), + input: json!({"p": "/src"}), + }, + ], + }, + ], + ..Default::default() + }; + let out = emit_request(&req, "deepseek-v4-flash"); + let messages = out["messages"].as_array().unwrap(); + assert_eq!( + messages[0]["reasoning_content"], + "the repo root is the place to look" + ); + assert_eq!(messages[1]["reasoning_content"], ""); + } + + /// Reasoning must survive a round trip, or replaying a conversation + /// this dialect produced would strip what the target demands back. + #[test] + fn reasoning_round_trips_through_canonical_form() { + let original = json!({ + "model": "deepseek-v4-flash", + "messages": [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "Looking.", + "reasoning_content": "they want a listing", + "tool_calls": [{"id":"call_1","type":"function", + "function":{"name":"ls","arguments":"{}"}}]}, + ] + }); + let canon = parse_request(&original); + assert_eq!( + canon.messages[1].blocks[0], + CanonBlock::Thinking("they want a listing".into()) + ); + let back = emit_request(&canon, "deepseek-v4-flash"); + assert_eq!(back["messages"][1]["reasoning_content"], "they want a listing"); + } + + /// A user turn never carries reasoning, whatever a client put on it. + #[test] + fn only_assistant_turns_carry_reasoning() { + let req = CanonRequest { + messages: vec![CanonMessage { + role: CanonRole::User, + blocks: vec![ + CanonBlock::Thinking("stray".into()), + CanonBlock::Text("hi".into()), + ], + }], + ..Default::default() + }; + let out = emit_request(&req, "deepseek-v4-flash"); + assert!(out["messages"][0].get("reasoning_content").is_none()); + } + + #[test] + fn decodes_streamed_and_whole_reasoning() { + assert_eq!( + decode_event(&json!({"choices":[{"delta":{"reasoning_content":"weighing"}}]})), + vec![CanonEvent::ThinkingDelta("weighing".into())] + ); + let whole = decode_full_response(&json!({ + "id":"cmpl_1", + "choices":[{"message":{"role":"assistant","reasoning_content":"weighing","content":"ok"}, + "finish_reason":"stop"}] + })); + assert!(whole.contains(&CanonEvent::ThinkingDelta("weighing".into()))); + assert!(whole.contains(&CanonEvent::TextDelta("ok".into()))); + } + #[test] fn omits_tool_choice_when_no_translatable_tools_remain() { let req = CanonRequest { diff --git a/crates/daemon/src/router/translate/responses.rs b/crates/daemon/src/router/translate/responses.rs index d8ab4afa..cd69a975 100644 --- a/crates/daemon/src/router/translate/responses.rs +++ b/crates/daemon/src/router/translate/responses.rs @@ -734,6 +734,9 @@ impl StreamEncoder { } match event { CanonEvent::Start { .. } => {} + // Reasoning is remembered by the proxy, not replayed to the + // harness: no client dialect asked for it (spec 0181). + CanonEvent::ThinkingDelta(_) => {} CanonEvent::TextDelta(text) => { if !self.message_open { out.push_str(&self.open_message()); @@ -895,6 +898,7 @@ pub fn encode_full(events: &[CanonEvent], model: &str) -> Value { for event in events { match event { CanonEvent::TextDelta(t) => text.push_str(t), + CanonEvent::ThinkingDelta(_) => {} CanonEvent::ToolStart { id, name, .. } => { calls.push((id.clone(), name.clone(), String::new())) } diff --git a/specs/0181-routed-turns-keep-their-reasoning.md b/specs/0181-routed-turns-keep-their-reasoning.md new file mode 100644 index 00000000..71a2cfd4 --- /dev/null +++ b/specs/0181-routed-turns-keep-their-reasoning.md @@ -0,0 +1,87 @@ +# 0181-routed-turns-keep-their-reasoning + +Status: accepted +Date: 2026-08-02 +Area: protocol +Scope: A routed turn must hand a thinking target back the reasoning that target produced, without the harness having to carry it. + +## Decision + +When a target refuses an assistant turn that does not carry back the +reasoning it produced for that turn, the router supplies it: + +- The router **captures** the reasoning a target streams, keyed by the + tool-call ids of the turn it belongs to, and keeps it for the session. +- On a later request it **restores** that reasoning onto each replayed + assistant turn that called a tool, in the field the target's dialect + defines for it. +- A turn whose reasoning the harness carried itself is left alone: the + harness's own account of its turn wins over the router's memory. +- A turn whose reasoning is no longer remembered is sent with an **empty** + reasoning. The router never writes reasoning the model did not produce. +- Reasoning is remembered for the session but never re-encoded into the + response the harness reads. It is the target's own record, not new + assistant content, and materializing it as text would put words in the + model's mouth. + +Which targets need this is **measured, not assumed** — the same rule as +effort support (spec 0160). A target is marked as requiring the echo only +after its refusal has been observed. + +An arm that must carry reasoning always rebuilds the request body, even +when the harness and the target speak the same dialect: byte-forwarding +cannot add a field. + +## Reason + +Reasoning models are increasingly stateful about their own thinking: the +provider will not re-derive reasoning it already produced, and refuses a +replayed tool-calling turn that arrives without it. DeepSeek's thinking +mode is the observed case — a turn carrying `tool_calls` is rejected +unless `reasoning_content` accompanies it, while the same turn with an +empty reasoning is accepted. + +The harness cannot solve this. Most harnesses speak a dialect with no +field for another vendor's reasoning, so whatever the target reasoned is +gone by the time the harness replays the conversation. A harness that +does have such a field is not the one that produced the reasoning either. +The router is the only participant that sees both the target's response +and the request that replays it, so it is the only one that can keep them +consistent. + +Depending on the target to recover its own reasoning is not a +substitute. A provider may resolve reasoning from a recently issued +tool-call id, but that recovery is a cache: it holds early in a +conversation and lapses later, which surfaces as a session that works for +several turns and then fails permanently mid-task. + +## Consequences + +- Sessions carry bounded per-session state that a stateless proxy + otherwise would not. It is capped, and the oldest turns are forgotten + first; a forgotten turn degrades to an empty echo, never to an error. +- A daemon restart loses the memory. Resumed conversations continue with + empty reasoning on their older turns rather than failing. +- Reasoning is provider-private text. It stays inside the router: it is + not persisted to the transcript, not shown as assistant output, and not + forwarded to a different target than the one that produced it. +- Adding a dialect means deciding how it carries reasoning in both + directions, not just how it carries text and tool calls. + +## Non-Goals + +- Rendering a routed model's reasoning to the user. Whether reasoning is + surfaced in a client is a separate decision from keeping the turn valid. +- Turning reasoning into portable content moved between providers. +- Reconstructing reasoning for a turn the router never saw. + +## Examples + +- A harness calls three tools over three turns against a thinking target. + Each replay carries that turn's own reasoning back; the conversation + continues instead of failing on the second or third round. +- The same conversation after a daemon restart: every prior turn carries + an empty reasoning, which the target accepts, and new turns start + accumulating reasoning again. +- A target with no such requirement sees byte-identical requests to + before; nothing is captured and nothing is added.