1use crate::approval::{ApprovalOutcome, request_approval};
2use crate::error::RuntimeError;
3use crate::event::{Event, FlowRunId, FlowStatus, NodeEvent};
4use crate::message::{Message, MessagePart, MessageRole};
5use crate::provider::LlmRequest;
6use crate::tool::{ApprovalLevel, BoxFut, Tier, Tool, ToolArgs, ToolCtx, ToolResult, ToolSpec};
7use crate::value::Value;
8use std::path::PathBuf;
9
10pub struct AgentSpawn;
11
12const DEFAULT_MAX_ITER: u64 = 20;
13const MAX_ITER_HARD_CAP: u64 = 200;
14
15impl Tool for AgentSpawn {
16 fn name(&self) -> &str {
17 "agent.spawn"
18 }
19
20 fn tier(&self) -> Tier {
21 Tier::Two
22 }
23
24 fn approval_level(&self, _args: &ToolArgs, _ctx: &ToolCtx) -> ApprovalLevel {
25 ApprovalLevel::Approve
26 }
27
28 fn description(&self) -> Option<&str> {
29 Some(
30 "Spawn an independent sub-agent to handle a focused sub-goal. The sub-agent runs its \
31 own message history and iteration counter, uses the same tool registry (or a subset \
32 you pick), and returns its final assistant text as this tool's result. Prefer this \
33 over doing large exploratory work directly when it would otherwise flood the main \
34 conversation with search output or scratch reasoning. Parameters: \
35 `goal` (required string), `tools` (optional list of tool-name strings — defaults \
36 to all tools available to you), `max_iterations` (optional int, default 20, capped \
37 at 200), `model` (optional model name — defaults to the last model this session \
38 used, then configured models, then claude-opus-4.7), `flow` (optional .at file path \
39 or command name; goal is passed as the first flow argument).",
40 )
41 }
42
43 fn input_schema(&self) -> serde_json::Value {
44 serde_json::json!({
45 "type": "object",
46 "properties": {
47 "goal": {"type": "string"},
48 "tools": {"type": "array", "items": {"type": "string"}},
49 "max_iterations": {"type": "integer"},
50 "model": {"type": "string"},
51 "flow": {"type": "string"}
52 },
53 "required": ["goal"]
54 })
55 }
56
57 fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
58 Box::pin(async move { run_sub_agent(args, ctx).await })
59 }
60}
61
62async fn run_sub_agent(args: ToolArgs, ctx: &ToolCtx) -> ToolResult {
63 let goal = extract_goal(&args)?;
64 if let Some(flow) = extract_flow(&args)? {
65 return run_flow_agent(&flow, goal, ctx).await;
66 }
67 let max_iter = extract_max_iter(&args);
68 let tool_filter = extract_tool_filter(&args)?;
69 let model = pick_model(&args, ctx);
70 let Some(providers) = ctx.providers.as_ref() else {
71 return Err(RuntimeError::ToolFailed(
72 "agent.spawn: no provider registry available on ctx".into(),
73 ));
74 };
75 let Some(provider) = providers.resolve(&model) else {
76 return Ok(Value::Str(format!(
77 "[sub-agent failed: no provider for model `{model}`]"
78 )));
79 };
80 let Some(registry) = ctx.registry.as_ref() else {
81 return Err(RuntimeError::ToolFailed(
82 "agent.spawn: no tool registry available on ctx".into(),
83 ));
84 };
85 let tool_specs = build_tool_specs(registry.as_ref(), tool_filter.as_deref());
86 let child_run_id = FlowRunId::now();
87 let task_id = ctx.task_registry.as_ref().map(|tr| {
88 tr.register(
89 crate::task_registry::TaskKind::Agent,
90 goal.clone(),
91 child_run_id.0.to_string(),
92 ctx.session_id.clone().unwrap_or_else(|| "anon".into()),
93 ctx.cancel.clone(),
94 )
95 });
96 emit_child_flow_start(ctx, &child_run_id, &goal);
97 let turn = ctx
98 .turn_id
99 .clone()
100 .unwrap_or_else(crate::event::TurnId::now);
101 let mut messages: Vec<Message> = vec![Message::user_text(turn.clone(), goal.clone())];
102 let mut final_text: Option<String> = None;
103 let mut failure_reason: Option<String> = None;
104 for iter in 0..max_iter {
105 if ctx.cancel.is_cancelled() {
106 failure_reason = Some("cancelled by parent".into());
107 break;
108 }
109 let req = LlmRequest {
110 model: model.clone(),
111 messages: messages.clone(),
112 system: None,
113 input: Value::Unit,
114 schema: None,
115 cache_prompt: false,
116 tools: tool_specs.clone(),
117 thinking_enabled: false,
118 stall_timeout_secs: 120,
119 };
120 let outcome = call_streaming_sub_agent(provider.as_ref(), req, ctx).await;
121 match outcome {
122 Ok(am) => {
123 emit_child_llm_call(ctx, &child_run_id, &model, &am);
124 let uses = extract_tool_uses(&am.message);
125 messages.push(am.message.clone());
126 emit_assistant_msg(ctx, &child_run_id, &am.message);
127 if uses.is_empty() {
128 final_text = Some(am.text_concat());
129 break;
130 }
131 let mut child_ctx = sanitize_child_ctx(ctx);
132 child_ctx.session_messages = Some(std::sync::Arc::new(messages.clone()));
133 let tool_results = dispatch_child_tools(&uses, registry.as_ref(), &child_ctx).await;
134 let turn_for_results = am.message.turn_id.clone();
135 let combined = Message {
136 turn_id: turn_for_results,
137 role: MessageRole::Tool,
138 parts: tool_results,
139 };
140 emit_tool_result_msg(ctx, &child_run_id, &combined);
141 messages.push(combined);
142 }
143 Err(e) => {
144 failure_reason = Some(format!("provider error at iter {iter}: {e}"));
145 break;
146 }
147 }
148 }
149 let status = if final_text.is_some() {
150 FlowStatus::Ok
151 } else {
152 FlowStatus::Errored {
153 message: failure_reason
154 .clone()
155 .unwrap_or_else(|| format!("hit max iterations {max_iter} without a final answer")),
156 }
157 };
158 emit_child_flow_end(ctx, &child_run_id, &status);
159 if let (Some(tr), Some(tid)) = (ctx.task_registry.as_ref(), &task_id) {
160 let ts = match &status {
161 FlowStatus::Ok => crate::task_registry::TaskStatus::Ok,
162 FlowStatus::Cancelled => crate::task_registry::TaskStatus::Killed,
163 FlowStatus::Errored { .. } => crate::task_registry::TaskStatus::Err,
164 };
165 tr.finish(tid, ts);
166 }
167 if let Some(text) = final_text {
168 Ok(Value::Str(text))
169 } else {
170 let reason = failure_reason
171 .unwrap_or_else(|| format!("hit max iterations {max_iter} without a final answer"));
172 let last = messages
173 .iter()
174 .rev()
175 .find(|m| matches!(m.role, MessageRole::Assistant))
176 .map(|m| m.text_concat())
177 .unwrap_or_default();
178 let partial = if last.is_empty() {
179 String::new()
180 } else {
181 format!("\n[partial output: {}]", truncate(&last, 400))
182 };
183 Ok(Value::Str(format!("[sub-agent failed: {reason}]{partial}")))
184 }
185}
186
187async fn run_flow_agent(flow_ref: &str, goal: String, ctx: &ToolCtx) -> ToolResult {
188 let Some(registry) = ctx.registry.as_ref() else {
189 return Err(RuntimeError::ToolFailed(
190 "agent.spawn: no tool registry available on ctx".into(),
191 ));
192 };
193 let Some(providers) = ctx.providers.as_ref() else {
194 return Err(RuntimeError::ToolFailed(
195 "agent.spawn: no provider registry available on ctx".into(),
196 ));
197 };
198 let (path, src) = read_flow_source(flow_ref).await?;
199 let file = atman_dsl::parse::parse_file(&src).map_err(|e| {
200 RuntimeError::ToolFailed(format!("agent.spawn: parse {}: {e}", path.display()))
201 })?;
202 let Some(flow) = file.flows.first() else {
203 return Err(RuntimeError::ToolFailed(format!(
204 "agent.spawn: no flow in {}",
205 path.display()
206 )));
207 };
208 let args = flow
209 .params
210 .first()
211 .map(|(ident, _)| vec![(ident.name.clone(), Value::Str(goal))])
212 .unwrap_or_default();
213 let flows = file
214 .flows
215 .iter()
216 .map(|flow| (flow.name.name.clone(), flow.clone()))
217 .collect();
218 let run_id = FlowRunId::now();
219 let task_id = ctx.task_registry.as_ref().map(|tr| {
220 tr.register(
221 crate::task_registry::TaskKind::Agent,
222 flow.name.name.clone(),
223 run_id.0.to_string(),
224 ctx.session_id.clone().unwrap_or_else(|| "anon".into()),
225 ctx.cancel.clone(),
226 )
227 });
228 emit_flow_agent_start(ctx, &run_id, &flow.name.name);
229 let mut child_ctx = sanitize_child_ctx(ctx);
230 child_ctx.session_messages = Some(std::sync::Arc::new(Vec::new()));
231 let out = crate::exec::exec_flow_with_siblings(
232 flow,
233 args,
234 registry.as_ref(),
235 &child_ctx,
236 providers.as_ref(),
237 &flows,
238 child_ctx.events.as_ref(),
239 child_ctx.turn_id.clone(),
240 Some(run_id.clone()),
241 None,
242 child_ctx.cancel.clone(),
243 None,
244 path.parent().map(|p| p.to_path_buf()),
245 )
246 .await;
247 let status = match &out {
248 Ok(_) => FlowStatus::Ok,
249 Err(e) => FlowStatus::Errored {
250 message: e.to_string(),
251 },
252 };
253 emit_child_flow_end(ctx, &run_id, &status);
254 if let (Some(tr), Some(tid)) = (ctx.task_registry.as_ref(), &task_id) {
255 let ts = match &status {
256 FlowStatus::Ok => crate::task_registry::TaskStatus::Ok,
257 FlowStatus::Cancelled => crate::task_registry::TaskStatus::Killed,
258 FlowStatus::Errored { .. } => crate::task_registry::TaskStatus::Err,
259 };
260 tr.finish(tid, ts);
261 }
262 out
263}
264
265fn extract_goal(args: &ToolArgs) -> Result<String, RuntimeError> {
266 match args.named("goal").or_else(|| args.positional.first()) {
267 Some(Value::Str(s)) if !s.trim().is_empty() => Ok(s.clone()),
268 Some(other) => Err(RuntimeError::TypeMismatch {
269 expected: "non-empty goal string".into(),
270 actual: other.kind_name().into(),
271 }),
272 None => Err(RuntimeError::MissingArg("agent.spawn.goal".into())),
273 }
274}
275
276fn extract_max_iter(args: &ToolArgs) -> u64 {
277 match args.named("max_iterations") {
278 Some(Value::Int(n)) if *n > 0 => (*n as u64).min(MAX_ITER_HARD_CAP),
279 _ => DEFAULT_MAX_ITER,
280 }
281}
282
283fn extract_flow(args: &ToolArgs) -> Result<Option<String>, RuntimeError> {
284 match args.named("flow") {
285 Some(Value::Str(s)) if !s.trim().is_empty() => Ok(Some(s.clone())),
286 Some(Value::Unit) | None => Ok(None),
287 Some(other) => Err(RuntimeError::TypeMismatch {
288 expected: "flow string".into(),
289 actual: other.kind_name().into(),
290 }),
291 }
292}
293
294fn extract_tool_filter(args: &ToolArgs) -> Result<Option<Vec<String>>, RuntimeError> {
295 match args.named("tools") {
296 Some(Value::List(items)) => {
297 let mut out = Vec::with_capacity(items.len());
298 for it in items {
299 match it {
300 Value::Str(s) => out.push(s.clone()),
301 other => {
302 return Err(RuntimeError::TypeMismatch {
303 expected: "string tool name".into(),
304 actual: other.kind_name().into(),
305 });
306 }
307 }
308 }
309 Ok(Some(out))
310 }
311 Some(Value::Unit) | None => Ok(None),
312 Some(other) => Err(RuntimeError::TypeMismatch {
313 expected: "list of tool names".into(),
314 actual: other.kind_name().into(),
315 }),
316 }
317}
318
319fn pick_model(args: &ToolArgs, ctx: &ToolCtx) -> String {
320 if let Some(Value::Str(s)) = args.named("model")
321 && !s.is_empty()
322 {
323 return crate::model_registry::resolve_alias(s);
324 }
325 if let Some(model) = &ctx.current_model {
326 return model.clone();
327 }
328 if let Some((name, _)) = crate::model_registry::all_model_entries().first() {
329 return name.clone();
330 }
331 "claude-opus-4.7".into()
332}
333
334async fn read_flow_source(flow_ref: &str) -> Result<(PathBuf, String), RuntimeError> {
335 for path in flow_candidates(flow_ref) {
336 match tokio::fs::read_to_string(&path).await {
337 Ok(src) => return Ok((path, src)),
338 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
339 Err(e) => {
340 return Err(RuntimeError::ToolFailed(format!(
341 "agent.spawn: read {}: {e}",
342 path.display()
343 )));
344 }
345 }
346 }
347 Err(RuntimeError::ToolFailed(format!(
348 "agent.spawn: flow `{flow_ref}` not found"
349 )))
350}
351
352fn flow_candidates(flow_ref: &str) -> Vec<PathBuf> {
353 let path = PathBuf::from(flow_ref);
354 if path.is_absolute() {
355 return vec![path];
356 }
357 let file_name = if flow_ref.ends_with(".at") {
358 flow_ref.to_string()
359 } else {
360 format!("{flow_ref}.at")
361 };
362 let mut out = Vec::new();
363 if let Some(home) = std::env::var_os("HOME") {
364 out.push(
365 PathBuf::from(home)
366 .join(".config")
367 .join("atman")
368 .join("commands")
369 .join(&file_name),
370 );
371 }
372 out.push(PathBuf::from(file_name));
373 out
374}
375
376async fn call_streaming_sub_agent(
377 provider: &dyn crate::provider::Provider,
378 req: LlmRequest,
379 ctx: &ToolCtx,
380) -> Result<crate::provider::AssistantMessage, RuntimeError> {
381 let model_name = req.model.clone();
382 let obs = provider.call_streaming(req);
383 let mut events = obs.events;
384 let output = obs.output;
385 tokio::pin!(output);
386 let result = loop {
387 tokio::select! {
388 ev = events.recv() => {
389 match ev {
390 Ok(event) => forward_stream_event(event, ctx, &model_name),
391 Err(tokio::sync::broadcast::error::RecvError::Closed) => {
392 break output.await;
394 }
395 Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
396 }
398 }
399 }
400 result = &mut output => break result,
401 }
402 };
403 while let Ok(ev) = events.try_recv() {
404 forward_stream_event(ev, ctx, &model_name);
405 }
406 result
407}
408
409fn forward_stream_event(ev: NodeEvent, ctx: &ToolCtx, model: &str) {
410 let Some(tx) = &ctx.stream_tx else {
411 return;
412 };
413 match ev {
414 NodeEvent::LlmChunk { text, .. } => {
415 let _ = tx.send(crate::stream::StreamFrame::LlmChunk {
416 text,
417 model: model.to_string(),
418 });
419 }
420 NodeEvent::ThinkingChunk { text } => {
421 let _ = tx.send(crate::stream::StreamFrame::ThinkingChunk { text });
422 }
423 NodeEvent::LlmDone { total_tokens } => {
424 let _ = tx.send(crate::stream::StreamFrame::LlmDone { total_tokens });
425 }
426 _ => {}
427 }
428}
429
430fn build_tool_specs(
431 registry: &crate::tool::ToolRegistry,
432 filter: Option<&[String]>,
433) -> Vec<ToolSpec> {
434 let mut specs = Vec::new();
435 for (name, tool) in registry.iter() {
436 if let Some(allow) = filter
437 && !allow.iter().any(|n| n == &name)
438 {
439 continue;
440 }
441 specs.push(crate::tool::tool_spec(tool.as_ref()));
442 }
443 specs
444}
445
446fn extract_tool_uses(msg: &Message) -> Vec<(String, String, Value)> {
447 let mut out = Vec::new();
448 for part in &msg.parts {
449 if let MessagePart::ToolUse { id, name, input } = part {
450 let value = Value::from_json(input.clone());
451 out.push((id.clone(), name.clone(), value));
452 }
453 }
454 out
455}
456
457async fn dispatch_child_tools(
458 uses: &[(String, String, Value)],
459 registry: &crate::tool::ToolRegistry,
460 ctx: &ToolCtx,
461) -> Vec<MessagePart> {
462 struct Ready {
463 idx: usize,
464 id: String,
465 name: String,
466 tool: std::sync::Arc<dyn crate::tool::Tool>,
467 call_args: ToolArgs,
468 }
469 let mut out: Vec<Option<MessagePart>> = vec![None; uses.len()];
470 let mut ready: Vec<Ready> = Vec::new();
471 for (idx, (id, name, input)) in uses.iter().enumerate() {
472 let Some(tool) = registry.get(name) else {
473 out[idx] = Some(MessagePart::ToolResult {
474 tool_use_id: id.clone(),
475 content: format!("sub-agent: unknown tool `{name}`"),
476 is_error: true,
477 });
478 continue;
479 };
480 let named = match input {
481 Value::Struct(fields) => fields.clone(),
482 Value::Unit => Vec::new(),
483 _ => Vec::new(),
484 };
485 ready.push(Ready {
486 idx,
487 id: id.clone(),
488 name: name.clone(),
489 tool,
490 call_args: ToolArgs {
491 positional: Vec::new(),
492 named,
493 },
494 });
495 }
496 for r in &ready {
498 emit_tool_use_start(ctx, &r.name, &r.id, &r.call_args);
499 }
500 let gates = ready.iter().map(|r| {
501 let level = r.tool.approval_level(&r.call_args, ctx);
502 request_approval(
503 ctx,
504 &r.id,
505 &r.name,
506 &r.call_args,
507 level,
508 Some(r.tool.as_ref()),
509 )
510 });
511 let outcomes = futures::future::join_all(gates).await;
512 for (r, gate) in ready.into_iter().zip(outcomes) {
513 let part = match gate {
514 ApprovalOutcome::Deny { reason } => MessagePart::ToolResult {
515 tool_use_id: r.id.clone(),
516 content: format!("sub-agent: tool `{}` denied — {reason}", r.name),
517 is_error: true,
518 },
519 ApprovalOutcome::Approve => match r.tool.call(r.call_args, ctx).await {
520 Ok(v) => MessagePart::ToolResult {
521 tool_use_id: r.id.clone(),
522 content: format_value(&v),
523 is_error: false,
524 },
525 Err(e) => MessagePart::ToolResult {
526 tool_use_id: r.id.clone(),
527 content: format!("{e}"),
528 is_error: true,
529 },
530 },
531 };
532 emit_tool_use_done(ctx, &r.name, &r.id, &part);
533 out[r.idx] = Some(part);
534 }
535 out.into_iter().flatten().collect()
536}
537
538fn emit_tool_use_start(ctx: &ToolCtx, name: &str, id: &str, args: &ToolArgs) {
539 if let Some(tx) = &ctx.stream_tx {
540 let _ = tx.send(crate::stream::StreamFrame::ToolUseStart {
541 tool: name.to_string(),
542 args_preview: preview_tool_args(args),
543 id: id.to_string(),
544 });
545 }
546}
547
548fn emit_tool_use_done(ctx: &ToolCtx, name: &str, id: &str, part: &MessagePart) {
549 if let Some(tx) = &ctx.stream_tx {
550 let (ok, preview) = match part {
551 MessagePart::ToolResult {
552 content, is_error, ..
553 } => (!is_error, truncate(content, 400)),
554 _ => (false, String::new()),
555 };
556 let _ = tx.send(crate::stream::StreamFrame::ToolUseDone {
557 tool: name.to_string(),
558 ok,
559 preview,
560 id: id.to_string(),
561 });
562 }
563}
564
565fn preview_tool_args(args: &ToolArgs) -> String {
566 let mut parts: Vec<String> = args.positional.iter().map(preview_value).collect();
567 for (k, v) in &args.named {
568 parts.push(format!("{k}={}", preview_value(v)));
569 }
570 truncate(&parts.join(", "), 4000)
571}
572
573fn preview_value(v: &Value) -> String {
574 match v {
575 Value::Str(s) => format!("{s:?}"),
576 Value::Int(n) => n.to_string(),
577 Value::Bool(b) => b.to_string(),
578 Value::Float(f) => f.to_string(),
579 Value::Unit => "()".into(),
580 Value::List(items) => format!("list[{}]", items.len()),
581 Value::Struct(items) => format!("struct[{}]", items.len()),
582 Value::Message(_) => "<message>".into(),
583 Value::Path(p) => p.display().to_string(),
584 Value::EditProposal(_) => "<edit proposal>".into(),
585 Value::Err(e) => format!("error({e})"),
586 }
587}
588
589fn format_value(v: &Value) -> String {
590 match v {
591 Value::Str(s) => s.clone(),
592 other => format!("{other:?}"),
593 }
594}
595
596fn emit_child_flow_start(ctx: &ToolCtx, run_id: &FlowRunId, goal: &str) {
597 let parent_run_id = ctx.flow_run_id.clone();
598 let parent_node_id = ctx.current_node_id.clone();
599 if let Some(sink) = &ctx.events {
600 sink.emit(Event::FlowStart {
601 run_id: run_id.clone(),
602 flow_name: "agent.sub".into(),
603 parent_run_id: parent_run_id.clone(),
604 parent_node_id: parent_node_id.clone(),
605 });
606 }
607 if let Some(tx) = &ctx.stream_tx {
608 let _ = tx.send(crate::stream::StreamFrame::FlowStart {
609 run_id: run_id.0.to_string(),
610 flow_name: format!("agent.sub · {}", truncate(goal, 60)),
611 parent_run_id: parent_run_id.as_ref().map(|r| r.0.to_string()),
612 parent_node_id,
613 });
614 }
615}
616
617fn emit_flow_agent_start(ctx: &ToolCtx, run_id: &FlowRunId, flow_name: &str) {
618 let parent_run_id = ctx.flow_run_id.clone();
619 let parent_node_id = ctx.current_node_id.clone();
620 if let Some(sink) = &ctx.events {
621 sink.emit(Event::FlowStart {
622 run_id: run_id.clone(),
623 flow_name: flow_name.into(),
624 parent_run_id: parent_run_id.clone(),
625 parent_node_id: parent_node_id.clone(),
626 });
627 }
628 if let Some(tx) = &ctx.stream_tx {
629 let _ = tx.send(crate::stream::StreamFrame::FlowStart {
630 run_id: run_id.0.to_string(),
631 flow_name: flow_name.into(),
632 parent_run_id: parent_run_id.as_ref().map(|r| r.0.to_string()),
633 parent_node_id,
634 });
635 }
636}
637
638fn emit_child_flow_end(ctx: &ToolCtx, run_id: &FlowRunId, status: &FlowStatus) {
639 if let Some(sink) = &ctx.events {
640 sink.emit(Event::FlowEnd {
641 run_id: run_id.clone(),
642 flow_name: "agent.sub".into(),
643 status: status.clone(),
644 });
645 }
646 if let Some(tx) = &ctx.stream_tx {
647 let _ = tx.send(crate::stream::StreamFrame::FlowDone {
648 run_id: run_id.0.to_string(),
649 flow_name: "agent.sub".into(),
650 ok: matches!(status, FlowStatus::Ok),
651 cancelled: false,
652 });
653 }
654}
655
656fn emit_child_llm_call(
657 ctx: &ToolCtx,
658 _run_id: &FlowRunId,
659 model: &str,
660 am: &crate::provider::AssistantMessage,
661) {
662 if let Some(sink) = &ctx.events {
663 sink.emit(Event::LlmCall {
664 model: model.into(),
665 provider: "sub".into(),
666 usage: am.token_usage.clone(),
667 wallclock_ms: 0,
668 ttft_ms: am.timing.ttft_ms,
669 tokens_per_second: am.timing.tokens_per_second(am.token_usage.output),
670 status: crate::event::LlmCallStatus::Ok,
671 run_id: None,
672 node_id: None,
673 });
674 }
675}
676
677fn emit_assistant_msg(ctx: &ToolCtx, run_id: &FlowRunId, message: &Message) {
678 let turn_id = message.turn_id.clone();
679 if let Some(sink) = &ctx.events {
680 sink.emit(Event::AssistantMsg {
681 turn_id: turn_id.clone(),
682 flow_run_id: Some(run_id.clone()),
683 message: message.clone(),
684 });
685 }
686 if let Some(tx) = &ctx.stream_tx {
687 let _ = tx.send(crate::stream::StreamFrame::AssistantMsg {
688 flow_run_id: Some(run_id.0.to_string()),
689 message: message.clone(),
690 });
691 }
692}
693
694fn emit_tool_result_msg(ctx: &ToolCtx, run_id: &FlowRunId, message: &Message) {
695 let message =
696 crate::tools::tool_output::maybe_truncate_tool_message(message, ctx.session_dir.as_deref());
697 let turn_id = message.turn_id.clone();
698 if let Some(sink) = &ctx.events {
699 sink.emit(Event::ToolResultMsg {
700 turn_id: turn_id.clone(),
701 flow_run_id: Some(run_id.clone()),
702 message: message.clone(),
703 });
704 }
705 if let Some(tx) = &ctx.stream_tx {
706 let _ = tx.send(crate::stream::StreamFrame::ToolResultMsg {
707 flow_run_id: Some(run_id.0.to_string()),
708 message: message.clone(),
709 });
710 }
711}
712
713fn sanitize_child_ctx(parent: &ToolCtx) -> ToolCtx {
714 let mut c = parent.clone();
715 c.session_runtime = None;
716 c.session_messages_handle = None;
717 c.compact_lock_handle = None;
718 c.forms = None;
719 c.on_memory_recent = None;
720 c
721}
722
723fn truncate(s: &str, n: usize) -> String {
724 let chars: Vec<char> = s.chars().collect();
725 if chars.len() <= n {
726 s.to_string()
727 } else {
728 chars.iter().take(n).collect::<String>() + "…"
729 }
730}