feat: align built-in tool contracts

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

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<Arc<dyn LlmClient>> {
+        self.inner
+            .fork_for_session(session_id)
+            .map(|inner| Arc::new(Self::new(inner, self.invocation.clone())) as Arc<dyn LlmClient>)
+    }
+
+    fn with_active_generation_timeout(&self, timeout: Duration) -> Option<Arc<dyn LlmClient>> {
+        self.inner
+            .with_active_generation_timeout(timeout)
+            .map(|inner| Arc::new(Self::new(inner, self.invocation.clone())) as Arc<dyn LlmClient>)
+    }
+
     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<dyn LlmClient> = Arc::new(SessionBindingClient {
+            bound_session: None,
+            observed_sessions: Arc::clone(&observed_sessions),
+        });
+        let invocation = InvocationContext::new(
+            Arc::<str>::from("parent-run"),
+            Arc::<str>::from("parent-session"),
+            CancellationToken::new(),
+            None,
+            InvocationGovernance::default(),
+        );
+        let governed: Arc<dyn LlmClient> = 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()]
+        );
+    }
 }

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 },
     };

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"))
                                     };
                                 }
                             }