diff --git a/docs/modules/harness/tool.md b/docs/modules/harness/tool.md index e24549e..6967a4f 100644 --- a/docs/modules/harness/tool.md +++ b/docs/modules/harness/tool.md @@ -254,6 +254,29 @@ assert!(run.messages.iter().any(|m| m.text().contains("unknown tool `missing`")) // A single UnknownToolCall event was recorded with recovery == "tool_error". ``` +## Invalid tool-argument recovery + +Two distinct failures can affect a provider-supplied call's arguments, and they +are handled separately: + +- **Schema-invalid** (well-formed JSON that violates the tool's input schema) is + governed by `RunPolicy::invalid_args: InvalidArgsPolicy`. `Fail` (default, + historical) aborts the turn; `ReturnToolError` injects a repairable tool-error + message (carrying the validation detail and the expected schema) and continues. +- **Unparseable** (malformed JSON the provider could not parse into arguments at + all) is surfaced by the provider as a `ToolCall` with `invalid: Some(reason)` + and the raw string preserved in `arguments`. Small local models (Ollama, LM + Studio, llama.cpp, vLLM) emit this occasionally. The agent loop **always** + recovers here — independent of `InvalidArgsPolicy`, since an unparseable + payload is a transport-level defect, not a schema violation — by injecting the + parse `reason` back to the model as an error tool result so it can retry. The + recovery emits `AgentEvent::InvalidToolArgs { call_id, tool_name, arguments, + error, recovery: "tool_error" }` and consumes one tool-call budget slot, so + `RunLimits::max_tool_calls` bounds the retry loop. Because the call always + resolves, a malformed argument blob can never become a never-resolving tool + call that stalls the loop. See the OpenAI provider README for how the wire + parser produces these invalid calls. + ## Tool policy enforcement Beyond the model-visible `ToolSchema`, each tool advertises a structured, diff --git a/examples/local_model_probe.rs b/examples/local_model_probe.rs index 48efd7a..84944bc 100644 --- a/examples/local_model_probe.rs +++ b/examples/local_model_probe.rs @@ -79,7 +79,6 @@ fn first_call(response: &ModelResponse) -> Option<&ToolCall> { response.message.tool_calls.first() } - fn msg_text(message: &AssistantMessage) -> String { message .content @@ -123,7 +122,10 @@ async fn tool_call(model: &OpenAiModel) -> Outcome { format!( "no tool_calls; finish={:?} text={:?}", resp.finish_reason, - msg_text(&resp.message).chars().take(120).collect::() + msg_text(&resp.message) + .chars() + .take(120) + .collect::() ), ), }, @@ -240,10 +242,7 @@ async fn parallel_tools(model: &OpenAiModel) -> Outcome { .iter() .map(|c| c.id.as_str()) .collect(); - let unique = ids - .iter() - .collect::>() - .len(); + let unique = ids.iter().collect::>().len(); if n >= 2 && unique == n { ok("parallel-tools", format!("{n} calls, ids={ids:?}")) } else if n >= 2 { @@ -251,7 +250,10 @@ async fn parallel_tools(model: &OpenAiModel) -> Outcome { } else { // Small models often serialize calls across turns; only flag // this as informational, not a harness bug. - ok("parallel-tools", format!("model made {n} call(s) (model behavior)")) + ok( + "parallel-tools", + format!("model made {n} call(s) (model behavior)"), + ) } } Err(e) => fail("parallel-tools", e.to_string()), @@ -313,16 +315,16 @@ async fn streaming_tools(model: &OpenAiModel) -> Outcome { "streaming-tools", format!("id={:?} args={}", call.id, call.arguments), ), - Some(call) => fail( - "streaming-tools", - format!("bad args: {}", call.arguments), - ), + Some(call) => fail("streaming-tools", format!("bad args: {}", call.arguments)), None => fail( "streaming-tools", format!( "no tool call; finish={:?} text={:?}", resp.finish_reason, - msg_text(&resp.message).chars().take(120).collect::() + msg_text(&resp.message) + .chars() + .take(120) + .collect::() ), ), }, @@ -345,7 +347,9 @@ async fn json_object(model: &OpenAiModel) -> Outcome { Ok(resp) => { let text = msg_text(&resp.message); match serde_json::from_str::(text.trim()) { - Ok(v) if v.is_object() => ok("json-object", text.chars().take(80).collect::()), + Ok(v) if v.is_object() => { + ok("json-object", text.chars().take(80).collect::()) + } _ => fail("json-object", format!("not a JSON object: {text:?}")), } } @@ -395,7 +399,10 @@ async fn thinking_leak(model: &OpenAiModel) -> Outcome { if text.contains("") || text.contains("") { fail( "thinking-leak", - format!(" leaked into text: {:?}", text.chars().take(120).collect::()), + format!( + " leaked into text: {:?}", + text.chars().take(120).collect::() + ), ) } else { ok("thinking-leak", text.chars().take(60).collect::()) diff --git a/src/graph/goals/test.rs b/src/graph/goals/test.rs index ab94639..aa4c17b 100644 --- a/src/graph/goals/test.rs +++ b/src/graph/goals/test.rs @@ -284,6 +284,7 @@ mod tool_tests { id: id.to_string(), name: name.to_string(), arguments: args, + invalid: None, } } diff --git a/src/graph/todos/test.rs b/src/graph/todos/test.rs index d3cdbf4..6c6c7d7 100644 --- a/src/graph/todos/test.rs +++ b/src/graph/todos/test.rs @@ -386,6 +386,7 @@ mod tool_tests { id: "c1".to_string(), name: "todo".to_string(), arguments: args, + invalid: None, } } diff --git a/src/harness/agent_loop/README.md b/src/harness/agent_loop/README.md index 02a81fd..53e567b 100644 --- a/src/harness/agent_loop/README.md +++ b/src/harness/agent_loop/README.md @@ -73,6 +73,27 @@ deterministic) and enabled per policy via `retry::RetryPolicy::with_backoff_sleep`. A real provider integration retries after a genuine, growing delay while unit tests stay sleep-free. +## Truncated-empty recovery + +Local reasoning models (for example `qwen3` via Ollama) intermittently spend +their entire token budget on the hidden reasoning channel and return +`finish_reason == "length"` with no visible text, no tool calls, and no +structured output — a result useless to every caller. Before finalizing such a +turn (and before structured extraction, which would otherwise fail on the empty +completion) the loop retries the model call up to +`runtime::RunPolicy::truncated_empty_retries` times (default `1`, so two +attempts total). Each retry drops the useless assistant row, doubles the +request's `max_tokens` when one was set — clamped at 4x the original cap, and a +deliberate override of the per-turn output cap that caused the truncation — or +re-issues unchanged when no budget was set (the failure is stochastic, so a +plain retry still helps). Each attempt counts as a model call and emits +`AgentEvent::RetryScheduled`. The retry runs *before* +`RunPolicy::error_on_empty_response`; only once the retries are exhausted does +that guard (if enabled) turn the still-blank final into +`TinyAgentsError::EmptyResponse`. Set `truncated_empty_retries` to `0` to +restore exact-replay behavior. The recovery lives in the shared `run_loop`, so +it applies identically to the unary and streaming paths. + ## Public surface - `AgentHarness::invoke(state, ctx_data, config, input) -> Result` — diff --git a/src/harness/agent_loop/run_loop.rs b/src/harness/agent_loop/run_loop.rs index e97b480..85ceb4a 100644 --- a/src/harness/agent_loop/run_loop.rs +++ b/src/harness/agent_loop/run_loop.rs @@ -47,6 +47,15 @@ impl AgentHarness { status.mark_running(HarnessPhase::Middleware); self.middleware.run_before_agent(ctx, state).await?; + // Truncated-empty recovery state (see `RunPolicy::truncated_empty_retries`). + // These persist across the retry `continue` within a single logical turn: + // `boosted_max_tokens` overrides the next request's cap, `truncation_base` + // records the original cap so growth stays clamped at 4x, and the counter + // bounds how many times we re-issue the call. + let mut truncated_empty_retries_used: u32 = 0; + let mut boosted_max_tokens: Option = None; + let mut truncation_base: Option = None; + loop { // Safe cancellation checkpoint: if an orchestrator requested // cooperative cancellation, stop before doing any further work @@ -97,6 +106,15 @@ impl AgentHarness { request.max_tokens = Some(request.max_tokens.map_or(cap, |current| current.min(cap))); } + // Truncated-empty recovery: a prior attempt this turn exhausted its + // token budget on the (hidden) reasoning channel and returned no + // usable content, so re-issue the call with a larger cap. The boost + // deliberately wins over the per-turn cap above — that cap is what + // truncated the response — and was already clamped to 4x the + // original budget when it was computed below. + if let Some(boost) = boosted_max_tokens { + request.max_tokens = Some(boost); + } status.mark_running(HarnessPhase::Middleware); self.middleware @@ -197,6 +215,10 @@ impl AgentHarness { .capture .model_io .then(|| serde_json::to_value(&request.messages).unwrap_or(Value::Null)); + // Snapshot the effective token cap before `request` moves into the + // model-wrap onion, so truncated-empty recovery can compute the next + // (doubled) budget from what was actually sent. + let attempt_max_tokens = request.max_tokens; let mut response = self .middleware .run_wrapped_model(ctx, state, request, &base) @@ -225,7 +247,7 @@ impl AgentHarness { .model_io .then(|| serde_json::to_value(&response.message).unwrap_or(Value::Null)); let record = ctx.emit(AgentEvent::ModelCompleted { - call_id, + call_id: call_id.clone(), started_at_ms: Some(model_started_at_ms), usage: response.usage, input: captured_input, @@ -272,6 +294,45 @@ impl AgentHarness { ); if tool_calls.is_empty() || structured_tool_hit { + // Truncated-empty recovery (runs before structured extraction, + // which would otherwise fail on the empty completion). A local + // reasoning model can burn the whole token budget on its hidden + // reasoning channel and return `finish_reason == "length"` with + // no visible text, no tool calls, and no structured output — a + // result useless to every caller. Retry the call (bumping the + // token budget when one was set) instead of surfacing the blank. + // A structured tool hit carries a real payload, so it is never + // treated as truncated-empty. + let truncated_empty = tool_calls.is_empty() + && response.finish_reason.as_deref() == Some("length") + && response.text().trim().is_empty(); + if truncated_empty + && truncated_empty_retries_used < self.policy.truncated_empty_retries + { + // Drop the useless empty assistant row appended above so the + // retry re-sends the identical transcript. + messages.pop(); + truncated_empty_retries_used += 1; + // Grow the token budget when the request set one: double it, + // clamped at 4x the original cap. An unset budget stays unset + // (a plain retry is still worthwhile — the failure is + // stochastic). + if let Some(sent) = attempt_max_tokens { + let base = *truncation_base.get_or_insert(sent); + let next = boosted_max_tokens + .unwrap_or(sent) + .saturating_mul(2) + .min(base.saturating_mul(4)); + boosted_max_tokens = Some(next); + } + let record = ctx.emit(AgentEvent::RetryScheduled { + call_id: call_id.clone(), + attempt: truncated_empty_retries_used as usize, + }); + status.set_last_event(record.id); + continue; + } + // Final response: optionally extract structured output using the // resolved plan (provider-native schema or tool-call arguments). if let Some((strategy, name, schema)) = &structured_plan { diff --git a/src/harness/agent_loop/test.rs b/src/harness/agent_loop/test.rs index b250735..eab6fdd 100644 --- a/src/harness/agent_loop/test.rs +++ b/src/harness/agent_loop/test.rs @@ -149,6 +149,29 @@ fn tool_call_response(id: &str, name: &str, arguments: serde_json::Value) -> Mod } } +/// Builds a tool-call response whose arguments the provider could not parse, +/// mirroring what the OpenAI provider produces for a small local model that +/// emitted malformed argument JSON: the raw string is preserved and the call is +/// marked [`ToolCall::invalid`]. +fn invalid_tool_call_response(id: &str, name: &str, raw: &str) -> ModelResponse { + let reason = format!( + "openai response contained invalid JSON arguments for tool call `{id}` (`{name}`): \ + EOF while parsing a value; raw arguments: {raw:?}" + ); + ModelResponse { + message: AssistantMessage { + id: Some(format!("msg-{id}")), + content: Vec::new(), + tool_calls: vec![ToolCall::invalid(id, name, raw, reason)], + usage: Some(Usage::new(7, 3)), + }, + usage: Some(Usage::new(7, 3)), + finish_reason: Some("tool_calls".to_string()), + raw: None, + resolved_model: None, + } +} + /// Builds a plain-text assistant response with explicit usage. fn text_response(text: &str, input: u64, output: u64) -> ModelResponse { ModelResponse { @@ -165,6 +188,26 @@ fn text_response(text: &str, input: u64, output: u64) -> ModelResponse { } } +/// Builds a *truncated empty* completion: `finish_reason == "length"` with no +/// text, no tool calls, and no structured output — the failure mode of a local +/// reasoning model that burned its whole token budget on the hidden reasoning +/// channel. `reasoning_tokens` is folded into `output_tokens` so the usage +/// mirrors a real length-truncated response. +fn truncated_empty_response(reasoning_tokens: u64) -> ModelResponse { + ModelResponse { + message: AssistantMessage { + id: None, + content: Vec::new(), + tool_calls: Vec::new(), + usage: Some(Usage::new(4, reasoning_tokens)), + }, + usage: Some(Usage::new(4, reasoning_tokens)), + finish_reason: Some("length".to_string()), + raw: None, + resolved_model: None, + } +} + /// Middleware that appends a user message to every model request. struct InjectMiddleware { text: &'static str, @@ -459,6 +502,166 @@ async fn empty_response_terminates_normally_when_guard_disabled() { assert_eq!(run.text(), Some(String::new())); } +#[tokio::test] +async fn truncated_empty_response_retries_then_succeeds() { + // A local reasoning model burns its whole token budget on the hidden + // reasoning channel and returns finish_reason="length" with empty content. + // The loop must recover automatically: retry the call with a doubled token + // budget and finish on the second, usable response. + let model = Arc::new(crate::harness::testkit::ScriptedModel::new(vec![ + truncated_empty_response(2048), + text_response("recovered", 4, 3), + ])); + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model("mock", Arc::clone(&model) as _); + + use crate::harness::testkit::EventRecorder; + let recorder = EventRecorder::new(); + let ctx = RunContext::new( + RunConfig::new("truncated-retry").with_max_turn_output_tokens(2048), + (), + ) + .with_events(recorder.sink()); + let run = harness + .invoke_in_context(&(), ctx, vec![Message::user("hi")]) + .await + .expect("the truncated-empty response should be retried, not surfaced"); + + assert_eq!(run.text(), Some("recovered".to_string())); + assert_eq!( + run.model_calls, 2, + "the retry counts as a second model call" + ); + assert!( + recorder + .events() + .iter() + .any(|e| matches!(e, AgentEvent::RetryScheduled { attempt: 1, .. })), + "the retry should be observable; got kinds {:?}", + recorder.kinds() + ); + // The first attempt sent the configured 2048 cap; the retry doubled it. + let sent: Vec> = model.requests().iter().map(|r| r.max_tokens).collect(); + assert_eq!( + sent, + vec![Some(2048), Some(4096)], + "the retried request should carry double the original token budget" + ); +} + +#[tokio::test] +async fn truncated_empty_retry_budget_stays_unset_when_request_had_none() { + // With no per-turn token cap the budget cannot be doubled, but the retry is + // still worthwhile because the failure is stochastic. The retried request + // simply carries `None` again. + let model = Arc::new(crate::harness::testkit::ScriptedModel::new(vec![ + truncated_empty_response(64), + text_response("recovered", 4, 3), + ])); + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model("mock", Arc::clone(&model) as _); + + let run = harness + .invoke_default(&(), vec![Message::user("hi")]) + .await + .expect("a plain retry still recovers a stochastic truncation"); + + assert_eq!(run.text(), Some("recovered".to_string())); + assert_eq!(run.model_calls, 2); + let sent: Vec> = model.requests().iter().map(|r| r.max_tokens).collect(); + assert_eq!(sent, vec![None, None]); +} + +#[tokio::test] +async fn truncated_empty_retries_exhausted_returns_blank_by_default() { + // When every attempt truncates, the retry budget is exhausted and the + // historical behavior applies: a blank final success (the empty-response + // guard is off by default). + let model = Arc::new(crate::harness::testkit::ScriptedModel::new(vec![ + truncated_empty_response(2048), + truncated_empty_response(4096), + ])); + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model("mock", Arc::clone(&model) as _); + + let ctx = RunContext::new( + RunConfig::new("truncated-exhausted").with_max_turn_output_tokens(2048), + (), + ); + let run = harness + .invoke_in_context(&(), ctx, vec![Message::user("hi")]) + .await + .expect("exhausted retries fall back to the blank-final behavior"); + + assert_eq!(run.text(), Some(String::new())); + assert_eq!(run.model_calls, 2, "one original attempt plus one retry"); +} + +#[tokio::test] +async fn truncated_empty_retries_exhausted_errors_when_guard_enabled() { + // With the empty-response guard on, exhausting the retries surfaces the + // typed EmptyResponse error rather than a blank success. + let model = Arc::new(crate::harness::testkit::ScriptedModel::new(vec![ + truncated_empty_response(2048), + truncated_empty_response(4096), + ])); + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model("mock", Arc::clone(&model) as _); + harness.with_policy(RunPolicy { + error_on_empty_response: true, + ..RunPolicy::default() + }); + + let err = harness + .invoke_default(&(), vec![Message::user("hi")]) + .await + .expect_err("exhausted retries with the guard on should fail"); + assert!( + matches!(err, TinyAgentsError::EmptyResponse), + "expected EmptyResponse, got {err:?}" + ); +} + +#[tokio::test] +async fn truncated_empty_retry_disabled_by_zero_policy() { + // truncated_empty_retries=0 restores exact-replay behavior: no retry, a + // single model call, blank final. + let model = Arc::new(crate::harness::testkit::ScriptedModel::new(vec![ + truncated_empty_response(2048), + ])); + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model("mock", Arc::clone(&model) as _); + harness.with_policy(RunPolicy { + truncated_empty_retries: 0, + ..RunPolicy::default() + }); + + let run = harness + .invoke_default(&(), vec![Message::user("hi")]) + .await + .expect("with retries disabled the blank final is returned"); + assert_eq!(run.model_calls, 1); + assert_eq!(run.text(), Some(String::new())); +} + +#[tokio::test] +async fn length_finish_with_text_is_not_treated_as_truncated_empty() { + // A length-truncated response that still carries visible text is a real + // answer and must not trigger the retry. + let mut partial = text_response("partial answer", 4, 2048); + partial.finish_reason = Some("length".to_string()); + let model = Arc::new(crate::harness::testkit::ScriptedModel::new(vec![partial])); + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model("mock", Arc::clone(&model) as _); + + let run = harness + .invoke_default(&(), vec![Message::user("hi")]) + .await + .expect("a length response with text is a normal final"); + assert_eq!(run.model_calls, 1); + assert_eq!(run.text(), Some("partial answer".to_string())); +} + #[tokio::test] async fn model_requests_tool_then_finishes() { let mut harness: AgentHarness<()> = AgentHarness::new(); @@ -761,6 +964,64 @@ async fn invalid_tool_arguments_return_tool_error_recovers() { ); } +#[tokio::test] +async fn malformed_tool_arguments_recover_as_error_tool_result() { + // A small local model emitted arguments the provider could not parse, so the + // response carries a `ToolCall::invalid` call. The loop must feed the parse + // error back to the model as a tool result and continue — *without* running + // the tool and *without* failing the run — so it can retry. This holds under + // the default `InvalidArgsPolicy::Fail` (which governs schema validation of + // well-formed args, not unparseable ones), proving the leniency is + // unconditional. + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model( + "mock", + Arc::new(MockModel::with_responses(vec![ + invalid_tool_call_response("call-x", "strict_lookup", "{\"query\":"), + text_response("recovered", 1, 1), + ])), + ); + let calls = Arc::new(Mutex::new(0)); + harness.register_tool(Arc::new(StrictLookupTool { + calls: Arc::clone(&calls), + })); + + use crate::harness::testkit::EventRecorder; + let recorder = EventRecorder::new(); + let ctx = + RunContext::new(RunConfig::new("malformed-args-run"), ()).with_events(recorder.sink()); + let run = harness + .invoke_in_context(&(), ctx, vec![Message::user("lookup")]) + .await + .expect("malformed args recover without failing the run"); + + assert_eq!(run.final_response.unwrap().text(), "recovered"); + assert_eq!( + *calls.lock().unwrap(), + 0, + "the tool implementation must not run on unparseable arguments" + ); + // The injected tool-error message carries the parse detail so the model can + // self-correct on the next turn. + let injected = run + .messages + .iter() + .any(|m| format!("{m:?}").contains("invalid JSON arguments")); + assert!( + injected, + "an error tool result should be injected into the transcript" + ); + // The recovery is surfaced as an `InvalidToolArgs` event. + assert!( + recorder + .events() + .iter() + .any(|e| matches!(e, AgentEvent::InvalidToolArgs { .. })), + "an InvalidToolArgs event should be emitted; got kinds {:?}", + recorder.kinds() + ); +} + #[tokio::test] async fn structured_output_is_extracted() { let mut harness: AgentHarness<()> = AgentHarness::new(); diff --git a/src/harness/agent_loop/tools.rs b/src/harness/agent_loop/tools.rs index c09aa15..0f4b419 100644 --- a/src/harness/agent_loop/tools.rs +++ b/src/harness/agent_loop/tools.rs @@ -146,6 +146,30 @@ impl AgentHarness { self.middleware.run_before_tool(ctx, state, call).await?; + // The provider marked this call's arguments unparseable (a small local + // model emitted malformed JSON). Rather than fail the run, inject a + // tool-error result carrying the parse detail and the raw arguments so + // the model can retry with corrected JSON — mirroring the schema-invalid + // recovery below and how mature frameworks (LangChain `invalid_tool_calls`, + // the AI SDK's invalid dynamic tool parts) surface the failure to the + // model. This consumed one tool-call budget slot above, bounding the loop, + // and always resolves the call so a stalled/never-resolving tool cannot + // hang the loop. Applied unconditionally (not gated by `InvalidArgsPolicy`, + // which governs *schema* validation of well-formed args): a call the + // provider could not even parse is a transport-level defect. + if let Some(detail) = call.invalid.clone() { + let call_id = CallId::new(call.id.clone()); + let record = ctx.emit(AgentEvent::InvalidToolArgs { + call_id, + tool_name: call.name.clone(), + arguments: call.arguments.clone(), + error: detail.clone(), + recovery: "tool_error".to_string(), + }); + status.set_last_event(record.id); + return Ok(ResolvedToolCall::ErrorMessage(detail)); + } + let tool = match self.tools.get(&call.name) { Some(tool) => tool, None => { diff --git a/src/harness/middleware/library/test.rs b/src/harness/middleware/library/test.rs index 03294da..3daee75 100644 --- a/src/harness/middleware/library/test.rs +++ b/src/harness/middleware/library/test.rs @@ -42,6 +42,7 @@ fn tool_call(name: &str) -> ToolCall { id: "call-1".to_string(), name: name.to_string(), arguments: json!({}), + invalid: None, } } @@ -1300,6 +1301,7 @@ async fn redaction_scrubs_tool_call_arguments_and_raw_payloads() { id: "c1".to_string(), name: "http".to_string(), arguments: json!({"header": "Bearer sk-secret", "nested": ["sk-secret"]}), + invalid: None, }); response.raw = Some(json!({"choices": [{"text": "token sk-secret"}]})); stack @@ -1320,6 +1322,7 @@ async fn redaction_scrubs_tool_call_arguments_and_raw_payloads() { id: "c2".to_string(), name: "http".to_string(), arguments: json!({"key": "sk-secret"}), + invalid: None, }; stack .run_before_tool(&mut ctx, &(), &mut call) diff --git a/src/harness/middleware/test.rs b/src/harness/middleware/test.rs index 0610c01..0363d3d 100644 --- a/src/harness/middleware/test.rs +++ b/src/harness/middleware/test.rs @@ -1246,6 +1246,7 @@ fn tool_call() -> ToolCall { id: "call-1".to_string(), name: "fake".to_string(), arguments: serde_json::Value::Null, + invalid: None, } } diff --git a/src/harness/model/mod.rs b/src/harness/model/mod.rs index cd5e314..6e5f58a 100644 --- a/src/harness/model/mod.rs +++ b/src/harness/model/mod.rs @@ -836,6 +836,7 @@ impl StreamAccumulator { name: name.unwrap_or_default(), arguments: serde_json::from_str(&args).unwrap_or(Value::Null), id, + invalid: None, }) .collect(); let message = AssistantMessage { diff --git a/src/harness/providers/mock.rs b/src/harness/providers/mock.rs index 24f5873..f781662 100644 --- a/src/harness/providers/mock.rs +++ b/src/harness/providers/mock.rs @@ -212,6 +212,7 @@ impl ChatModel for MockModel { id: format!("mock-tool-{call_id}"), name: name.clone(), arguments: arguments.clone(), + invalid: None, }; let usage = Usage::new(input_tokens, 5); let message = AssistantMessage { diff --git a/src/harness/providers/openai/README.md b/src/harness/providers/openai/README.md index a0e711b..408def0 100644 --- a/src/harness/providers/openai/README.md +++ b/src/harness/providers/openai/README.md @@ -64,11 +64,119 @@ Streaming responses are decoded by a small state machine (`SseState` / two HTTP chunks is never corrupted into replacement characters. - Index-less streamed tool-call fragments (some providers omit the tool-call index on continuation chunks) are correlated by id rather than position. +- Parallel tool-call fragments that all arrive with `index: 0` (Ollama's `/v1` + bug, ollama/ollama#15457) are kept distinct: when an explicit index carries a + non-empty id that conflicts with the id already recorded at that slot, a fresh + slot is opened instead of merging two calls onto one. +- A later fragment with an empty `function.name` never overwrites a name already + recorded for the slot (LM Studio, lmstudio-bug-tracker#649). - A trailing partial line without a final newline (providers that terminate the last SSE event without a trailing newline) is still flushed. - Mid-stream error payloads (`data: {"error": ...}`) surface as a stream error instead of being silently swallowed. +## Local-server compatibility + +Local OpenAI-compatible runtimes (LM Studio, llama.cpp server, and others) +implement a stricter subset of the Chat Completions request surface than hosted +OpenAI and reject two request shapes with an HTTP 400: + +- **Named `tool_choice`** — `{"type":"function","function":{"name":...}}` is + refused (`Invalid tool_choice type: 'object'`); only `none`/`auto`/`required` + are accepted. +- **`response_format: {"type":"json_object"}`** — refused + (`'response_format.type' must be 'json_schema' or 'text'`). + +The provider degrades both without dropping semantics: + +- A named tool choice becomes `tool_choice: "required"` **and** the wire `tools` + array is filtered down to just the named tool, so the model still has exactly + one tool to call. If the named tool is not in the list, `tools` is left intact + and `"required"` is still sent. +- A `json_object` response format becomes a permissive `json_schema` + (`{"name":"json_object","schema":{"type":"object"},"strict":false}`), which + still guarantees a JSON object. + +Two knobs control this per instance (default `true` = the hosted-OpenAI shapes): + +- `.with_named_tool_choice(false)` — always degrade named tool choice. +- `.with_json_object_format(false)` — always degrade `json_object`. + +**Zero-config auto-degrade:** even with the knobs left on, a first attempt that +fails with an HTTP 400 whose error body implicates `tool_choice` (for a named +tool-choice request) or `response_format` (for a `json_object` request) is +retried **once** with the offending shape degraded. The retry covers both the +unary and streaming paths, and never loops — at most one degraded retry per +call, and only for the shape the request actually used. This means unmodified +`from_env()` / preset instances "just work" against LM Studio and friends +without any per-provider configuration. + +## Inline reasoning-tag extraction + +Reasoning models served through OpenAI-compatible local runtimes (qwen3 and +deepseek-r1 distills via Ollama `/v1`, LM Studio, llama.cpp) often emit their +chain-of-thought **inline** in the normal `content` string wrapped in +`…`, rather than on the `reasoning_content` / `reasoning` +side-channel. `reasoning_tags.rs` moves that inline chain-of-thought onto the +reasoning channel (a leading `ContentBlock::Thinking` block) and strips the +tags from the visible text, matching how the side-channel is normalized. + +- **Enabled by default for non-hosted base URLs** with the plain `think` tag: + unhandled leakage is the common local-model failure, while hosted OpenAI + never emits inline `` and extraction there would strip legitimate + content that mentions a literal tag, so the hosted default stays off. Toggle + with `OpenAiModel::with_reasoning_tag_extraction(config)`: pass `None` to + disable everywhere (content passes through verbatim) or + `Some(ReasoningTagExtraction::new(tag))` to force it on — including for the + hosted base URL — and customize the tag name / separator. +- **Streaming-safe.** Content deltas are scanned incrementally; any trailing + bytes that could be the prefix of an opening/closing tag are held back until + the next delta resolves them (the Vercel AI SDK's `getPotentialStartIndex` + trick), so a tag split across deltas is neither leaked nor mangled. A lone + `<` or a look-alike such as `` is released as soon as it is + disproven, never held forever. +- **DeepSeek-R1 template mode.** When the output BEGINS mid-reasoning with no + opening tag and only a closing `` appears, set + `ReasoningTagExtraction::default().with_start_with_reasoning(true)` (the AI + SDK's `startWithReasoning`); off by default. +- **Side-channel interaction.** Inline-tag reasoning and side-channel reasoning + both feed the single leading `Thinking` block; when both appear, side-channel + reasoning leads and inline reasoning follows, joined by the configured + separator. The streamed terminal `Completed` response recomputes the split + from the raw accumulated content so it is consistent with the non-streaming + path (visible segments concatenate; the block-bordering whitespace is + trimmed). Live deltas carry no separator/trim, so the authoritative clean + text is the terminal response. + +## Local-model tolerance (tool calls) + +Small local servers (Ollama, LM Studio, llama.cpp, vLLM) routinely violate the +OpenAI tool-call contract. A single non-conforming field used to fail +deserialization of the *whole* response, so the model appeared "broken". The +wire structs (`types.rs`) and the translator (`convert.rs`) are deliberately +lenient — leniency is additive to the serde shapes (`#[serde(default)]` / +custom deserializers), so hosted OpenAI is unaffected: + +- **Missing/empty `id`** — Ollama omitted the tool-call `id` until v0.12.11. An + empty or absent id is treated as absent and a stable `tool-{index}` fallback + is synthesized (the same one the streaming path uses) so tool results still + correlate back to the call. +- **Missing `type`** — defaulted rather than required. +- **`arguments` as an object** — some servers send `function.arguments` as a + JSON object instead of the OpenAI-standard stringified JSON; it is normalized + to the parsed arguments either way. +- **Malformed `arguments` JSON** — when arguments cannot be parsed even after the + conservative chat-template-marker repair (`recover_tool_arguments`), the call + does **not** fail the model call. It is surfaced as a `ToolCall` with + `invalid: Some(reason)` and the raw string preserved in `arguments`. The agent + loop feeds `reason` back to the model as an error tool result (an + `AgentEvent::InvalidToolArgs` recovery, mirroring LangChain's + `invalid_tool_calls` and the AI SDK's invalid dynamic tool parts) so the model + can retry, and — because the call always resolves — a malformed blob can never + become a never-resolving tool call that stalls the loop. This leniency is + unconditional: unlike `InvalidArgsPolicy` (which governs *schema* validation of + well-formed arguments), an unparseable payload is a transport-level defect. + ## Error handling Non-2xx responses are normalized through `parse_error_body` / `provider_error` @@ -102,6 +210,7 @@ via `provider_failure_message`. Malformed JSON bodies surface as | `mod.rs` | Module wiring: shared imports/constants and re-exports (`OpenAiModel`). | | `transport.rs` | `OpenAiModel` construction, provider presets, request building, and the `ChatModel` impl (`invoke`/`stream`). | | `convert.rs` | Request/response translation between harness types and the OpenAI wire format. | +| `reasoning_tags.rs` | Streaming-safe inline `…` reasoning-tag extraction (`ReasoningTagExtraction`, `ReasoningTagStream`, `extract_reasoning`). | | `sse.rs` | SSE stream parsing and incremental accumulation (`SseState`, `OpenAiStreamAcc`, `sse_next`). | | `types.rs` | Wire (de)serialization shapes (`ModelListWire`, `ModelListing`, request/response bodies). | | `test.rs` | Unit tests (SSE boundary decoding, tool-call correlation, error mapping, presets). | diff --git a/src/harness/providers/openai/convert.rs b/src/harness/providers/openai/convert.rs index 6fd71a8..a46c68a 100644 --- a/src/harness/providers/openai/convert.rs +++ b/src/harness/providers/openai/convert.rs @@ -148,6 +148,26 @@ pub(super) fn translate_tool_choice(choice: &ToolChoice) -> Value { } } +/// The degraded `response_format` wire form for a [`ResponseFormat::JsonObject`] +/// request against a server that rejects `{"type":"json_object"}` (LM Studio, +/// and likely other local OpenAI-compatible runtimes). +/// +/// Maps to a permissive `json_schema` that constrains output to *some* JSON +/// object without pinning its shape: an empty object schema with `strict:false`, +/// so the model is free to choose keys. Verified accepted by LM Studio; a strict +/// or fully-specified schema would over-constrain the free-form "just JSON" +/// intent that [`ResponseFormat::JsonObject`] expresses. +pub(super) fn degraded_json_object_format() -> Value { + json!({ + "type": "json_schema", + "json_schema": { + "name": "json_object", + "schema": { "type": "object" }, + "strict": false, + } + }) +} + /// Translates a [`ResponseFormat`] into the OpenAI `response_format` JSON value. /// /// Returns `None` for [`ResponseFormat::Text`] so the field is omitted entirely. @@ -182,7 +202,24 @@ pub(super) fn translate_response_format(format: &ResponseFormat) -> Option Result { + parse_chat_response(value, None) +} + +/// Like [`parse_response`], but also normalizes reasoning into a leading +/// [`ContentBlock::Thinking`] block. Side-channel reasoning +/// (`reasoning_content` / `reasoning`) is always extracted; inline +/// `…` tags in the visible content are extracted only when +/// `reasoning_tags` is `Some`. When both are present, side-channel reasoning +/// leads and inline reasoning follows, joined by the configured separator. +pub(super) fn parse_chat_response( + value: Value, + reasoning_tags: Option<&ReasoningTagExtraction>, +) -> Result { let parsed: ChatCompletionResponse = serde_json::from_value(value.clone())?; let choice = parsed.choices.into_iter().next().ok_or_else(|| { @@ -190,29 +227,66 @@ pub(super) fn parse_response(value: Value) -> Result { })?; let mut content = Vec::new(); - if let Some(text) = choice.message.content.filter(|t| !t.is_empty()) { - content.push(ContentBlock::Text(text)); + + // Side-channel reasoning first, normalized the same way as the stream path. + let mut reasoning = String::new(); + for value in [choice.message.reasoning_content, choice.message.reasoning] + .into_iter() + .flatten() + { + if let Some(fragment) = reasoning_value_text(value) { + reasoning.push_str(&fragment); + } + } + + // Inline `` extraction on the visible content, when enabled. + let visible = match ( + choice.message.content.filter(|t| !t.is_empty()), + reasoning_tags, + ) { + (Some(text), Some(config)) => { + let (visible, inline) = extract_reasoning(config, &text); + if !inline.is_empty() { + if !reasoning.is_empty() { + reasoning.push_str(config.separator()); + } + reasoning.push_str(&inline); + } + visible + } + (Some(text), None) => text, + (None, _) => String::new(), + }; + + if !reasoning.is_empty() { + content.push(ContentBlock::Thinking { + text: reasoning, + signature: None, + }); + } + if !visible.is_empty() { + content.push(ContentBlock::Text(visible)); } let tool_calls = choice .message .tool_calls .into_iter() - .map(|call| { - Ok(ToolCall { - id: call.id.clone(), - name: call.function.name.clone(), - // Tool arguments arrive as a JSON string. Invalid JSON is a - // provider/model error, not an empty/default argument payload. - arguments: parse_tool_arguments( - "openai response", - &call.id, - &call.function.name, - &call.function.arguments, - )?, - }) + .enumerate() + .map(|(index, call)| { + // Local servers routinely omit `id`; synthesize the same + // `tool-{index}` fallback the streaming path uses so the agent loop + // can still correlate the tool result back to this call. An empty + // id is treated as absent. + tool_call_from_wire( + "openai response", + index, + &call.id, + &call.function.name, + &call.function.arguments, + ) }) - .collect::>>()?; + .collect::>(); let usage = parsed.usage.map(convert_usage); @@ -243,12 +317,47 @@ pub(super) fn tool_call_id(slot: usize, id: &str) -> String { } } -pub(super) fn parse_tool_arguments( +/// Builds a provider-neutral [`ToolCall`] from the wire fields, tolerating the +/// defects small local models produce. +/// +/// `slot` is the tool call's position in the response (used to synthesize a +/// stable `tool-{slot}` id when the provider omits one — Ollama did so until +/// v0.12.11). When the arguments cannot be parsed even after repair, the call is +/// marked [`ToolCall::invalid`] with the raw arguments preserved rather than +/// failing the whole model call: the agent loop feeds the error back to the +/// model as a tool result so it can retry (mirroring LangChain and the AI SDK), +/// and — because the call still resolves — a malformed argument blob can never +/// become a never-resolving tool call that stalls the loop. +pub(super) fn tool_call_from_wire( context: &str, - call_id: &str, + slot: usize, + id: &str, name: &str, raw: &str, -) -> Result { +) -> ToolCall { + let call_id = tool_call_id(slot, id); + match parse_tool_arguments(raw) { + Ok(arguments) => ToolCall { + id: call_id, + name: name.to_string(), + arguments, + invalid: None, + }, + Err(detail) => { + let reason = format!( + "{context} contained invalid JSON arguments for tool call `{call_id}` (`{name}`): {detail}; raw arguments: {raw:?}" + ); + ToolCall::invalid(call_id, name, raw, reason) + } + } +} + +/// Parses a tool-call arguments string into JSON, returning `Err(detail)` with a +/// short human-readable reason when the string cannot be recovered. +/// +/// This never fails the model call itself; the caller +/// ([`tool_call_from_wire`]) turns an `Err` into an [`ToolCall::invalid`] call. +pub(super) fn parse_tool_arguments(raw: &str) -> std::result::Result { // Some OpenAI-compatible backends emit an empty arguments string for a // zero-argument tool call. That is a well-formed "no arguments" payload, not // malformed JSON, so map it to an empty object instead of failing the call. @@ -270,13 +379,7 @@ pub(super) fn parse_tool_arguments( if let Some(value) = recover_tool_arguments(raw) { return Ok(value); } - // Still unparseable after repair: fail fast with a clear, - // non-retryable error. Surfacing the failure keeps the malformed - // call from becoming a never-resolving tool call that stalls the - // agent loop (and the parent's join/reduce) indefinitely. - Err(TinyAgentsError::Model(format!( - "{context} contained invalid JSON arguments for tool call `{call_id}` (`{name}`): {err}; raw arguments: {raw:?}" - ))) + Err(err.to_string()) } } } diff --git a/src/harness/providers/openai/mod.rs b/src/harness/providers/openai/mod.rs index 9e0a697..470664c 100644 --- a/src/harness/providers/openai/mod.rs +++ b/src/harness/providers/openai/mod.rs @@ -17,6 +17,15 @@ //! the translation logic and the HTTP transport, keeping OpenAI-specific JSON //! out of the rest of the harness. //! +//! Local OpenAI-compatible runtimes (LM Studio, llama.cpp server, …) reject a +//! named `tool_choice` object and a `json_object` response format with an HTTP +//! 400. The transport degrades both to shapes they accept — `tool_choice` +//! `"required"` with the `tools` array filtered to the named tool, and a +//! permissive `json_schema` — either eagerly via +//! [`OpenAiModel::with_named_tool_choice`] / [`OpenAiModel::with_json_object_format`] +//! or automatically as a single retry when a 400 body implicates the shape. See +//! the module `README.md` "Local-server compatibility" section. +//! //! # Example //! //! ```no_run @@ -67,18 +76,21 @@ const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 600; mod convert; mod prompt_tools; +mod reasoning_tags; mod responses; mod sse; mod transport; +pub use reasoning_tags::ReasoningTagExtraction; pub use transport::{AuthStyle, OpenAiModel}; use convert::*; +use reasoning_tags::*; use sse::*; #[cfg(test)] use transport::{ - auth_headers, effective_temperature, glob_match, merge_provider_options, - merge_system_into_user, request_timeout, + Degrade, auth_headers, degrade_for_400, effective_temperature, glob_match, + merge_provider_options, merge_system_into_user, request_timeout, }; #[cfg(test)] diff --git a/src/harness/providers/openai/prompt_tools.rs b/src/harness/providers/openai/prompt_tools.rs index a830e61..a922b1a 100644 --- a/src/harness/providers/openai/prompt_tools.rs +++ b/src/harness/providers/openai/prompt_tools.rs @@ -134,6 +134,7 @@ fn parse_one(inner: &str, index: usize) -> Option { id: format!("call_{index}"), name, arguments, + invalid: None, }) } diff --git a/src/harness/providers/openai/reasoning_tags.rs b/src/harness/providers/openai/reasoning_tags.rs new file mode 100644 index 0000000..6cae9ee --- /dev/null +++ b/src/harness/providers/openai/reasoning_tags.rs @@ -0,0 +1,481 @@ +//! Streaming-safe extraction of inline `…` reasoning tags. +//! +//! Reasoning models served through OpenAI-compatible local runtimes (qwen3 and +//! deepseek-r1 distills via Ollama `/v1`, LM Studio, llama.cpp) frequently emit +//! their chain-of-thought **inline** in the normal `content` string, wrapped in +//! `…`, instead of on the `reasoning_content` / `reasoning` +//! side-channel the adapter already normalizes (see +//! [`reasoning_value_text`](super::reasoning_value_text)). Left untouched, that +//! chain-of-thought leaks straight into the visible assistant text. +//! +//! This module moves the tagged text onto the reasoning channel +//! ([`ContentBlock::Thinking`](crate::harness::message::ContentBlock::Thinking)) +//! and strips the tags from the visible text. It provides two entry points that +//! share the same tag-matching logic so the streamed deltas and the final +//! response agree on what is reasoning and what is visible: +//! +//! * [`ReasoningTagStream`] — an incremental state machine for the SSE path. It +//! buffers and holds back any trailing bytes that *could* be the prefix of an +//! opening or closing tag ([`potential_start_index`], the Vercel AI SDK's +//! `getPotentialStartIndex` trick) so a tag split across deltas is neither +//! leaked as visible text nor mangled. +//! * [`extract_reasoning`] — a whole-string extractor for the non-streaming +//! path (and the authoritative terminal streamed response). +//! +//! # Side-channel interaction +//! +//! Inline-tag reasoning and side-channel reasoning both feed the single leading +//! `Thinking` block. When a response carries both, side-channel reasoning leads +//! and inline-extracted reasoning follows, joined by the configured separator. +//! In practice a given model uses one convention or the other, not both. + +/// Options controlling inline `…` reasoning extraction on the +/// OpenAI-compatible provider. Construct with [`ReasoningTagExtraction::new`] or +/// [`ReasoningTagExtraction::default`] (the `think` tag) and pass to +/// [`OpenAiModel::with_reasoning_tag_extraction`](super::OpenAiModel::with_reasoning_tag_extraction). +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ReasoningTagExtraction { + /// Tag name without angle brackets, e.g. `think` → `` / ``. + tag_name: String, + /// Separator inserted between multiple extracted reasoning sections (and + /// between side-channel and inline reasoning). Defaults to a newline. + separator: String, + /// When `true`, the output BEGINS mid-reasoning with no opening tag and only + /// a closing `` appears — the DeepSeek-R1 template mode (the AI SDK's + /// `startWithReasoning`). Everything before the first closing tag is + /// reasoning. Defaults to `false`. + start_with_reasoning: bool, +} + +impl Default for ReasoningTagExtraction { + fn default() -> Self { + Self { + tag_name: "think".to_string(), + separator: "\n".to_string(), + start_with_reasoning: false, + } + } +} + +impl ReasoningTagExtraction { + /// Extraction for a custom tag name (no angle brackets), newline separator, + /// opening-tag-gated (not DeepSeek mode). + pub fn new(tag_name: impl Into) -> Self { + Self { + tag_name: tag_name.into(), + ..Self::default() + } + } + + /// Overrides the separator joining multiple reasoning sections. + pub fn with_separator(mut self, separator: impl Into) -> Self { + self.separator = separator.into(); + self + } + + /// Enables DeepSeek-R1 template mode: the stream begins mid-reasoning with no + /// opening tag, and only a closing `` marks the end of reasoning. + pub fn with_start_with_reasoning(mut self, enabled: bool) -> Self { + self.start_with_reasoning = enabled; + self + } + + /// The literal opening tag, e.g. ``. + fn opening_tag(&self) -> String { + format!("<{}>", self.tag_name) + } + + /// The literal closing tag, e.g. ``. + fn closing_tag(&self) -> String { + format!("", self.tag_name) + } + + /// The separator joining reasoning sections. + pub(super) fn separator(&self) -> &str { + &self.separator + } +} + +/// Finds where `searched` begins within `text`, as either a full occurrence or a +/// partial prefix at the very end of `text`. +/// +/// Returns the byte offset of the earliest position at which `searched` could +/// start: +/// * `Some(idx)` where `text[idx..]` fully contains `searched` (a complete tag), +/// or where `text[idx..]` is a non-empty proper prefix of `searched` (a +/// partial tag straddling the end that must be held back for more input); +/// * `None` when no suffix of `text` could begin `searched` — the whole `text` +/// is safe to release. +/// +/// This is the Rust port of the Vercel AI SDK's `getPotentialStartIndex`. It +/// only ever returns positions on UTF-8 character boundaries (the tags are +/// ASCII, and it scans `char_indices`), so callers may slice `text` at the +/// result safely. +pub(super) fn potential_start_index(text: &str, searched: &str) -> Option { + if searched.is_empty() { + return None; + } + if let Some(idx) = text.find(searched) { + return Some(idx); + } + // No full occurrence: look for a suffix of `text` that is a prefix of + // `searched`, scanning shortest-suffix-first (end inward) to match the AI + // SDK. Iterate char-boundary starts so any returned index is sliceable. + let starts: Vec = text.char_indices().map(|(i, _)| i).collect(); + for &i in starts.iter().rev() { + let suffix = &text[i..]; + if searched.starts_with(suffix) { + return Some(i); + } + } + None +} + +/// Incremental extractor for the streamed content path. +/// +/// Feed each `content` delta to [`push`](Self::push); it appends any resolved +/// visible / reasoning text to the supplied buffers and retains any trailing +/// partial-tag bytes internally until the next delta (or [`finish`](Self::finish) +/// at end of stream) resolves them. +#[derive(Clone, Debug)] +pub(super) struct ReasoningTagStream { + opening: String, + closing: String, + /// Bytes received but not yet released (may end in a partial tag). + buffer: String, + /// Whether the machine is currently inside a reasoning section. + in_reasoning: bool, +} + +impl ReasoningTagStream { + /// Builds a stream extractor from the configured tags. In DeepSeek mode the + /// machine starts already inside a reasoning section. + pub(super) fn new(config: &ReasoningTagExtraction) -> Self { + Self { + opening: config.opening_tag(), + closing: config.closing_tag(), + buffer: String::new(), + in_reasoning: config.start_with_reasoning, + } + } + + /// Folds one content delta into the machine, appending resolved text to + /// `visible` / `reasoning`. A trailing partial tag is held in `buffer`. + pub(super) fn push(&mut self, delta: &str, visible: &mut String, reasoning: &mut String) { + self.buffer.push_str(delta); + loop { + let tag = if self.in_reasoning { + &self.closing + } else { + &self.opening + }; + match potential_start_index(&self.buffer, tag) { + // Neither a complete tag nor a partial-tag suffix: everything + // buffered is safe to release. + None => { + let released = std::mem::take(&mut self.buffer); + self.emit(&released, visible, reasoning); + break; + } + Some(idx) => { + // Text before the (partial or full) tag belongs to the + // current channel. + let before = self.buffer[..idx].to_string(); + self.emit(&before, visible, reasoning); + if idx + tag.len() <= self.buffer.len() { + // Complete tag: drop it, toggle channel, keep scanning + // the remainder for the opposite tag. + self.buffer = self.buffer[idx + tag.len()..].to_string(); + self.in_reasoning = !self.in_reasoning; + } else { + // Partial tag at the buffer tail: hold it for more input. + self.buffer = self.buffer[idx..].to_string(); + break; + } + } + } + } + } + + /// Releases any buffered partial-tag tail at end of stream into the current + /// channel — a partial tag that never completed is real content. + /// + /// The streaming provider path does not call this: it recomputes the + /// authoritative split from the raw accumulated content in + /// [`into_response`](super::OpenAiStreamAcc), which also applies the + /// non-streaming trimming so the terminal response matches exactly. This is + /// the state machine's flush contract, exercised by the unit tests. + #[cfg(test)] + pub(super) fn finish(&mut self, visible: &mut String, reasoning: &mut String) { + let released = std::mem::take(&mut self.buffer); + self.emit(&released, visible, reasoning); + } + + fn emit(&self, text: &str, visible: &mut String, reasoning: &mut String) { + if text.is_empty() { + return; + } + if self.in_reasoning { + reasoning.push_str(text); + } else { + visible.push_str(text); + } + } +} + +/// Extracts inline reasoning from a complete `content` string. +/// +/// Returns `(visible_text, reasoning_text)`. Reasoning sections are trimmed and +/// joined with the configured separator; the visible text has the tagged +/// sections (and the whitespace that surrounded them) removed. When the content +/// contains no reasoning at all, it is returned verbatim as the visible text so +/// plain responses are preserved byte-for-byte. +pub(super) fn extract_reasoning( + config: &ReasoningTagExtraction, + content: &str, +) -> (String, String) { + let opening = config.opening_tag(); + let closing = config.closing_tag(); + let mut in_reasoning = config.start_with_reasoning; + let mut rest = content; + let mut visible_parts: Vec<&str> = Vec::new(); + let mut reasoning_parts: Vec<&str> = Vec::new(); + + loop { + let tag = if in_reasoning { &closing } else { &opening }; + match rest.find(tag.as_str()) { + Some(idx) => { + let (before, after) = rest.split_at(idx); + if in_reasoning { + reasoning_parts.push(before); + } else { + visible_parts.push(before); + } + rest = &after[tag.len()..]; + in_reasoning = !in_reasoning; + } + None => { + if in_reasoning { + reasoning_parts.push(rest); + } else { + visible_parts.push(rest); + } + break; + } + } + } + + // No reasoning was ever entered: return the content untouched so plain text + // is preserved exactly (no whitespace normalization). + if !config.start_with_reasoning && reasoning_parts.is_empty() { + return (content.to_string(), String::new()); + } + + // Visible text: concatenate the surviving segments (matching the streaming + // deltas, which carry no separator) and trim the whitespace that bordered + // the removed sections at the ends. + let visible = visible_parts.concat().trim().to_string(); + // Reasoning: join distinct sections with the configured separator, trimming + // each so tag-adjacent whitespace does not bloat the thinking block. + let reasoning = reasoning_parts + .iter() + .map(|part| part.trim()) + .filter(|part| !part.is_empty()) + .collect::>() + .join(&config.separator); + + (visible, reasoning) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Drives a sequence of content deltas through the stream machine and + /// returns the accumulated `(visible, reasoning)`. + fn run_stream(config: &ReasoningTagExtraction, deltas: &[&str]) -> (String, String) { + let mut machine = ReasoningTagStream::new(config); + let mut visible = String::new(); + let mut reasoning = String::new(); + for delta in deltas { + machine.push(delta, &mut visible, &mut reasoning); + } + machine.finish(&mut visible, &mut reasoning); + (visible, reasoning) + } + + #[test] + fn potential_start_index_finds_full_and_partial_matches() { + assert_eq!(potential_start_index("abc", ""), Some(2)); + // Partial tag straddling the end is held from its first byte. + assert_eq!(potential_start_index("hello"), Some(5)); + assert_eq!(potential_start_index("<", ""), Some(0)); + // A lone trailing `<` with a non-matching continuation is not held. + assert_eq!(potential_start_index(""), None); + assert_eq!(potential_start_index("plain text", ""), None); + // `` is not `` and must not be held back. + assert_eq!(potential_start_index("", ""), None); + } + + #[test] + fn tag_fully_inside_one_delta() { + let cfg = ReasoningTagExtraction::default(); + let (visible, reasoning) = run_stream(&cfg, &["Hiponderthere"]); + assert_eq!(visible, "Hithere"); + assert_eq!(reasoning, "ponder"); + } + + #[test] + fn tag_split_across_deltas_at_every_boundary() { + let cfg = ReasoningTagExtraction::default(); + let source = "beforesecretafter"; + let open = ""; + // Split the opening tag at every interior byte boundary. + for cut in 0..=open.len() { + let prefix = format!("before{}", &open[..cut]); + let suffix = format!("{}secretafter", &open[cut..]); + let (visible, reasoning) = run_stream(&cfg, &[&prefix, &suffix]); + assert_eq!(visible, "beforeafter", "cut at {cut}"); + assert_eq!(reasoning, "secret", "cut at {cut}"); + } + // Sanity: the whole thing in one delta agrees. + let (visible, reasoning) = run_stream(&cfg, &[source]); + assert_eq!(visible, "beforeafter"); + assert_eq!(reasoning, "secret"); + } + + #[test] + fn closing_tag_split_across_deltas() { + let cfg = ReasoningTagExtraction::default(); + let close = ""; + for cut in 0..=close.len() { + let first = format!("reason{}", &close[..cut]); + let second = format!("{}visible", &close[cut..]); + let (visible, reasoning) = run_stream(&cfg, &[&first, &second]); + assert_eq!(visible, "visible", "cut at {cut}"); + assert_eq!(reasoning, "reason", "cut at {cut}"); + } + } + + #[test] + fn lone_angle_bracket_is_not_held_forever() { + let cfg = ReasoningTagExtraction::default(); + // A `<` that turns out not to open a tag is released once disproven. + let (visible, reasoning) = run_stream(&cfg, &["a < b < c"]); + assert_eq!(visible, "a < b < c"); + assert_eq!(reasoning, ""); + + // `<` held at a delta boundary, then disproven by the next delta. + let (visible, reasoning) = run_stream(&cfg, &["value <", "= 3"]); + assert_eq!(visible, "value <= 3"); + assert_eq!(reasoning, ""); + } + + #[test] + fn thinker_lookalike_tag_is_not_extracted() { + let cfg = ReasoningTagExtraction::default(); + let (visible, reasoning) = run_stream(&cfg, &["use here"]); + assert_eq!(visible, "use here"); + assert_eq!(reasoning, ""); + + // Even when split so ` here"]); + assert_eq!(visible, "use here"); + assert_eq!(reasoning, ""); + } + + #[test] + fn think_section_spanning_many_deltas() { + let cfg = ReasoningTagExtraction::default(); + let deltas = [ + "Answer: ", + "", + "step one, ", + "step two, ", + "step three", + "", + "42", + ]; + let (visible, reasoning) = run_stream(&cfg, &deltas); + assert_eq!(visible, "Answer: 42"); + assert_eq!(reasoning, "step one, step two, step three"); + } + + #[test] + fn text_after_think_is_visible() { + let cfg = ReasoningTagExtraction::default(); + let (visible, reasoning) = run_stream(&cfg, &["hmm", "the final answer"]); + assert_eq!(visible, "the final answer"); + assert_eq!(reasoning, "hmm"); + } + + #[test] + fn start_with_reasoning_mode_stream() { + // DeepSeek-R1 template: output begins mid-reasoning, only a closing tag. + let cfg = ReasoningTagExtraction::default().with_start_with_reasoning(true); + let (visible, reasoning) = + run_stream(&cfg, &["chain of ", "thought", "final answer"]); + assert_eq!(visible, "final answer"); + assert_eq!(reasoning, "chain of thought"); + } + + #[test] + fn start_with_reasoning_without_closing_tag_is_all_reasoning() { + let cfg = ReasoningTagExtraction::default().with_start_with_reasoning(true); + let (visible, reasoning) = run_stream(&cfg, &["still ", "thinking"]); + assert_eq!(visible, ""); + assert_eq!(reasoning, "still thinking"); + } + + #[test] + fn unclosed_think_streams_remainder_as_reasoning() { + let cfg = ReasoningTagExtraction::default(); + let (visible, reasoning) = run_stream(&cfg, &["ok never closed"]); + assert_eq!(visible, "ok "); + assert_eq!(reasoning, "never closed"); + } + + #[test] + fn non_streaming_extraction_strips_surrounding_whitespace() { + let cfg = ReasoningTagExtraction::default(); + let (visible, reasoning) = + extract_reasoning(&cfg, "deliberate\n\nThe answer is 42."); + assert_eq!(visible, "The answer is 42."); + assert_eq!(reasoning, "deliberate"); + } + + #[test] + fn non_streaming_plain_text_is_preserved_verbatim() { + let cfg = ReasoningTagExtraction::default(); + let (visible, reasoning) = extract_reasoning(&cfg, "line one\nline two"); + assert_eq!(visible, "line one\nline two"); + assert_eq!(reasoning, ""); + } + + #[test] + fn non_streaming_multiple_sections_join_reasoning_with_separator() { + let cfg = ReasoningTagExtraction::default(); + let (visible, reasoning) = extract_reasoning(&cfg, "ar1br2c"); + // Visible segments concatenate (consistent with the streamed deltas); + // reasoning sections join with the separator. + assert_eq!(visible, "abc"); + assert_eq!(reasoning, "r1\nr2"); + } + + #[test] + fn non_streaming_start_with_reasoning() { + let cfg = ReasoningTagExtraction::default().with_start_with_reasoning(true); + let (visible, reasoning) = + extract_reasoning(&cfg, "reasoning here\n\nvisible answer"); + assert_eq!(visible, "visible answer"); + assert_eq!(reasoning, "reasoning here"); + } + + #[test] + fn custom_tag_name_is_honored() { + let cfg = ReasoningTagExtraction::new("reasoning"); + let (visible, reasoning) = run_stream(&cfg, &["a", "bc"]); + assert_eq!(visible, "ac"); + assert_eq!(reasoning, "b"); + } +} diff --git a/src/harness/providers/openai/sse.rs b/src/harness/providers/openai/sse.rs index 5421765..71cf50e 100644 --- a/src/harness/providers/openai/sse.rs +++ b/src/harness/providers/openai/sse.rs @@ -26,16 +26,44 @@ pub(super) struct ToolCallBuild { pub(super) struct OpenAiStreamAcc { id: Option, text: String, - /// Accumulated reasoning/thinking fragments, preserved on the final - /// message as a leading [`ContentBlock::Thinking`] block (unsigned — the - /// OpenAI-compatible path has no provider signature). + /// Accumulated **side-channel** reasoning/thinking fragments + /// (`reasoning_content` / `reasoning`), preserved on the final message as a + /// leading [`ContentBlock::Thinking`] block (unsigned — the + /// OpenAI-compatible path has no provider signature). Inline `` tag + /// reasoning is not accumulated here; it is recomputed from `text` in + /// [`into_response`](Self::into_response) so the terminal response matches + /// the non-streaming path exactly. reasoning: String, tool_calls: Vec, usage: Option, finish_reason: Option, + /// Inline `` extraction config (`None` disables it). Applied to the + /// terminal response via [`extract_reasoning`]. + reasoning_tags: Option, + /// Live per-delta inline-tag splitter, so streamed content deltas route + /// chain-of-thought onto the reasoning channel instead of leaking it. + extractor: Option, + /// Wire `index` → current accumulator slot. Normally the identity mapping, + /// but when a backend reuses one index for several parallel calls (Ollama's + /// `/v1` bug, ollama/ollama#15457) the conflict handling in + /// [`resolve_slot`](Self::resolve_slot) repoints the index at the freshly + /// opened slot, so id-less argument continuations for that index keep + /// following the call most recently opened there. + index_slots: std::collections::HashMap, } impl OpenAiStreamAcc { + /// Builds an accumulator with the given inline reasoning-tag extraction + /// config (`None` disables inline extraction). + pub(super) fn new(reasoning_tags: Option) -> Self { + let extractor = reasoning_tags.as_ref().map(ReasoningTagStream::new); + Self { + reasoning_tags, + extractor, + ..Self::default() + } + } + /// Folds one parsed chunk into the accumulator and pushes the corresponding /// neutral [`ModelStreamItem`]s onto `pending`. fn ingest(&mut self, chunk: ChatCompletionChunk, pending: &mut VecDeque) { @@ -63,12 +91,40 @@ impl OpenAiStreamAcc { })); } if let Some(content) = choice.delta.content.filter(|c| !c.is_empty()) { + // Retain the raw content so the terminal response can recompute + // an authoritative inline-tag split (see `into_response`). self.text.push_str(&content); - pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { - text: content, - reasoning: String::new(), - tool_call: None, - })); + match self.extractor.as_mut() { + // Inline extraction on: scan this delta, holding back any + // trailing partial tag, and route the split onto the two + // channels as separate deltas. + Some(extractor) => { + let mut visible = String::new(); + let mut reasoning = String::new(); + extractor.push(&content, &mut visible, &mut reasoning); + if !reasoning.is_empty() { + pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { + text: String::new(), + reasoning, + tool_call: None, + })); + } + if !visible.is_empty() { + pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { + text: visible, + reasoning: String::new(), + tool_call: None, + })); + } + } + None => { + pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { + text: content, + reasoning: String::new(), + tool_call: None, + })); + } + } } for fragment in choice.delta.tool_calls { let idx = self.resolve_slot(&fragment); @@ -106,13 +162,45 @@ impl OpenAiStreamAcc { /// carrying a known id reuses that slot, and an id-less continuation fragment /// (arguments only) appends to the most recent slot — so parallel calls no /// longer all collapse onto slot 0. + /// + /// One exception guards Ollama's `/v1` parallel-tool-call bug + /// (ollama/ollama#15457), where every parallel call arrives with `index: 0`: + /// when an explicit index carries a non-empty id that *conflicts* with the id + /// already recorded at that slot, the fragment is a distinct new call, not a + /// continuation, so it opens a fresh slot (or reuses an existing slot already + /// opened for that id) instead of silently merging two calls onto one. The + /// index is then repointed at that slot, so subsequent id-less argument + /// continuations arriving under the same reused index follow the call most + /// recently opened there instead of the original occupant. fn resolve_slot(&mut self, fragment: &ToolCallChunkWire) -> usize { if let Some(index) = fragment.index { let idx = index as usize; while self.tool_calls.len() <= idx { self.tool_calls.push(ToolCallBuild::default()); } - return idx; + let current = *self.index_slots.entry(index).or_insert(idx); + if let Some(id) = fragment.id.as_deref().filter(|id| !id.is_empty()) { + let occupant = &self.tool_calls[current]; + if !occupant.id.is_empty() && occupant.id != id { + // Conflict: a second distinct call reusing index 0. Reuse a + // slot already opened for this id if one exists (so its later + // continuation fragments still land correctly), else open a + // fresh slot rather than overwriting the occupant. Either + // way, repoint the index cursor so id-less argument + // continuations for this index follow the call most + // recently opened here rather than the original occupant. + let slot = match self.tool_calls.iter().position(|slot| slot.id == id) { + Some(pos) => pos, + None => { + self.tool_calls.push(ToolCallBuild::default()); + self.tool_calls.len() - 1 + } + }; + self.index_slots.insert(index, slot); + return slot; + } + } + return current; } if let Some(id) = fragment.id.as_deref().filter(|id| !id.is_empty()) { if let Some(pos) = self.tool_calls.iter().position(|slot| slot.id == id) { @@ -128,18 +216,40 @@ impl OpenAiStreamAcc { } /// Consumes the accumulator into the final, merged [`ModelResponse`]. - fn into_response(self) -> Result { + /// + /// Infallible: a tool call whose reassembled arguments cannot be parsed is + /// surfaced as an [`ToolCall::invalid`] call (raw arguments preserved) rather + /// than failing the whole stream, so the agent loop can feed the error back + /// to the model and the call still resolves instead of stalling the loop. + fn into_response(self) -> ModelResponse { let mut content = Vec::new(); - // Reasoning streamed on the side channel leads the final message as a - // `Thinking` block so it survives persistence and provider replay. - if !self.reasoning.is_empty() { + // Recompute the inline-tag split over the raw accumulated content so the + // terminal response is byte-identical to the non-streaming path, then + // combine it with side-channel reasoning. Both feed one leading + // `Thinking` block (side-channel leads) so it survives persistence and + // provider replay. + let (visible_text, reasoning) = match &self.reasoning_tags { + Some(config) => { + let (visible, inline) = extract_reasoning(config, &self.text); + let mut reasoning = self.reasoning.clone(); + if !inline.is_empty() { + if !reasoning.is_empty() { + reasoning.push_str(config.separator()); + } + reasoning.push_str(&inline); + } + (visible, reasoning) + } + None => (self.text.clone(), self.reasoning.clone()), + }; + if !reasoning.is_empty() { content.push(ContentBlock::Thinking { - text: self.reasoning, + text: reasoning, signature: None, }); } - if !self.text.is_empty() { - content.push(ContentBlock::Text(self.text)); + if !visible_text.is_empty() { + content.push(ContentBlock::Text(visible_text)); } // Enumerate over the full slot vector *before* filtering so the synthetic // fallback id (`tool-{idx}`) matches the one streamed in `ToolCallDelta` @@ -150,28 +260,21 @@ impl OpenAiStreamAcc { .into_iter() .enumerate() .filter(|(_, b)| !b.name.is_empty() || !b.args.is_empty()) - .map(|(idx, b)| { - let id = tool_call_id(idx, &b.id); - Ok(ToolCall { - id: id.clone(), - name: b.name.clone(), - arguments: parse_tool_arguments("openai stream", &id, &b.name, &b.args)?, - }) - }) - .collect::>>()?; + .map(|(idx, b)| tool_call_from_wire("openai stream", idx, &b.id, &b.name, &b.args)) + .collect::>(); let message = AssistantMessage { id: self.id, content, tool_calls, usage: self.usage, }; - Ok(ModelResponse { + ModelResponse { message, usage: self.usage, finish_reason: self.finish_reason, raw: None, resolved_model: None, - }) + } } } @@ -325,20 +428,11 @@ pub(super) async fn sse_next(mut state: SseState) -> Option<(ModelStreamItem, Ss return None; } state.terminal_emitted = true; - return match std::mem::take(&mut state.acc).into_response() { - Ok(response) => Some((ModelStreamItem::Completed(response), state)), - Err(error) => { - let provider_error = ProviderError { - provider: state.provider.clone(), - model: Some(state.model.clone()), - code: Some("invalid_tool_arguments".to_string()), - message: error.to_string(), - retryable: false, - ..ProviderError::default() - }; - Some((ModelStreamItem::ProviderFailed(provider_error), state)) - } - }; + // Reconstruction is infallible: malformed tool arguments become an + // `ToolCall::invalid` call inside the response (not a stream + // failure), so the agent loop recovers instead of aborting the run. + let response = std::mem::take(&mut state.acc).into_response(); + return Some((ModelStreamItem::Completed(response), state)); } match state.bytes.next().await { Some(Ok(chunk)) => { diff --git a/src/harness/providers/openai/test.rs b/src/harness/providers/openai/test.rs index 5ae6268..41035a6 100644 --- a/src/harness/providers/openai/test.rs +++ b/src/harness/providers/openai/test.rs @@ -193,6 +193,7 @@ fn translates_assistant_tool_calls_to_stringified_arguments() { id: "call-1".to_string(), name: "get_weather".to_string(), arguments: json!({ "city": "Paris" }), + invalid: None, }], usage: None, }), @@ -289,7 +290,11 @@ fn parses_openai_response_with_content_tool_call_and_usage() { } #[test] -fn parse_response_errors_on_invalid_tool_argument_json() { +fn parse_response_marks_invalid_tool_argument_json_instead_of_failing() { + // Malformed argument JSON must not fail the whole model call (which made + // small local models appear "broken"). Instead the call is surfaced as an + // `invalid` ToolCall with the raw arguments preserved so the agent loop can + // feed the error back to the model. let body = json!({ "id": "chatcmpl-badargs", "choices": [ @@ -312,12 +317,81 @@ fn parse_response_errors_on_invalid_tool_argument_json() { ] }); - let err = parse_response(body).expect_err("invalid arguments must fail"); - let message = err.to_string(); - assert!(matches!(err, TinyAgentsError::Model(_))); - assert!(message.contains("call-bad"), "{message}"); - assert!(message.contains("lookup"), "{message}"); - assert!(message.contains("raw arguments"), "{message}"); + let response = parse_response(body).expect("malformed args must not fail the call"); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].id, "call-bad"); + assert_eq!(calls[0].name, "lookup"); + // Raw arguments preserved verbatim as a JSON string value. + assert_eq!(calls[0].arguments, json!("{\"q\":")); + let reason = calls[0].invalid.as_deref().expect("call marked invalid"); + assert!(reason.contains("call-bad"), "{reason}"); + assert!(reason.contains("lookup"), "{reason}"); + assert!(reason.contains("raw arguments"), "{reason}"); +} + +#[test] +fn parses_id_less_tool_call_with_synthesized_fallback_id() { + // Ollama's /v1 endpoint omitted the tool-call `id` entirely until v0.12.11, + // and some servers omit `type`. Neither may fail deserialization; a + // missing/empty id gets the same `tool-{index}` fallback the streaming path + // uses so the agent loop can still correlate the result. + let body = json!({ + "id": "chatcmpl-noid", + "choices": [{ + "message": { + "role": "assistant", + "tool_calls": [ + // No `id`, no `type`. + { "function": { "name": "ping", "arguments": "{}" } }, + // Explicit empty id is treated as absent. + { "id": "", "type": "function", "function": { "name": "pong", "arguments": "{\"n\":1}" } } + ] + }, + "finish_reason": "tool_calls" + }] + }); + + let response = parse_response(body).unwrap(); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 2); + assert_eq!(calls[0].id, "tool-0"); + assert_eq!(calls[0].name, "ping"); + assert_eq!(calls[0].arguments, json!({})); + assert_eq!(calls[1].id, "tool-1"); + assert_eq!(calls[1].name, "pong"); + assert_eq!(calls[1].arguments, json!({ "n": 1 })); +} + +#[test] +fn parses_object_form_tool_arguments() { + // Some OpenAI-compatible servers send `function.arguments` as a JSON object + // instead of the OpenAI-standard stringified JSON. It must normalize to the + // same parsed arguments, not fail the response. + let body = json!({ + "id": "chatcmpl-obj", + "choices": [{ + "message": { + "role": "assistant", + "tool_calls": [{ + "id": "call-obj", + "type": "function", + "function": { "name": "get_weather", "arguments": { "city": "Paris" } } + }] + }, + "finish_reason": "tool_calls" + }] + }); + + let response = parse_response(body).unwrap(); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].name, "get_weather"); + assert_eq!(calls[0].arguments, json!({ "city": "Paris" })); + assert!( + calls[0].invalid.is_none(), + "object args are valid, not invalid" + ); } #[test] @@ -456,6 +530,39 @@ fn parse_error_body_classifies_retryability_by_http_status() { assert!(server_error.retryable, "5xx must be retryable"); } +#[test] +fn reasoning_tag_extraction_defaults_off_for_hosted_openai_only() { + // Hosted OpenAI never emits inline `` reasoning; unconditional + // extraction there would silently strip legitimate content mentioning a + // literal tag. The built-in default therefore only takes effect for + // non-hosted base URLs, while an explicit override always wins. + let hosted = OpenAiModel::new("k"); + assert!( + hosted.effective_reasoning_tags().is_none(), + "hosted default must not extract inline tags" + ); + + let local = OpenAiModel::compatible("k", "http://localhost:1234/v1", "m"); + assert!( + local.effective_reasoning_tags().is_some(), + "compat endpoints get extraction by default" + ); + assert!(OpenAiModel::ollama().effective_reasoning_tags().is_some()); + + let forced = OpenAiModel::new("k") + .with_reasoning_tag_extraction(Some(ReasoningTagExtraction::default())); + assert!( + forced.effective_reasoning_tags().is_some(), + "explicit override forces extraction on for the hosted base URL" + ); + + let disabled = OpenAiModel::ollama().with_reasoning_tag_extraction(None); + assert!( + disabled.effective_reasoning_tags().is_none(), + "explicit None disables extraction everywhere" + ); +} + #[test] fn compatible_presets_set_base_url_and_default_model() { let deepseek = OpenAiModel::deepseek("k"); @@ -669,19 +776,53 @@ async fn sse_stream_preserves_reasoning_content_as_side_channel() { } #[tokio::test] -async fn sse_stream_invalid_tool_argument_json_fails_terminally() { - use futures::StreamExt; - +async fn sse_stream_invalid_tool_argument_json_reconstructs_as_invalid_call() { + // A streamed tool call whose reassembled arguments are malformed must not + // fail the stream terminally: it reconstructs as an `invalid` ToolCall (raw + // arguments preserved) so the agent loop can feed the error back to the + // model and the call still resolves instead of stalling the loop. let raw: Vec> = vec![ b"data: {\"id\":\"chatcmpl-1\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-bad\",\"function\":{\"name\":\"lookup\",\"arguments\":\"{\\\"q\\\":\"}}]}}]}\n\n".to_vec(), b"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n".to_vec(), b"data: [DONE]\n\n".to_vec(), ]; + + let items = collect_sse(raw).await; + + // No terminal failure is emitted; the stream completes normally. + assert!( + !items + .iter() + .any(|item| matches!(item, ModelStreamItem::ProviderFailed(_))), + "malformed args must not emit ProviderFailed" + ); + assert!(matches!(items.last(), Some(ModelStreamItem::Completed(_)))); + + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged + .finish() + .expect("stream must reconstruct an invalid call, not fail"); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].id, "call-bad"); + assert_eq!(calls[0].name, "lookup"); + assert_eq!(calls[0].arguments, json!("{\"q\":")); + let reason = calls[0].invalid.as_deref().expect("call marked invalid"); + assert!(reason.contains("call-bad"), "{reason}"); + assert!(reason.contains("lookup"), "{reason}"); +} + +/// Drives an SSE byte stream through the parser and returns every item. +async fn collect_sse(raw: Vec>) -> Vec { + use futures::StreamExt; + let bytes = futures::stream::iter( raw.into_iter() .map(|v| Ok::(bytes::Bytes::from(v))), ); - let state = SseState { bytes: Box::pin(bytes), buf: Vec::new(), @@ -693,32 +834,15 @@ async fn sse_stream_invalid_tool_argument_json_fails_terminally() { finished: false, terminal_emitted: false, }; - let items: Vec = futures::stream::unfold(state, sse_next).collect().await; - - let failed = items - .iter() - .find_map(|item| match item { - ModelStreamItem::ProviderFailed(error) => Some(error), - _ => None, - }) - .expect("invalid arguments should emit ProviderFailed"); - assert_eq!(failed.code.as_deref(), Some("invalid_tool_arguments")); - assert!(!failed.retryable); - assert!(failed.message.contains("call-bad"), "{}", failed.message); - assert!(failed.message.contains("lookup"), "{}", failed.message); - - let mut merged = StreamAccumulator::new(); - for item in &items { - merged.push(item); - } - let err = merged - .finish() - .expect_err("provider failure must reach accumulator"); - assert!(err.to_string().contains("invalid_tool_arguments")); + futures::stream::unfold(state, sse_next).collect().await } -/// Drives an SSE byte stream through the parser and returns every item. -async fn collect_sse(raw: Vec>) -> Vec { +/// Like [`collect_sse`], but with a specific inline reasoning-tag extraction +/// config (`None` disables inline extraction). +async fn collect_sse_with( + raw: Vec>, + reasoning_tags: Option, +) -> Vec { use futures::StreamExt; let bytes = futures::stream::iter( @@ -729,7 +853,7 @@ async fn collect_sse(raw: Vec>) -> Vec { bytes: Box::pin(bytes), buf: Vec::new(), pending: std::collections::VecDeque::new(), - acc: OpenAiStreamAcc::default(), + acc: OpenAiStreamAcc::new(reasoning_tags), provider: "openai".to_string(), model: "gpt-4.1-mini".to_string(), started: false, @@ -739,6 +863,194 @@ async fn collect_sse(raw: Vec>) -> Vec { futures::stream::unfold(state, sse_next).collect().await } +/// Concatenates the reasoning fragments across every streamed `MessageDelta`. +fn stream_reasoning(items: &[ModelStreamItem]) -> String { + items + .iter() + .filter_map(|item| match item { + ModelStreamItem::MessageDelta(delta) => Some(delta.reasoning.clone()), + _ => None, + }) + .collect() +} + +/// Concatenates the text of every leading `Thinking` block on a response. +fn response_reasoning(response: &ModelResponse) -> String { + response + .message + .content + .iter() + .filter_map(|block| block.as_thinking().map(|(text, _)| text.to_string())) + .collect() +} + +/// Concatenates the visible-text fragments across every streamed `MessageDelta`. +fn stream_text(items: &[ModelStreamItem]) -> String { + items + .iter() + .filter_map(|item| match item { + ModelStreamItem::MessageDelta(delta) => Some(delta.text.clone()), + _ => None, + }) + .collect() +} + +/// Builds a one-content-delta SSE body carrying `content`, then a stop chunk. +fn content_chunks(deltas: &[&str]) -> Vec> { + let mut raw: Vec> = deltas + .iter() + .map(|d| { + format!( + "data: {}\n\n", + json!({ "choices": [{ "delta": { "content": d } }] }) + ) + .into_bytes() + }) + .collect(); + raw.push(b"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n".to_vec()); + raw.push(b"data: [DONE]\n\n".to_vec()); + raw +} + +#[tokio::test] +async fn sse_stream_extracts_inline_think_tags() { + let items = collect_sse_with( + content_chunks(&["reasoning here", "the answer"]), + Some(ReasoningTagExtraction::default()), + ) + .await; + + // Live deltas keep chain-of-thought off the visible channel. + assert_eq!(stream_reasoning(&items), "reasoning here"); + assert_eq!(stream_text(&items), "the answer"); + + // The terminal response carries a leading Thinking block + clean text. + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged.finish().unwrap(); + assert_eq!(response.text(), "the answer"); + assert_eq!(response_reasoning(&response), "reasoning here"); + assert_eq!( + response.message.content.first(), + Some(&crate::harness::message::ContentBlock::Thinking { + text: "reasoning here".into(), + signature: None, + }) + ); +} + +#[tokio::test] +async fn sse_stream_inline_think_tag_split_across_deltas() { + // The opening tag is split across three network chunks: ``. + let items = collect_sse_with( + content_chunks(&["beforesecretafter"]), + Some(ReasoningTagExtraction::default()), + ) + .await; + + assert_eq!(stream_reasoning(&items), "secret"); + assert_eq!(stream_text(&items), "beforeafter"); + + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged.finish().unwrap(); + assert_eq!(response.text(), "beforeafter"); + assert_eq!(response_reasoning(&response), "secret"); +} + +#[tokio::test] +async fn sse_stream_disabled_extraction_leaks_think_tags() { + // With extraction disabled the inline tags pass straight through — the + // documented opt-out behavior. + let items = collect_sse_with(content_chunks(&["cotanswer"]), None).await; + + assert_eq!(stream_text(&items), "cotanswer"); + assert_eq!(stream_reasoning(&items), ""); +} + +#[tokio::test] +async fn sse_stream_side_channel_and_inline_reasoning_combine() { + // A response carrying BOTH a side-channel reasoning fragment and an inline + // section: side-channel leads, inline follows, joined by the + // configured separator. + let raw: Vec> = vec![ + b"data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"side\"}}]}\n\n".to_vec(), + b"data: {\"choices\":[{\"delta\":{\"content\":\"inlinedone\"},\"finish_reason\":\"stop\"}]}\n\n".to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + let items = collect_sse_with(raw, Some(ReasoningTagExtraction::default())).await; + + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged.finish().unwrap(); + assert_eq!(response.text(), "done"); + assert_eq!(response_reasoning(&response), "side\ninline"); +} + +#[tokio::test] +async fn sse_stream_start_with_reasoning_deepseek_template() { + // DeepSeek-R1 template: output begins mid-reasoning, only a closing tag. + let items = collect_sse_with( + content_chunks(&["chain of thought", "final"]), + Some(ReasoningTagExtraction::default().with_start_with_reasoning(true)), + ) + .await; + + assert_eq!(stream_reasoning(&items), "chain of thought"); + assert_eq!(stream_text(&items), "final"); + + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged.finish().unwrap(); + assert_eq!(response.text(), "final"); + assert_eq!(response_reasoning(&response), "chain of thought"); +} + +#[test] +fn parse_chat_response_extracts_inline_think_and_side_channel() { + let body = json!({ + "id": "chatcmpl-think", + "choices": [ + { + "message": { + "role": "assistant", + "reasoning_content": "side", + "content": "inline\n\nThe answer is 42." + }, + "finish_reason": "stop" + } + ] + }); + + let cfg = ReasoningTagExtraction::default(); + let response = parse_chat_response(body, Some(&cfg)).unwrap(); + assert_eq!(response.text(), "The answer is 42."); + // Side-channel leads, inline follows, separator-joined. + assert_eq!(response_reasoning(&response), "side\ninline"); +} + +#[test] +fn parse_chat_response_without_config_leaves_inline_tags_in_text() { + let body = json!({ + "id": "chatcmpl-plain", + "choices": [ + { "message": { "role": "assistant", "content": "xy" }, "finish_reason": "stop" } + ] + }); + + let response = parse_chat_response(body, None).unwrap(); + assert_eq!(response.text(), "xy"); + assert_eq!(response_reasoning(&response), ""); +} + #[tokio::test] async fn sse_stream_reassembles_multibyte_char_split_across_chunks() { // A 4-byte emoji in the content payload, split down the middle across two @@ -900,6 +1212,101 @@ async fn sse_stream_correlates_indexless_parallel_tool_calls_by_id() { assert_eq!(calls[1].arguments, json!({ "y": 2 })); } +#[tokio::test] +async fn sse_stream_index_zero_parallel_tool_calls_do_not_merge() { + // Ollama's /v1 endpoint emitted parallel tool calls all carrying `index: 0` + // (ollama/ollama#15457). Two distinct ids at the same index must open two + // separate slots instead of silently merging into one. Interleaved + // continuation fragments (still `index: 0`, carrying the id) must route back + // to the correct already-open slot rather than the occupant of slot 0. + let raw: Vec> = vec![ + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-a\",\"function\":{\"name\":\"alpha\",\"arguments\":\"{\\\"x\\\":\"}}]}}]}\n\n".to_vec(), + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-b\",\"function\":{\"name\":\"beta\",\"arguments\":\"{\\\"y\\\":\"}}]}}]}\n\n".to_vec(), + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-a\",\"function\":{\"arguments\":\"1}\"}}]}}]}\n\n".to_vec(), + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-b\",\"function\":{\"arguments\":\"2}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n".to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + + let items = collect_sse(raw).await; + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged.finish().unwrap(); + let calls = response.tool_calls(); + assert_eq!( + calls.len(), + 2, + "index-0 parallel calls must not merge: {calls:?}" + ); + assert_eq!(calls[0].id, "call-a"); + assert_eq!(calls[0].name, "alpha"); + assert_eq!(calls[0].arguments, json!({ "x": 1 })); + assert_eq!(calls[1].id, "call-b"); + assert_eq!(calls[1].name, "beta"); + assert_eq!(calls[1].arguments, json!({ "y": 2 })); +} + +#[tokio::test] +async fn sse_stream_index_zero_id_less_continuations_follow_latest_call() { + // Worst-case combination of two local-server defects: parallel calls all + // reuse `index: 0` (ollama/ollama#15457) AND continuation fragments carry + // no id (the standard OpenAI streaming shape). An id-less continuation + // under a reused index must follow the call most recently opened at that + // index — not the original occupant of slot 0. + let raw: Vec> = vec![ + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-a\",\"function\":{\"name\":\"alpha\",\"arguments\":\"{\\\"x\\\":1}\"}}]}}]}\n\n".to_vec(), + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-b\",\"function\":{\"name\":\"beta\",\"arguments\":\"{\\\"y\\\":\"}}]}}]}\n\n".to_vec(), + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"2}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n".to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + + let items = collect_sse(raw).await; + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged.finish().unwrap(); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 2, "expected two distinct calls: {calls:?}"); + assert_eq!(calls[0].id, "call-a"); + assert_eq!(calls[0].arguments, json!({ "x": 1 })); + assert!(calls[0].invalid.is_none(), "call-a must stay valid"); + assert_eq!(calls[1].id, "call-b"); + assert_eq!( + calls[1].arguments, + json!({ "y": 2 }), + "id-less continuation must land on call-b, the latest call at index 0" + ); + assert!(calls[1].invalid.is_none(), "call-b must stay valid"); +} + +#[tokio::test] +async fn sse_stream_empty_name_continuation_never_overwrites_recorded_name() { + // LM Studio (lmstudio-bug-tracker#649) can send a later tool-call fragment + // whose `function.name` is an empty string. It must never clobber the name + // already recorded from the call-opening fragment. + let raw: Vec> = vec![ + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call-1\",\"function\":{\"name\":\"lookup\",\"arguments\":\"{\\\"q\\\":\"}}]}}]}\n\n".to_vec(), + b"data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"name\":\"\",\"arguments\":\"1}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n".to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + + let items = collect_sse(raw).await; + let mut merged = StreamAccumulator::new(); + for item in &items { + merged.push(item); + } + let response = merged.finish().unwrap(); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 1); + assert_eq!( + calls[0].name, "lookup", + "an empty-name continuation must not overwrite the recorded name" + ); + assert_eq!(calls[0].arguments, json!({ "q": 1 })); +} + #[tokio::test] async fn sse_stream_indexless_fallback_ids_match_between_delta_and_final() { // No `index` and no `id` at all (arguments-only continuation on the same @@ -1331,18 +1738,21 @@ fn recovers_first_call_when_fragments_are_concatenated() { } #[test] -fn still_fails_fast_on_genuinely_malformed_tool_args() { +fn marks_genuinely_malformed_tool_args_invalid_instead_of_hanging() { // The exact corruption seen in the wild carries a stray `]` *inside* the - // JSON — not just a leaked delimiter — so it cannot be safely repaired. It - // must fail fast (a clear, non-retryable model error) rather than hang. - let err = parse_response(tool_call_body( + // JSON — not just a leaked delimiter — so it cannot be safely repaired. + // Rather than fail the call (or hang on a never-resolving one), it is + // surfaced as an `invalid` ToolCall the agent loop can bounce back to the + // model with a clear error. + let response = parse_response(tool_call_body( r#"{"arguments":{"body":"hi"]}}"#, )) - .expect_err("unrepairable args must fail closed"); - assert!(matches!(err, TinyAgentsError::Model(_)), "got {err:?}"); - let message = err.to_string(); - assert!(message.contains("call-1"), "{message}"); - assert!(message.contains("raw arguments"), "{message}"); + .expect("unrepairable args must resolve as an invalid call, not fail"); + let calls = response.tool_calls(); + assert_eq!(calls.len(), 1); + let reason = calls[0].invalid.as_deref().expect("call marked invalid"); + assert!(reason.contains("call-1"), "{reason}"); + assert!(reason.contains("raw arguments"), "{reason}"); } #[test] @@ -1475,3 +1885,180 @@ fn merge_provider_options_prefers_overrides_and_handles_nulls() { json!(["top_k", 40]) ); } + +// --------------------------------------------------------------------------- +// Local-server request-shape degradation (named tool_choice, json_object) +// --------------------------------------------------------------------------- + +#[test] +fn degrades_named_tool_choice_to_required_and_filters_tools() { + // The endpoint rejects the object form: send `"required"` and drop every + // tool except the named one so the model has no other tool to pick. + let model = OpenAiModel::new("k") + .with_model("m") + .with_named_tool_choice(false); + let request = ModelRequest::new(vec![Message::user("hi")]) + .with_tools(vec![ + ToolSchema::new("get_weather", "w", json!({})), + ToolSchema::new("get_time", "t", json!({})), + ]) + .with_tool_choice(ToolChoice::Tool("get_weather".to_string())); + + let value = serde_json::to_value(model.translate_request(&request).unwrap()).unwrap(); + + assert_eq!(value["tool_choice"], json!("required")); + let tools = value["tools"].as_array().unwrap(); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0]["function"]["name"], json!("get_weather")); +} + +#[test] +fn degraded_named_tool_choice_keeps_tools_when_named_tool_absent() { + // Named tool not in the list: leave `tools` intact but still send "required". + let model = OpenAiModel::new("k") + .with_model("m") + .with_named_tool_choice(false); + let request = ModelRequest::new(vec![Message::user("hi")]) + .with_tools(vec![ToolSchema::new("get_time", "t", json!({}))]) + .with_tool_choice(ToolChoice::Tool("missing".to_string())); + + let value = serde_json::to_value(model.translate_request(&request).unwrap()).unwrap(); + + assert_eq!(value["tool_choice"], json!("required")); + let tools = value["tools"].as_array().unwrap(); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0]["function"]["name"], json!("get_time")); +} + +#[test] +fn degrades_json_object_to_permissive_json_schema() { + let model = OpenAiModel::new("k") + .with_model("m") + .with_json_object_format(false); + let request = ModelRequest::new(vec![Message::user("hi")]) + .with_response_format(ResponseFormat::JsonObject); + + let value = serde_json::to_value(model.translate_request(&request).unwrap()).unwrap(); + + assert_eq!( + value["response_format"], + json!({ + "type": "json_schema", + "json_schema": { + "name": "json_object", + "schema": { "type": "object" }, + "strict": false, + } + }) + ); +} + +#[test] +fn baseline_knobs_default_to_supported_wire_shapes() { + // Default knobs preserve the hosted-OpenAI object/json_object wire shapes, so + // an accidental default flip would fail here. + let model = model(); + assert_eq!(model.baseline_degrade(), Degrade::default()); + + let request = ModelRequest::new(vec![Message::user("hi")]) + .with_tools(vec![ToolSchema::new("t", "d", json!({}))]) + .with_tool_choice(ToolChoice::Tool("t".to_string())) + .with_response_format(ResponseFormat::JsonObject); + let value = serde_json::to_value(model.translate_request(&request).unwrap()).unwrap(); + + assert_eq!( + value["tool_choice"], + json!({ "type": "function", "function": { "name": "t" } }) + ); + assert_eq!(value["response_format"], json!({ "type": "json_object" })); +} + +#[test] +fn degrade_for_400_targets_only_the_shape_the_request_used() { + let named = ModelRequest::new(vec![Message::user("hi")]) + .with_tools(vec![ToolSchema::new("t", "d", json!({}))]) + .with_tool_choice(ToolChoice::Tool("t".to_string())); + assert_eq!( + degrade_for_400( + "Invalid tool_choice type: 'object'. Supported string values: none, auto, required", + &named, + Degrade::default(), + ), + Some(Degrade { + named_tool_choice: true, + json_object: false, + }) + ); + + let json = ModelRequest::new(vec![Message::user("hi")]) + .with_response_format(ResponseFormat::JsonObject); + assert_eq!( + degrade_for_400( + "'response_format.type' must be 'json_schema' or 'text'", + &json, + Degrade::default(), + ), + Some(Degrade { + named_tool_choice: false, + json_object: true, + }) + ); +} + +#[test] +fn degrade_for_400_ignores_unrelated_or_already_degraded_failures() { + let named = ModelRequest::new(vec![Message::user("hi")]) + .with_tools(vec![ToolSchema::new("t", "d", json!({}))]) + .with_tool_choice(ToolChoice::Tool("t".to_string())); + + // A tool_choice message but the request used `Required` (not a named tool): + // there is nothing to degrade, so no retry. + let required = ModelRequest::new(vec![Message::user("hi")]) + .with_tools(vec![ToolSchema::new("t", "d", json!({}))]) + .with_tool_choice(ToolChoice::Required); + assert_eq!( + degrade_for_400("Invalid tool_choice type", &required, Degrade::default()), + None + ); + + // An unrelated 400 never triggers a degraded retry. + assert_eq!( + degrade_for_400("context length exceeded", &named, Degrade::default()), + None + ); + + // Already degraded on the first attempt -> no repeat (prevents a retry loop). + assert_eq!( + degrade_for_400( + "Invalid tool_choice type", + &named, + Degrade { + named_tool_choice: true, + json_object: false, + }, + ), + None + ); +} + +#[test] +fn degrade_for_400_unions_with_existing_baseline_degrade() { + // A json_object 400 while the named-tool-choice degrade is already baked on + // keeps that baseline flag set for the retry. + let json = ModelRequest::new(vec![Message::user("hi")]) + .with_response_format(ResponseFormat::JsonObject); + assert_eq!( + degrade_for_400( + "response_format must be json_schema", + &json, + Degrade { + named_tool_choice: true, + json_object: false, + }, + ), + Some(Degrade { + named_tool_choice: true, + json_object: true, + }) + ); +} diff --git a/src/harness/providers/openai/transport.rs b/src/harness/providers/openai/transport.rs index 9623d16..dbec9bc 100644 --- a/src/harness/providers/openai/transport.rs +++ b/src/harness/providers/openai/transport.rs @@ -54,6 +54,18 @@ pub struct OpenAiModel { /// the `system` role is dropped — for OpenAI-compatible endpoints that reject /// a `system` role. `false` by default (system messages pass through). merge_system_into_user: bool, + /// Whether the endpoint accepts a **named** `tool_choice` + /// (`{"type":"function","function":{"name":…}}`). `true` by default. When + /// `false`, a [`ToolChoice::Tool`] request is degraded to + /// `tool_choice:"required"` with the `tools` array filtered to the named tool + /// — some local runtimes (LM Studio, llama.cpp server) 400 on the object form. + /// See [`Self::with_named_tool_choice`]. + named_tool_choice_supported: bool, + /// Whether the endpoint accepts `response_format:{"type":"json_object"}`. + /// `true` by default. When `false`, a [`ResponseFormat::JsonObject`] request is + /// degraded to a permissive `json_schema` wire form — some local runtimes 400 + /// on `json_object`. See [`Self::with_json_object_format`]. + json_object_format_supported: bool, /// Default model id used when a request does not override it. model: String, /// Provider family identifier used in profiles and normalized errors. @@ -81,6 +93,19 @@ pub struct OpenAiModel { /// A `User-Agent` header override (e.g. the Codex CLI UA). `None` uses /// reqwest's default. See [`Self::with_user_agent`]. user_agent: Option, + /// Inline `…` reasoning-tag extraction config. `Some` moves + /// inline chain-of-thought onto the reasoning channel; `None` passes content + /// through untouched. Until [`Self::with_reasoning_tag_extraction`] is + /// called (`reasoning_tags_overridden == false`) this default only takes + /// effect for non-hosted base URLs — see + /// [`Self::effective_reasoning_tags`]. + reasoning_tags: Option, + /// Whether [`Self::with_reasoning_tag_extraction`] was called explicitly. + /// When `false`, extraction auto-enables only for OpenAI-compatible + /// endpoints (`base_url != DEFAULT_BASE_URL`): hosted OpenAI never emits + /// inline `` reasoning, and unconditional extraction would silently + /// strip legitimate content that mentions a literal `` tag. + reasoning_tags_overridden: bool, } /// The auth headers `(name, value)` for a given [`AuthStyle`] + credential. @@ -288,6 +313,8 @@ impl OpenAiModel { temperature_unsupported: Vec::new(), temperature_override: None, merge_system_into_user: false, + named_tool_choice_supported: true, + json_object_format_supported: true, model: DEFAULT_MODEL.to_string(), provider: "openai".to_string(), base_url: DEFAULT_BASE_URL.to_string(), @@ -297,6 +324,13 @@ impl OpenAiModel { responses_omit_max_output_tokens: false, extra_query_params: Vec::new(), user_agent: None, + // Inline `` extraction defaults ON, but until overridden it + // only takes effect for non-hosted base URLs (see + // `effective_reasoning_tags`): unhandled leakage is the common + // local-model failure, while hosted OpenAI never emits inline + // `` and must not strip literal mentions of the tag. + reasoning_tags: Some(ReasoningTagExtraction::default()), + reasoning_tags_overridden: false, } } @@ -378,6 +412,69 @@ impl OpenAiModel { self } + /// Declares whether the endpoint accepts a **named** `tool_choice` + /// (`{"type":"function","function":{"name":…}}`). `true` by default. + /// + /// Pass `false` for local OpenAI-compatible runtimes (LM Studio, llama.cpp + /// server, and others) that only accept the string forms `none`/`auto`/ + /// `required` and 400 on the object form. A [`ToolChoice::Tool`] request is + /// then degraded to `tool_choice:"required"` with the wire `tools` array + /// filtered down to just the named tool, preserving the "must call *this* + /// tool" semantics. Independent of this flag, a 400 whose body implicates + /// `tool_choice` triggers the same degraded retry automatically (once). + pub fn with_named_tool_choice(mut self, supported: bool) -> Self { + self.named_tool_choice_supported = supported; + self + } + + /// Declares whether the endpoint accepts + /// `response_format:{"type":"json_object"}`. `true` by default. + /// + /// Pass `false` for local OpenAI-compatible runtimes that only accept + /// `json_schema`/`text` and 400 on `json_object`. A + /// [`ResponseFormat::JsonObject`] request is then degraded to a permissive + /// `json_schema` wire form (an empty object schema with `strict:false`). + /// Independent of this flag, a 400 whose body implicates `response_format` + /// triggers the same degraded retry automatically (once). + pub fn with_json_object_format(mut self, supported: bool) -> Self { + self.json_object_format_supported = supported; + self + } + + /// Configures inline `…` reasoning-tag extraction for + /// OpenAI-compatible reasoning models that embed chain-of-thought in the + /// visible `content` string (qwen3, deepseek-r1 distills via Ollama `/v1`, + /// LM Studio, llama.cpp) rather than on the `reasoning_content` / + /// `reasoning` side-channel. + /// + /// Until this method is called, extraction defaults ON with the plain + /// `think` tag for **non-hosted base URLs only** (hosted OpenAI never emits + /// inline `` reasoning, and extraction there would silently strip + /// legitimate content that mentions a literal tag). Pass `None` to disable + /// it everywhere (content passes through verbatim), or `Some(config)` to + /// force it on — including for the hosted base URL — and customize the tag + /// name, separator, or DeepSeek-R1 template mode via + /// [`ReasoningTagExtraction`]. Extracted reasoning surfaces as a leading + /// [`ContentBlock::Thinking`](crate::harness::message::ContentBlock::Thinking) + /// block on both the unary and streamed paths, consistent with the + /// side-channel normalization. + pub fn with_reasoning_tag_extraction(mut self, config: Option) -> Self { + self.reasoning_tags = config; + self.reasoning_tags_overridden = true; + self + } + + /// The reasoning-tag extraction config in effect for a call: the explicit + /// [`Self::with_reasoning_tag_extraction`] override when one was given, + /// otherwise the built-in `think` default gated to OpenAI-compatible + /// endpoints (`base_url != DEFAULT_BASE_URL`). + pub(super) fn effective_reasoning_tags(&self) -> Option<&ReasoningTagExtraction> { + if !self.reasoning_tags_overridden && self.base_url == DEFAULT_BASE_URL { + return None; + } + self.reasoning_tags.as_ref() + } + /// Overrides whether this model advertises **native** tool calling on its /// [`profile`](ChatModel::profile). Many self-hosted / local OpenAI-compatible /// runtimes (Ollama and others) reject the OpenAI `tools` parameter with an @@ -678,12 +775,37 @@ impl OpenAiModel { &self.base_url } + /// The baseline request-shape degradations to apply for this instance, + /// derived from its capability knobs. A `true` field means "degrade this + /// shape on the wire". + pub(super) fn baseline_degrade(&self) -> Degrade { + Degrade { + named_tool_choice: !self.named_tool_choice_supported, + json_object: !self.json_object_format_supported, + } + } + /// Translates a provider-neutral [`ModelRequest`] into the OpenAI wire - /// request body. The per-request `model` override wins over the instance - /// default. + /// request body, applying this instance's baseline request-shape + /// degradations (see [`Self::baseline_degrade`]). + /// + /// Test-only: production paths go through [`Self::build_chat_body`] / + /// [`Self::post_chat_with_degrade`], which thread an explicit [`Degrade`]. + #[cfg(test)] pub(super) fn translate_request( &self, request: &ModelRequest, + ) -> Result { + self.translate_request_with(request, self.baseline_degrade()) + } + + /// Translates a provider-neutral [`ModelRequest`] into the OpenAI wire + /// request body, applying the given request-shape `degrade`. The per-request + /// `model` override wins over the instance default. + pub(super) fn translate_request_with( + &self, + request: &ModelRequest, + degrade: Degrade, ) -> Result { // Prompt-guided tools: a model without native tool calling that is still // handed tools gets the tool protocol embedded in its system prompt and no @@ -713,7 +835,7 @@ impl OpenAiModel { .map(translate_message) .collect::>>()?; - let tools: Vec = if prompt_guided_tools { + let mut tools: Vec = if prompt_guided_tools { Vec::new() } else { request @@ -733,14 +855,32 @@ impl OpenAiModel { // tool_choice is only meaningful when tools are declared. let tool_choice = if tools.is_empty() { None + } else if let (true, ToolChoice::Tool(name)) = + (degrade.named_tool_choice, &request.tool_choice) + { + // The endpoint rejects a named `tool_choice` object. Preserve the + // "must call *this* tool" semantics by sending `"required"` and, when + // the named tool is actually declared, filtering the wire `tools` + // down to just it so the model has no other tool to pick. If the named + // tool is absent, leave `tools` intact (mirrors the un-degraded path, + // which would also send an unmatched name) and still send "required". + if tools.iter().any(|t| t.function.name == *name) { + tools.retain(|t| t.function.name == *name); + } + Some(json!("required")) } else { Some(translate_tool_choice(&request.tool_choice)) }; - let response_format = request - .response_format - .as_ref() - .and_then(translate_response_format); + let response_format = request.response_format.as_ref().and_then(|format| { + if degrade.json_object && matches!(format, ResponseFormat::JsonObject) { + // The endpoint rejects `{"type":"json_object"}`; use a permissive + // `json_schema` that still guarantees a JSON object. + Some(degraded_json_object_format()) + } else { + translate_response_format(format) + } + }); let model = request.model.clone().unwrap_or_else(|| self.model.clone()); // The o-series reasoning models reject `max_tokens` and require @@ -931,6 +1071,57 @@ impl OpenAiModel { self.send_checked(builder, what, &url).await } + /// Builds the chat-completions wire body for `request` under the given + /// `degrade`, setting the streaming fields when `streaming` is `true`. + fn build_chat_body( + &self, + request: &ModelRequest, + degrade: Degrade, + streaming: bool, + ) -> Result { + let mut body = self.translate_request_with(request, degrade)?; + if streaming { + body.stream = true; + body.stream_options = Some(json!({ "include_usage": true })); + } + Ok(body) + } + + /// Posts a chat-completions request with automatic single-shot degraded + /// retry for local-server request-shape rejections. + /// + /// The first attempt applies this instance's baseline degradations. If it + /// fails with an HTTP 400 whose body implicates a named `tool_choice` or a + /// `json_object` `response_format` that the request actually used — and that + /// shape was not already degraded — the request is rebuilt with that shape + /// degraded and sent exactly once more. Any other error (or a request that + /// used neither shape) surfaces unchanged. Shared by [`Self::invoke`] and + /// [`Self::stream`], so the retry covers the streaming path too. + async fn post_chat_with_degrade( + &self, + request: &ModelRequest, + streaming: bool, + what: &str, + ) -> Result { + let baseline = self.baseline_degrade(); + let body = self.build_chat_body(request, baseline, streaming)?; + match self + .post_json(&body, request.timeout_ms, streaming, what) + .await + { + Ok(response) => Ok(response), + Err(TinyAgentsError::Provider(err)) + if err.status == Some(400) + && let Some(degrade) = degrade_for_400(&err.message, request, baseline) => + { + let retry = self.build_chat_body(request, degrade, streaming)?; + self.post_json(&retry, request.timeout_ms, streaming, what) + .await + } + Err(e) => Err(e), + } + } + fn provider_error( &self, message: impl Into, @@ -992,6 +1183,57 @@ impl OpenAiModel { } } +/// Request-shape degradations to apply when building an OpenAI wire body. +/// +/// Each field, when `true`, replaces a request shape that some local +/// OpenAI-compatible servers reject with an equivalent one they accept. The +/// baseline comes from the instance's capability knobs +/// ([`OpenAiModel::baseline_degrade`]) from the instance's capability knobs; a +/// 400 whose error body implicates one of these shapes turns the corresponding +/// field on for a single degraded retry (see [`degrade_for_400`]). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct Degrade { + /// Degrade a named `tool_choice` to `"required"` + a filtered `tools` array. + pub named_tool_choice: bool, + /// Degrade `response_format:{"type":"json_object"}` to a permissive + /// `json_schema`. + pub json_object: bool, +} + +/// Computes the additional degradation to apply after an HTTP 400, or `None` +/// when the failure is not an auto-degradable request-shape rejection. +/// +/// Returns `Some(degrade)` only when the 400 error body implicates a shape the +/// request actually used *and* that shape was not already degraded on the +/// original attempt — so a degraded retry is issued at most once and only when +/// it could plausibly help. The returned [`Degrade`] is the union of `already` +/// and the newly implicated shape, so the retry keeps any baseline degradations. +/// +/// Pure, so the 400-detection policy is unit-testable without a network call. +pub(super) fn degrade_for_400( + message: &str, + request: &ModelRequest, + already: Degrade, +) -> Option { + let lower = message.to_ascii_lowercase(); + let mut degrade = already; + + if !already.named_tool_choice + && lower.contains("tool_choice") + && matches!(request.tool_choice, ToolChoice::Tool(_)) + { + degrade.named_tool_choice = true; + } + if !already.json_object + && lower.contains("response_format") + && matches!(request.response_format, Some(ResponseFormat::JsonObject)) + { + degrade.json_object = true; + } + + (degrade != already).then_some(degrade) +} + /// Resolves the per-request timeout to apply to an outbound HTTP call. /// /// An explicit [`ModelRequest::timeout_ms`] always wins. Otherwise a unary call @@ -1091,10 +1333,8 @@ impl ChatModel for OpenAiModel { if self.responses_api_primary { return self.invoke_responses(&request).await; } - let body = self.translate_request(&request)?; - let response = self - .post_json(&body, request.timeout_ms, false, "request") + .post_chat_with_degrade(&request, false, "request") .await?; let text = response.text().await.map_err(|e| { @@ -1102,7 +1342,7 @@ impl ChatModel for OpenAiModel { })?; let value: Value = serde_json::from_str(&text)?; - let response = parse_response(value)?; + let response = parse_chat_response(value, self.effective_reasoning_tags())?; // Prompt-guided tools: recover the model's `` blocks into // `message.tool_calls` when native tool calling was suppressed. if !self.profile.tool_calling && !request.tools.is_empty() { @@ -1149,12 +1389,8 @@ impl ChatModel for OpenAiModel { ]; return Ok(Box::pin(futures::stream::iter(items))); } - let mut body = self.translate_request(&request)?; - body.stream = true; - body.stream_options = Some(json!({ "include_usage": true })); - let response = self - .post_json(&body, request.timeout_ms, true, "stream request") + .post_chat_with_degrade(&request, true, "stream request") .await?; // Leniency: some OpenAI-compatible endpoints (and test mocks) ignore @@ -1174,7 +1410,7 @@ impl ChatModel for OpenAiModel { TinyAgentsError::Model(format!("openai non-stream stream-body read failed: {e}")) })?; let value: Value = serde_json::from_str(&text)?; - let mut parsed = parse_response(value)?; + let mut parsed = parse_chat_response(value, self.effective_reasoning_tags())?; if !self.profile.tool_calling && !request.tools.is_empty() { parsed = prompt_tools::apply_to_response(parsed); } @@ -1203,7 +1439,7 @@ impl ChatModel for OpenAiModel { bytes: Box::pin(bytes), buf: Vec::new(), pending: VecDeque::new(), - acc: OpenAiStreamAcc::default(), + acc: OpenAiStreamAcc::new(self.effective_reasoning_tags().cloned()), provider: self.provider.clone(), model: self.model.clone(), started: false, diff --git a/src/harness/providers/openai/types.rs b/src/harness/providers/openai/types.rs index 66fdd3c..a352252 100644 --- a/src/harness/providers/openai/types.rs +++ b/src/harness/providers/openai/types.rs @@ -140,10 +140,53 @@ pub struct FunctionChunkWire { #[serde(default)] pub name: Option, /// Incremental stringified-JSON arguments fragment. - #[serde(default)] + /// + /// OpenAI streams this as a string fragment, but some OpenAI-compatible + /// local servers emit the completed arguments as a JSON **object** in a + /// single delta. [`deserialize_optional_arguments`] normalizes either shape + /// to the stringified form the accumulator concatenates. + #[serde(default, deserialize_with = "deserialize_optional_arguments")] pub arguments: Option, } +/// Normalizes a tool-call `function.arguments` payload to the stringified-JSON +/// form the provider expects. +/// +/// OpenAI sends `arguments` as a JSON **string**, but OpenAI-compatible local +/// servers (LM Studio, some vLLM/llama.cpp builds) send it as a JSON **object** +/// instead, or omit it entirely. A single unexpected shape used to fail +/// deserialization of the *whole* response, so the model appeared "broken". +/// This accepts a string, an object/array (re-serialized to its JSON text), or +/// a null/absent value (an empty string). +fn deserialize_arguments<'de, D>(deserializer: D) -> std::result::Result +where + D: serde::Deserializer<'de>, +{ + Ok(match Option::::deserialize(deserializer)? { + None | Some(Value::Null) => String::new(), + Some(Value::String(text)) => text, + // An object/array (or any non-string scalar) is re-serialized to its + // canonical JSON text so downstream string parsing is uniform. + Some(other) => other.to_string(), + }) +} + +/// The streaming counterpart of [`deserialize_arguments`], preserving the +/// `Option` wrapper used by incremental deltas (a fragment that carries no +/// `arguments` key stays `None` rather than becoming an empty string). +fn deserialize_optional_arguments<'de, D>( + deserializer: D, +) -> std::result::Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + Ok(match Option::::deserialize(deserializer)? { + None | Some(Value::Null) => None, + Some(Value::String(text)) => Some(text), + Some(other) => Some(other.to_string()), + }) +} + /// The `content` of a [`ChatMessageWire`]. /// /// OpenAI accepts either a plain string (the common case) or an array of typed @@ -228,9 +271,18 @@ pub struct FunctionSchemaWire { #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct ToolCallWire { /// Provider-assigned call id. + /// + /// Defaulted because local servers violate the OpenAI contract: Ollama's + /// `/v1` endpoint omitted `id` entirely until v0.12.11. An empty string is + /// treated as absent by the translator, which synthesizes a stable + /// `tool-{index}` fallback so result correlation still works. + #[serde(default)] pub id: String, /// Always `"function"`. - #[serde(rename = "type")] + /// + /// Defaulted because some OpenAI-compatible servers omit `type`; a single + /// missing field used to fail deserialization of the whole response. + #[serde(rename = "type", default)] pub kind: String, /// The function name and stringified-JSON arguments. pub function: FunctionCallWire, @@ -240,8 +292,15 @@ pub struct ToolCallWire { #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct FunctionCallWire { /// Function name. + #[serde(default)] pub name: String, /// Arguments encoded as a JSON **string** (OpenAI sends stringified JSON). + /// + /// Some OpenAI-compatible local servers send a JSON **object** instead of a + /// string, or omit the field; [`deserialize_arguments`] normalizes any of + /// those shapes to the stringified form so one non-conforming field can no + /// longer fail the whole response. + #[serde(default, deserialize_with = "deserialize_arguments")] pub arguments: String, } diff --git a/src/harness/runtime/types.rs b/src/harness/runtime/types.rs index 1164809..12420c9 100644 --- a/src/harness/runtime/types.rs +++ b/src/harness/runtime/types.rs @@ -178,6 +178,24 @@ pub struct RunPolicy { /// rely on empty finals; opt in to turn a silent blank success into a typed /// error the caller can re-prompt on. pub error_on_empty_response: bool, + /// Number of automatic retries when a model call returns a *truncated + /// empty* completion — `finish_reason == "length"` with no visible text, no + /// tool calls, and no structured output. + /// + /// This is the failure mode of local reasoning models (for example + /// `qwen3` via Ollama) that intermittently spend the entire token budget on + /// the hidden reasoning channel and emit nothing usable. Because such a + /// response is useless to *every* caller and the failure is stochastic, + /// retrying — with a doubled token budget when the request set one (capped + /// at 4x the original), or a plain retry when it did not — is strictly + /// better than surfacing a blank success. + /// + /// The retry runs *before* [`Self::error_on_empty_response`]; only once + /// these retries are exhausted does that guard (if enabled) apply. + /// + /// Defaults to `1` (one retry, two attempts total). Set to `0` to disable + /// for exact-replay callers that must not re-issue a call. + pub truncated_empty_retries: u32, } impl Default for RunPolicy { @@ -198,6 +216,10 @@ impl Default for RunPolicy { }, // Opt-in: preserve the historical blank-final behavior by default. error_on_empty_response: false, + // On by default: a truncated-empty completion is useless to every + // caller, so one stochastic-failure retry is strictly better than a + // blank final. + truncated_empty_retries: 1, } } } diff --git a/src/harness/structured/test.rs b/src/harness/structured/test.rs index 8a5be61..cf0b0c6 100644 --- a/src/harness/structured/test.rs +++ b/src/harness/structured/test.rs @@ -72,6 +72,7 @@ fn tool_call_strategy_reads_matching_call() { id: "tc-1".to_string(), name: "extract_answer".to_string(), arguments: json!({"answer": "yes"}), + invalid: None, }); let output = extractor.extract(&response).unwrap(); assert_eq!(output.value["answer"], "yes"); diff --git a/src/harness/tool/mod.rs b/src/harness/tool/mod.rs index 297b46e..d8258f5 100644 --- a/src/harness/tool/mod.rs +++ b/src/harness/tool/mod.rs @@ -112,8 +112,33 @@ impl ToolCall { id: id.into(), name: name.into(), arguments, + invalid: None, } } + + /// Creates a tool call the provider could not parse: `raw` (the unparseable + /// arguments string) is preserved as the arguments value and `reason` + /// records why parsing failed. The agent loop feeds `reason` back to the + /// model as an error tool result instead of failing the run. + pub fn invalid( + id: impl Into, + name: impl Into, + raw: impl Into, + reason: impl Into, + ) -> Self { + Self { + id: id.into(), + name: name.into(), + arguments: Value::String(raw.into()), + invalid: Some(reason.into()), + } + } + + /// Returns `true` when the model emitted unparseable arguments for this call + /// (see [`Self::invalid`]). + pub fn is_invalid(&self) -> bool { + self.invalid.is_some() + } } impl ToolResult { diff --git a/src/harness/tool/types.rs b/src/harness/tool/types.rs index 3edb7b5..971a1cb 100644 --- a/src/harness/tool/types.rs +++ b/src/harness/tool/types.rs @@ -70,8 +70,23 @@ pub struct ToolCall { /// Name of the tool to invoke. pub name: String, /// Arguments supplied by the model, as raw JSON. + /// + /// When [`Self::invalid`] is set, this preserves the raw (unparseable) + /// arguments string the model emitted (as a JSON string value) instead of a + /// parsed object. #[serde(default)] pub arguments: Value, + /// Set when the model emitted arguments that could not be parsed as JSON. + /// + /// Small local models (Ollama, LM Studio, llama.cpp, vLLM) occasionally emit + /// malformed argument JSON. Rather than fail the whole model call, the + /// provider surfaces the call with this error message and the raw arguments + /// preserved in [`Self::arguments`]; the agent loop feeds the message back to + /// the model as an error tool result so it can retry (mirroring LangChain's + /// `invalid_tool_calls` and the AI SDK's invalid dynamic tool parts). `None` + /// on a well-formed call. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub invalid: Option, } /// The outcome of executing a [`ToolCall`]. diff --git a/src/repl/session/builtins/batched.rs b/src/repl/session/builtins/batched.rs index 636d7b7..813a23e 100644 --- a/src/repl/session/builtins/batched.rs +++ b/src/repl/session/builtins/batched.rs @@ -124,6 +124,7 @@ pub(super) fn tool_call_batched_impl( id: call_id.as_str().to_string(), name: name.clone(), arguments: arguments.clone(), + invalid: None, }; async move { let start = Instant::now(); diff --git a/src/repl/session/builtins/capabilities.rs b/src/repl/session/builtins/capabilities.rs index 48264dc..01f0f47 100644 --- a/src/repl/session/builtins/capabilities.rs +++ b/src/repl/session/builtins/capabilities.rs @@ -64,6 +64,7 @@ pub(super) fn tool_call_impl( id: call_id.as_str().to_string(), name: tool_name.clone(), arguments: arguments.clone(), + invalid: None, }; emit_call_started(ctx, &call_id, ReplCallKind::Tool, &tool_name); let start = Instant::now(); diff --git a/tests/e2e_harness_provider_contracts.rs b/tests/e2e_harness_provider_contracts.rs index 5f3baee..1fd5cae 100644 --- a/tests/e2e_harness_provider_contracts.rs +++ b/tests/e2e_harness_provider_contracts.rs @@ -446,6 +446,7 @@ fn structured_output_supports_provider_schema_and_tool_fallbacks() { id: "tool-1".into(), name: "score".into(), arguments: json!({ "score": 7 }), + invalid: None, }], usage: None, },