From 1786c5eb2eeac8e98eaa484005f4e13b41c52450 Mon Sep 17 00:00:00 2001 From: TheArchitectit Date: Wed, 29 Apr 2026 10:54:04 -0500 Subject: [PATCH] feat: inbox-based team monitoring - Agents post completion/failure to team inbox on termination (.clawd-agents/mailbox/team/{team_id}/{agent_id}-{ts}.json) - Team watcher reads from inbox instead of polling .json files - New TeamStatus action=inbox reads team messages from the inbox - AgentOutput carries team_id, persisted in manifest - AgentInput accepts team_id from TeamCreate - TeamCreate passes team_id to each spawned agent - Inbox cleaned up when all agents finish Co-authored-by: GLM 5.1 FP8 via Crush --- rust/crates/tools/src/lane_completion.rs | 1 + rust/crates/tools/src/lib.rs | 168 +++++++++++++++++------ 2 files changed, 127 insertions(+), 42 deletions(-) diff --git a/rust/crates/tools/src/lane_completion.rs b/rust/crates/tools/src/lane_completion.rs index 947d58d9..5b190498 100644 --- a/rust/crates/tools/src/lane_completion.rs +++ b/rust/crates/tools/src/lane_completion.rs @@ -117,6 +117,7 @@ mod tests { derived_state: "working".to_string(), current_blocker: None, error: None, + team_id: None, } } diff --git a/rust/crates/tools/src/lib.rs b/rust/crates/tools/src/lib.rs index 9ea8ebb1..38dd7d03 100644 --- a/rust/crates/tools/src/lib.rs +++ b/rust/crates/tools/src/lib.rs @@ -1171,8 +1171,8 @@ pub fn mvp_tool_specs() -> Vec { "team_id": { "type": "string", "description": "Team ID to check" }, "action": { "type": "string", - "enum": ["status", "summary", "events"], - "description": "status=live snapshot, summary=final results, events=timeline" + "enum": ["status", "summary", "events", "inbox"], + "description": "status=live snapshot, summary=final results, events=timeline, inbox=team messages from agents" } }, "required": ["team_id"], @@ -1855,6 +1855,7 @@ fn run_team_create(input: TeamCreateInput) -> Result { subagent_type: subagent_type.map(|s| s.to_string()), name: Some(format!("{}-agent-{}", slugify_agent_name(&input.name), i + 1)), model: model_override.map(|s| s.to_string()), + team_id: Some(team_id.clone()), }; match execute_agent_with_spawn(agent_input, spawn_agent_job) { @@ -2041,7 +2042,30 @@ fn run_team_status(input: TeamStatusInput) -> Result { "events": events, })) } - other => Err(format!("unknown TeamStatus action: {other}. Use status, summary, or events")), + "inbox" => { + let inbox_dir = agent_mailbox_dir().join("team").join(&input.team_id); + let mut messages: Vec = Vec::new(); + if inbox_dir.exists() { + if let Ok(entries) = std::fs::read_dir(&inbox_dir) { + for entry in entries.filter_map(|e| e.ok()) { + let path = entry.path(); + if path.extension().is_some_and(|ext| ext == "json") { + if let Ok(data) = std::fs::read_to_string(&path) { + if let Ok(msg) = serde_json::from_str::(&data) { + messages.push(msg); + } + } + } + } + } + } + to_pretty_json(json!({ + "team_id": input.team_id, + "inbox_count": messages.len(), + "messages": messages, + })) + } + other => Err(format!("unknown TeamStatus action: {other}. Use status, summary, events, or inbox")), } } @@ -2049,65 +2073,71 @@ fn run_team_status(input: TeamStatusInput) -> Result { /// Prints agent completion/failure events to stderr. Exits when all agents are done. fn spawn_team_watcher(team_id: &str, agent_ids: &[String]) { let team_id = team_id.to_string(); - let agent_ids = agent_ids.to_vec(); + let total = agent_ids.len(); let thread_name = format!("claw-team-watch-{team_id}"); std::thread::Builder::new() .name(thread_name) .spawn(move || { - let store_dir = match agent_store_dir() { - Ok(d) => d, - Err(_) => return, - }; - let team_dir = store_dir.join("teams"); + let inbox_dir = agent_mailbox_dir().join("team").join(&team_id); + let _ = std::fs::create_dir_all(&inbox_dir); + + let team_dir = agent_store_dir().map(|d| d.join("teams")).unwrap_or_default(); let events_path = team_dir.join(format!("{team_id}-events.jsonl")); let mut completed: BTreeSet = BTreeSet::new(); let mut failed: BTreeSet = BTreeSet::new(); - let total = agent_ids.len(); + let mut seen: BTreeSet = BTreeSet::new(); - eprintln!("[team] {team_id}: watching {total} agents..."); + eprintln!("[team] {team_id}: watching {total} agents via inbox..."); loop { - let mut all_done = true; - for id in &agent_ids { - if completed.contains(id) || failed.contains(id) { - continue; - } - let agent_json = store_dir.join(format!("{id}.json")); - if let Ok(data) = std::fs::read_to_string(&agent_json) { - if let Ok(parsed) = serde_json::from_str::(&data) { - let status = parsed.get("status").and_then(|v| v.as_str()).unwrap_or("unknown"); - match status { - "completed" => { - completed.insert(id.clone()); - let name = parsed.get("name").and_then(|v| v.as_str()).unwrap_or(id); - let subagent_type = parsed.get("subagent_type").and_then(|v| v.as_str()).unwrap_or("unknown"); - let result_preview = get_agent_result_preview(&store_dir, id); - eprintln!("[team] {team_id}: done {name} ({subagent_type}) completed - {}/{total}{}", completed.len() + failed.len(), if result_preview.is_empty() { String::new() } else { format!(" - {}", &result_preview[..result_preview.len().min(120)]) }); - append_team_event(&events_path, &team_id, id, "completed", &name, Some(&result_preview)); - } - "failed" => { - failed.insert(id.clone()); - let name = parsed.get("name").and_then(|v| v.as_str()).unwrap_or(id); - let error = parsed.get("error").and_then(|v| v.as_str()).unwrap_or("unknown error"); - eprintln!("[team] {team_id}: FAIL {name} failed - {error}"); - append_team_event(&events_path, &team_id, id, "failed", &name, Some(error)); - } - _ => { - all_done = false; + let mut new_events = false; + if let Ok(entries) = std::fs::read_dir(&inbox_dir) { + for entry in entries.filter_map(|e| e.ok()) { + let path = entry.path(); + let name = path.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(); + if seen.contains(&name) { + continue; + } + if !path.extension().is_some_and(|ext| ext == "json") { + continue; + } + if let Ok(data) = std::fs::read_to_string(&path) { + if let Ok(parsed) = serde_json::from_str::(&data) { + seen.insert(name); + new_events = true; + let agent_id = parsed.get("agent_id").and_then(|v| v.as_str()).unwrap_or("unknown"); + let event = parsed.get("event").and_then(|v| v.as_str()).unwrap_or("unknown"); + let agent_name = parsed.get("name").and_then(|v| v.as_str()).unwrap_or(agent_id); + let subagent_type = parsed.get("subagent_type").and_then(|v| v.as_str()).unwrap_or("unknown"); + let result_preview = parsed.get("result_preview").and_then(|v| v.as_str()).unwrap_or(""); + + match event { + "agent_completed" => { + completed.insert(agent_id.to_string()); + let done_count = completed.len() + failed.len(); + eprintln!("[team] {team_id}: done {agent_name} ({subagent_type}) completed - {done_count}/{total}{}", if result_preview.is_empty() { String::new() } else { format!(" - {}", &result_preview[..result_preview.len().min(120)]) }); + append_team_event(&events_path, &team_id, agent_id, "completed", agent_name, Some(result_preview)); + } + "agent_failed" => { + failed.insert(agent_id.to_string()); + let error = parsed.get("error").and_then(|v| v.as_str()).unwrap_or("unknown error"); + eprintln!("[team] {team_id}: FAIL {agent_name} failed - {error}"); + append_team_event(&events_path, &team_id, agent_id, "failed", agent_name, Some(error)); + } + _ => {} } } } - } else { - all_done = false; } } + let all_done = completed.len() + failed.len() >= total; if all_done { break; } - std::thread::sleep(std::time::Duration::from_secs(2)); + std::thread::sleep(std::time::Duration::from_secs(1)); } eprintln!("[team] {team_id}: all agents finished - {}/{} completed, {}/{} failed", completed.len(), total, failed.len(), total); @@ -2122,6 +2152,9 @@ fn spawn_team_watcher(team_id: &str, agent_ids: &[String]) { } } } + + // Clean up inbox + let _ = std::fs::remove_dir_all(&inbox_dir); }) .ok(); } @@ -3368,6 +3401,8 @@ struct AgentInput { subagent_type: Option, name: Option, model: Option, + #[serde(default)] + team_id: Option, } #[derive(Debug, Deserialize)] @@ -3773,6 +3808,8 @@ struct AgentOutput { derived_state: String, #[serde(skip_serializing_if = "Option::is_none")] error: Option, + #[serde(rename = "teamId", skip_serializing_if = "Option::is_none")] + team_id: Option, } #[derive(Debug, Clone)] @@ -3781,6 +3818,7 @@ struct AgentJob { prompt: String, system_prompt: Vec, allowed_tools: BTreeSet, + team_id: Option, } #[derive(Debug, Clone, Serialize, PartialEq, Eq)] @@ -4677,6 +4715,7 @@ where let created_at = iso8601_now(); let system_prompt = build_agent_system_prompt(&normalized_subagent_type, &model)?; let allowed_tools = allowed_tools_for_subagent(&normalized_subagent_type); + let team_id = input.team_id.clone(); let output_contents = format!( "# Agent Task @@ -4711,6 +4750,7 @@ where current_blocker: None, derived_state: String::from("working"), error: None, + team_id: input.team_id.clone(), }; write_agent_manifest(&manifest)?; @@ -4720,6 +4760,7 @@ where prompt: input.prompt, system_prompt, allowed_tools, + team_id: input.team_id.clone(), }; if let Err(error) = spawn_fn(job) { let error = format!("failed to spawn sub-agent: {error}"); @@ -4973,7 +5014,40 @@ fn persist_agent_terminal_state( )); } } - write_agent_manifest(&next_manifest) + write_agent_manifest(&next_manifest)?; + + // If this agent belongs to a team, post completion to the team inbox + if let Some(ref tid) = next_manifest.team_id { + let _ = post_agent_completion_to_team_inbox(&next_manifest, tid, status, result); + } + + Ok(()) +} + +fn post_agent_completion_to_team_inbox( + manifest: &AgentOutput, + team_id: &str, + status: &str, + result: Option<&str>, +) -> Result<(), String> { + let mailbox_dir = agent_mailbox_dir().join("team").join(team_id); + std::fs::create_dir_all(&mailbox_dir).map_err(|e| e.to_string())?; + let ts = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + let entry = json!({ + "event": format!("agent_{status}"), + "agent_id": manifest.agent_id, + "name": manifest.name, + "subagent_type": manifest.subagent_type, + "status": status, + "result_preview": result.map(|r| r.chars().take(2000).collect::()), + "timestamp": ts, + }); + let msg_file = mailbox_dir.join(format!("{}-{ts}.json", manifest.agent_id)); + std::fs::write(&msg_file, serde_json::to_string_pretty(&entry).map_err(|e| e.to_string())?) + .map_err(|e| e.to_string()) } const MIN_LANE_SUMMARY_WORDS: usize = 7; @@ -9277,6 +9351,7 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("ship-audit".to_string()), model: None, + team_id: None, }, move |job| { *captured_for_spawn @@ -9358,6 +9433,7 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("complete-task".to_string()), model: Some("claude-sonnet-4-6".to_string()), + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9415,6 +9491,7 @@ mod tests { subagent_type: Some("Verification".to_string()), name: Some("fail-task".to_string()), model: None, + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9462,6 +9539,7 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("summary-floor".to_string()), model: None, + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9507,6 +9585,7 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("recovery-lane".to_string()), model: None, + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9555,6 +9634,7 @@ mod tests { subagent_type: Some("Verification".to_string()), name: Some("review-lane".to_string()), model: None, + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9595,6 +9675,7 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("backlog-scan".to_string()), model: None, + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9641,6 +9722,7 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("artifact-lane".to_string()), model: None, + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9711,6 +9793,7 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("cron-closeout".to_string()), model: None, + team_id: None, }, |job| { persist_agent_terminal_state( @@ -9752,6 +9835,7 @@ mod tests { subagent_type: None, name: Some("spawn-error".to_string()), model: None, + team_id: None, }, |_| Err(String::from("thread creation failed")), )