use crate::error::RuntimeError;
use crate::event::{Event, FlowRunId, FlowStatus};
use crate::message::Message;
use crate::tool::{ApprovalLevel, BoxFut, Tier, Tool, ToolArgs, ToolCtx, ToolResult};
use crate::value::Value;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
pub struct AgentSpawn;
#[derive(Debug, Clone)]
pub enum FlowRunStatus {
Running {
started_at: chrono::DateTime<chrono::Utc>,
},
Ok {
ended_at: chrono::DateTime<chrono::Utc>,
final_text: String,
},
Err {
ended_at: chrono::DateTime<chrono::Utc>,
message: String,
},
Killed {
ended_at: chrono::DateTime<chrono::Utc>,
},
}
impl FlowRunStatus {
pub fn is_running(&self) -> bool {
matches!(self, Self::Running { .. })
}
pub fn kind_str(&self) -> &'static str {
match self {
Self::Running { .. } => "running",
Self::Ok { .. } => "ok",
Self::Err { .. } => "err",
Self::Killed { .. } => "killed",
}
}
}
#[derive(Debug, Clone)]
pub enum FlowEvent {
AssistantDone { text: String },
Exited { status: FlowRunStatus },
}
pub struct FlowEntry {
pub handle: String,
pub goal: String,
pub status: Arc<Mutex<FlowRunStatus>>,
pub output: Arc<Mutex<String>>,
pub cancel: tokio_util::sync::CancellationToken,
pub stream_tx: tokio::sync::broadcast::Sender<FlowEvent>,
pub messages: Arc<Mutex<Vec<Message>>>,
pub iteration: Arc<std::sync::atomic::AtomicU64>,
pub child_run_id: FlowRunId,
pub model: String,
pub started_at: chrono::DateTime<chrono::Utc>,
pub compact_lock: Arc<tokio::sync::Mutex<()>>,
pub interjection_tx: tokio::sync::broadcast::Sender<crate::injection::Injection>,
pub pending_injections: Arc<std::sync::Mutex<Vec<crate::injection::Injection>>>,
pub injection_notify: Arc<tokio::sync::Notify>,
pub frame_tx: tokio::sync::broadcast::Sender<crate::stream::StreamFrame>,
}
impl crate::watch::Watchable for FlowEntry {
fn watch_output(
self: Arc<Self>,
pattern: String,
cancel: tokio_util::sync::CancellationToken,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = crate::watch::WatchResult> + Send>>
{
let stream_tx = self.stream_tx.clone();
let output = self.output.clone();
let status = self.status.clone();
Box::pin(async move {
{
let existing = output.lock().unwrap().clone();
if existing.find(&pattern).is_some() {
return crate::watch::WatchResult::Matched {
row: None,
col: None,
text: existing,
};
}
}
let mut rx = stream_tx.subscribe();
{
let st = status.lock().unwrap().clone();
if !st.is_running() {
if let FlowRunStatus::Ok { final_text, .. } = &st {
if final_text.find(&pattern).is_some() {
return crate::watch::WatchResult::Matched {
row: None,
col: None,
text: final_text.clone(),
};
}
}
return crate::watch::WatchResult::SourceExited;
}
}
loop {
tokio::select! {
_ = cancel.cancelled() => return crate::watch::WatchResult::Cancelled,
result = rx.recv() => match result {
Ok(FlowEvent::AssistantDone { text }) => {
if text.find(&pattern).is_some() {
return crate::watch::WatchResult::Matched {
row: None,
col: None,
text,
};
}
}
Ok(FlowEvent::Exited { status }) => {
if let FlowRunStatus::Ok { final_text, .. } = &status {
if final_text.find(&pattern).is_some() {
return crate::watch::WatchResult::Matched {
row: None,
col: None,
text: final_text.clone(),
};
}
}
return crate::watch::WatchResult::SourceExited;
}
Err(_) => return crate::watch::WatchResult::SourceExited,
}
}
}
})
}
}
#[derive(Default)]
pub struct FlowRegistry {
entries: Mutex<std::collections::HashMap<String, Arc<FlowEntry>>>,
}
impl FlowRegistry {
pub fn new() -> Self {
Self::default()
}
pub fn create_entry(
&self,
handle: String,
goal: String,
model: String,
child_run_id: FlowRunId,
) -> Arc<FlowEntry> {
let (stream_tx, _) = tokio::sync::broadcast::channel(64);
let entry = Arc::new(FlowEntry {
handle: handle.clone(),
goal,
status: Arc::new(Mutex::new(FlowRunStatus::Running {
started_at: chrono::Utc::now(),
})),
output: Arc::new(Mutex::new(String::new())),
cancel: tokio_util::sync::CancellationToken::new(),
stream_tx,
messages: Arc::new(Mutex::new(Vec::new())),
iteration: Arc::new(std::sync::atomic::AtomicU64::new(0)),
child_run_id,
model,
started_at: chrono::Utc::now(),
compact_lock: Arc::new(tokio::sync::Mutex::new(())),
interjection_tx: tokio::sync::broadcast::channel(32).0,
pending_injections: Arc::new(std::sync::Mutex::new(Vec::new())),
injection_notify: Arc::new(tokio::sync::Notify::new()),
frame_tx: tokio::sync::broadcast::channel(256).0,
});
self.entries
.lock()
.unwrap()
.insert(handle, Arc::clone(&entry));
entry
}
pub fn lookup(&self, handle: &str) -> Result<Arc<FlowEntry>, RuntimeError> {
self.entries
.lock()
.unwrap()
.get(handle)
.map(Arc::clone)
.ok_or_else(|| RuntimeError::ToolFailed(format!("agent: handle '{handle}' not found")))
}
pub fn remove(&self, handle: &str) {
self.entries.lock().unwrap().remove(handle);
}
}
impl Tool for AgentSpawn {
fn name(&self) -> &str {
"flow.spawn"
}
fn tier(&self) -> Tier {
Tier::Two
}
fn approval_level(&self, _args: &ToolArgs, _ctx: &ToolCtx) -> ApprovalLevel {
ApprovalLevel::Approve
}
fn description(&self) -> Option<&str> {
Some(
"Spawn a DSL flow as an independent sub-agent with its own message history and \
iteration counter. All named args except `flow` and `async` pass through to the \
flow as parameters — call flow.list first to discover available flows and their \
parameter signatures.\n\n\
Flow reference syntax: `file@flow_name`\n\
- \"subagent.at@subagent\" — run the `subagent` flow in subagent.at\n\
- \"subagent@research_loop\" — .at suffix optional\n\
- \"subagent.at\" — no @, takes first non-describe flow\n\
- \"/abs/path/my.at@main\" — absolute path\n\n\
Default flow is `subagent.at` (research/verify/implement/review roles). \
Required: `flow`. `async` is optional (default true). Other named args pass through to the flow. \
Use flow.status/flow.output/flow.kill to manage async sub-agents by handle. \
Best practice: call flow.list to see available flows and params, then pass \
matching named args. Missing params use flow-defined defaults.",
)
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({
"type": "object",
"properties": {
"flow": {"type": "string", "description": "Flow reference (e.g. \"subagent.at@subagent\")."},
"arguments": {
"type": "object",
"additionalProperties": true,
"description": "Target flow parameters as key-value pairs. You MUST call flow.list first to discover the flow's description and parameter signatures (names, types, required/optional), then construct this object accordingly. Example: arguments={\"goal\":\"read Cargo.toml\",\"role\":\"research\"}"
},
"async": {"type": "boolean", "default": true, "description": "If true (default), run in background and return a handle. If false, block until done."},
"inherit_context": {"type": "boolean", "default": false, "description": "If true, seed the sub-agent's context with a snapshot of the parent's messages."}
},
"required": ["flow"]
})
}
fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move {
let is_async = args
.named("async")
.and_then(|v| {
if let Value::Bool(b) = v {
Some(*b)
} else {
None
}
})
.unwrap_or(true);
if is_async {
run_sub_agent_async(args, ctx).await
} else {
run_sub_agent(args, ctx).await
}
})
}
}
async fn run_sub_agent(args: ToolArgs, ctx: &ToolCtx) -> ToolResult {
let flow = extract_flow(&args)?.unwrap_or_else(|| "subagent.at".to_string());
run_flow_agent(&flow, &args, ctx, FlowRunId::now()).await
}
async fn run_sub_agent_async(args: ToolArgs, ctx: &ToolCtx) -> ToolResult {
let arg_fields: Vec<(String, Value)> = match args.named("arguments") {
Some(Value::Struct(fields)) => fields.clone(),
_ => Vec::new(),
};
let display_label = arg_fields
.iter()
.find_map(|(_, v)| {
if let Value::Str(s) = v {
Some(s.clone())
} else {
None
}
})
.unwrap_or_default();
let inherit_context = args
.named("inherit_context")
.map(|v| matches!(v, Value::Bool(true)))
.unwrap_or(false);
let flow_registry = ctx.flow_registry.clone().ok_or_else(|| {
RuntimeError::ToolFailed("flow.spawn: no agent registry available on ctx".into())
})?;
let flow_ref = extract_flow(&args)?.unwrap_or_else(|| "subagent.at".to_string());
let handle = format!("agent_{}", uuid::Uuid::now_v7().simple());
let child_run_id = FlowRunId::now();
let entry = flow_registry.create_entry(
handle.clone(),
display_label,
String::new(),
child_run_id.clone(),
);
if inherit_context {
if let Some(parent) = &ctx.session_messages_handle {
let snapshot = parent.lock().unwrap().clone();
*entry.messages.lock().unwrap() = snapshot;
}
}
let task_registry = ctx.task_registry.clone();
let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".into());
let task_id = task_registry.as_ref().map(|tr| {
tr.register(
crate::task_registry::TaskKind::Flow,
entry.goal.clone(),
handle.clone(),
session_id,
entry.cancel.clone(),
)
});
let entry_clone = Arc::clone(&entry);
let ctx_clone = ctx.clone();
let parent_stream_tx = ctx.stream_tx.clone();
let child_run_id_str = child_run_id.0.to_string();
tokio::spawn(async move {
if let Some(tx) = &parent_stream_tx {
let _ = tx.send(crate::stream::StreamFrame::SubAgentStarted {
handle: entry_clone.handle.clone(),
goal: entry_clone.goal.clone(),
child_run_id: child_run_id_str.clone(),
model: entry_clone.model.clone(),
});
}
let mut ctx_for_flow = ctx_clone;
ctx_for_flow.cancel = entry_clone.cancel.clone();
ctx_for_flow.agent_entry = Some(Arc::clone(&entry_clone));
ctx_for_flow.compact_lock_handle = Some(Arc::clone(&entry_clone.compact_lock));
let result = run_flow_agent(&flow_ref, &args, &ctx_for_flow, child_run_id.clone()).await;
let status = match &result {
Ok(Value::Str(s)) => FlowRunStatus::Ok {
ended_at: chrono::Utc::now(),
final_text: s.clone(),
},
Ok(_) => FlowRunStatus::Ok {
ended_at: chrono::Utc::now(),
final_text: String::new(),
},
Err(e) => FlowRunStatus::Err {
ended_at: chrono::Utc::now(),
message: e.to_string(),
},
};
*entry_clone.status.lock().unwrap() = status.clone();
if let Some(tx) = &parent_stream_tx {
let final_text = match &status {
FlowRunStatus::Ok { final_text, .. } => final_text.clone(),
_ => String::new(),
};
let _ = tx.send(crate::stream::StreamFrame::SubAgentDone {
handle: entry_clone.handle.clone(),
status: status.kind_str().to_string(),
final_text,
});
}
let _ = entry_clone.stream_tx.send(FlowEvent::Exited { status });
if let (Some(tr), Some(tid)) = (&task_registry, &task_id) {
let ts = match result {
Ok(_) => crate::task_registry::TaskStatus::Ok,
Err(_) => crate::task_registry::TaskStatus::Err,
};
tr.finish(tid, ts);
}
});
Ok(Value::Struct(vec![
("handle".into(), Value::Str(handle)),
("status".into(), Value::Str("running".into())),
]))
}
pub struct AgentStatus;
impl Tool for AgentStatus {
fn name(&self) -> &str {
"flow.status"
}
fn tier(&self) -> Tier {
Tier::Zero
}
fn description(&self) -> Option<&str> {
Some("Check the status of an async sub-agent. Returns handle, status, goal, and timing.")
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({
"type": "object",
"properties": {"handle": {"type": "string"}},
"required": ["handle"]
})
}
fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move {
let handle = extract_string(&args, "handle", 0)?;
let reg = ctx
.flow_registry
.clone()
.ok_or_else(|| RuntimeError::ToolFailed("flow.status: no agent registry".into()))?;
let entry = reg.lookup(&handle)?;
let st = entry.status.lock().unwrap().clone();
let goal = entry.goal.clone();
let mut fields = vec![
("handle".into(), Value::Str(handle)),
("status".into(), Value::Str(st.kind_str().into())),
("goal".into(), Value::Str(goal)),
];
if let FlowRunStatus::Err { message, .. } = &st {
fields.push(("error".into(), Value::Str(message.clone())));
}
Ok(Value::Struct(fields))
})
}
}
pub struct AgentOutput;
impl Tool for AgentOutput {
fn name(&self) -> &str {
"flow.output"
}
fn tier(&self) -> Tier {
Tier::Zero
}
fn description(&self) -> Option<&str> {
Some("Read accumulated assistant text from an async sub-agent.")
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({
"type": "object",
"properties": {
"handle": {"type": "string"},
"cursor": {"type": "integer", "default": 0},
"limit": {"type": "integer", "default": 4096}
},
"required": ["handle"]
})
}
fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move {
let handle = extract_string(&args, "handle", 0)?;
let cursor = args
.named("cursor")
.and_then(|v| {
if let Value::Int(n) = v {
Some(*n)
} else {
None
}
})
.unwrap_or(0)
.max(0) as usize;
let limit = args
.named("limit")
.and_then(|v| {
if let Value::Int(n) = v {
Some(*n)
} else {
None
}
})
.unwrap_or(4096)
.max(1) as usize;
let reg = ctx
.flow_registry
.clone()
.ok_or_else(|| RuntimeError::ToolFailed("flow.output: no agent registry".into()))?;
let entry = reg.lookup(&handle)?;
let output = entry.output.lock().unwrap().clone();
let chunk = output.chars().skip(cursor).take(limit).collect::<String>();
let next_cursor = cursor + chunk.chars().count();
let eof = next_cursor >= output.chars().count();
Ok(Value::Struct(vec![
("handle".into(), Value::Str(handle)),
("chunk".into(), Value::Str(chunk)),
("cursor".into(), Value::Int(cursor as i64)),
("next_cursor".into(), Value::Int(next_cursor as i64)),
("eof".into(), Value::Bool(eof)),
]))
})
}
}
pub struct AgentKill;
impl Tool for AgentKill {
fn name(&self) -> &str {
"flow.kill"
}
fn tier(&self) -> Tier {
Tier::Four
}
fn description(&self) -> Option<&str> {
Some("Cancel a running async sub-agent by handle.")
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({
"type": "object",
"properties": {"handle": {"type": "string"}},
"required": ["handle"]
})
}
fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move {
let handle = extract_string(&args, "handle", 0)?;
let reg = ctx
.flow_registry
.clone()
.ok_or_else(|| RuntimeError::ToolFailed("flow.kill: no agent registry".into()))?;
let entry = reg.lookup(&handle)?;
entry.cancel.cancel();
Ok(Value::Unit)
})
}
}
pub struct FlowInterject;
impl Tool for FlowInterject {
fn name(&self) -> &str {
"flow.interject"
}
fn tier(&self) -> Tier {
Tier::Two
}
fn approval_level(&self, _args: &ToolArgs, _ctx: &ToolCtx) -> ApprovalLevel {
ApprovalLevel::Approve
}
fn description(&self) -> Option<&str> {
Some(
"Interject a text message into a running FlowRun (root or sub-agent) by handle. \
L1 (nudge): inject text into context, flow continues. \
L2 (course_correct): inject text, cancel current LLM call, flow continues with correction. \
L3 (redirect): cancel current LLM call, redirect to a different flow. \
L4 (hard_stop): kill the FlowRun immediately. \
Use handle \"root\" to interject into the main agent.",
)
}
fn input_schema(&self) -> serde_json::Value {
serde_json::json!({
"type": "object",
"properties": {
"handle": {"type": "string", "description": "Target FlowRun handle (e.g. from flow.spawn return, or \"root\")."},
"text": {"type": "string", "description": "Interjection text."},
"level": {"type": "string", "enum": ["l1_nudge", "l2_course_correct", "l3_redirect", "l4_hard_stop"], "default": "l1_nudge", "description": "Interjection level."},
"redirect_target": {"type": "string", "description": "Required for L3 redirect: the flow to redirect to."}
},
"required": ["handle", "text"]
})
}
fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
Box::pin(async move {
let handle = extract_string(&args, "handle", 0)?;
let text = extract_string(&args, "text", 1)?;
let level_str = args
.named("level")
.and_then(|v| {
if let Value::Str(s) = v {
Some(s.clone())
} else {
None
}
})
.unwrap_or_else(|| "l1_nudge".to_string());
let level = match level_str.as_str() {
"l2_course_correct" => crate::injection::InjectionLevel::L2CourseCorrect,
"l3_redirect" => crate::injection::InjectionLevel::L3Redirect,
"l4_hard_stop" => crate::injection::InjectionLevel::L4HardStop,
_ => crate::injection::InjectionLevel::L1Nudge,
};
let redirect_target = args.named("redirect_target").and_then(|v| {
if let Value::Str(s) = v {
Some(s.clone())
} else {
None
}
});
let reg = ctx.flow_registry.clone().ok_or_else(|| {
RuntimeError::ToolFailed("flow.interject: no agent registry".into())
})?;
let entry = reg.lookup(&handle)?;
let inj = crate::injection::Injection::with_level(
crate::event::TurnId::now(),
text,
level,
redirect_target,
);
entry.pending_injections.lock().unwrap().push(inj);
entry.injection_notify.notify_one();
Ok(Value::Unit)
})
}
}
fn extract_string(args: &ToolArgs, name: &str, pos: usize) -> Result<String, RuntimeError> {
let value = match args.named(name) {
Some(v) => v,
None => args.positional(pos)?,
};
match value {
Value::Str(s) => Ok(s.clone()),
other => Err(RuntimeError::TypeMismatch {
expected: "string".into(),
actual: other.kind_name().into(),
}),
}
}
async fn run_flow_agent(
flow_ref: &str,
args: &ToolArgs,
ctx: &ToolCtx,
run_id: FlowRunId,
) -> ToolResult {
let Some(registry) = ctx.registry.as_ref() else {
return Err(RuntimeError::ToolFailed(
"flow.spawn: no tool registry available on ctx".into(),
));
};
let Some(providers) = ctx.providers.as_ref() else {
return Err(RuntimeError::ToolFailed(
"flow.spawn: no provider registry available on ctx".into(),
));
};
let (file_part, flow_name) = match flow_ref.split_once('@') {
Some((f, n)) => (f, Some(n.to_string())),
None => (flow_ref, None),
};
let (path, src) = read_flow_source(file_part).await?;
let file = atman_dsl::parse::parse_file(&src).map_err(|e| {
RuntimeError::ToolFailed(format!("flow.spawn: parse {}: {e}", path.display()))
})?;
let flow = match &flow_name {
Some(name) => file
.flows
.iter()
.find(|f| f.name.name == *name)
.ok_or_else(|| {
let available: Vec<&str> =
file.flows.iter().map(|f| f.name.name.as_str()).collect();
RuntimeError::ToolFailed(format!(
"flow.spawn: flow `{name}` not found in {}. available: {}",
path.display(),
available.join(", ")
))
})?,
None => file
.flows
.iter()
.find(|f| f.name.name != "describe")
.ok_or_else(|| {
RuntimeError::ToolFailed(format!(
"flow.spawn: no entry flow in {} (all flows are describe())",
path.display()
))
})?,
};
let mut flow_args: Vec<(String, Value)> = Vec::new();
if let Some(Value::Struct(fields)) = args.named("arguments") {
for (key, value) in fields {
if key == "flow" || key == "async" || key == "inherit_context" {
continue;
}
if flow.params.iter().any(|p| p.name.name == *key) {
flow_args.push((key.clone(), value.clone()));
}
}
}
for (key, value) in &args.named {
if key == "flow" || key == "async" || key == "inherit_context" || key == "arguments" {
continue;
}
if flow.params.iter().any(|p| p.name.name == *key)
&& !flow_args.iter().any(|(k, _)| k == key)
{
flow_args.push((key.clone(), value.clone()));
}
}
let flows = file
.flows
.iter()
.map(|flow| (flow.name.name.clone(), flow.clone()))
.collect();
emit_flow_agent_start(ctx, &run_id, &flow.name.name);
let mut child_ctx = sanitize_child_ctx(ctx);
let inherit = args
.named("inherit_context")
.map(|v| matches!(v, Value::Bool(true)))
.unwrap_or(false);
child_ctx.session_messages_handle = match &ctx.agent_entry {
Some(entry) => Some(std::sync::Arc::clone(&entry.messages)),
None => {
let handle: std::sync::Arc<std::sync::Mutex<Vec<_>>> =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
if inherit {
if let Some(parent) = &ctx.session_messages_handle {
*handle.lock().unwrap() = parent.lock().unwrap().clone();
}
}
Some(handle)
}
};
if ctx.agent_entry.is_none() {
child_ctx.compact_lock_handle = Some(std::sync::Arc::new(tokio::sync::Mutex::new(())));
}
let out = crate::exec::exec_flow_with_siblings(
flow,
flow_args,
registry.as_ref(),
&child_ctx,
providers.as_ref(),
&flows,
child_ctx.events.as_ref(),
child_ctx.turn_id.clone(),
Some(run_id.clone()),
None,
child_ctx.cancel.clone(),
None,
path.parent().map(|p| p.to_path_buf()),
)
.await;
let status = match &out {
Ok(_) => FlowStatus::Ok,
Err(e) => FlowStatus::Errored {
message: e.to_string(),
},
};
emit_child_flow_end(ctx, &run_id, &status);
out
}
fn extract_flow(args: &ToolArgs) -> Result<Option<String>, RuntimeError> {
match args.named("flow") {
Some(Value::Str(s)) if !s.trim().is_empty() => Ok(Some(s.clone())),
Some(Value::Unit) | None => Ok(None),
Some(other) => Err(RuntimeError::TypeMismatch {
expected: "flow string".into(),
actual: other.kind_name().into(),
}),
}
}
async fn read_flow_source(flow_ref: &str) -> Result<(PathBuf, String), RuntimeError> {
for path in flow_candidates(flow_ref) {
match tokio::fs::read_to_string(&path).await {
Ok(src) => return Ok((path, src)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
Err(e) => {
return Err(RuntimeError::ToolFailed(format!(
"flow.spawn: read {}: {e}",
path.display()
)));
}
}
}
Err(RuntimeError::ToolFailed(format!(
"flow.spawn: flow `{flow_ref}` not found"
)))
}
fn flow_candidates(flow_ref: &str) -> Vec<PathBuf> {
let path = PathBuf::from(flow_ref);
if path.is_absolute() {
return vec![path];
}
let file_name = if flow_ref.ends_with(".at") {
flow_ref.to_string()
} else {
format!("{flow_ref}.at")
};
let mut out = Vec::new();
if let Some(home) = std::env::var_os("HOME") {
out.push(
PathBuf::from(home)
.join(".config")
.join("atman")
.join("commands")
.join(&file_name),
);
}
out.push(PathBuf::from(file_name));
out
}
fn emit_flow_agent_start(ctx: &ToolCtx, run_id: &FlowRunId, flow_name: &str) {
let parent_run_id = ctx.flow_run_id.clone();
let parent_node_id = ctx.current_node_id.clone();
if let Some(sink) = &ctx.events {
sink.emit(Event::FlowStart {
run_id: run_id.clone(),
flow_name: flow_name.into(),
parent_run_id: parent_run_id.clone(),
parent_node_id: parent_node_id.clone(),
spawned: true,
});
}
if let Some(tx) = &ctx.stream_tx {
let _ = tx.send(crate::stream::StreamFrame::FlowStart {
run_id: run_id.0.to_string(),
flow_name: flow_name.into(),
parent_run_id: parent_run_id.as_ref().map(|r| r.0.to_string()),
parent_node_id,
});
}
}
fn emit_child_flow_end(ctx: &ToolCtx, run_id: &FlowRunId, status: &FlowStatus) {
if let Some(sink) = &ctx.events {
sink.emit(Event::FlowEnd {
run_id: run_id.clone(),
flow_name: "agent.sub".into(),
status: status.clone(),
});
}
if let Some(tx) = &ctx.stream_tx {
let _ = tx.send(crate::stream::StreamFrame::FlowDone {
run_id: run_id.0.to_string(),
flow_name: "agent.sub".into(),
ok: matches!(status, FlowStatus::Ok),
cancelled: false,
});
}
}
fn sanitize_child_ctx(parent: &ToolCtx) -> ToolCtx {
let mut c = parent.clone();
c.session_runtime = None;
c.history_segment = crate::tool::HistorySegment::Spawned;
c.session_messages_handle = None;
c.compact_lock_handle = None;
c.forms = None;
c.on_memory_recent = None;
c
}