pub(super) use super::super::profiles;
pub(super) use super::super::{
SerializableSubagentsOutput, SubagentProviderOverride, SubagentRunConfig, SubagentStatus,
SubagentTask, SubagentTaskResult, SubagentsArgs, SubagentsOutput, SubagentsSummary,
dispatch_subagents, run_subagents, run_subagents_with_wait,
};
pub(super) use super::super::{
activity::UsageAccumulator,
activity::{SubagentActivitySink, subagent_task_activity_metadata},
cwd::resolve_child_cwd,
dto::{MAX_SUBAGENT_TASK_CONTEXT_BYTES, MAX_SUBAGENT_TASK_INTENT_BYTES},
scheduler::{
PreparedSubagentTask, SCHEDULER_EVENT_CHANNEL_BOUND, SchedulerTaskReporter,
SchedulerWaitConfig, SharedTaskQueue, SharedTaskResults, SubagentScheduler,
TaskActivityFinisher, WorkerEvent, output_from_results, send_terminal_worker_event,
},
worker::{
SNAPSHOT_BYTE_LIMIT, SUBAGENT_PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
SUBAGENT_RESULT_ERROR_CHAR_LIMIT, SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT,
SUBAGENT_TRUNCATION_MARKER, SubagentRunInput, can_retry_subagent_provider,
child_agent_for_task, failed_result, failed_result_with_session_and_output,
failed_subagent_session_snapshot, run_one_subagent, subagent_compaction_instructions,
subagent_prompt, truncate_string_field, truncate_subagents_output,
},
};
pub(super) use crate::{
agent::{AgentOutputSink, AgentSession},
cancellation::AgentCancellation,
instructions::{InstructionFile, InstructionSourceKind},
output::{
ActivityEvent, ActivityId, ActivityKind, ActivityMetadata, ActivitySender, ActivityStatus,
OutputEvent,
},
providers::{Provider, ProviderError, ProviderEvent, ProviderRequest, ToolCall, Usage},
sessions::{SessionEvent, SessionManager},
skills::{DiscoveredSkill, SkillDiscovery, filter_enabled_skills},
tools::{ToolResult, ToolRuntime},
};
#[cfg(unix)]
pub(super) use crate::{
config::{HookDefinition, HookFailurePolicy, HookSettings},
hooks::HookRuntime,
};
pub(super) use serde_json::{Value, json};
pub(super) use std::{
collections::{BTreeMap, BTreeSet, HashSet},
path::{Path, PathBuf},
sync::atomic::{AtomicBool, AtomicUsize, Ordering},
sync::{Arc, Mutex},
thread,
time::Duration,
};
pub(super) struct CountingProvider {
pub(super) active: AtomicUsize,
pub(super) max_active: AtomicUsize,
pub(super) fail_on: String,
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
impl CountingProvider {
pub(super) fn new(fail_on: &str) -> Self {
Self {
active: AtomicUsize::new(0),
max_active: AtomicUsize::new(0),
fail_on: fail_on.to_string(),
requests: Mutex::new(Vec::new()),
}
}
}
impl Provider for CountingProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.lock().unwrap().push(request.clone());
let now = self.active.fetch_add(1, Ordering::SeqCst) + 1;
self.max_active.fetch_max(now, Ordering::SeqCst);
thread::sleep(Duration::from_millis(20));
self.active.fetch_sub(1, Ordering::SeqCst);
let messages = request.messages();
let user = messages
.last()
.map(|m| m.content.as_str())
.unwrap_or_default();
if user.contains(&self.fail_on) {
anyhow::bail!("planned failure");
}
on_event(ProviderEvent::TextDelta(
user.lines().next().unwrap_or_default().to_string(),
))?;
on_event(ProviderEvent::Done)?;
Ok(())
}
}
pub(super) struct TokenUsageProvider;
pub(super) struct LargeOutputProvider;
impl Provider for LargeOutputProvider {
fn stream_cancellable(
&self,
_request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
on_event(ProviderEvent::TextDelta(
"x".repeat(SUBAGENT_RESULT_OUTPUT_CHAR_LIMIT + 1024),
))?;
on_event(ProviderEvent::Done)?;
Ok(())
}
}
pub(super) struct AutoContinueChildProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
impl AutoContinueChildProvider {
pub(super) fn new() -> Self {
Self {
requests: Mutex::new(Vec::new()),
}
}
}
impl Provider for AutoContinueChildProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let attempt = {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
requests.len()
};
if attempt == 1 {
on_event(ProviderEvent::TextDelta("partial ".to_string()))?;
return Err(ProviderError::stream_failed_incomplete(
"provider stream ended prematurely after partial response; response is incomplete: provider stream no semantic progress before timeout",
)
.into());
}
on_event(ProviderEvent::TextDelta("done".to_string()))?;
on_event(ProviderEvent::Done)
}
}
pub(super) struct ChildCompactionProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
tool_call_count: usize,
partial_output: Option<String>,
}
impl ChildCompactionProvider {
pub(super) fn new_with_tool_call_count_and_partial_output(
tool_call_count: usize,
partial_output: Option<&str>,
) -> Self {
Self {
requests: Mutex::new(Vec::new()),
tool_call_count,
partial_output: partial_output.map(str::to_string),
}
}
}
impl Provider for ChildCompactionProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let attempt = {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
requests.len()
};
if attempt <= self.tool_call_count {
if let Some(partial_output) = &self.partial_output {
on_event(ProviderEvent::TextDelta(partial_output.clone()))?;
}
let id = match attempt {
1 => "child_compaction_read".to_string(),
2 => "child_limit_read".to_string(),
attempt => format!("child_limit_read_{attempt}"),
};
on_event(ProviderEvent::ToolCall(ToolCall {
id,
name: "read".to_string(),
arguments: json!({"paths":["large-child.txt"]}),
}))?;
} else {
on_event(ProviderEvent::TextDelta(
"child continued after compaction".to_string(),
))?;
}
on_event(ProviderEvent::Done)
}
}
pub(super) fn start_child_compaction_server_for_requests_with_expected_scope(
expected_requests: usize,
summary: &str,
expected_scope: Option<Value>,
) -> (String, std::thread::JoinHandle<anyhow::Result<()>>) {
use std::io::{Read, Write};
use std::net::TcpListener;
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let address = listener.local_addr().unwrap();
let summary = summary.to_string();
let handle = std::thread::spawn(move || {
let payload = format!(
"data: {}\n\ndata: {}\n\ndata: [DONE]\n\n",
json!({"choices":[{"delta":{"content":summary}}]}),
json!({"choices":[{"finish_reason":"stop"}], "usage":{"prompt_tokens":17,"completion_tokens":3,"total_tokens":29,"prompt_tokens_details":{"cached_tokens":5}}}),
);
for _ in 0..expected_requests {
let deadline = std::time::Instant::now() + Duration::from_secs(2);
let connection = loop {
match listener.accept() {
Ok(connection) => break Some(connection),
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
if std::time::Instant::now() >= deadline {
break None;
}
thread::sleep(Duration::from_millis(10));
}
Err(error) => return Err(error.into()),
}
};
let Some((mut stream, _)) = connection else {
break;
};
stream.set_nonblocking(false)?;
stream.set_read_timeout(Some(Duration::from_secs(2)))?;
let mut request = Vec::new();
let mut buffer = [0_u8; 8192];
let body_end = loop {
let count = stream.read(&mut buffer)?;
if count == 0 {
anyhow::bail!("child compaction test server received premature EOF");
}
request.extend_from_slice(&buffer[..count]);
let Some(header_end) = request.windows(4).position(|part| part == b"\r\n\r\n")
else {
continue;
};
let headers = std::str::from_utf8(&request[..header_end])?;
let content_length = headers
.lines()
.find_map(|line| {
let (name, value) = line.split_once(':')?;
name.eq_ignore_ascii_case("content-length")
.then(|| value.trim().parse::<usize>().ok())
.flatten()
})
.ok_or_else(|| anyhow::anyhow!("missing Content-Length"))?;
break header_end + 4 + content_length;
};
while request.len() < body_end {
let count = stream.read(&mut buffer)?;
if count == 0 {
anyhow::bail!("child compaction test server received truncated body");
}
request.extend_from_slice(&buffer[..count]);
}
if let Some(expected_scope) = &expected_scope {
let header_end = request
.windows(4)
.position(|part| part == b"\r\n\r\n")
.ok_or_else(|| anyhow::anyhow!("missing HTTP request headers"))?;
let body = serde_json::from_slice::<Value>(&request[header_end + 4..body_end])?;
let instruction = body
.get("messages")
.and_then(Value::as_array)
.and_then(|messages| messages.last())
.and_then(|message| message.get("content"))
.and_then(Value::as_str)
.ok_or_else(|| anyhow::anyhow!("missing child compaction instruction"))?;
let json_start = instruction
.find('{')
.ok_or_else(|| anyhow::anyhow!("missing child compaction scope document"))?;
if !instruction[..json_start].contains(
"following single JSON document extends to the end of this instruction",
) {
anyhow::bail!(
"child compaction scope document has no end-of-instruction prefix"
);
}
let scope = serde_json::from_str::<Value>(&instruction[json_start..])?;
let actual_payload = scope.get("original_task").ok_or_else(|| {
anyhow::anyhow!("child compaction scope has no original task")
})?;
if actual_payload != expected_scope {
anyhow::bail!(
"child compaction request did not preserve the original task payload"
);
}
}
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
payload.len(),
payload
);
stream.write_all(response.as_bytes())?;
stream.flush()?;
}
Ok(())
});
(format!("http://{address}/v1"), handle)
}
pub(super) struct PartialThenFailProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
impl PartialThenFailProvider {
pub(super) fn new() -> Self {
Self {
requests: Mutex::new(Vec::new()),
}
}
}
pub(super) fn retryable_provider_failure() -> anyhow::Error {
anyhow::Error::new(ProviderError::stream_failed_incomplete(
"provider stream ended with failed or incomplete response: code=server_error",
))
}
pub(super) struct RetryThenSucceedProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
impl RetryThenSucceedProvider {
pub(super) fn new() -> Self {
Self {
requests: Mutex::new(Vec::new()),
}
}
}
impl Provider for RetryThenSucceedProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let attempt = {
let mut requests = self.requests.lock().unwrap();
requests.push(request);
requests.len()
};
if attempt == 1 {
return Err(retryable_provider_failure());
}
on_event(ProviderEvent::TextDelta("retry succeeded".to_string()))?;
on_event(ProviderEvent::Done)?;
Ok(())
}
}
pub(super) struct AlwaysRetryableFailProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
impl AlwaysRetryableFailProvider {
pub(super) fn new() -> Self {
Self {
requests: Mutex::new(Vec::new()),
}
}
}
impl Provider for AlwaysRetryableFailProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
_on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.lock().unwrap().push(request);
Err(retryable_provider_failure())
}
}
pub(super) struct SchemaRetryProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
pub(super) always_invalid: bool,
pub(super) invalid_output: String,
pub(super) usage: Option<Usage>,
}
impl SchemaRetryProvider {
pub(super) fn new(always_invalid: bool) -> Self {
Self::new_with_invalid_output(always_invalid, "not json")
}
pub(super) fn new_with_invalid_output(always_invalid: bool, invalid_output: &str) -> Self {
Self {
requests: Mutex::new(Vec::new()),
always_invalid,
invalid_output: invalid_output.to_string(),
usage: None,
}
}
}
impl Provider for SchemaRetryProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let attempt = self.requests.lock().unwrap().len();
self.requests.lock().unwrap().push(request);
if self.always_invalid || attempt == 0 {
on_event(ProviderEvent::TextDelta(self.invalid_output.clone()))?;
} else {
on_event(ProviderEvent::TextDelta(
json!({
"phase": "IMPLEMENT",
"status": "COMPLETE",
"summary": "implemented",
"artifacts": [],
"verification": ["cargo test"],
"risks": [],
"changed_files": ["src/lib.rs"],
"tests_run": ["cargo test"],
"implementation_notes": ["fixed"]
})
.to_string(),
))?;
}
if let Some(usage) = &self.usage {
on_event(ProviderEvent::Usage(usage.clone()))?;
}
on_event(ProviderEvent::Done)?;
Ok(())
}
}
impl Provider for PartialThenFailProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.lock().unwrap().push(request);
on_event(ProviderEvent::Usage(Usage {
input: 40,
output: 2,
cache_read: 0,
cache_write: 0,
total: 123,
reasoning_tokens: None,
}))?;
on_event(ProviderEvent::TextDelta(
"checkpoint before failure".to_string(),
))?;
anyhow::bail!("planned partial failure")
}
}
impl Provider for TokenUsageProvider {
fn stream_cancellable(
&self,
_request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
on_event(ProviderEvent::Usage(Usage {
input: 40,
output: 2,
cache_read: 0,
cache_write: 0,
total: 123,
reasoning_tokens: None,
}))?;
on_event(ProviderEvent::TextDelta("done".to_string()))?;
on_event(ProviderEvent::Done)?;
Ok(())
}
}
pub(super) fn config(provider: Arc<dyn Provider>, cwd: &Path) -> SubagentRunConfig {
SubagentRunConfig {
parent_agent: AgentSession::new("model", &[], &crate::skills::SkillDiscovery::default()),
provider,
provider_override: None,
parent_tools: ToolRuntime::new(cwd).unwrap(),
parent_cwd: cwd.to_path_buf(),
cancellation: AgentCancellation::default(),
profiles: BTreeMap::new(),
subagent_profiles_prompt: None,
sessions_root: None,
parent_session_id: None,
depth: 0,
parent_activity_id: None,
activity_sender: None,
inherited_hooks: None,
semantic_progress_timeout: SUBAGENT_PROVIDER_STREAM_NO_SEMANTIC_PROGRESS_TIMEOUT,
schema_validation_max_retries: 2,
compaction: None,
}
}
pub(super) fn completed_result(id: &str, total_tokens: Option<u64>) -> SubagentTaskResult {
SubagentTaskResult {
id: id.to_string(),
status: SubagentStatus::Completed,
intent: format!("intent {id}"),
agent: None,
identity: None,
cwd: PathBuf::from("."),
session_id: None,
session_path: None,
total_tokens,
usage: None,
changed_files: Vec::new(),
output: "done".to_string(),
structured_output: None,
output_truncated: false,
error: None,
}
}
pub(super) struct WritingProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
pub(super) fail_on_continuation: bool,
}
impl WritingProvider {
pub(super) fn new() -> Self {
Self::new_with_failure(false)
}
pub(super) fn new_failing_on_continuation() -> Self {
Self::new_with_failure(true)
}
pub(super) fn new_with_failure(fail_on_continuation: bool) -> Self {
Self {
requests: Mutex::new(Vec::new()),
fail_on_continuation,
}
}
}
impl Provider for WritingProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let has_tool_result = !request.tool_results().is_empty();
self.requests.lock().unwrap().push(request);
if has_tool_result {
if self.fail_on_continuation {
return Err(retryable_provider_failure());
}
on_event(ProviderEvent::TextDelta("done".to_string()))?;
} else {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "write_1".to_string(),
name: "write".to_string(),
arguments: json!({"path":"child.txt","content":"made by child"}),
}))?;
}
on_event(ProviderEvent::Done)?;
Ok(())
}
}
#[cfg(unix)]
pub(super) struct PolicyToolProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
#[cfg(unix)]
impl PolicyToolProvider {
pub(super) fn new() -> Self {
Self {
requests: Mutex::new(Vec::new()),
}
}
}
#[cfg(unix)]
impl Provider for PolicyToolProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let has_tool_result = !request.tool_results().is_empty();
self.requests.lock().unwrap().push(request);
if has_tool_result {
on_event(ProviderEvent::TextDelta("child done".to_string()))?;
} else {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "write_policy_1".to_string(),
name: "write".to_string(),
arguments: json!({"path":"policy.txt","content":"target ran"}),
}))?;
}
on_event(ProviderEvent::Done)?;
Ok(())
}
}
pub(super) fn profile(id: &str, prompt: &str) -> profiles::SubagentProfile {
profiles::SubagentProfile {
id: id.to_string(),
name: id.to_string(),
description: format!("{id} description"),
model: None,
reasoning: None,
path: PathBuf::from(format!("{id}.md")),
prompt: prompt.to_string(),
output_schema: None,
disabled_tools: HashSet::new(),
}
}
pub(super) struct NestedSubagentsProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
pub(super) respect_schema_hiding: bool,
}
pub(super) struct ChildCancelingProvider {
pub(super) cancel: Arc<AtomicBool>,
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
pub(super) struct DrainOnCancellationProvider {
pub(super) entered: Mutex<Option<std::sync::mpsc::Sender<()>>>,
pub(super) release: Mutex<std::sync::mpsc::Receiver<()>>,
pub(super) observed_child_cancellation: AtomicBool,
pub(super) requests: Mutex<Vec<ProviderRequest>>,
}
impl DrainOnCancellationProvider {
pub(super) fn new(
entered: std::sync::mpsc::Sender<()>,
release: std::sync::mpsc::Receiver<()>,
) -> Self {
Self {
entered: Mutex::new(Some(entered)),
release: Mutex::new(release),
observed_child_cancellation: AtomicBool::new(false),
requests: Mutex::new(Vec::new()),
}
}
}
pub(super) struct StallingAfterToolResultProvider {
pub(super) requests: Mutex<Vec<ProviderRequest>>,
pub(super) saw_continuation: AtomicBool,
pub(super) saw_cancellation: AtomicBool,
pub(super) attempted_post_cancel_event: AtomicBool,
pub(super) unblock: AtomicBool,
}
impl StallingAfterToolResultProvider {
pub(super) fn new() -> Self {
Self {
requests: Mutex::new(Vec::new()),
saw_continuation: AtomicBool::new(false),
saw_cancellation: AtomicBool::new(false),
attempted_post_cancel_event: AtomicBool::new(false),
unblock: AtomicBool::new(false),
}
}
pub(super) fn unblock(&self) {
self.unblock.store(true, Ordering::SeqCst);
}
}
impl Provider for StallingAfterToolResultProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let has_tool_result = !request.tool_results().is_empty();
self.requests.lock().unwrap().push(request);
if has_tool_result {
self.saw_continuation.store(true, Ordering::SeqCst);
while !self.unblock.load(Ordering::SeqCst) {
if cancellation.is_canceled() {
self.saw_cancellation.store(true, Ordering::SeqCst);
self.attempted_post_cancel_event
.store(true, Ordering::SeqCst);
on_event(ProviderEvent::ToolCall(ToolCall {
id: "bash_after_cancel".to_string(),
name: "bash".to_string(),
arguments: json!({"command":"printf unsafe","timeout":5}),
}))?;
return Ok(());
}
thread::sleep(Duration::from_millis(10));
}
on_event(ProviderEvent::Done)?;
} else {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "bash_stall_1".to_string(),
name: "bash".to_string(),
arguments: json!({"command":"printf 'child bash ok'","timeout":5}),
}))?;
on_event(ProviderEvent::Done)?;
}
Ok(())
}
}
impl ChildCancelingProvider {
pub(super) fn new(cancel: Arc<AtomicBool>) -> Self {
Self {
cancel,
requests: Mutex::new(Vec::new()),
}
}
}
impl Provider for ChildCancelingProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.lock().unwrap().push(request);
on_event(ProviderEvent::TextDelta("before cancel".to_string()))?;
self.cancel.store(true, Ordering::SeqCst);
on_event(ProviderEvent::TextDelta("after cancel".to_string()))?;
on_event(ProviderEvent::Done)?;
Ok(())
}
}
impl Provider for DrainOnCancellationProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
self.requests.lock().unwrap().push(request);
if let Some(entered) = self.entered.lock().unwrap().take() {
let _ = entered.send(());
}
while !cancellation.is_canceled() {
thread::sleep(Duration::from_millis(5));
}
self.observed_child_cancellation
.store(true, Ordering::SeqCst);
self.release
.lock()
.unwrap()
.recv_timeout(Duration::from_secs(1))
.unwrap();
on_event(ProviderEvent::Done)?;
Ok(())
}
}
impl NestedSubagentsProvider {
pub(super) fn new() -> Self {
Self {
requests: Mutex::new(Vec::new()),
respect_schema_hiding: true,
}
}
pub(super) fn ignoring_schema_hiding() -> Self {
Self {
requests: Mutex::new(Vec::new()),
respect_schema_hiding: false,
}
}
}
impl Provider for NestedSubagentsProvider {
fn stream_cancellable(
&self,
request: ProviderRequest,
_cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let has_tool_result = !request.tool_results().is_empty();
self.requests.lock().unwrap().push(request.clone());
if self.respect_schema_hiding && !request.subagents_tool_enabled() {
on_event(ProviderEvent::TextDelta("schema hidden".to_string()))?;
} else if has_tool_result {
on_event(ProviderEvent::TextDelta("nested complete".to_string()))?;
} else {
on_event(ProviderEvent::ToolCall(ToolCall {
id: "nested_1".to_string(),
name: "subagents".to_string(),
arguments: json!({"tasks":[{"intent":"nested"}],"concurrency":1}),
}))?;
}
on_event(ProviderEvent::Done)?;
Ok(())
}
}