From 03634761ef21e9d138198fdced2c5e201198cfdd Mon Sep 17 00:00:00 2001 From: Edwin Date: Sun, 2 Aug 2026 11:36:07 -0700 Subject: [PATCH] Give a routed turn back the reasoning its target produced MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A DeepSeek-routed session dies mid-task with "The `reasoning_content` in the thinking mode must be passed back to the API." DeepSeek's thinking mode refuses a replayed assistant turn that carries tool calls without the reasoning it produced for that turn — measured: the same turn is accepted the moment the field is present, even when it is empty. The router could not supply it. The chat-completions dialect dropped reasoning in both directions: responses were decoded without it, and canonical thinking blocks were discarded when emitting a request. The harness cannot cover for that either — Anthropic-speaking Claude Code has no field for another vendor's reasoning, so by the time it replays the conversation the reasoning is gone. DeepSeek recovers it from a recently issued tool-call id for a while, which is why a session works for several turns and then fails for good. So the router keeps it. Reasoning a target streams is captured against that turn's tool-call ids and restored onto the turn when the harness replays it. A turn the harness carried reasoning for is left alone; a turn no longer remembered is sent with an empty reasoning, which the target accepts — nothing is invented. The memory is bounded per session and the reasoning is never re-encoded into what the harness reads. Marked per target, measured rather than assumed (spec 0160's rule), and an arm that needs it always rebuilds the request body, since byte forwarding cannot add a field. Verified against the live API: the exact body the router now emits is accepted, and the same body with the reasoning stripped reproduces the 400. --- crates/daemon/src/router.rs | 147 ++++++++++- crates/daemon/src/router/proxy.rs | 239 +++++++++++++++++- crates/daemon/src/router/translate.rs | 39 ++- .../daemon/src/router/translate/anthropic.rs | 2 + .../src/router/translate/openai_chat.rs | 133 +++++++++- .../daemon/src/router/translate/responses.rs | 4 + .../0181-routed-turns-keep-their-reasoning.md | 87 +++++++ 7 files changed, 630 insertions(+), 21 deletions(-) create mode 100644 specs/0181-routed-turns-keep-their-reasoning.md 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.