diff --git a/CHANGELOG.md b/CHANGELOG.md index a923c97b..1201f213 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added strict `generate_object` argument validation, root JSON Schema support + for references, composition, constants, arrays, and scalars, and active + generation-timeout propagation through governed LLM clients. +- Added typed HTTP cancellation, transport, and retry-exhaustion status errors, + including the SDK-visible `transport` tool error kind. + +### Changed + +- Required `parallel_task` calls to contain 2-32 independent foreground + branches and reject invalid timeout and partial-success thresholds. +- Kept branch retry ownership inside provider and child runtimes; the parallel + fan-out layer no longer replays failures based on rendered error messages. +- Synchronized the Node `ToolErrorKind` declarations with every Rust wire + variant, including cancellation, partial failure, rate limiting, and + transport failures. + +### Fixed + +- Preserved local `$defs` and `definitions` references when provider-facing + structured-output schemas wrap root arrays or scalar values. +- Preserved typed transport and cancellation failures across OpenAI-compatible + and Anthropic blocking and streaming request paths. + ## [6.5.1] - 2026-07-28 ### Added diff --git a/core/src/agent/llm_invoker.rs b/core/src/agent/llm_invoker.rs index 4523071d..98e9ca86 100644 --- a/core/src/agent/llm_invoker.rs +++ b/core/src/agent/llm_invoker.rs @@ -15,6 +15,7 @@ use crate::llm::{ use async_trait::async_trait; use std::future::Future; use std::sync::Arc; +use std::time::Duration; use tokio::sync::mpsc; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -171,6 +172,18 @@ impl LlmClient for LlmInvoker { self.inner.model_generation_concurrency() } + fn fork_for_session(&self, session_id: &str) -> Option> { + self.inner + .fork_for_session(session_id) + .map(|inner| Arc::new(Self::new(inner, self.invocation.clone())) as Arc) + } + + fn with_active_generation_timeout(&self, timeout: Duration) -> Option> { + self.inner + .with_active_generation_timeout(timeout) + .map(|inner| Arc::new(Self::new(inner, self.invocation.clone())) as Arc) + } + async fn complete( &self, messages: &[Message], @@ -510,4 +523,34 @@ mod tests { vec!["child-session".to_string()] ); } + + #[tokio::test] + async fn governed_client_keeps_provider_session_forking_available() { + let observed_sessions = Arc::new(Mutex::new(Vec::new())); + let client: Arc = Arc::new(SessionBindingClient { + bound_session: None, + observed_sessions: Arc::clone(&observed_sessions), + }); + let invocation = InvocationContext::new( + Arc::::from("parent-run"), + Arc::::from("parent-session"), + CancellationToken::new(), + None, + InvocationGovernance::default(), + ); + let governed: Arc = Arc::new(LlmInvoker::new(client, invocation)); + let forked = governed + .fork_for_session("nested-child-session") + .expect("governed client must preserve provider session forking"); + + forked + .complete(&[Message::user("hello")], None, &[]) + .await + .unwrap(); + + assert_eq!( + *observed_sessions.lock().unwrap(), + vec!["nested-child-session".to_string()] + ); + } } diff --git a/core/src/agent_api/tests.rs b/core/src/agent_api/tests.rs index daca3829..ec993d9e 100644 --- a/core/src/agent_api/tests.rs +++ b/core/src/agent_api/tests.rs @@ -5767,6 +5767,7 @@ async fn test_registered_parallel_task_inherits_final_confirmation_manager() { serde_json::json!({"file_path": "note.txt"}), ), scripted_text_response("child read complete"), + scripted_text_response("control child complete"), ])); let worker = crate::subagent::WorkerAgentSpec::custom( "needs-parent-hitl", @@ -5777,6 +5778,7 @@ async fn test_registered_parallel_task_inherits_final_confirmation_manager() { let opts = SessionOptions::new() .with_llm_client(client) .with_worker_agent(worker) + .with_max_parallel_tasks(1) .with_confirmation_policy(crate::hitl::ConfirmationPolicy::enabled()); let session = agent .session_async(dir.path().to_string_lossy().to_string(), Some(opts)) @@ -5786,12 +5788,20 @@ async fn test_registered_parallel_task_inherits_final_confirmation_manager() { let (_rx, join) = session.tool_with_events( "parallel_task", serde_json::json!({ - "tasks": [{ - "agent": "needs-parent-hitl", - "description": "Read a note", - "prompt": "Read note.txt", - "max_steps": 3 - }] + "tasks": [ + { + "agent": "needs-parent-hitl", + "description": "Read a note", + "prompt": "Read note.txt", + "max_steps": 3 + }, + { + "agent": "needs-parent-hitl", + "description": "Independent control", + "prompt": "Return a short control result.", + "max_steps": 1 + } + ] }), ); @@ -5840,10 +5850,11 @@ async fn test_dynamic_workflow_parallel_explore_can_use_readonly_web_tools() { serde_json::json!({"url": "not-a-url"}), ), scripted_text_response("explore web fetch completed"), + scripted_text_response("independent web review completed"), ])); let opts = SessionOptions::new() .with_llm_client(client) - .with_max_parallel_tasks(2) + .with_max_parallel_tasks(1) .with_manual_delegation_enabled(true); let session = agent .session_async(dir.path().to_string_lossy().to_string(), Some(opts)) @@ -5863,12 +5874,20 @@ async function run(ctx, inputs) { step_id: "web_research", step_name: "parallel_task", input: { - tasks: [{ - agent: "explore", - description: "Fetch web evidence", - prompt: "Use web_fetch on the requested URL and summarize the result.", - max_steps: 3, - }], + tasks: [ + { + agent: "explore", + description: "Fetch web evidence", + prompt: "Use web_fetch on the requested URL and summarize the result.", + max_steps: 3, + }, + { + agent: "explore", + description: "Review web evidence", + prompt: "Return a short independent review.", + max_steps: 1, + }, + ], }, retry: { max_attempts: 1, delay_ms: 0 }, }; @@ -5934,13 +5953,14 @@ async fn test_dynamic_workflow_parallel_deep_research_inherits_parent_permission serde_json::json!({"command": "echo inherited-dynamic-workflow-deep-research"}), ), scripted_text_response("deep-research child bash completed"), + scripted_text_response("deep-research control child completed"), ])); let policy = crate::permissions::PermissionPolicy::new().allow("bash(*)"); let opts = SessionOptions::new() .with_llm_client(client) .with_permission_policy(policy) .with_workspace_backend(services) - .with_max_parallel_tasks(2) + .with_max_parallel_tasks(1) .with_manual_delegation_enabled(true); let session = agent .session_async(dir.path().to_string_lossy().to_string(), Some(opts)) @@ -5960,12 +5980,20 @@ async function run(ctx, inputs) { step_id: "deep_research", step_name: "parallel_task", input: { - tasks: [{ - agent: "deep-research", - description: "Use inherited parent tool policy", - prompt: "Run the harmless bash command requested by this regression test.", - max_steps: 3, - }], + tasks: [ + { + agent: "deep-research", + description: "Use inherited parent tool policy", + prompt: "Run the harmless bash command requested by this regression test.", + max_steps: 3, + }, + { + agent: "deep-research", + description: "Verify inherited policy independently", + prompt: "Return a short control result.", + max_steps: 1, + }, + ], }, retry: { max_attempts: 1, delay_ms: 0 }, }; @@ -6032,12 +6060,13 @@ async fn test_dynamic_workflow_parallel_deep_research_inherits_parent_write_perm }), ), scripted_text_response("deep-research child write completed"), + scripted_text_response("deep-research write control completed"), ])); let policy = crate::permissions::PermissionPolicy::new().allow("write(*)"); let opts = SessionOptions::new() .with_llm_client(client) .with_permission_policy(policy) - .with_max_parallel_tasks(2) + .with_max_parallel_tasks(1) .with_manual_delegation_enabled(true); let session = agent .session_async(dir.path().to_string_lossy().to_string(), Some(opts)) @@ -6057,12 +6086,20 @@ async function run(ctx, inputs) { step_id: "deep_research", step_name: "parallel_task", input: { - tasks: [{ - agent: "deep-research", - description: "Use inherited parent write policy", - prompt: "Write the requested child evidence file.", - max_steps: 3, - }], + tasks: [ + { + agent: "deep-research", + description: "Use inherited parent write policy", + prompt: "Write the requested child evidence file.", + max_steps: 3, + }, + { + agent: "deep-research", + description: "Verify inherited write policy independently", + prompt: "Return a short control result.", + max_steps: 1, + }, + ], }, retry: { max_attempts: 1, delay_ms: 0 }, }; diff --git a/core/src/llm/anthropic.rs b/core/src/llm/anthropic.rs index e36be04e..274e6dcf 100644 --- a/core/src/llm/anthropic.rs +++ b/core/src/llm/anthropic.rs @@ -209,7 +209,17 @@ impl AnthropicClient { )) } } - Err(e) => AttemptOutcome::Fatal(e), + Err(e) => { + if crate::llm::http::is_retryable_http_failure(&e) { + AttemptOutcome::Retryable { + status: reqwest::StatusCode::SERVICE_UNAVAILABLE, + body: format!("network error: {e}"), + retry_after: None, + } + } else { + AttemptOutcome::Fatal(e) + } + } } } }) @@ -351,28 +361,24 @@ impl AnthropicClient { async move { let resp = tokio::select! { _ = cancel_token.cancelled() => { - return AttemptOutcome::Fatal(anyhow::anyhow!("HTTP request cancelled")); + return AttemptOutcome::Fatal(anyhow::Error::new( + crate::llm::HttpClientError::cancelled( + "Anthropic streaming HTTP request", + ), + )); } result = http.post_streaming(url, headers, request_body, cancel_token.clone()) => { match result { Ok(r) => r, Err(e) => { - // A transient network error (timeout, reset, - // mid-flight drop — common on throttled - // endpoints) carries no HTTP status. Retry it - // with backoff like 429/5xx instead of failing - // the turn; a real fatal error still bails. - return if crate::retry::is_transient_error(&e) { + return if crate::llm::http::is_retryable_http_failure(&e) { AttemptOutcome::Retryable { status: reqwest::StatusCode::SERVICE_UNAVAILABLE, body: format!("network error: {e}"), retry_after: None, } } else { - AttemptOutcome::Fatal(anyhow::anyhow!( - "HTTP request failed: {}", - e - )) + AttemptOutcome::Fatal(e.context("HTTP request failed")) }; } } diff --git a/core/src/llm/http.rs b/core/src/llm/http.rs index 8a7181a1..90c49307 100644 --- a/core/src/llm/http.rs +++ b/core/src/llm/http.rs @@ -9,6 +9,55 @@ use std::sync::Arc; use std::time::Duration; use tokio_util::sync::CancellationToken; +/// Typed failures emitted before an HTTP response exists. +/// +/// Retry and tool policies inspect this enum rather than rendered diagnostics. +#[derive(Debug, thiserror::Error)] +pub enum HttpClientError { + #[error("{operation} was cancelled")] + Cancelled { operation: String }, + #[error("{operation} transport failed: {message}")] + Transport { operation: String, message: String }, + #[error("{operation} request was invalid: {message}")] + InvalidRequest { operation: String, message: String }, +} + +impl HttpClientError { + pub fn cancelled(operation: impl Into) -> Self { + Self::Cancelled { + operation: operation.into(), + } + } + + pub fn transport(operation: impl Into, message: impl Into) -> Self { + Self::Transport { + operation: operation.into(), + message: message.into(), + } + } + + fn from_reqwest(operation: &str, error: reqwest::Error) -> Self { + if error.is_builder() { + Self::InvalidRequest { + operation: operation.to_string(), + message: error.to_string(), + } + } else { + Self::transport(operation, error.to_string()) + } + } + + pub fn is_retryable(&self) -> bool { + matches!(self, Self::Transport { .. }) + } +} + +pub(crate) fn is_retryable_http_failure(error: &anyhow::Error) -> bool { + error + .downcast_ref::() + .is_some_and(HttpClientError::is_retryable) +} + /// HTTP response from a non-streaming POST request pub struct HttpResponse { pub status: u16, @@ -150,15 +199,19 @@ impl HttpClient for ReqwestHttpClient { let response = tokio::select! { _ = cancel_token.cancelled() => { - anyhow::bail!("HTTP request cancelled"); + return Err(anyhow::Error::new(HttpClientError::cancelled("HTTP request"))); } result = request.send() => { - result.context(format!("Failed to send request to {}", url))? + result.map_err(|error| { + anyhow::Error::new(HttpClientError::from_reqwest("HTTP request", error)) + })? } }; let status = response.status().as_u16(); - let response_body = response.text().await?; + let response_body = response.text().await.map_err(|error| { + anyhow::Error::new(HttpClientError::from_reqwest("HTTP response body", error)) + })?; let response_bytes = response_body.len() as u64; let duration_ms = start.elapsed().as_secs_f64() * 1000.0; @@ -197,10 +250,17 @@ impl HttpClient for ReqwestHttpClient { let response = tokio::select! { _ = cancel_token.cancelled() => { - anyhow::bail!("HTTP streaming request cancelled"); + return Err(anyhow::Error::new(HttpClientError::cancelled( + "HTTP streaming request", + ))); } result = request.send() => { - result.context(format!("Failed to send streaming request to {}", url))? + result.map_err(|error| { + anyhow::Error::new(HttpClientError::from_reqwest( + "HTTP streaming request", + error, + )) + })? } }; @@ -225,9 +285,11 @@ impl HttpClient for ReqwestHttpClient { }); if (200..300).contains(&status) { - let byte_stream = response - .bytes_stream() - .map(|r| r.map_err(|e| anyhow::anyhow!("Stream error: {}", e))); + let byte_stream = response.bytes_stream().map(|result| { + result.map_err(|error| { + anyhow::Error::new(HttpClientError::from_reqwest("HTTP response stream", error)) + }) + }); Ok(StreamingHttpResponse { status, retry_after, @@ -327,6 +389,23 @@ mod tests { use std::sync::{Mutex, OnceLock}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; + #[test] + fn retryable_http_failure_requires_a_typed_transport_error() { + let prose = anyhow::anyhow!( + "Human-readable text says timeout, connection reset, and TLS handshake." + ); + assert!(!is_retryable_http_failure(&prose)); + + let transport = anyhow::Error::new(HttpClientError::transport( + "stream request", + "opaque diagnostic", + )); + assert!(is_retryable_http_failure(&transport)); + + let cancelled = anyhow::Error::new(HttpClientError::cancelled("stream request")); + assert!(!is_retryable_http_failure(&cancelled)); + } + fn proxy_env_lock() -> &'static Mutex<()> { static LOCK: OnceLock> = OnceLock::new(); LOCK.get_or_init(|| Mutex::new(())) diff --git a/core/src/llm/mod.rs b/core/src/llm/mod.rs index a3aeae19..126f3e1c 100644 --- a/core/src/llm/mod.rs +++ b/core/src/llm/mod.rs @@ -25,7 +25,7 @@ pub use error::NonRetryableLlmError; pub use factory::{create_client_with_config, LlmConfig}; pub use http::{ clear_http_metrics_callback, default_http_client, set_http_metrics_callback, HttpClient, - HttpMetricsCallback, HttpMetricsRecord, HttpResponse, StreamingHttpResponse, + HttpClientError, HttpMetricsCallback, HttpMetricsRecord, HttpResponse, StreamingHttpResponse, }; pub use openai::OpenAiClient; pub(crate) use token_estimation::{estimate_message_tokens, estimate_prompt_tokens}; @@ -34,6 +34,7 @@ pub use zhipu::ZhipuClient; use anyhow::Result; use async_trait::async_trait; +use std::time::Duration; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -60,6 +61,19 @@ pub trait LlmClient: Send + Sync { None } + /// Return a view of this client configured for one active generation + /// deadline. The caller still owns and enforces the outer deadline. + /// + /// Composite and account-backed clients can use this budget to configure + /// their underlying transport without inferring timeout intent from error + /// text. Stateless clients may keep the default. + fn with_active_generation_timeout( + &self, + _timeout: Duration, + ) -> Option> { + None + } + /// Complete a conversation (non-streaming) async fn complete( &self, diff --git a/core/src/llm/openai.rs b/core/src/llm/openai.rs index de21f432..9b181c08 100644 --- a/core/src/llm/openai.rs +++ b/core/src/llm/openai.rs @@ -426,7 +426,15 @@ impl OpenAiClient { } Err(e) => { tracing::error!("HTTP error: {e:?}"); - AttemptOutcome::Fatal(e) + if crate::llm::http::is_retryable_http_failure(&e) { + AttemptOutcome::Retryable { + status: reqwest::StatusCode::SERVICE_UNAVAILABLE, + body: format!("network error: {e}"), + retry_after: None, + } + } else { + AttemptOutcome::Fatal(e) + } } } } diff --git a/core/src/llm/openai/streaming.rs b/core/src/llm/openai/streaming.rs index 952329d2..af59fac7 100644 --- a/core/src/llm/openai/streaming.rs +++ b/core/src/llm/openai/streaming.rs @@ -28,28 +28,24 @@ impl OpenAiClient { // Wrap in tokio::select! so cancellation aborts the HTTP request mid-flight let resp = tokio::select! { _ = cancel_token.cancelled() => { - return AttemptOutcome::Fatal(anyhow::anyhow!("HTTP request cancelled")); + return AttemptOutcome::Fatal(anyhow::Error::new( + crate::llm::HttpClientError::cancelled( + "OpenAI streaming HTTP request", + ), + )); } result = http.post_streaming(url, headers, request, cancel_token.clone()) => { match result { Ok(r) => r, Err(e) => { - // Transient network error (timeout, reset, - // mid-flight drop — common on throttled - // endpoints): retry with backoff like 429/5xx - // instead of failing the turn. GLM and other - // OpenAI-compatible endpoints hit this most. - return if crate::retry::is_transient_error(&e) { + return if crate::llm::http::is_retryable_http_failure(&e) { AttemptOutcome::Retryable { status: reqwest::StatusCode::SERVICE_UNAVAILABLE, body: format!("network error: {e}"), retry_after: None, } } else { - AttemptOutcome::Fatal(anyhow::anyhow!( - "HTTP request failed: {}", - e - )) + AttemptOutcome::Fatal(e.context("HTTP request failed")) }; } } diff --git a/core/src/llm/structured.rs b/core/src/llm/structured.rs index 77d6385b..5149128a 100644 --- a/core/src/llm/structured.rs +++ b/core/src/llm/structured.rs @@ -121,34 +121,18 @@ enum SchemaEnvelope { impl SchemaEnvelope { fn for_schema(schema: &Value) -> Self { - if schema_is_object_like(schema) { - Self::Direct - } else if schema.get("type").and_then(Value::as_str) == Some("array") { - Self::Elements - } else { - Self::Value + match schema_root_kind(schema, schema, &mut Vec::new(), 0) { + Some(SchemaRootKind::Object) => Self::Direct, + Some(SchemaRootKind::Array) => Self::Elements, + Some(SchemaRootKind::Other) | None => Self::Value, } } fn response_schema(self, schema: &Value) -> Value { match self { Self::Direct => schema.clone(), - Self::Elements => serde_json::json!({ - "type": "object", - "required": ["elements"], - "additionalProperties": false, - "properties": { - "elements": schema - } - }), - Self::Value => serde_json::json!({ - "type": "object", - "required": ["value"], - "additionalProperties": false, - "properties": { - "value": schema - } - }), + Self::Elements => wrap_response_schema("elements", schema), + Self::Value => wrap_response_schema("value", schema), } } @@ -190,11 +174,137 @@ impl SchemaEnvelope { } } -fn schema_is_object_like(schema: &Value) -> bool { - schema.get("type").and_then(Value::as_str) == Some("object") - || schema.get("properties").is_some() - || schema.get("required").is_some() - || schema.get("additionalProperties").is_some() +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SchemaRootKind { + Object, + Array, + Other, +} + +fn schema_root_kind( + schema: &Value, + root: &Value, + active_refs: &mut Vec, + depth: usize, +) -> Option { + if depth > 64 { + return None; + } + let object = schema.as_object()?; + + if let Some(kind) = object.get("type").and_then(schema_type_kind) { + return Some(kind); + } + if let Some(value) = object.get("const") { + return Some(value_kind(value)); + } + if let Some(values) = object.get("enum").and_then(Value::as_array) { + if let Some(kind) = common_value_kind(values) { + return Some(kind); + } + } + if let Some(reference) = object.get("$ref").and_then(Value::as_str) { + if let Some(pointer) = reference.strip_prefix('#') { + if !active_refs.iter().any(|active| active == reference) { + if let Some(target) = root.pointer(pointer) { + active_refs.push(reference.to_string()); + let kind = schema_root_kind(target, root, active_refs, depth + 1); + active_refs.pop(); + if kind.is_some() { + return kind; + } + } + } + } + } + if let Some(all_of) = object.get("allOf").and_then(Value::as_array) { + if let Some(kind) = all_of + .iter() + .find_map(|branch| schema_root_kind(branch, root, active_refs, depth + 1)) + { + return Some(kind); + } + } + for keyword in ["anyOf", "oneOf"] { + if let Some(branches) = object.get(keyword).and_then(Value::as_array) { + let kinds = branches + .iter() + .map(|branch| schema_root_kind(branch, root, active_refs, depth + 1)) + .collect::>>(); + if let Some(kinds) = kinds { + if let Some(first) = kinds.first().copied() { + if kinds.iter().all(|kind| *kind == first) { + return Some(first); + } + } + } + } + } + + // Preserve the established object-schema behavior for schemas that rely + // on object-only keywords without an explicit `type`. + if ["properties", "required", "additionalProperties"] + .iter() + .any(|keyword| object.contains_key(*keyword)) + { + return Some(SchemaRootKind::Object); + } + None +} + +fn schema_type_kind(value: &Value) -> Option { + match value { + Value::String(value) => Some(type_name_kind(value)), + Value::Array(values) if values.len() == 1 => values[0].as_str().map(type_name_kind), + _ => None, + } +} + +fn type_name_kind(value: &str) -> SchemaRootKind { + match value { + "object" => SchemaRootKind::Object, + "array" => SchemaRootKind::Array, + _ => SchemaRootKind::Other, + } +} + +fn value_kind(value: &Value) -> SchemaRootKind { + match value { + Value::Object(_) => SchemaRootKind::Object, + Value::Array(_) => SchemaRootKind::Array, + _ => SchemaRootKind::Other, + } +} + +fn common_value_kind(values: &[Value]) -> Option { + let first = values.first().map(value_kind)?; + values + .iter() + .all(|value| value_kind(value) == first) + .then_some(first) +} + +fn wrap_response_schema(field: &str, schema: &Value) -> Value { + let mut embedded = schema.clone(); + let mut wrapper = serde_json::json!({ + "type": "object", + "required": [field], + "additionalProperties": false, + "properties": {} + }); + + // A local reference such as `#/$defs/item` resolves from the provider- + // facing document root. Hoist root definitions when the requested schema + // must be wrapped so those references retain their original meaning. + if let Some(embedded_object) = embedded.as_object_mut() { + for keyword in ["$defs", "definitions"] { + if let Some(definitions) = embedded_object.remove(keyword) { + wrapper[keyword] = definitions; + } + } + } + wrapper["properties"][field] = embedded; + wrapper } // --------------------------------------------------------------------------- diff --git a/core/src/retry.rs b/core/src/retry.rs index 2983590c..792f5597 100644 --- a/core/src/retry.rs +++ b/core/src/retry.rs @@ -125,6 +125,32 @@ pub enum AttemptOutcome { Fatal(anyhow::Error), } +/// A retry loop that exhausted a typed HTTP response status. +/// +/// The rendered body is diagnostic only; callers use the retained status for +/// policy decisions. +#[derive(Debug, thiserror::Error)] +#[error("LLM API request failed after {attempts} attempts. Last status: {status} Body: {body}")] +pub(crate) struct RetryExhaustedError { + attempts: u32, + status: StatusCode, + body: String, +} + +impl RetryExhaustedError { + pub(crate) fn new(attempts: u32, status: StatusCode, body: impl Into) -> Self { + Self { + attempts, + status, + body: body.into(), + } + } + + pub(crate) fn status(&self) -> StatusCode { + self.status + } +} + /// Execute an async operation with retry logic. /// /// The `operation` closure is called on each attempt and must return an `AttemptOutcome`. @@ -177,45 +203,11 @@ where // All retries exhausted let status = last_status.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - anyhow::bail!( - "LLM API request failed after {} attempts. Last status: {} Body: {}", + Err(anyhow::Error::new(RetryExhaustedError::new( config.max_retries + 1, status, last_body, - ) -} - -/// Heuristic: is this a transient *network* error worth retrying — a timeout, -/// connection reset/refused/closed, broken pipe, DNS failure, or a request that -/// dropped mid-flight? These carry no HTTP status (so `is_retryable_status` -/// can't see them), yet Claude Code retries them just like 429/5xx. We only have -/// the error's rendered text (a `CodeError`/`anyhow::Error` chain) to classify. -pub fn is_transient_error(e: &E) -> bool { - let m = e.to_string().to_lowercase(); - [ - "timed out", - "timeout", - "connection reset", - "connection refused", - "connection closed", - "connection aborted", - "connection error", - "broken pipe", - "reset by peer", - "error sending request", - "incomplete message", - "unexpected eof", - "dns error", - "unreachable", - "tls handshake", - "request error", - "body error", - "decoding response", - "channel closed", - "stream closed", - ] - .iter() - .any(|p| m.contains(p)) + ))) } #[cfg(test)] @@ -224,18 +216,26 @@ mod tests { use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; - #[test] - fn transient_error_classification() { - let t = |s: &str| is_transient_error(&anyhow::anyhow!("{s}")); - // Transient network errors → retry. - assert!(t("error sending request for url: operation timed out")); - assert!(t("connection reset by peer")); - assert!(t("LLM error: connection closed before message completed")); - assert!(t("tls handshake eof")); - // Real application errors → do NOT retry. - assert!(!t("invalid api key")); - assert!(!t("model not found")); - assert!(!t("context length exceeded")); + #[tokio::test] + async fn retry_exhaustion_preserves_typed_http_status() { + let config = RetryConfig { + max_retries: 0, + ..RetryConfig::default() + }; + let error = with_retry::<(), _, _>(&config, |_| async { + AttemptOutcome::Retryable { + status: StatusCode::TOO_MANY_REQUESTS, + body: "opaque provider response".to_string(), + retry_after: None, + } + }) + .await + .expect_err("the only attempt must fail"); + + let exhausted = error + .downcast_ref::() + .expect("retry exhaustion must retain its typed status"); + assert_eq!(exhausted.status(), StatusCode::TOO_MANY_REQUESTS); } // ======================================================================== diff --git a/core/src/security/mod.rs b/core/src/security/mod.rs index 5129ab22..9acb34b6 100644 --- a/core/src/security/mod.rs +++ b/core/src/security/mod.rs @@ -96,6 +96,9 @@ pub fn sanitize_agent_event( op: text(provider, op), duration_ms: *duration_ms, }, + ToolErrorKind::Transport { op } => ToolErrorKind::Transport { + op: text(provider, op), + }, ToolErrorKind::Cancelled { op } => ToolErrorKind::Cancelled { op: text(provider, op), }, @@ -622,6 +625,14 @@ mod tests { duration_ms: 42, }, ), + ( + ToolErrorKind::Transport { + op: format!("operation: {secret}"), + }, + ToolErrorKind::Transport { + op: format!("operation: {redacted}"), + }, + ), ]; for (error_kind, expected) in cases { diff --git a/core/src/tools/builtin/generate_object.rs b/core/src/tools/builtin/generate_object.rs index d8037797..66fc0d9d 100644 --- a/core/src/tools/builtin/generate_object.rs +++ b/core/src/tools/builtin/generate_object.rs @@ -1,8 +1,8 @@ //! Built-in `generate_object` tool for structured JSON output. //! //! This tool allows the agent (or users via `session.tool()`) to generate a -//! JSON object that conforms to a given JSON Schema. It supports streaming -//! partial objects via `ToolStreamEvent::OutputDelta`. +//! JSON value that conforms to a given JSON Schema. It supports streaming +//! partial values via `ToolStreamEvent::OutputDelta`. use crate::llm::structured::{self, PartialObjectCallback, StructuredMode, StructuredRequest}; use crate::llm::{ @@ -12,6 +12,7 @@ use crate::llm::{ use crate::tools::types::{Tool, ToolContext, ToolErrorKind, ToolOutput, ToolStreamEvent}; use anyhow::Result; use async_trait::async_trait; +use serde::Deserialize; use serde_json::Value; use std::future::Future; use std::sync::Arc; @@ -19,6 +20,8 @@ use std::time::{Duration, Instant}; const MAX_SCHEMA_BYTES: usize = 64 * 1024; const MAX_SCHEMA_DEPTH: usize = 32; +const MAX_SCHEMA_NAME_BYTES: usize = 59; +const MAX_SCHEMA_DESCRIPTION_BYTES: usize = 4 * 1024; const MAX_PROMPT_BYTES: usize = 128 * 1024; const MAX_SYSTEM_BYTES: usize = 32 * 1024; const DEFAULT_TIMEOUT_MS: u64 = 120_000; @@ -26,6 +29,27 @@ const MAX_TIMEOUT_MS: u64 = 600_000; const PARTIAL_EVENT_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100); const MAX_PARTIAL_EVENT_BYTES: usize = 64 * 1024; +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct GenerateObjectParams { + schema: Value, + #[serde(default)] + schema_name: Option, + #[serde(default)] + schema_description: Option, + prompt: String, + #[serde(default)] + system: Option, + #[serde(default)] + mode: Option, + #[serde(default)] + max_repair_attempts: Option, + #[serde(default)] + include_raw_text: bool, + #[serde(default)] + timeout_ms: Option, +} + pub struct GenerateObjectTool { llm_client: Arc, admission: ModelGenerationAdmission, @@ -48,10 +72,11 @@ impl Tool for GenerateObjectTool { } fn description(&self) -> &str { - "Generate a JSON object that strictly conforms to a provided JSON Schema. \ + "Generate a JSON value that strictly conforms to a provided JSON Schema. \ Use when you need structured output: extracting fields from text, classifying \ data, converting natural language to typed records, or producing machine-readable \ - results. Returns the validated object on success." + results. The root value may be an object, array, or scalar. Returns the validated \ + value on success." } fn parameters(&self) -> Value { @@ -62,24 +87,31 @@ impl Tool for GenerateObjectTool { "properties": { "schema": { "type": "object", - "description": "JSON Schema that the output object must conform to" + "description": "JSON Schema that the output value must conform to" }, "schema_name": { "type": "string", "description": "Short name for the schema (used internally for tool naming)", + "minLength": 1, + "maxLength": MAX_SCHEMA_NAME_BYTES, + "pattern": "^[A-Za-z0-9_-]+$", "default": "result" }, "schema_description": { "type": "string", - "description": "Optional description of what the schema represents" + "description": "Optional description of what the schema represents", + "maxLength": MAX_SCHEMA_DESCRIPTION_BYTES }, "prompt": { "type": "string", - "description": "The prompt describing what object to generate or extract" + "description": "The prompt describing what value to generate or extract", + "minLength": 1, + "maxLength": MAX_PROMPT_BYTES }, "system": { "type": "string", - "description": "Optional system prompt to guide generation" + "description": "Optional system prompt to guide generation", + "maxLength": MAX_SYSTEM_BYTES }, "mode": { "type": "string", @@ -103,24 +135,38 @@ impl Tool for GenerateObjectTool { "type": "integer", "minimum": 1000, "maximum": MAX_TIMEOUT_MS, - "description": "Active generation deadline in milliseconds. Admission queue wait is excluded. Default 120000; maximum 600000." + "description": "Active generation deadline in milliseconds. Admission queue wait is excluded. Default 120000; maximum 600000.", + "default": DEFAULT_TIMEOUT_MS } } }) } async fn execute(&self, args: &Value, ctx: &ToolContext) -> Result { - let schema = match args.get("schema") { - Some(s) if s.is_object() => s.clone(), - Some(_) => { - return Ok(ToolOutput::error( - "'schema' must be a JSON object (a valid JSON Schema)", - )); - } - None => { - return Ok(ToolOutput::error("'schema' parameter is required")); + let params: GenerateObjectParams = match serde_json::from_value(args.clone()) { + Ok(params) => params, + Err(error) => { + return Ok(invalid_argument(format!( + "Invalid generate_object parameters: {error}" + ))); } }; + let GenerateObjectParams { + schema, + schema_name, + schema_description, + prompt, + system, + mode, + max_repair_attempts, + include_raw_text, + timeout_ms, + } = params; + if !schema.is_object() { + return Ok(invalid_argument( + "'schema' must be a JSON object (a valid JSON Schema)".to_string(), + )); + } let schema_bytes = serde_json::to_vec(&schema)?.len(); if schema_bytes > MAX_SCHEMA_BYTES { return Ok(invalid_argument(format!( @@ -138,55 +184,36 @@ impl Tool for GenerateObjectTool { ))); } - let prompt = match args.get("prompt").and_then(|v| v.as_str()) { - Some(p) if !p.is_empty() => p.to_string(), - _ => { - return Ok(ToolOutput::error( - "'prompt' parameter is required and must be non-empty", - )); - } - }; + if prompt.trim().is_empty() { + return Ok(invalid_argument( + "'prompt' parameter is required and must contain non-whitespace text".to_string(), + )); + } if prompt.len() > MAX_PROMPT_BYTES { return Ok(invalid_argument(format!( "'prompt' exceeds the {MAX_PROMPT_BYTES} byte limit" ))); } - // Validate schema has at minimum a "type" or "properties" or "anyOf" field - if schema.get("type").is_none() - && schema.get("properties").is_none() - && schema.get("anyOf").is_none() - && schema.get("oneOf").is_none() - && schema.get("enum").is_none() + let schema_name = schema_name.unwrap_or_else(|| "result".to_string()); + if schema_name.is_empty() + || schema_name.len() > MAX_SCHEMA_NAME_BYTES + || !schema_name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) { - return Ok(ToolOutput::error( - "'schema' should contain at least one of: type, properties, anyOf, oneOf, or enum", - )); + return Ok(invalid_argument(format!( + "'schema_name' must match ^[A-Za-z0-9_-]+$ and contain at most {MAX_SCHEMA_NAME_BYTES} bytes" + ))); + } + if schema_description + .as_ref() + .is_some_and(|value| value.len() > MAX_SCHEMA_DESCRIPTION_BYTES) + { + return Ok(invalid_argument(format!( + "'schema_description' exceeds the {MAX_SCHEMA_DESCRIPTION_BYTES} byte limit" + ))); } - - let schema_name: String = args - .get("schema_name") - .and_then(|v| v.as_str()) - .unwrap_or("result") - .chars() - .filter(|c| c.is_alphanumeric() || *c == '_' || *c == '-') - .take(64) - .collect(); - let schema_name = if schema_name.is_empty() { - "result".to_string() - } else { - schema_name - }; - - let schema_description = args - .get("schema_description") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()); - - let system = args - .get("system") - .and_then(|v| v.as_str()) - .map(|s| s.to_string()); if system .as_ref() .is_some_and(|value| value.len() > MAX_SYSTEM_BYTES) @@ -196,15 +223,15 @@ impl Tool for GenerateObjectTool { ))); } - let requested_mode = args.get("mode").and_then(|v| v.as_str()).unwrap_or("auto"); - let mode = match requested_mode { + let requested_mode = mode.unwrap_or_else(|| "auto".to_string()); + let structured_mode = match requested_mode.as_str() { "strict" => StructuredMode::Strict, "json" => StructuredMode::Json, "tool" => StructuredMode::Tool, "prompt" => StructuredMode::Prompt, "auto" => StructuredMode::Auto, other => { - return Ok(ToolOutput::error(format!( + return Ok(invalid_argument(format!( "'mode' must be one of auto, strict, json, tool, or prompt; got '{other}'" ))); } @@ -214,20 +241,19 @@ impl Tool for GenerateObjectTool { // the client's native capability. Unsupported native modes safely fall // back to prompt+schema parsing instead of sending provider parameters // that some OpenAI-compatible endpoints hang on. - let max_repair_attempts = args - .get("max_repair_attempts") - .and_then(|v| v.as_u64()) - .unwrap_or(2) - .min(5) as u8; - let include_raw_text = args - .get("include_raw_text") - .and_then(|v| v.as_bool()) - .unwrap_or(false); - let timeout_ms = args - .get("timeout_ms") - .and_then(|value| value.as_u64()) - .unwrap_or(DEFAULT_TIMEOUT_MS) - .clamp(1_000, MAX_TIMEOUT_MS); + let max_repair_attempts = max_repair_attempts.unwrap_or(2); + if max_repair_attempts > 5 { + return Ok(invalid_argument( + "'max_repair_attempts' must be between 0 and 5".to_string(), + )); + } + let max_repair_attempts = max_repair_attempts as u8; + let timeout_ms = timeout_ms.unwrap_or(DEFAULT_TIMEOUT_MS); + if !(1_000..=MAX_TIMEOUT_MS).contains(&timeout_ms) { + return Ok(invalid_argument(format!( + "'timeout_ms' must be between 1000 and {MAX_TIMEOUT_MS}" + ))); + } let req = StructuredRequest { prompt, @@ -235,13 +261,17 @@ impl Tool for GenerateObjectTool { schema, schema_name: schema_name.clone(), schema_description, - mode, + mode: structured_mode, max_repair_attempts, }; let llm_client = ctx .llm_client() .unwrap_or_else(|| Arc::clone(&self.llm_client)); + let active_timeout = Duration::from_millis(timeout_ms); + let llm_client = llm_client + .with_active_generation_timeout(active_timeout) + .unwrap_or(llm_client); let admission = ctx .model_generation_admission() .unwrap_or_else(|| self.admission.clone()); @@ -285,7 +315,7 @@ impl Tool for GenerateObjectTool { &admission, preadmitted, &cancellation, - Duration::from_millis(timeout_ms), + active_timeout, generation, ) .await; @@ -363,20 +393,15 @@ impl Tool for GenerateObjectTool { }), ), GenerationStop::Failed(error) => { - let message = error.to_string(); - let lower = message.to_ascii_lowercase(); - let kind = (lower.contains("rate limit") - || lower.contains("too many requests")) - .then_some(ToolErrorKind::RateLimited { - retry_after_ms: None, - }); + let message = format!("{error:#}"); + let kind = generation_failure_kind(&error); (format!("generate_object failed: {message}"), kind) } }; let output = ToolOutput::error(message).with_metadata(serde_json::json!({ "schema_name": schema_name, "requested_mode": requested_mode, - "mode_requested": mode, + "mode_requested": structured_mode, "timeout_ms": timeout_ms, "generation_admission": admission_metadata, })); @@ -395,6 +420,33 @@ enum GenerationStop { Failed(anyhow::Error), } +fn generation_failure_kind(error: &anyhow::Error) -> Option { + if let Some(exhausted) = error.downcast_ref::() { + let status = exhausted.status(); + return if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + Some(ToolErrorKind::RateLimited { + retry_after_ms: None, + }) + } else if status == reqwest::StatusCode::REQUEST_TIMEOUT || status.is_server_error() { + Some(ToolErrorKind::Transport { + op: "generate_object".to_string(), + }) + } else { + None + }; + } + + match error.downcast_ref::() { + Some(crate::llm::HttpClientError::Cancelled { .. }) => Some(ToolErrorKind::Cancelled { + op: "generate_object".to_string(), + }), + Some(crate::llm::HttpClientError::Transport { .. }) => Some(ToolErrorKind::Transport { + op: "generate_object".to_string(), + }), + Some(crate::llm::HttpClientError::InvalidRequest { .. }) | None => None, + } +} + struct GenerationExecution { result: std::result::Result, queue_wait: Duration, @@ -877,3 +929,7 @@ mod tests { drop(replacement); } } + +#[cfg(test)] +#[path = "generate_object_contract_tests.rs"] +mod contract_tests; diff --git a/core/src/tools/builtin/generate_object_contract_tests.rs b/core/src/tools/builtin/generate_object_contract_tests.rs new file mode 100644 index 00000000..22243acc --- /dev/null +++ b/core/src/tools/builtin/generate_object_contract_tests.rs @@ -0,0 +1,374 @@ +use super::*; +use crate::llm::structured::NativeStructuredSupport; +use crate::llm::{ContentBlock, LlmResponse, Message, StreamEvent, TokenUsage, ToolDefinition}; +use crate::tools::ToolExecutor; +use async_trait::async_trait; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +struct OneResponseClient { + response: Mutex>, +} + +impl OneResponseClient { + fn new(input: Value) -> Self { + Self { + response: Mutex::new(Some(LlmResponse { + message: Message { + role: "assistant".to_string(), + content: vec![ContentBlock::ToolUse { + id: "contract-call".to_string(), + name: "emit_result".to_string(), + input, + }], + reasoning_content: None, + }, + usage: TokenUsage::default(), + stop_reason: Some("tool_use".to_string()), + token_logprobs: Vec::new(), + meta: None, + })), + } + } +} + +#[async_trait] +impl LlmClient for OneResponseClient { + async fn complete( + &self, + _messages: &[Message], + _system: Option<&str>, + _tools: &[ToolDefinition], + ) -> anyhow::Result { + self.response + .lock() + .expect("contract response lock") + .take() + .ok_or_else(|| anyhow::anyhow!("contract response already consumed")) + } + + async fn complete_streaming( + &self, + _messages: &[Message], + _system: Option<&str>, + _tools: &[ToolDefinition], + _cancel_token: CancellationToken, + ) -> anyhow::Result> { + anyhow::bail!("contract tests use blocking generation") + } + + fn native_structured_support(&self) -> NativeStructuredSupport { + NativeStructuredSupport::ForcedTool + } +} + +#[derive(Clone)] +struct TimeoutAwareClient { + observed_timeout_ms: Arc, + response: Arc>>, +} + +#[async_trait] +impl LlmClient for TimeoutAwareClient { + fn with_active_generation_timeout(&self, timeout: Duration) -> Option> { + self.observed_timeout_ms.store( + u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX), + Ordering::SeqCst, + ); + Some(Arc::new(self.clone())) + } + + async fn complete( + &self, + _messages: &[Message], + _system: Option<&str>, + _tools: &[ToolDefinition], + ) -> anyhow::Result { + self.response + .lock() + .expect("timeout response lock") + .take() + .ok_or_else(|| anyhow::anyhow!("timeout response already consumed")) + } + + async fn complete_streaming( + &self, + _messages: &[Message], + _system: Option<&str>, + _tools: &[ToolDefinition], + _cancel_token: CancellationToken, + ) -> anyhow::Result> { + anyhow::bail!("timeout contract uses blocking generation") + } + + fn native_structured_support(&self) -> NativeStructuredSupport { + NativeStructuredSupport::ForcedTool + } +} + +async fn run_contract_case(schema: Value, provider_input: Value) -> ToolOutput { + let workspace = tempfile::tempdir().expect("contract workspace"); + let tool = GenerateObjectTool::new(Arc::new(OneResponseClient::new(provider_input))); + tool.execute( + &serde_json::json!({ + "schema": schema, + "prompt": "Return the requested value.", + "mode": "tool" + }), + &ToolContext::new(workspace.path().to_path_buf()), + ) + .await + .expect("generate_object contract execution") +} + +#[tokio::test] +async fn generate_object_accepts_standard_root_schema_keywords() { + let cases = [ + ( + serde_json::json!({ + "$defs": { + "record": { + "type": "object", + "additionalProperties": false, + "properties": {"answer": {"const": "ok"}}, + "required": ["answer"] + } + }, + "$ref": "#/$defs/record" + }), + serde_json::json!({"answer": "ok"}), + serde_json::json!({"answer": "ok"}), + ), + ( + serde_json::json!({ + "allOf": [ + {"type": "object"}, + { + "additionalProperties": false, + "properties": {"answer": {"const": "ok"}}, + "required": ["answer"] + } + ] + }), + serde_json::json!({"answer": "ok"}), + serde_json::json!({"answer": "ok"}), + ), + ( + serde_json::json!({"const": "fixed"}), + serde_json::json!({"value": "fixed"}), + serde_json::json!("fixed"), + ), + ( + serde_json::json!({ + "$defs": { + "list": { + "type": "array", + "items": {"type": "string"}, + "minItems": 1 + } + }, + "$ref": "#/$defs/list" + }), + serde_json::json!({"elements": ["ready"]}), + serde_json::json!(["ready"]), + ), + ]; + + for (schema, provider_input, expected) in cases { + let output = run_contract_case(schema, provider_input).await; + assert!(output.success, "unexpected contract failure: {output:#?}"); + let payload: Value = serde_json::from_str(&output.content).expect("JSON tool output"); + assert_eq!(payload["object"], expected); + } +} + +#[tokio::test] +async fn generate_object_rejects_parameters_outside_its_declared_contract() { + let valid_schema = serde_json::json!({ + "type": "object", + "properties": {"value": {"type": "string"}}, + "required": ["value"] + }); + let base = || { + serde_json::json!({ + "schema": valid_schema.clone(), + "prompt": "Return a value.", + "mode": "tool" + }) + }; + let mut invalid_arguments = Vec::new(); + + let mut whitespace_prompt = base(); + whitespace_prompt["prompt"] = Value::String(" \n\t ".to_string()); + invalid_arguments.push(whitespace_prompt); + + let mut unicode_name = base(); + unicode_name["schema_name"] = Value::String("结果".to_string()); + invalid_arguments.push(unicode_name); + + let mut long_name = base(); + long_name["schema_name"] = Value::String("a".repeat(60)); + invalid_arguments.push(long_name); + + let mut long_description = base(); + long_description["schema_description"] = Value::String("d".repeat(4 * 1024 + 1)); + invalid_arguments.push(long_description); + + let mut excess_repairs = base(); + excess_repairs["max_repair_attempts"] = serde_json::json!(6); + invalid_arguments.push(excess_repairs); + + let mut short_timeout = base(); + short_timeout["timeout_ms"] = serde_json::json!(999); + invalid_arguments.push(short_timeout); + + let mut unknown_field = base(); + unknown_field["unexpected"] = serde_json::json!(true); + invalid_arguments.push(unknown_field); + + for args in invalid_arguments { + let workspace = tempfile::tempdir().expect("validation workspace"); + let tool = GenerateObjectTool::new(Arc::new(OneResponseClient::new( + serde_json::json!({"value": "must not be used"}), + ))); + let output = tool + .execute(&args, &ToolContext::new(workspace.path().to_path_buf())) + .await + .expect("validation output"); + assert!(!output.success, "arguments should be rejected: {args:#}"); + assert!( + matches!( + output.error_kind, + Some(ToolErrorKind::InvalidArgument { .. }) + ), + "expected InvalidArgument for {args:#}, got {output:#?}" + ); + } +} + +#[test] +fn generate_object_parameter_schema_declares_runtime_limits() { + let tool = GenerateObjectTool::new(Arc::new(OneResponseClient::new(Value::Null))); + let parameters = tool.parameters(); + let properties = ¶meters["properties"]; + + assert_eq!(properties["prompt"]["minLength"], 1); + assert_eq!(properties["schema_name"]["maxLength"], 59); + assert_eq!(properties["schema_name"]["pattern"], "^[A-Za-z0-9_-]+$"); + assert_eq!(properties["schema_description"]["maxLength"], 4 * 1024); +} + +#[test] +fn generation_failure_kind_uses_typed_errors_instead_of_diagnostic_prose() { + let prose = anyhow::anyhow!("rate limit: too many requests; status 429"); + assert_eq!(generation_failure_kind(&prose), None); + + let rate_limited = anyhow::Error::new(crate::retry::RetryExhaustedError::new( + 1, + reqwest::StatusCode::TOO_MANY_REQUESTS, + "opaque", + )); + assert_eq!( + generation_failure_kind(&rate_limited), + Some(ToolErrorKind::RateLimited { + retry_after_ms: None, + }) + ); + + let unavailable = anyhow::Error::new(crate::retry::RetryExhaustedError::new( + 1, + reqwest::StatusCode::SERVICE_UNAVAILABLE, + "opaque", + )); + assert_eq!( + generation_failure_kind(&unavailable), + Some(ToolErrorKind::Transport { + op: "generate_object".to_string(), + }) + ); + + let transport = anyhow::Error::new(crate::llm::HttpClientError::transport( + "HTTP request", + "opaque diagnostic", + )); + assert_eq!( + generation_failure_kind(&transport), + Some(ToolErrorKind::Transport { + op: "generate_object".to_string(), + }) + ); + + let cancelled = anyhow::Error::new(crate::llm::HttpClientError::cancelled("HTTP request")); + assert_eq!( + generation_failure_kind(&cancelled), + Some(ToolErrorKind::Cancelled { + op: "generate_object".to_string(), + }) + ); +} + +#[tokio::test] +async fn generate_object_propagates_active_timeout_through_the_invocation_gateway() { + let workspace = tempfile::tempdir().expect("timeout workspace"); + let observed_timeout_ms = Arc::new(AtomicU64::new(0)); + let raw_client: Arc = Arc::new(TimeoutAwareClient { + observed_timeout_ms: Arc::clone(&observed_timeout_ms), + response: Arc::new(Mutex::new(Some(LlmResponse { + message: Message { + role: "assistant".to_string(), + content: vec![ContentBlock::ToolUse { + id: "timeout-call".to_string(), + name: "emit_result".to_string(), + input: serde_json::json!({"value": "ok"}), + }], + reasoning_content: None, + }, + usage: TokenUsage::default(), + stop_reason: Some("tool_use".to_string()), + token_logprobs: Vec::new(), + meta: None, + }))), + }); + let agent = crate::agent::AgentLoop::new( + Arc::clone(&raw_client), + Arc::new(ToolExecutor::new( + workspace.path().to_string_lossy().to_string(), + )), + ToolContext::new(workspace.path().to_path_buf()), + crate::agent::AgentConfig::default(), + ); + let cancellation = CancellationToken::new(); + let governed = agent.scoped_llm_client_for_parts(Some("timeout-session"), &None, &cancellation); + let context = ToolContext::new(workspace.path().to_path_buf()) + .with_session_id("timeout-session") + .with_cancellation(cancellation) + .with_llm_client(governed); + let tool = GenerateObjectTool::new(raw_client); + let requested_timeout_ms = 45_000; + + let output = tool + .execute( + &serde_json::json!({ + "schema": { + "type": "object", + "additionalProperties": false, + "properties": {"value": {"type": "string"}}, + "required": ["value"] + }, + "prompt": "Return a value.", + "mode": "tool", + "timeout_ms": requested_timeout_ms + }), + &context, + ) + .await + .expect("timeout-aware output"); + + assert!(output.success, "{}", output.content); + assert_eq!( + observed_timeout_ms.load(Ordering::SeqCst), + requested_timeout_ms + ); +} diff --git a/core/src/tools/task.rs b/core/src/tools/task.rs index 7c323796..9d7f7b23 100644 --- a/core/src/tools/task.rs +++ b/core/src/tools/task.rs @@ -44,7 +44,7 @@ const MAX_TASK_SOURCE_TOOL_BYTES: usize = 64; const MAX_TASK_SOURCE_VALUE_BYTES: usize = 4 * 1024; const MAX_PARALLEL_TASK_SOURCE_ANCHORS: usize = MAX_TASK_SOURCE_ANCHORS; const TASK_TOOL_DESCRIPTION: &str = "Delegate a bounded task to a specialized child run. Choose the canonical worker name from the live agent catalog. Custom agents from agent_dirs and .a3s/agents are supported; .claude/agents is read for compatibility."; -const PARALLEL_TASK_TOOL_DESCRIPTION: &str = "Fan out 2 or more INDEPENDENT subtasks as delegated child runs that execute concurrently; results are returned when all complete. By default any failed child makes the tool fail; evidence-gathering callers may set allow_partial_failure=true to continue when at least one child succeeds. Transient provider failures may be retried once only for explicitly read-only branches; successful and potentially mutating branches are never replayed. Use this only when the work genuinely splits into branches that can be investigated or implemented separately. Do not use it for trivial, conversational, single-step, or dependent work. Choose canonical worker names from the live agent catalog."; +const PARALLEL_TASK_TOOL_DESCRIPTION: &str = "Fan out 2 or more INDEPENDENT subtasks as delegated child runs that execute concurrently; results are returned when all complete. By default any failed child makes the tool fail; evidence-gathering callers may set allow_partial_failure=true to continue when at least one child succeeds. Child output never authorizes branch replay; provider and child runtimes own any typed retry policy below this boundary. Use this only when the work genuinely splits into branches that can be investigated or implemented separately. Do not use it for trivial, conversational, single-step, or dependent work. Choose canonical worker names from the live agent catalog."; /// Task tool parameters #[derive(Debug, Clone, Serialize, Deserialize)] @@ -963,26 +963,62 @@ impl Tool for ParallelTaskTool { async fn execute(&self, args: &serde_json::Value, ctx: &ToolContext) -> Result { let started_at = std::time::Instant::now(); - let params: ParallelTaskParams = - serde_json::from_value(args.clone()).context("Invalid parallel task parameters")?; + let params: ParallelTaskParams = match serde_json::from_value(args.clone()) { + Ok(params) => params, + Err(error) => { + return Ok(invalid_parallel_task_argument(format!( + "Invalid parallel_task parameters: {error}" + ))); + } + }; let parent_cancellation = ctx.cancellation_token(); let executor = self.executor.scoped_for_invocation(ctx); - if params.tasks.is_empty() { - return Ok(ToolOutput::error("No tasks provided".to_string())); + if params.tasks.len() < 2 { + return Ok(invalid_parallel_task_argument( + "parallel_task requires at least 2 independent tasks".to_string(), + )); } if params.tasks.len() > MAX_PARALLEL_TASKS_PER_CALL { - return Ok(ToolOutput::error(format!( + return Ok(invalid_parallel_task_argument(format!( "parallel_task accepts at most {MAX_PARALLEL_TASKS_PER_CALL} tasks" ))); } + if let Some((index, _)) = params + .tasks + .iter() + .enumerate() + .find(|(_, task)| task.background) + { + return Ok(invalid_parallel_task_argument(format!( + "parallel_task task {} cannot set background=true; every branch is already executed concurrently and collected by the parent call", + index + 1 + ))); + } + if params.timeout_ms == Some(0) { + return Ok(invalid_parallel_task_argument( + "parallel_task timeout_ms must be at least 1".to_string(), + )); + } + if let Some(min_success_count) = params.min_success_count { + if !params.allow_partial_failure { + return Ok(invalid_parallel_task_argument( + "parallel_task min_success_count requires allow_partial_failure=true" + .to_string(), + )); + } + if min_success_count == 0 || min_success_count > params.tasks.len() { + return Ok(invalid_parallel_task_argument(format!( + "parallel_task min_success_count must be between 1 and the task count ({})", + params.tasks.len() + ))); + } + } let task_count = params.tasks.len(); - let tasks = params.tasks.clone(); - - let mut run = executor + let run = executor .execute_parallel_for_tool( - tasks.clone(), + params.tasks.clone(), ctx.agent_event_tx.clone(), parallel_execution::ParallelToolOptions { parent_session_id: ctx.session_id.as_deref(), @@ -993,19 +1029,6 @@ impl Tool for ParallelTaskTool { }, ) .await; - let retry_summary = executor - .retry_transient_parallel_failures( - &tasks, - parallel_execution::ParallelRetryOptions { - event_tx: ctx.agent_event_tx.clone(), - parent_session_id: ctx.session_id.as_deref(), - parent_cancellation: &parent_cancellation, - total_timeout_ms: params.timeout_ms, - started_at, - }, - &mut run, - ) - .await; let results = run.results; // Format results with compact per-task excerpts for parent context. @@ -1032,7 +1055,6 @@ impl Tool for ParallelTaskTool { "truncated_for_context": truncated, "artifact_id": task_artifact_id(result), "artifact_uri": task_artifact_uri(result), - "retry_attempts": retry_summary.attempts_by_index.get(i).copied().unwrap_or_default(), })); output.push_str(&format!( "--- Task {} ({}) {} ---\n{}\n\n", @@ -1047,12 +1069,6 @@ impl Tool for ParallelTaskTool { let failed_count = results.len().saturating_sub(success_count); let all_success = failed_count == 0; let partial_failure = failed_count > 0 && success_count > 0; - if retry_summary.recovered_task_count > 0 { - output.push_str(&format!( - "Recovered {} transient read-only child failure(s) by retrying only failed branches.\n", - retry_summary.recovered_task_count - )); - } if params.allow_partial_failure && partial_failure { output.push_str(&format!( "Partial failure tolerated: {success_count} succeeded, {failed_count} failed.\n" @@ -1095,14 +1111,16 @@ impl Tool for ParallelTaskTool { "timed_out": run.timed_out, "min_success_count": params.min_success_count, "returned_early": run.returned_early, - "retry_attempt_count": retry_summary.retry_attempt_count, - "retried_task_count": retry_summary.retried_task_count, - "recovered_task_count": retry_summary.recovered_task_count, "duration_ms": started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64, "results": metadata_results, }))) } } +fn invalid_parallel_task_argument(message: String) -> ToolOutput { + ToolOutput::error(&message) + .with_error_kind(crate::tools::ToolErrorKind::InvalidArgument { message }) +} + #[cfg(test)] mod tests; diff --git a/core/src/tools/task/parallel_execution.rs b/core/src/tools/task/parallel_execution.rs index d10e580d..bc889841 100644 --- a/core/src/tools/task/parallel_execution.rs +++ b/core/src/tools/task/parallel_execution.rs @@ -1,12 +1,6 @@ use super::*; use std::collections::HashMap; -const MAX_TRANSIENT_PARALLEL_RETRIES: u8 = 1; -#[cfg(not(test))] -const PARALLEL_RETRY_BASE_DELAY_MS: u64 = 750; -#[cfg(test)] -const PARALLEL_RETRY_BASE_DELAY_MS: u64 = 1; - pub(super) struct ParallelToolOptions<'a> { pub(super) parent_session_id: Option<&'a str>, pub(super) timeout_ms: Option, @@ -15,22 +9,6 @@ pub(super) struct ParallelToolOptions<'a> { pub(super) parent_cancellation: Option<&'a CancellationToken>, } -pub(super) struct ParallelRetryOptions<'a> { - pub(super) event_tx: Option>, - pub(super) parent_session_id: Option<&'a str>, - pub(super) parent_cancellation: &'a CancellationToken, - pub(super) total_timeout_ms: Option, - pub(super) started_at: std::time::Instant, -} - -#[derive(Debug, Default)] -pub(super) struct ParallelRetrySummary { - pub(super) attempts_by_index: Vec, - pub(super) retry_attempt_count: usize, - pub(super) retried_task_count: usize, - pub(super) recovered_task_count: usize, -} - impl TaskExecutor { /// Execute multiple tasks in parallel. /// @@ -312,201 +290,6 @@ impl TaskExecutor { min_success_count, } } - - /// Retry only failed branches that are both side-effect-safe and clearly - /// transient. Successful branches are retained in place. This second - /// boundary is intentionally narrower than the provider and child-loop - /// retry layers: it exists for account-backed providers that can reject an - /// individual child after a concurrent burst, without replaying maker or - /// implementation work that may already have changed the workspace. - pub(super) async fn retry_transient_parallel_failures( - self: &Arc, - tasks: &[TaskParams], - options: ParallelRetryOptions<'_>, - run: &mut ParallelTaskRun, - ) -> ParallelRetrySummary { - let ParallelRetryOptions { - event_tx, - parent_session_id, - parent_cancellation, - total_timeout_ms, - started_at, - } = options; - let mut summary = ParallelRetrySummary { - attempts_by_index: vec![0; tasks.len()], - ..ParallelRetrySummary::default() - }; - if run.timed_out || run.returned_early || parent_cancellation.is_cancelled() { - return summary; - } - - for attempt in 0..MAX_TRANSIENT_PARALLEL_RETRIES { - let retry_indexes = tasks - .iter() - .zip(run.results.iter()) - .enumerate() - .filter_map(|(index, (task, result))| { - (!result.success - && self.is_parallel_retry_safe(task) - && is_transient_parallel_failure(&result.output)) - .then_some(index) - }) - .collect::>(); - if retry_indexes.is_empty() { - break; - } - - let delay = parallel_retry_delay(attempt); - if let Some(remaining) = remaining_parallel_timeout(total_timeout_ms, started_at) { - if remaining.is_zero() || delay >= remaining { - break; - } - } - let retry_allowed = tokio::select! { - biased; - _ = parent_cancellation.cancelled() => false, - _ = tokio::time::sleep(delay) => true, - }; - if !retry_allowed { - break; - } - - let retry_timeout_ms = match remaining_parallel_timeout(total_timeout_ms, started_at) { - Some(remaining) if remaining.is_zero() => break, - Some(remaining) => { - Some(remaining.as_millis().min(u128::from(u64::MAX)).max(1) as u64) - } - None => None, - }; - let retry_tasks = retry_indexes - .iter() - .filter_map(|index| tasks.get(*index).cloned()) - .collect::>(); - let retry_run = self - .execute_parallel_for_tool( - retry_tasks, - event_tx.clone(), - ParallelToolOptions { - parent_session_id, - timeout_ms: retry_timeout_ms, - min_success_count: None, - allow_partial_failure: true, - parent_cancellation: Some(parent_cancellation), - }, - ) - .await; - - summary.retry_attempt_count = summary - .retry_attempt_count - .saturating_add(retry_indexes.len()); - for (index, retry_result) in retry_indexes.into_iter().zip(retry_run.results) { - summary.attempts_by_index[index] = - summary.attempts_by_index[index].saturating_add(1); - if retry_result.success && !run.results[index].success { - summary.recovered_task_count = summary.recovered_task_count.saturating_add(1); - } - run.results[index] = retry_result; - } - if retry_run.timed_out { - run.timed_out = true; - break; - } - } - - summary.retried_task_count = summary - .attempts_by_index - .iter() - .filter(|attempts| **attempts > 0) - .count(); - summary - } - - fn is_parallel_retry_safe(&self, task: &TaskParams) -> bool { - let Some(agent) = self.registry.get(&task.agent) else { - return false; - }; - if agent.tool_free { - return true; - } - - // A retry starts a fresh child run. Require an explicit deny for every - // mutation-capable tool; Ask is not safe because a parent confirmation - // policy may auto-approve it. This keeps retries to read-only agents. - let empty_args = serde_json::json!({}); - ["write", "edit", "patch", "batch", "bash", "git"] - .iter() - .all(|tool| agent.permissions.is_denied(tool, &empty_args)) - } -} - -fn remaining_parallel_timeout( - total_timeout_ms: Option, - started_at: std::time::Instant, -) -> Option { - total_timeout_ms.map(|timeout_ms| { - std::time::Duration::from_millis(timeout_ms.max(1)).saturating_sub(started_at.elapsed()) - }) -} - -fn parallel_retry_delay(attempt: u8) -> std::time::Duration { - let base = PARALLEL_RETRY_BASE_DELAY_MS.saturating_mul(1_u64 << u32::from(attempt.min(3))); - let jitter_range = (base / 4).max(1); - let entropy = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|duration| u64::from(duration.subsec_nanos())) - .unwrap_or_default(); - let jitter = entropy % (jitter_range.saturating_mul(2).saturating_add(1)); - std::time::Duration::from_millis(base.saturating_sub(jitter_range).saturating_add(jitter)) -} - -fn is_transient_parallel_failure(message: &str) -> bool { - let lower = message.to_ascii_lowercase(); - if [ - "cancelled", - "canceled", - "permission denied", - "not permitted", - "unknown agent", - "invalid api key", - "unauthorized", - "forbidden", - "quota exhausted", - "context length", - "structured output failed", - ] - .iter() - .any(|marker| lower.contains(marker)) - { - return false; - } - - crate::retry::is_transient_error(&lower) - || [ - "rate limit", - "too many requests", - "overloaded", - "server busy", - "temporarily unavailable", - "service unavailable", - "bad gateway", - "gateway timeout", - "status 408", - "status: 408", - "status 429", - "status: 429", - "status 500", - "status: 500", - "status 502", - "status: 502", - "status 503", - "status: 503", - "status 504", - "status: 504", - "status 529", - "status: 529", - ] - .iter() - .any(|marker| lower.contains(marker)) } async fn settle_cancelled_parallel_tasks( @@ -792,21 +575,4 @@ mod tests { ); assert!(active_indexes.is_empty()); } - - #[test] - fn transient_retry_classifier_rejects_deterministic_failures() { - assert!(is_transient_parallel_failure( - "LLM request failed with status 529: overloaded" - )); - assert!(is_transient_parallel_failure("connection reset by peer")); - assert!(!is_transient_parallel_failure( - "permission denied while writing src/main.rs" - )); - assert!(!is_transient_parallel_failure( - "structured output failed schema validation" - )); - assert!(!is_transient_parallel_failure( - "invalid API key returned status 503" - )); - } } diff --git a/core/src/tools/task/parallel_params.rs b/core/src/tools/task/parallel_params.rs index ce23c10c..1452f9f2 100644 --- a/core/src/tools/task/parallel_params.rs +++ b/core/src/tools/task/parallel_params.rs @@ -56,11 +56,6 @@ pub(super) fn parallel_task_params_schema_for_agents( "type": "string", "description": "Required. Detailed instruction for the delegated child run." }, - "background": { - "type": "boolean", - "description": "Optional. Run this delegated child task in the background. Default: false.", - "default": false - }, "max_steps": { "type": "integer", "description": "Optional. Maximum number of tool/model steps for this delegated child task." @@ -72,7 +67,7 @@ pub(super) fn parallel_task_params_schema_for_agents( }, "required": ["agent", "description", "prompt"] }, - "minItems": 1, + "minItems": 2, "maxItems": MAX_PARALLEL_TASKS_PER_CALL }, "allow_partial_failure": { diff --git a/core/src/tools/task/tests.rs b/core/src/tools/task/tests.rs index 92ac3bc4..97362662 100644 --- a/core/src/tools/task/tests.rs +++ b/core/src/tools/task/tests.rs @@ -840,7 +840,7 @@ fn test_parallel_task_params_schema() { schema["properties"]["tasks"]["maxItems"], MAX_PARALLEL_TASKS_PER_CALL ); - assert_eq!(schema["properties"]["tasks"]["minItems"], 1); + assert_eq!(schema["properties"]["tasks"]["minItems"], 2); assert_eq!( schema["properties"]["allow_partial_failure"]["type"], "boolean" @@ -870,8 +870,7 @@ fn test_parallel_task_params_schema_items() { assert!(item_required.contains(&serde_json::json!("agent"))); assert!(item_required.contains(&serde_json::json!("description"))); assert!(item_required.contains(&serde_json::json!("prompt"))); - assert_eq!(items["properties"]["background"]["type"], "boolean"); - assert_eq!(items["properties"]["background"]["default"], false); + assert!(items["properties"].get("background").is_none()); assert_eq!(items["properties"]["max_steps"]["type"], "integer"); assert_eq!(items["properties"]["output_schema"]["type"], "object"); } @@ -2976,6 +2975,70 @@ async fn concurrent_parallel_batches_share_eight_slots_and_execute_each_branch_o ); } +#[tokio::test] +async fn parallel_task_tool_rejects_arguments_that_conflict_with_its_contract() { + let workspace = tempfile::tempdir().unwrap(); + let executor = Arc::new(TaskExecutor::new( + test_registry_with_text_worker(), + Arc::new(StaticLlmClient::new("must not be used")), + workspace.path().to_string_lossy().to_string(), + )); + let tool = ParallelTaskTool::new(executor); + let ctx = ToolContext::new(workspace.path().to_path_buf()); + let task = |description: &str| { + serde_json::json!({ + "agent": "worker", + "description": description, + "prompt": format!("Run {description}") + }) + }; + + let invalid_arguments = [ + serde_json::json!({"tasks": [task("only branch")]}), + serde_json::json!({ + "tasks": [ + task("foreground branch"), + { + "agent": "worker", + "description": "background branch", + "prompt": "Run independently", + "background": true + } + ] + }), + serde_json::json!({ + "min_success_count": 1, + "tasks": [task("first branch"), task("second branch")] + }), + serde_json::json!({ + "allow_partial_failure": true, + "min_success_count": 3, + "tasks": [task("first branch"), task("second branch")] + }), + serde_json::json!({ + "allow_partial_failure": true, + "min_success_count": 0, + "tasks": [task("first branch"), task("second branch")] + }), + serde_json::json!({ + "timeout_ms": 0, + "tasks": [task("first branch"), task("second branch")] + }), + ]; + + for args in invalid_arguments { + let output = tool.execute(&args, &ctx).await.unwrap(); + assert!(!output.success, "arguments should fail: {args:#}"); + assert!( + matches!( + output.error_kind, + Some(crate::tools::ToolErrorKind::InvalidArgument { .. }) + ), + "expected InvalidArgument for {args:#}, got {output:#?}" + ); + } +} + #[tokio::test] async fn cancelling_a_batch_waiting_for_provider_capacity_returns_promptly() { let workspace = tempfile::tempdir().unwrap(); @@ -2996,11 +3059,19 @@ async fn cancelling_a_batch_waiting_for_provider_capacity_returns_promptly() { first_tool .execute( &serde_json::json!({ - "tasks": [{ - "agent": "worker", - "description": "Hold provider capacity", - "prompt": "blocking branch" - }] + "allow_partial_failure": true, + "tasks": [ + { + "agent": "worker", + "description": "Hold provider capacity", + "prompt": "blocking branch" + }, + { + "agent": "missing-agent", + "description": "Control branch", + "prompt": "Return without using provider capacity" + } + ] }), &first_context, ) @@ -3019,11 +3090,18 @@ async fn cancelling_a_batch_waiting_for_provider_capacity_returns_promptly() { second_tool .execute( &serde_json::json!({ - "tasks": [{ - "agent": "worker", - "description": "Wait for provider capacity", - "prompt": "waiting branch" - }] + "tasks": [ + { + "agent": "worker", + "description": "Wait for provider capacity", + "prompt": "waiting branch" + }, + { + "agent": "missing-agent", + "description": "Cancelled control branch", + "prompt": "Return without using provider capacity" + } + ] }), &second_context, ) @@ -3059,7 +3137,7 @@ async fn cancelling_a_batch_waiting_for_provider_capacity_returns_promptly() { } #[tokio::test] -async fn parallel_task_retries_only_the_failed_read_only_branch() { +async fn parallel_task_does_not_replay_a_read_only_branch_from_error_prose() { let workspace = tempfile::tempdir().unwrap(); let client = Arc::new(TransientThenSuccessLlmClient::new(2)); let executor = Arc::new(TaskExecutor::new( @@ -3073,11 +3151,19 @@ async fn parallel_task_retries_only_the_failed_read_only_branch() { let output = tool .execute( &serde_json::json!({ - "tasks": [{ - "agent": "explore", - "description": "Inspect retry behavior", - "prompt": "Read the relevant files and report evidence." - }] + "allow_partial_failure": true, + "tasks": [ + { + "agent": "explore", + "description": "Inspect retry behavior", + "prompt": "Read the relevant files and report evidence." + }, + { + "agent": "missing-agent", + "description": "Independent control branch", + "prompt": "Return a control result." + } + ] }), &ctx, ) @@ -3085,20 +3171,23 @@ async fn parallel_task_retries_only_the_failed_read_only_branch() { .unwrap(); assert!( - output.success, - "retry should recover the branch: {output:?}" + !output.success, + "untyped failure must remain failed: {output:?}" ); - assert!(output.content.contains("recovered:"), "{output:?}"); let metadata = output.metadata.expect("parallel metadata"); - assert_eq!(metadata["retry_attempt_count"], 1); - assert_eq!(metadata["retried_task_count"], 1); - assert_eq!(metadata["recovered_task_count"], 1); - assert_eq!(metadata["results"][0]["retry_attempts"], 1); - assert_eq!(client.calls(), 3, "two child attempts plus one retry"); + assert!(metadata.get("retry_attempt_count").is_none()); + assert!(metadata.get("retried_task_count").is_none()); + assert!(metadata.get("recovered_task_count").is_none()); + assert!(metadata["results"][0].get("retry_attempts").is_none()); + assert_eq!( + client.calls(), + 2, + "only the child LLM boundary may consume its configured attempt budget" + ); } #[tokio::test] -async fn parallel_retry_preserves_successful_siblings_without_replaying_them() { +async fn parallel_task_preserves_successful_siblings_without_prose_driven_replay() { let workspace = tempfile::tempdir().unwrap(); let client = Arc::new(SelectiveTransientLlmClient::new()); let executor = Arc::new(TaskExecutor::new( @@ -3131,21 +3220,19 @@ async fn parallel_retry_preserves_successful_siblings_without_replaying_them() { .unwrap(); assert!( - output.success, - "the isolated retry should recover: {output:#?}" + !output.success, + "one branch must remain failed: {output:#?}" ); let metadata = output.metadata.expect("parallel metadata"); - assert_eq!(metadata["retry_attempt_count"], 1); - assert_eq!(metadata["retried_task_count"], 1); - assert_eq!(metadata["recovered_task_count"], 1); - assert_eq!(metadata["results"][0]["retry_attempts"], 0); - assert_eq!(metadata["results"][1]["retry_attempts"], 1); + assert!(metadata.get("retry_attempt_count").is_none()); + assert!(metadata.get("retried_task_count").is_none()); + assert!(metadata.get("recovered_task_count").is_none()); + assert!(metadata["results"][0].get("retry_attempts").is_none()); + assert!(metadata["results"][1].get("retry_attempts").is_none()); assert!(metadata["results"][0]["output_excerpt"] .as_str() .is_some_and(|output| output.contains("stable branch"))); - assert!(metadata["results"][1]["output_excerpt"] - .as_str() - .is_some_and(|output| output.contains("flaky branch"))); + assert_eq!(metadata["results"][1]["success"], false); assert_eq!( client.attempts_for("stable"), 1, @@ -3153,8 +3240,8 @@ async fn parallel_retry_preserves_successful_siblings_without_replaying_them() { ); assert_eq!( client.attempts_for("flaky"), - 3, - "the child circuit breaker may try twice before one bounded branch retry" + 2, + "the task layer must not create an extra attempt from rendered error text" ); } @@ -3173,11 +3260,18 @@ async fn parallel_task_does_not_retry_a_mutating_branch() { let output = tool .execute( &serde_json::json!({ - "tasks": [{ - "agent": "general", - "description": "Potentially mutate the workspace", - "prompt": "Implement the requested change." - }] + "tasks": [ + { + "agent": "general", + "description": "Potentially mutate the workspace", + "prompt": "Implement the requested change." + }, + { + "agent": "missing-agent", + "description": "Independent failed control", + "prompt": "Return a control result." + } + ] }), &ctx, ) @@ -3186,10 +3280,10 @@ async fn parallel_task_does_not_retry_a_mutating_branch() { assert!(!output.success, "a mutating branch must not be replayed"); let metadata = output.metadata.expect("parallel metadata"); - assert_eq!(metadata["retry_attempt_count"], 0); - assert_eq!(metadata["retried_task_count"], 0); - assert_eq!(metadata["recovered_task_count"], 0); - assert_eq!(metadata["results"][0]["retry_attempts"], 0); + assert!(metadata.get("retry_attempt_count").is_none()); + assert!(metadata.get("retried_task_count").is_none()); + assert!(metadata.get("recovered_task_count").is_none()); + assert!(metadata["results"][0].get("retry_attempts").is_none()); assert_eq!( client.calls(), 2, @@ -3592,12 +3686,19 @@ async fn parallel_task_tool_returns_structured_child_output_when_schema_requeste let output = tool .execute( &serde_json::json!({ - "tasks": [{ - "agent": "worker", - "description": "Structured verdict", - "prompt": "Return a verdict.", - "output_schema": verdict_schema() - }] + "tasks": [ + { + "agent": "worker", + "description": "Structured verdict", + "prompt": "Return a verdict.", + "output_schema": verdict_schema() + }, + { + "agent": "worker", + "description": "Independent plain result", + "prompt": "Return a plain result." + } + ] }), &ctx, ) @@ -3822,23 +3923,34 @@ async fn parallel_task_metadata_carries_successful_child_read_anchor() { serde_json::json!({"file_path": "source.md"}), ), MockLlmClient::text_response("Evidence came from source.md."), + MockLlmClient::text_response("Independent source review complete."), ])); - let executor = Arc::new(TaskExecutor::new( - test_registry_with_writer(), - llm, - workspace.path().to_string_lossy().to_string(), - )); + let executor = Arc::new( + TaskExecutor::new( + test_registry_with_writer(), + llm, + workspace.path().to_string_lossy().to_string(), + ) + .with_max_parallel_tasks(1), + ); let tool = ParallelTaskTool::new(executor); let ctx = ToolContext::new(workspace.path().to_path_buf()); let output = tool .execute( &serde_json::json!({ - "tasks": [{ - "agent": "writer", - "description": "Read source", - "prompt": "Read source.md and report what it says." - }] + "tasks": [ + { + "agent": "writer", + "description": "Read source", + "prompt": "Read source.md and report what it says." + }, + { + "agent": "writer", + "description": "Review source independently", + "prompt": "Return a short independent review." + } + ] }), &ctx, ) diff --git a/core/src/tools/types.rs b/core/src/tools/types.rs index 03459ee7..f09413d9 100644 --- a/core/src/tools/types.rs +++ b/core/src/tools/types.rs @@ -683,6 +683,8 @@ pub enum ToolErrorKind { Unsupported { message: String }, /// The operation's outer timeout fired before the backend responded. Timeout { op: String, duration_ms: u64 }, + /// A typed network or upstream HTTP transport failure. + Transport { op: String }, /// The caller or owning session cancelled the operation. Cancelled { op: String }, /// A collection operation completed but one or more children failed. diff --git a/sdk/node/extra-types.d.ts b/sdk/node/extra-types.d.ts index 703612b1..92170a6b 100644 --- a/sdk/node/extra-types.d.ts +++ b/sdk/node/extra-types.d.ts @@ -24,6 +24,10 @@ export type ToolErrorKind = | { type: 'invalid_argument'; message: string } | { type: 'unsupported'; message: string } | { type: 'timeout'; op: string; duration_ms: number } + | { type: 'transport'; op: string } + | { type: 'cancelled'; op: string } + | { type: 'partial_failure'; failed: number; total: number } + | { type: 'rate_limited'; retry_after_ms: number | null } export type VerificationStatus = 'passed' | 'failed' | 'needs_review' | 'skipped' diff --git a/sdk/node/test-types.ts b/sdk/node/test-types.ts index 477c786d..2376e7fe 100644 --- a/sdk/node/test-types.ts +++ b/sdk/node/test-types.ts @@ -138,5 +138,12 @@ function _describe(err: ToolErrorKind): string { return err.message case 'timeout': return `${err.op} after ${err.duration_ms}ms` + case 'transport': + case 'cancelled': + return err.op + case 'partial_failure': + return `${err.failed}/${err.total} failed` + case 'rate_limited': + return err.retry_after_ms === null ? 'rate limited' : `retry after ${err.retry_after_ms}ms` } } diff --git a/website/docs/v6/en/guide/examples/structured-output.mdx b/website/docs/v6/en/guide/examples/structured-output.mdx index 3e9492f0..8c2ab344 100644 --- a/website/docs/v6/en/guide/examples/structured-output.mdx +++ b/website/docs/v6/en/guide/examples/structured-output.mdx @@ -1,15 +1,15 @@ --- title: 'Structured Output' -description: 'Generate schema-validated JSON objects with the generate_object tool.' +description: 'Generate schema-validated JSON values with the generate_object tool.' --- import { Tab, Tabs } from '@rspress/core/theme'; # Structured Output -The built-in `generate_object` tool asks the configured LLM for a JSON object, +The built-in `generate_object` tool asks the configured LLM for a JSON value, validates the response against a JSON Schema you supply, and returns the -validated object only on a zero-exit result. Use it whenever you need +validated value only on a zero-exit result. Use it whenever you need machine-readable results: extraction, classification, config generation, or feeding another program. @@ -161,7 +161,9 @@ print(obj["sentiment"], obj["confidence"]) # "negative" 0.97 ## Nested schemas and arrays -Schemas can nest objects and arrays to any depth, and the runtime validates the whole structure. This models real config files, manifests, or API payloads in one call. +Schemas can nest objects and arrays within the runtime's bounded schema-depth +limit, and the runtime validates the whole structure. This models real config +files, manifests, or API payloads in one call. @@ -276,15 +278,16 @@ The built-in validator supports: - `type` (including nullable arrays like `["string", "null"]`) - `required`, `properties`, `additionalProperties` - `enum`, `const` -- `anyOf`, `oneOf` +- `$ref` with local `$defs` or `definitions` +- `allOf`, `anyOf`, `oneOf` - `minLength`, `maxLength`, `pattern` - `minimum`, `maximum`, `exclusiveMinimum`, `exclusiveMaximum` - `minItems`, `maxItems`, `items` -- Nested object and array validation +- Nested object and array validation, including root arrays and scalar values ## Notes -- The validated value is on the `object` key of the parsed `result.output`. Pass `mode: 'tool'` for the current cross-provider default, or `mode: 'prompt'` for a prompt-only fallback. `auto`, `strict`, and `json` currently resolve to `tool` in this runtime path. +- The validated value is on the `object` key of the parsed `result.output`. `auto` selects forced-tool mode when the provider declares that capability and otherwise uses the prompt+schema fallback. Explicit `strict` and `json` modes use provider-native response formats only when supported, then safely fall back to forced-tool or prompt mode. - List every field you depend on in `required` — the runtime enforces it, so missing or mistyped fields fail validation instead of silently returning partial data. - `generate_object` is a built-in tool registered independently of built-in skills. - Direct `session.tool(...)` calls are host control-plane calls. Use `permissionPolicy` when you let the model choose tools inside `send` / `run` / `stream`; use host authorization before direct SDK calls. diff --git a/website/docs/v6/en/guide/tasks.mdx b/website/docs/v6/en/guide/tasks.mdx index a10933e9..6a3b72fa 100644 --- a/website/docs/v6/en/guide/tasks.mdx +++ b/website/docs/v6/en/guide/tasks.mdx @@ -88,6 +88,13 @@ if (batch.exitCode !== 0) throw new Error(batch.output); console.log(batch.output); ``` +Each `parallel_task` call requires at least two independent foreground tasks +and accepts at most 32. A branch cannot request `background`; the parent call +already owns concurrent execution and result collection. By default every +branch must succeed. Set `allow_partial_failure` only for evidence-gathering +work that can use incomplete results; `min_success_count` is available only in +that mode and must not exceed the submitted task count. + `session.task(...)` and `session.tasks(...)` return `ToolResult` values from the `task` and `parallel_task` tools. Read `output` for the compact child summary and check `exitCode` before treating the result as successful. `maxParallelTasks` diff --git a/website/docs/v6/en/guide/tools.mdx b/website/docs/v6/en/guide/tools.mdx index 45198801..ed7f76e7 100644 --- a/website/docs/v6/en/guide/tools.mdx +++ b/website/docs/v6/en/guide/tools.mdx @@ -129,6 +129,13 @@ indices while treating the orchestration as completed, so callers retry only failed items. `parallel_task` has the same 32-task bound and settles cancelled children before publishing terminal state. +`parallel_task` requires 2-32 independent foreground branches. Each branch +accepts `agent`, `description`, `prompt`, optional `max_steps`, and optional +`output_schema`; `background` is not a parallel-branch parameter. The optional +`min_success_count` is valid only with `allow_partial_failure: true` and must be +between 1 and the submitted task count. Provider and child runtimes own typed +retry policy; the fan-out layer never replays a branch based on error text. + `web_search` reports `complete`, `partial`, or `failed` in metadata. Empty results with engine errors are failures rather than successful empty searches. Timeout, cancellation, invalid-argument, partial-failure, and rate-limit errors @@ -136,9 +143,10 @@ carry structured error kinds. ## Structured Output with `generate_object` -The `generate_object` tool asks the configured LLM for a JSON object, validates -the response against a JSON Schema, and returns the validated object only on a -zero-exit result. It works in two ways: +The `generate_object` tool asks the configured LLM for a JSON value, validates +the response against a JSON Schema, and returns the validated value only on a +zero-exit result. Root objects, arrays, enums, constants, composition keywords, +and local `$ref` definitions are supported. It works in two ways: 1. **Agent-driven**: The LLM sees `generate_object` in its tool list and calls it autonomously when structured output is needed. 2. **Direct call**: Your application calls `session.tool('generate_object', ...)` to bypass model-driven tool selection. The tool itself still calls the configured LLM. @@ -169,23 +177,25 @@ const { object } = JSON.parse(result.output); ### Parameters -| Parameter | Type | Required | Description | -| --------------------- | ------- | -------- | ------------------------------------------------------------------------ | -| `schema` | object | yes | JSON Schema used to validate the output | -| `prompt` | string | yes | What to generate or extract | -| `schema_name` | string | no | Short name (default: "result") | -| `schema_description` | string | no | Description for the synthetic tool | -| `system` | string | no | Optional system prompt | -| `mode` | string | no | `"auto"` / `"strict"` / `"json"` / `"tool"` / `"prompt"` (default: auto) | -| `max_repair_attempts` | integer | no | 0–5 (default: 2) | -| `timeout_ms` | integer | no | Independent generation deadline, 1,000–600,000 ms (default: 120,000) | +| Parameter | Type | Required | Description | +| --------------------- | ------- | -------- | -------------------------------------------------------------------------------- | +| `schema` | object | yes | JSON Schema used to validate the output value | +| `prompt` | string | yes | Non-whitespace generation or extraction instruction | +| `schema_name` | string | no | 1-59 ASCII letters, digits, `_`, or `-` (default: `result`) | +| `schema_description` | string | no | Synthetic-tool description, at most 4,096 bytes | +| `system` | string | no | Optional system prompt, at most 32,768 bytes | +| `mode` | string | no | `"auto"` / `"strict"` / `"json"` / `"tool"` / `"prompt"` (default: auto) | +| `max_repair_attempts` | integer | no | 0-5 (default: 2) | +| `include_raw_text` | boolean | no | Include the provider text or tool arguments used for extraction (default: false) | +| `timeout_ms` | integer | no | Active generation deadline, 1,000-600,000 ms (default: 120,000) | ### Modes -- **tool**: Injects a synthetic tool whose parameters are the schema. This is the current cross-provider default. +- **tool**: Forces a synthetic tool whose parameters are the schema when the provider supports forced tool calls. - **prompt**: Appends schema instructions to the prompt. It is useful as a prompt-only fallback, but depends more on model compliance. -- **auto**: Currently resolves to `tool`. -- **strict** / **json**: Accepted by the API, but currently resolve to `tool` because provider-level `response_format` wiring is not enabled in this runtime path. +- **auto**: Selects forced-tool mode when available, otherwise prompt mode. +- **strict**: Uses provider-native strict JSON Schema when supported, otherwise safely falls back to forced-tool or prompt mode. +- **json**: Uses provider-native JSON-object mode when supported, otherwise safely falls back to forced-tool or prompt mode. ### Streaming