From eda5afeb52791132c4dd67f1e24073c07401a4a6 Mon Sep 17 00:00:00 2001 From: Siddhesh Sonar Date: Thu, 1 Oct 2026 16:19:53 +0530 Subject: [PATCH] fix(claude-code): show model thinking and cache reads, price cache reads in agent configs --- docs/EDITORS.md | 8 ++ src/account/console.rs | 2 + src/anthropic/translate.rs | 260 +++++++++++++++++++++++++++++++--- src/harness/agents.rs | 5 +- src/harness/catalog_models.rs | 12 +- src/harness/opencode.rs | 4 + tests/cpp_case_exceptions.txt | 1 + tests/test_wally_harness.rs | 3 + tests/test_wally_opencode.rs | 25 ++++ tests/test_wally_unit/shim.rs | 24 ++-- 10 files changed, 303 insertions(+), 41 deletions(-) diff --git a/docs/EDITORS.md b/docs/EDITORS.md index 67bf745e..5181b0c3 100644 --- a/docs/EDITORS.md +++ b/docs/EDITORS.md @@ -40,6 +40,14 @@ which is what lets an agent on the far side run the tools it was given rather than describe them. `opencode`, `hermes` and `openclaw` speak OpenAI already, so they talk to the endpoint directly. +The model's reasoning shows up as a thinking block with an empty signature, +because these models can't sign it. Claude Code accepts that, and wally drops +the block again when Claude Code sends it back. Cache reads show up as +`cache_read_input_tokens`, taken out of `input_tokens` so nothing is counted +twice. Cache writes aren't reported, because the endpoint doesn't report them. +OpenCode, OpenClaw and Prime Agent get the catalog's cache-read price, so their +cost display charges a cached prompt at the cached rate. + Nothing wally writes for a tool outlives the run. The variables go in the child process. DeepSeek Harness is handed a patch in a temp directory, deleted when it exits; its own `$DSH_HOME` is never written to, and the API key never enters diff --git a/src/account/console.rs b/src/account/console.rs index c1f208d9..09bc1d39 100644 --- a/src/account/console.rs +++ b/src/account/console.rs @@ -255,6 +255,7 @@ pub struct CatalogPrice { pub id: String, pub input_per_mtok: i64, pub output_per_mtok: i64, + pub cached_input_per_mtok: i64, } pub const LOGIN_MAX_WAIT_SECONDS: i32 = 10; @@ -1909,6 +1910,7 @@ impl ConsoleClient { id: model.id, input_per_mtok: model.input_per_mtok, output_per_mtok: model.output_per_mtok, + cached_input_per_mtok: model.cached_input_per_mtok, }) .collect(); (IdentityResult::Ok, prices, String::new()) diff --git a/src/anthropic/translate.rs b/src/anthropic/translate.rs index 1e7c3020..6829d36b 100644 --- a/src/anthropic/translate.rs +++ b/src/anthropic/translate.rs @@ -22,6 +22,8 @@ pub struct StreamState { pub opened: bool, pub block_open: bool, pub text_index: i32, + pub thinking_open: bool, + pub thinking_index: i32, pub next_index: i32, pub failed: bool, pub closed: bool, @@ -30,6 +32,9 @@ pub struct StreamState { pub stop_reason: String, pub input_tokens: i32, pub output_tokens: i32, + /// `prompt_tokens_details.cached_tokens` when upstream reported it; None + /// keeps the usage block exactly as it was before cache reporting. + pub cached_tokens: Option, pub input_estimate: i32, pub output_chars: i32, pub tool_calls: BTreeMap, @@ -44,6 +49,8 @@ impl Default for StreamState { opened: false, block_open: false, text_index: -1, + thinking_open: false, + thinking_index: -1, next_index: 0, failed: false, closed: false, @@ -52,6 +59,7 @@ impl Default for StreamState { stop_reason: String::new(), input_tokens: 0, output_tokens: 0, + cached_tokens: None, input_estimate: 0, output_chars: 0, tool_calls: BTreeMap::new(), @@ -277,6 +285,62 @@ fn estimate_tokens_from_chars(chars: usize) -> i32 { } } +/// `usage.prompt_tokens_details.cached_tokens`, when upstream sent it. +fn cached_tokens(usage: &Value) -> Option { + usage + .get("prompt_tokens_details") + .and_then(|details| details.get("cached_tokens")) + .and_then(Value::as_i64) + .map(|v| v as i32) +} + +/// Anthropic usage from OpenAI counts. Anthropic counts cache reads apart +/// from `input_tokens` while OpenAI's `prompt_tokens` includes them, so the +/// cached part moves to `cache_read_input_tokens` and the total is unchanged. +fn anthropic_usage(prompt: i32, output: i32, cached: Option) -> Value { + match cached { + Some(cached) => { + let cached = cached.clamp(0, prompt.max(0)); + json!({ + "input_tokens": prompt - cached, + "output_tokens": output, + "cache_read_input_tokens": cached, + }) + } + None => json!({"input_tokens": prompt, "output_tokens": output}), + } +} + +fn read_stream_usage(usage: &Value, state: &mut StreamState) { + state.input_tokens = count(usage, "prompt_tokens", state.input_tokens); + state.output_tokens = count(usage, "completion_tokens", state.output_tokens); + if let Some(cached) = cached_tokens(usage) { + state.cached_tokens = Some(cached); + } +} + +fn close_text_block(state: &mut StreamState) -> String { + if !state.block_open { + return String::new(); + } + state.block_open = false; + event( + "content_block_stop", + &json!({"type": "content_block_stop", "index": state.text_index}), + ) +} + +fn close_thinking_block(state: &mut StreamState) -> String { + if !state.thinking_open { + return String::new(); + } + state.thinking_open = false; + event( + "content_block_stop", + &json!({"type": "content_block_stop", "index": state.thinking_index}), + ) +} + // ---- public API ------------------------------------------------------- /// One Anthropic `/v1/messages` request body translated to OpenAI @@ -457,9 +521,17 @@ pub fn response_to_anthropic(openai: &Value, model: &str) -> Value { .to_string(); let mut content = Vec::new(); + // The model's reasoning, shown the way Anthropic shows its own. The + // signature is empty: no upstream here signs its reasoning, and the + // block is dropped again when Claude Code sends the history back. + let reasoning = field(&message, "reasoning_content"); + if !reasoning.is_empty() { + content.push(json!({"type": "thinking", "thinking": reasoning, "signature": ""})); + } if !text.is_empty() { content.push(json!({"type": "text", "text": text})); } + let blocks_before_tools = content.len(); if let Some(calls) = message.get("tool_calls").and_then(Value::as_array) { for call in calls { if !call.is_object() { @@ -482,7 +554,7 @@ pub fn response_to_anthropic(openai: &Value, model: &str) -> Value { })); } } - let tool_called = content.len() > if text.is_empty() { 0 } else { 1 }; + let tool_called = content.len() > blocks_before_tools; if content.is_empty() { content.push(json!({"type": "text", "text": ""})); } @@ -503,10 +575,11 @@ pub fn response_to_anthropic(openai: &Value, model: &str) -> Value { "content": content, "stop_reason": if stop.is_empty() { Value::Null } else { json!(stop) }, "stop_sequence": Value::Null, - "usage": { - "input_tokens": count(&usage, "prompt_tokens", 0), - "output_tokens": count(&usage, "completion_tokens", 0), - }, + "usage": anthropic_usage( + count(&usage, "prompt_tokens", 0), + count(&usage, "completion_tokens", 0), + cached_tokens(&usage), + ), }) } @@ -609,8 +682,7 @@ pub fn stream_chunk_to_anthropic(chunk: &Value, state: &mut StreamState) -> Stri ); } if let Some(usage) = chunk.get("usage").filter(|u| u.is_object()) { - state.input_tokens = count(usage, "prompt_tokens", state.input_tokens); - state.output_tokens = count(usage, "completion_tokens", state.output_tokens); + read_stream_usage(usage, state); } return out; } @@ -629,8 +701,7 @@ pub fn stream_chunk_to_anthropic(chunk: &Value, state: &mut StreamState) -> Stri } if let Some(usage) = chunk.get("usage").filter(|u| u.is_object()) { - state.input_tokens = count(usage, "prompt_tokens", state.input_tokens); - state.output_tokens = count(usage, "completion_tokens", state.output_tokens); + read_stream_usage(usage, state); } if let Some(d) = choice.get("delta") { @@ -732,10 +803,38 @@ pub fn stream_chunk_to_anthropic(chunk: &Value, state: &mut StreamState) -> Stri } } - // Reasoning tokens count toward the output estimate but never surface as - // their own block; Anthropic has no equivalent block type here. - if let Some(reasoning) = delta.get("reasoning_content").and_then(Value::as_str) { + // Reasoning streams as a thinking block, the way Anthropic streams its + // own. Blocks never overlap: a thinking delta closes an open text block, + // and text closes an open thinking block, each opening at the next index. + // The signature is empty; see response_to_anthropic. + if let Some(reasoning) = delta + .get("reasoning_content") + .and_then(Value::as_str) + .filter(|r| !r.is_empty()) + { state.output_chars += reasoning.len() as i32; + out += &close_text_block(state); + if !state.thinking_open { + state.thinking_open = true; + state.thinking_index = state.next_index; + state.next_index += 1; + out += &event( + "content_block_start", + &json!({ + "type": "content_block_start", + "index": state.thinking_index, + "content_block": {"type": "thinking", "thinking": "", "signature": ""}, + }), + ); + } + out += &event( + "content_block_delta", + &json!({ + "type": "content_block_delta", + "index": state.thinking_index, + "delta": {"type": "thinking_delta", "thinking": reasoning}, + }), + ); } let text = delta @@ -748,6 +847,7 @@ pub fn stream_chunk_to_anthropic(chunk: &Value, state: &mut StreamState) -> Stri } state.output_chars += text.len() as i32; + out += &close_thinking_block(state); if !state.block_open { state.block_open = true; state.text_index = state.next_index; @@ -810,14 +910,8 @@ pub fn stream_close_to_anthropic(state: &mut StreamState) -> String { } state.closed = true; - let mut out = String::new(); - if state.block_open { - state.block_open = false; - out += &event( - "content_block_stop", - &json!({"type": "content_block_stop", "index": state.text_index}), - ); - } + let mut out = close_thinking_block(state); + out += &close_text_block(state); let mut index = state.next_index; let mut emitted_tool_block = false; @@ -867,7 +961,7 @@ pub fn stream_close_to_anthropic(state: &mut StreamState) -> String { &json!({ "type": "message_delta", "delta": {"stop_reason": stop, "stop_sequence": Value::Null}, - "usage": {"input_tokens": input_final, "output_tokens": output_final}, + "usage": anthropic_usage(input_final, output_final, state.cached_tokens), }), ); out += &event("message_stop", &json!({"type": "message_stop"})); @@ -1108,4 +1202,128 @@ mod tests { assert!(out.contains("\"model\":\"claude-3-5-sonnet-20241022\"")); assert_eq!(state.model, "claude-3-5-sonnet-20241022"); } + + fn chunk(delta: Value, finish: Value) -> Value { + json!({"id": "c1", "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]}) + } + + fn streamed(chunks: &[Value]) -> String { + let mut state = StreamState::new(); + state.model = "m".to_string(); + let mut out = String::new(); + for c in chunks { + out += &stream_chunk_to_anthropic(c, &mut state); + } + out + &stream_close_to_anthropic(&mut state) + } + + /// The SSE stream as (event type, index, block or delta type) rows. + fn shape(sse: &str) -> Vec<(String, i64, String)> { + sse.lines() + .filter_map(|line| line.strip_prefix("data: ")) + .map(|data| serde_json::from_str::(data).unwrap()) + .map(|e| { + let kind = e["content_block"]["type"] + .as_str() + .or(e["delta"]["type"].as_str()) + .unwrap_or("") + .to_string(); + ( + e["type"].as_str().unwrap().to_string(), + e["index"].as_i64().unwrap_or(-1), + kind, + ) + }) + .collect() + } + + fn row(event: &str, index: i64, kind: &str) -> (String, i64, String) { + (event.to_string(), index, kind.to_string()) + } + + #[test] + fn reasoning_then_text_streams_a_thinking_block_then_a_text_block() { + let out = streamed(&[ + chunk(json!({"reasoning_content": "let me "}), Value::Null), + chunk(json!({"reasoning_content": "check"}), Value::Null), + chunk(json!({"content": "51"}), json!("stop")), + ]); + let rows = shape(&out); + assert_eq!( + rows[1..7], + [ + row("content_block_start", 0, "thinking"), + row("content_block_delta", 0, "thinking_delta"), + row("content_block_delta", 0, "thinking_delta"), + row("content_block_stop", 0, ""), + row("content_block_start", 1, "text"), + row("content_block_delta", 1, "text_delta"), + ] + ); + assert_eq!(rows[7], row("content_block_stop", 1, "")); + assert!(out.contains(r#""thinking":"let me ""#)); + } + + #[test] + fn reasoning_only_closes_the_thinking_block_and_opens_no_text_block() { + let out = streamed(&[ + chunk(json!({"reasoning_content": "hmm"}), Value::Null), + chunk(json!({}), json!("length")), + ]); + let rows = shape(&out); + assert!(rows.contains(&row("content_block_stop", 0, ""))); + assert!(!rows.iter().any(|r| r.2 == "text")); + assert!(out.contains(r#""stop_reason":"max_tokens""#)); + } + + #[test] + fn non_streaming_reasoning_becomes_a_thinking_block_before_the_text() { + let reply = response_to_anthropic( + &json!({"choices": [{"message": {"reasoning_content": "think", "content": "51"}, "finish_reason": "stop"}]}), + "m", + ); + assert_eq!( + reply["content"], + json!([ + {"type": "thinking", "thinking": "think", "signature": ""}, + {"type": "text", "text": "51"}, + ]) + ); + assert_eq!(reply["stop_reason"], json!("end_turn")); + } + + #[test] + fn cached_tokens_move_from_input_to_cache_read() { + let out = streamed(&[ + chunk(json!({"content": "hi"}), json!("stop")), + json!({"id": "c1", "choices": [], "usage": {"prompt_tokens": 3779, "completion_tokens": 9, "prompt_tokens_details": {"cached_tokens": 3776}}}), + ]); + assert!(out.contains( + r#""usage":{"cache_read_input_tokens":3776,"input_tokens":3,"output_tokens":9}"# + )); + } + + #[test] + fn absent_cached_tokens_leave_usage_as_it_was() { + let reply = response_to_anthropic( + &json!({"choices": [{"message": {"content": "hi"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 10, "completion_tokens": 2}}), + "m", + ); + assert_eq!( + reply["usage"], + json!({"input_tokens": 10, "output_tokens": 2}) + ); + } + + #[test] + fn zero_cached_tokens_report_a_zero_cache_read() { + let reply = response_to_anthropic( + &json!({"choices": [{"message": {"content": "hi"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 10, "completion_tokens": 2, "prompt_tokens_details": {"cached_tokens": 0}}}), + "m", + ); + assert_eq!( + reply["usage"], + json!({"input_tokens": 10, "output_tokens": 2, "cache_read_input_tokens": 0}) + ); + } } diff --git a/src/harness/agents.rs b/src/harness/agents.rs index 238be0be..a282995e 100644 --- a/src/harness/agents.rs +++ b/src/harness/agents.rs @@ -766,7 +766,7 @@ pub fn build_open_claw_config( entry["cost"] = json!({ "input": model.input_per_mtok as f64 * PER_MICRO, "output": model.output_per_mtok as f64 * PER_MICRO, - "cacheRead": 0, + "cacheRead": model.cached_input_per_mtok as f64 * PER_MICRO, "cacheWrite": 0, }); } @@ -878,7 +878,7 @@ pub fn build_prime_agent_extension( entry["cost"] = json!({ "input": model.input_per_mtok as f64 * PER_MICRO, "output": model.output_per_mtok as f64 * PER_MICRO, - "cacheRead": 0, + "cacheRead": model.cached_input_per_mtok as f64 * PER_MICRO, "cacheWrite": 0, }); } @@ -1693,6 +1693,7 @@ mod tests { max_output: 16_384, input_per_mtok: 0, output_per_mtok: 0, + cached_input_per_mtok: 0, }]; let built = build_prime_agent_extension("https://example.test/v1", KEY_VARIABLE, &models); let json = built diff --git a/src/harness/catalog_models.rs b/src/harness/catalog_models.rs index 2f725d01..47be29ef 100644 --- a/src/harness/catalog_models.rs +++ b/src/harness/catalog_models.rs @@ -19,6 +19,8 @@ pub struct CatalogModel { pub max_output: i64, pub input_per_mtok: i64, pub output_per_mtok: i64, + /// What a prompt token read from the server's cache costs. + pub cached_input_per_mtok: i64, } /// Moves the entry whose id is `primary` to the front, or inserts a bare one @@ -53,25 +55,26 @@ pub fn catalog_models_with( let (_, models, _) = console.fetch_models(console_url, access_token); let (_, prices, _) = console.fetch_catalog(console_url, access_token); - let price_for = |id: &str| -> (i64, i64) { + let price_for = |id: &str| -> (i64, i64, i64) { prices .iter() .find(|p| p.id == id) - .map(|p| (p.input_per_mtok, p.output_per_mtok)) - .unwrap_or((0, 0)) + .map(|p| (p.input_per_mtok, p.output_per_mtok, p.cached_input_per_mtok)) + .unwrap_or((0, 0, 0)) }; for info in &models { if info.id.is_empty() { continue; } - let (in_price, out_price) = price_for(&info.id); + let (in_price, out_price, cached_price) = price_for(&info.id); out.push(CatalogModel { id: info.id.clone(), context_window: info.context_window, max_output: info.max_output_tokens, input_per_mtok: in_price, output_per_mtok: out_price, + cached_input_per_mtok: cached_price, }); } primary_first(&mut out, primary); @@ -108,6 +111,7 @@ pub fn catalog_models_for(endpoint: &Endpoint, primary: &str) -> Vec 0 { + entry["cost"]["cache_read"] = + json!(model.cached_input_per_mtok as f64 / 1_000_000.0); + } } entries[model.id.as_str()] = entry; } diff --git a/tests/cpp_case_exceptions.txt b/tests/cpp_case_exceptions.txt index 81e3fc72..489e3b0d 100644 --- a/tests/cpp_case_exceptions.txt +++ b/tests/cpp_case_exceptions.txt @@ -1,3 +1,4 @@ # C++ test cases deliberately not ported as a same-named Rust test, one per line: # — test_wally_opencode.cpp config_declares_the_harness — ported as opencode_config_declares_the_harness in tests/test_wally_opencode.rs +test_wally_unit.cpp reasoning_content_counts_without_an_unsigned_block — replaced by reasoning_content_streams_as_thinking_and_counts_toward_output in tests/test_wally_unit/shim.rs (#99: reasoning now streams as a thinking block) diff --git a/tests/test_wally_harness.rs b/tests/test_wally_harness.rs index 5fd634d4..a83197df 100644 --- a/tests/test_wally_harness.rs +++ b/tests/test_wally_harness.rs @@ -344,6 +344,7 @@ fn openclaw_config_selects_our_provider_and_model() { max_output: 8192, input_per_mtok: 300000, output_per_mtok: 1200000, + cached_input_per_mtok: 75000, }]; let config: Value = serde_json::from_str(&harness::build_open_claw_config( "", @@ -380,6 +381,7 @@ fn openclaw_config_selects_our_provider_and_model() { assert_eq!(models[0]["maxTokens"], json!(8192)); assert_eq!(models[0]["cost"]["input"], json!(0.3)); assert_eq!(models[0]["cost"]["output"], json!(1.2)); + assert_eq!(models[0]["cost"]["cacheRead"], json!(0.075)); } // A local server needs no credential and is handed none by Resolve, but an @@ -1155,6 +1157,7 @@ fn declare_catalog() -> Vec { max_output: 0, input_per_mtok: 0, output_per_mtok: 0, + cached_input_per_mtok: 0, }] } diff --git a/tests/test_wally_opencode.rs b/tests/test_wally_opencode.rs index 5a21194c..ab4f7398 100644 --- a/tests/test_wally_opencode.rs +++ b/tests/test_wally_opencode.rs @@ -233,6 +233,7 @@ fn config_injects_limit_and_cost() { max_output: 0, input_per_mtok: 600000, output_per_mtok: 2200000, + cached_input_per_mtok: 0, }], )) .expect("parse config"); @@ -304,6 +305,7 @@ fn opencode_config_declares_the_harness() { max_output: 0, input_per_mtok: 0, output_per_mtok: 0, + cached_input_per_mtok: 0, }]; let configs = [ harness::build_open_code_config("glm-5.3-flash", "http://127.0.0.1:52431/v1", "", &catalog), @@ -322,3 +324,26 @@ fn opencode_config_declares_the_harness() { ); } } + +#[test] +fn config_prices_cache_reads_only_when_the_catalog_has_a_cached_price() { + let config = |cached: i64| -> Value { + let j: Value = serde_json::from_str(&harness::build_open_code_cloud_config( + "glm-5.3-flash", + "https://x/v1", + "tok", + &[CatalogModel { + id: "glm-5.3-flash".to_string(), + context_window: 1048576, + max_output: 0, + input_per_mtok: 600000, + output_per_mtok: 2200000, + cached_input_per_mtok: cached, + }], + )) + .expect("parse config"); + j["provider"]["runanywhere"]["models"]["glm-5.3-flash"]["cost"].clone() + }; + assert_eq!(config(110000)["cache_read"], json!(0.11)); + assert!(config(0).get("cache_read").is_none()); +} diff --git a/tests/test_wally_unit/shim.rs b/tests/test_wally_unit/shim.rs index 36b66ab3..776fc2d7 100644 --- a/tests/test_wally_unit/shim.rs +++ b/tests/test_wally_unit/shim.rs @@ -195,13 +195,14 @@ fn stream_usage_falls_back_when_endpoint_never_reports_it() { } #[test] -fn reasoning_content_counts_without_an_unsigned_block() { +fn reasoning_content_streams_as_thinking_and_counts_toward_output() { // Some endpoints stream their thinking as `delta.reasoning_content`, // separate from `delta.content`, and on a tight max_tokens budget it can - // be the ONLY thing the model emits. Those characters must count toward - // the fallback estimate so the turn is not mistaken for a free one -- but - // they must NOT go out as a `thinking` block, which Anthropic's stream - // requires a signature_delta to close and an OpenAI endpoint cannot sign. + // be the ONLY thing the model emits. It goes out as a thinking block with + // an empty signature: our upstreams cannot sign reasoning, and Claude Code + // accepts the unsigned block (checked live against Claude Code 2.1.286 + // over two turns, 1 Oct 2026). The characters still count toward the + // fallback estimate so the turn is not mistaken for a free one. let mut state = tr::StreamState::new(); let thinking_chunk: serde_json::Value = serde_json::from_str( r#"{"id":"c1","choices":[{"delta":{"reasoning_content":"counting to five"}}]}"#, @@ -214,22 +215,17 @@ fn reasoning_content_counts_without_an_unsigned_block() { tr::stream_chunk_to_anthropic(&finish_chunk, &mut state); let closing = tr::stream_close_to_anthropic(&mut state); - // No unsigned thinking block on the wire, in either half of the stream. assert!( - !opening.contains("\"type\":\"thinking\"") && !opening.contains("thinking_delta"), - "reasoning must not surface as a thinking block; got: {opening}" + opening.contains("\"type\":\"thinking\"") && opening.contains("thinking_delta"), + "reasoning should open a thinking block; got: {opening}" ); assert!( - !closing.contains("\"type\":\"thinking\""), - "reasoning must not surface as a thinking block; got: {closing}" + closing.contains("\"type\":\"content_block_stop\""), + "the thinking block should be closed; got: {closing}" ); // "counting to five" is 17 characters -> a 4-token fallback estimate. // The real point: not 0. A turn that spent its whole budget thinking // must not report as though nothing happened. - assert!( - !closing.contains("\"output_tokens\":0"), - "thinking-only turn still reports 0 output_tokens; got: {closing}" - ); assert!( closing.contains("\"output_tokens\":4"), "expected the 4-token character estimate for 17 characters; got: {closing}"