From 9c4e7a1b7da60a4f74e294fa58bb98507cf2eb61 Mon Sep 17 00:00:00 2001 From: Wylabb <77673282+Wylabb@users.noreply.github.com> Date: Sun, 5 Apr 2026 08:43:44 +0200 Subject: [PATCH] Port teamwork task and agent parity slice --- .gitignore | 1 + rust/.gitignore | 1 + rust/Cargo.lock | 1 + rust/crates/channel-gateway-core/src/lib.rs | 6 +- .../channel-gateway-core/src/protocol.rs | 36 + rust/crates/claw-profile-worker/Cargo.toml | 1 + rust/crates/claw-profile-worker/src/server.rs | 126 ++- rust/crates/claw-telegram/src/gateway.rs | 248 +++++- .../crates/claw-telegram/src/worker_client.rs | 39 +- rust/crates/runtime/src/bash.rs | 50 ++ rust/crates/runtime/src/lib.rs | 21 +- rust/crates/runtime/src/runtime_task_store.rs | 413 +++++++++ rust/crates/runtime/src/task_list_store.rs | 369 ++++++++ rust/crates/runtime/src/teamwork_store.rs | 384 +++++++++ rust/crates/runtime/src/workflow_state.rs | 176 ++++ rust/crates/tools/.gitignore | 1 + rust/crates/tools/src/lane_completion.rs | 47 +- rust/crates/tools/src/lib.rs | 792 +++++++++++------- 18 files changed, 2386 insertions(+), 326 deletions(-) create mode 100644 rust/crates/runtime/src/runtime_task_store.rs create mode 100644 rust/crates/runtime/src/task_list_store.rs create mode 100644 rust/crates/runtime/src/teamwork_store.rs create mode 100644 rust/crates/runtime/src/workflow_state.rs diff --git a/.gitignore b/.gitignore index cfa4cf3..d05fb9a 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ __pycache__/ archive/ .omx/ .clawd-agents/ +.clawd-state/ # Claude Code local artifacts .claude/settings.local.json .claude/sessions/ diff --git a/rust/.gitignore b/rust/.gitignore index 19e1a8e..5c8fcf5 100644 --- a/rust/.gitignore +++ b/rust/.gitignore @@ -1,3 +1,4 @@ target/ .omx/ .clawd-agents/ +.clawd-state/ diff --git a/rust/Cargo.lock b/rust/Cargo.lock index f2456f0..67d3c53 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -282,6 +282,7 @@ dependencies = [ "channel-gateway-core", "futures-core", "reqwest", + "runtime", "serde", "serde_json", "subtle", diff --git a/rust/crates/channel-gateway-core/src/lib.rs b/rust/crates/channel-gateway-core/src/lib.rs index 3d6273e..f943f97 100644 --- a/rust/crates/channel-gateway-core/src/lib.rs +++ b/rust/crates/channel-gateway-core/src/lib.rs @@ -10,8 +10,10 @@ pub use manifest::{ ProfileRecord, WorkerDefaults, WorkerSpec, }; pub use protocol::{ - GeneratedFileDescriptor, InboundAttachment, TurnSource, WorkerApprovalDecision, - WorkerStatusResponse, WorkerTurnAccepted, WorkerTurnEvent, WorkerTurnRequest, + GeneratedFileDescriptor, InboundAttachment, TurnSource, WorkerAgentListResponse, + WorkerApprovalDecision, WorkerMailboxSummaryResponse, WorkerStatusResponse, + WorkerTaskListResponse, WorkerTaskSnapshotResponse, WorkerTeamSnapshotResponse, + WorkerTurnAccepted, WorkerTurnEvent, WorkerTurnRequest, }; pub use runtime_host::{ ApprovalDecision, ApprovalRequestPayload, ApprovalResponder, AttachmentKind, AttachmentRef, diff --git a/rust/crates/channel-gateway-core/src/protocol.rs b/rust/crates/channel-gateway-core/src/protocol.rs index 132a40f..c192e89 100644 --- a/rust/crates/channel-gateway-core/src/protocol.rs +++ b/rust/crates/channel-gateway-core/src/protocol.rs @@ -1,4 +1,5 @@ use serde::{Deserialize, Serialize}; +use runtime::{MailboxSummary, RuntimeTaskRecord, TaskListRecord, TeamRecord}; use crate::runtime_host::{ApprovalRequestPayload, AttachmentKind}; @@ -97,4 +98,39 @@ pub struct WorkerStatusResponse { pub permission_mode: String, pub default_cwd: String, pub busy: bool, + #[serde(default)] + pub task_list_id: String, + #[serde(default)] + pub active_team: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkerTaskListResponse { + pub task_list_id: String, + pub tasks: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkerTaskSnapshotResponse { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub task: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub runtime_task: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkerTeamSnapshotResponse { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub team: Option, + pub task_list_id: String, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkerAgentListResponse { + pub agents: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkerMailboxSummaryResponse { + pub mailbox: MailboxSummary, } diff --git a/rust/crates/claw-profile-worker/Cargo.toml b/rust/crates/claw-profile-worker/Cargo.toml index ddc8002..0ee19a0 100644 --- a/rust/crates/claw-profile-worker/Cargo.toml +++ b/rust/crates/claw-profile-worker/Cargo.toml @@ -11,6 +11,7 @@ axum = { version = "0.7", features = ["multipart"] } base64 = "0.22" channel-gateway-core = { path = "../channel-gateway-core" } futures-core = "0.3" +runtime = { path = "../runtime" } serde = { version = "1", features = ["derive"] } serde_json.workspace = true subtle = "2" diff --git a/rust/crates/claw-profile-worker/src/server.rs b/rust/crates/claw-profile-worker/src/server.rs index 44a7d48..af3e495 100644 --- a/rust/crates/claw-profile-worker/src/server.rs +++ b/rust/crates/claw-profile-worker/src/server.rs @@ -16,7 +16,13 @@ use base64::Engine as _; use channel_gateway_core::{ ApprovalDecision, ApprovalResponder, AttachmentRef, GeneratedFileDescriptor, HostError, RuntimeEvent, RuntimeHost, RuntimeHostConfig, SessionApprovalState, WorkerApprovalDecision, - WorkerStatusResponse, WorkerTurnAccepted, WorkerTurnEvent, WorkerTurnRequest, + WorkerAgentListResponse, WorkerMailboxSummaryResponse, WorkerStatusResponse, + WorkerTaskListResponse, WorkerTaskSnapshotResponse, WorkerTeamSnapshotResponse, + WorkerTurnAccepted, WorkerTurnEvent, WorkerTurnRequest, +}; +use runtime::{ + current_task_list_id, RuntimeTaskKind, RuntimeTaskRecord, RuntimeTaskStore, TaskListStore, + TeamStore, }; use serde::Deserialize; use subtle::ConstantTimeEq; @@ -41,6 +47,13 @@ fn app_router(config: WorkerConfig, runtime: Arc) -> Router { .route("/healthz", get(health)) .route("/v1/status", get(status)) .route("/v1/session/reset", post(reset_session)) + .route("/v1/tasks", get(list_tasks)) + .route("/v1/tasks/:task_id", get(get_task)) + .route("/v1/tasks/:task_id/stop", post(stop_task)) + .route("/v1/team", get(get_team)) + .route("/v1/agents", get(list_agents)) + .route("/v1/agents/:agent_id", get(get_agent)) + .route("/v1/mailbox", get(get_mailbox)) .route("/v1/turns", post(post_turn)) .route("/v1/turns/:turn_id/events", get(stream_events)) .route("/v1/turns/:turn_id/approval", post(post_approval)) @@ -229,9 +242,120 @@ async fn status( permission_mode: state.config.permission_mode.as_str().to_string(), default_cwd: state.config.default_cwd.display().to_string(), busy, + task_list_id: current_task_list_id().unwrap_or_else(|_| state.config.profile_id.clone()), + active_team: TeamStore::new() + .current_team() + .ok() + .flatten() + .map(|team| team.team_name), })) } +async fn list_tasks( + State(state): State>, + headers: HeaderMap, +) -> Result, StatusCode> { + authorize(&headers, &state.config.auth_token)?; + let store = TaskListStore::current().map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let tasks = store.list(false).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + Ok(Json(WorkerTaskListResponse { + task_list_id: store.task_list_id().to_string(), + tasks, + })) +} + +async fn get_task( + State(state): State>, + headers: HeaderMap, + AxumPath(task_id): AxumPath, +) -> Result, StatusCode> { + authorize(&headers, &state.config.auth_token)?; + let task = TaskListStore::current() + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .get(&task_id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let runtime_task = RuntimeTaskStore::new() + .get(&task_id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + if task.is_none() && runtime_task.is_none() { + return Err(StatusCode::NOT_FOUND); + } + Ok(Json(WorkerTaskSnapshotResponse { task, runtime_task })) +} + +async fn stop_task( + State(state): State>, + headers: HeaderMap, + AxumPath(task_id): AxumPath, +) -> Result { + authorize(&headers, &state.config.auth_token)?; + match RuntimeTaskStore::new().stop(&task_id) { + Ok(Some(_)) => Ok(StatusCode::ACCEPTED), + Ok(None) => Err(StatusCode::NOT_FOUND), + Err(_) => Err(StatusCode::BAD_REQUEST), + } +} + +async fn get_team( + State(state): State>, + headers: HeaderMap, +) -> Result, StatusCode> { + authorize(&headers, &state.config.auth_token)?; + let task_list_id = current_task_list_id().unwrap_or_else(|_| state.config.profile_id.clone()); + let team = TeamStore::new() + .current_team() + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + Ok(Json(WorkerTeamSnapshotResponse { team, task_list_id })) +} + +async fn list_agents( + State(state): State>, + headers: HeaderMap, +) -> Result, StatusCode> { + authorize(&headers, &state.config.auth_token)?; + let agents = RuntimeTaskStore::new() + .list() + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .into_iter() + .filter(|task| task.kind == RuntimeTaskKind::Agent) + .collect(); + Ok(Json(WorkerAgentListResponse { agents })) +} + +async fn get_agent( + State(state): State>, + headers: HeaderMap, + AxumPath(agent_id): AxumPath, +) -> Result, StatusCode> { + authorize(&headers, &state.config.auth_token)?; + let task = RuntimeTaskStore::new() + .get(&agent_id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .filter(|task| task.kind == RuntimeTaskKind::Agent) + .ok_or(StatusCode::NOT_FOUND)?; + Ok(Json(task)) +} + +async fn get_mailbox( + State(state): State>, + headers: HeaderMap, +) -> Result, StatusCode> { + authorize(&headers, &state.config.auth_token)?; + let mailbox = match TeamStore::new() + .current_team() + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + { + Some(team) => TeamStore::new() + .mailbox_summary(&team.team_name, 20) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?, + None => runtime::MailboxSummary { + team_name: None, + recent_messages: Vec::new(), + }, + }; + Ok(Json(WorkerMailboxSummaryResponse { mailbox })) +} + async fn reset_session( State(state): State>, headers: HeaderMap, diff --git a/rust/crates/claw-telegram/src/gateway.rs b/rust/crates/claw-telegram/src/gateway.rs index a59e9e4..91fea5a 100644 --- a/rust/crates/claw-telegram/src/gateway.rs +++ b/rust/crates/claw-telegram/src/gateway.rs @@ -6,8 +6,11 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use channel_gateway_core::{ AttachmentKind, AttachmentRef, GatewayManifest, GatewaySettings, ManifestError, ProfileId, - ProfileRecord, TurnSource, WorkerApprovalDecision, WorkerDefaults, WorkerTurnEvent, + ProfileRecord, TurnSource, WorkerAgentListResponse, WorkerApprovalDecision, + WorkerDefaults, WorkerMailboxSummaryResponse, WorkerTaskListResponse, + WorkerTaskSnapshotResponse, WorkerTeamSnapshotResponse, WorkerTurnEvent, }; +use runtime::RuntimeTaskRecord; use tokio::sync::Mutex; use crate::config::GatewayConfig; @@ -220,17 +223,100 @@ impl TelegramGateway { let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; let status = client.status().await?; let text = format!( - "Status\nProfile: {}\nContainer: {}\nMessages: {}\nModel: {}\nPermission mode: {}\nWorking directory: {}\nBusy: {}", + "Status\nProfile: {}\nContainer: {}\nMessages: {}\nModel: {}\nPermission mode: {}\nWorking directory: {}\nTask list: {}\nActive team: {}\nBusy: {}", status.profile_id, profile.worker.container_name, status.message_count, status.model, status.permission_mode, status.default_cwd, + status.task_list_id, + status.active_team.as_deref().unwrap_or("(none)"), if status.busy { "yes" } else { "no" } ); self.api.send_message(message.chat.id, &text, None).await?; } + Command::Tasks => { + let worker = self + .worker_manager + .ensure_profile_worker(manifest, profile) + .await?; + let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; + let tasks = client.list_tasks().await?; + self.api + .send_message(message.chat.id, &render_task_list(&tasks), None) + .await?; + } + Command::Task(task_id) => { + let worker = self + .worker_manager + .ensure_profile_worker(manifest, profile) + .await?; + let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; + let task = client.get_task(&task_id).await?; + self.api + .send_message(message.chat.id, &render_task_snapshot(&task_id, &task), None) + .await?; + } + Command::Team => { + let worker = self + .worker_manager + .ensure_profile_worker(manifest, profile) + .await?; + let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; + let team = client.team().await?; + self.api + .send_message(message.chat.id, &render_team_snapshot(&team), None) + .await?; + } + Command::Agents => { + let worker = self + .worker_manager + .ensure_profile_worker(manifest, profile) + .await?; + let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; + let agents = client.agents().await?; + self.api + .send_message(message.chat.id, &render_agents_snapshot(&agents), None) + .await?; + } + Command::Agent(agent_id) => { + let worker = self + .worker_manager + .ensure_profile_worker(manifest, profile) + .await?; + let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; + let agent = client.agent(&agent_id).await?; + self.api + .send_message(message.chat.id, &render_agent_snapshot(&agent), None) + .await?; + } + Command::StopTask(task_id) => { + let worker = self + .worker_manager + .ensure_profile_worker(manifest, profile) + .await?; + let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; + client.stop_task(&task_id).await?; + self.api + .send_message( + message.chat.id, + &format!("Stop requested for runtime task `{task_id}`."), + None, + ) + .await?; + } + Command::Messages => { + let worker = self + .worker_manager + .ensure_profile_worker(manifest, profile) + .await?; + let client = WorkerClient::new(&worker.base_url, &self.config.worker_auth_token)?; + let mailbox = client.mailbox().await?; + self.api + .send_message(message.chat.id, &render_mailbox_summary(&mailbox), None) + .await?; + } Command::New => { if self.is_profile_busy(profile.profile_id.as_str()).await { self.api @@ -1056,31 +1142,185 @@ fn push_wrapped_line( } fn render_help() -> String { - "Commands\n/start or /help - show help\n/status - show the routed worker status\n/new - start a fresh session\n/cancel - cancel the active turn\n\nSend a normal message, photo, or document to talk to your profile worker.".to_string() + "Commands\n/start or /help - show help\n/status - show the routed worker status\n/tasks - list task-list items\n/task - inspect a task-list or runtime task\n/team - show the active team context\n/agents - list spawned agents\n/agent - inspect a spawned agent\n/messages - show recent team mailbox messages\n/stop_task - stop a runtime task\n/new - start a fresh session\n/cancel - cancel the active turn\n\nSend a normal message, photo, or document to talk to your profile worker.".to_string() } enum Command { Start, Help, Status, + Tasks, + Task(String), + Team, + Agents, + Agent(String), + StopTask(String), + Messages, New, Cancel, } fn parse_command(message: &Message) -> Option { let text = message.text_or_caption()?.trim(); - let command = text.split_whitespace().next()?; + let mut parts = text.split_whitespace(); + let command = parts.next()?; let command = command.split('@').next().unwrap_or(command); match command { "/start" => Some(Command::Start), "/help" => Some(Command::Help), "/status" => Some(Command::Status), + "/tasks" => Some(Command::Tasks), + "/task" => parts.next().map(|value| Command::Task(value.to_string())), + "/team" => Some(Command::Team), + "/agents" => Some(Command::Agents), + "/agent" => parts.next().map(|value| Command::Agent(value.to_string())), + "/stop_task" => parts.next().map(|value| Command::StopTask(value.to_string())), + "/messages" => Some(Command::Messages), "/new" => Some(Command::New), "/cancel" => Some(Command::Cancel), _ => None, } } +fn render_task_list(response: &WorkerTaskListResponse) -> String { + if response.tasks.is_empty() { + return format!("Tasks\nTask list: {}\nNo tasks.", response.task_list_id); + } + let mut lines = vec![format!("Tasks\nTask list: {}", response.task_list_id)]; + for task in &response.tasks { + let blocked = if task.blocked_by.is_empty() { + String::new() + } else { + format!(" blocked by {}", task.blocked_by.join(", ")) + }; + lines.push(format!( + "{} [{}] {}{}", + task.id, task.status, task.subject, blocked + )); + } + lines.join("\n") +} + +fn render_task_snapshot(task_id: &str, response: &WorkerTaskSnapshotResponse) -> String { + if let Some(task) = &response.task { + return format!( + "Task {}\nSubject: {}\nStatus: {}\nOwner: {}\nDescription: {}\nBlocks: {}\nBlocked by: {}", + task.id, + task.subject, + task.status, + task.owner.as_deref().unwrap_or("(unassigned)"), + task.description, + if task.blocks.is_empty() { + "(none)".to_string() + } else { + task.blocks.join(", ") + }, + if task.blocked_by.is_empty() { + "(none)".to_string() + } else { + task.blocked_by.join(", ") + } + ); + } + if let Some(runtime_task) = &response.runtime_task { + return render_runtime_task(task_id, runtime_task); + } + format!("Task `{task_id}` was not found.") +} + +fn render_team_snapshot(response: &WorkerTeamSnapshotResponse) -> String { + let Some(team) = &response.team else { + return format!( + "Team\nNo active team.\nTask list: {}", + response.task_list_id + ); + }; + let members = team + .members + .iter() + .map(|member| member.name.clone()) + .collect::>(); + format!( + "Team\nName: {}\nLead: {}\nTask list: {}\nMembers: {}\nDescription: {}", + team.team_name, + team.lead_agent_id, + response.task_list_id, + if members.is_empty() { + "(none)".to_string() + } else { + members.join(", ") + }, + team.description.as_deref().unwrap_or("(none)") + ) +} + +fn render_agents_snapshot(response: &WorkerAgentListResponse) -> String { + if response.agents.is_empty() { + return "Agents\nNo spawned agents.".to_string(); + } + let mut lines = vec!["Agents".to_string()]; + for agent in &response.agents { + lines.push(format!( + "{} [{}] {}", + agent.task_id, + agent.status, + agent + .agent_name + .as_deref() + .or(agent.agent_id.as_deref()) + .unwrap_or("agent") + )); + } + lines.join("\n") +} + +fn render_agent_snapshot(agent: &RuntimeTaskRecord) -> String { + render_runtime_task( + agent.agent_id.as_deref().unwrap_or(agent.task_id.as_str()), + agent, + ) +} + +fn render_runtime_task(label: &str, task: &RuntimeTaskRecord) -> String { + format!( + "Runtime task {}\nKind: {:?}\nStatus: {}\nDescription: {}\nTeam: {}\nOutput file: {}", + label, + task.kind, + task.status, + task.description, + task.team_name.as_deref().unwrap_or("(none)"), + task.output_file.as_deref().unwrap_or("(none)") + ) +} + +fn render_mailbox_summary(response: &WorkerMailboxSummaryResponse) -> String { + if response.mailbox.recent_messages.is_empty() { + return format!( + "Messages\nTeam: {}\nNo recent team messages.", + response.mailbox.team_name.as_deref().unwrap_or("(none)") + ); + } + let mut lines = vec![format!( + "Messages\nTeam: {}", + response.mailbox.team_name.as_deref().unwrap_or("(none)") + )]; + for item in &response.mailbox.recent_messages { + let preview = item + .envelope + .summary + .clone() + .unwrap_or_else(|| match &item.envelope.message { + serde_json::Value::String(value) => value.clone(), + other => other.to_string(), + }); + lines.push(format!( + "{} -> {}: {}", + item.envelope.from, item.recipient, preview + )); + } + lines.join("\n") +} + #[derive(Debug, Clone, PartialEq, Eq)] struct ApprovalAction { turn_id: String, diff --git a/rust/crates/claw-telegram/src/worker_client.rs b/rust/crates/claw-telegram/src/worker_client.rs index 97f5c85..e02854e 100644 --- a/rust/crates/claw-telegram/src/worker_client.rs +++ b/rust/crates/claw-telegram/src/worker_client.rs @@ -3,9 +3,12 @@ use std::path::{Path, PathBuf}; use base64::Engine as _; use channel_gateway_core::{ - AttachmentRef, GeneratedFileDescriptor, TurnSource, WorkerApprovalDecision, - WorkerStatusResponse, WorkerTurnAccepted, WorkerTurnEvent, WorkerTurnRequest, + AttachmentRef, GeneratedFileDescriptor, TurnSource, WorkerAgentListResponse, + WorkerApprovalDecision, WorkerMailboxSummaryResponse, WorkerStatusResponse, + WorkerTaskListResponse, WorkerTaskSnapshotResponse, WorkerTeamSnapshotResponse, + WorkerTurnAccepted, WorkerTurnEvent, WorkerTurnRequest, }; +use runtime::RuntimeTaskRecord; use futures_util::StreamExt; use serde::Serialize; use tokio::sync::mpsc; @@ -55,6 +58,38 @@ impl WorkerClient { .await } + pub async fn list_tasks(&self) -> Result { + self.get_json("/v1/tasks").await + } + + pub async fn get_task( + &self, + task_id: &str, + ) -> Result { + self.get_json(&format!("/v1/tasks/{task_id}")).await + } + + pub async fn stop_task(&self, task_id: &str) -> Result<(), WorkerClientError> { + self.post_no_content(&format!("/v1/tasks/{task_id}/stop"), &serde_json::json!({})) + .await + } + + pub async fn team(&self) -> Result { + self.get_json("/v1/team").await + } + + pub async fn agents(&self) -> Result { + self.get_json("/v1/agents").await + } + + pub async fn agent(&self, agent_id: &str) -> Result { + self.get_json(&format!("/v1/agents/{agent_id}")).await + } + + pub async fn mailbox(&self) -> Result { + self.get_json("/v1/mailbox").await + } + pub async fn post_turn( &self, prompt: String, diff --git a/rust/crates/runtime/src/bash.rs b/rust/crates/runtime/src/bash.rs index ef9ff8f..477a2b6 100644 --- a/rust/crates/runtime/src/bash.rs +++ b/rust/crates/runtime/src/bash.rs @@ -1,5 +1,6 @@ use std::env; use std::io; +use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::time::Duration; @@ -64,6 +65,14 @@ pub struct BashCommandOutput { pub sandbox_status: Option, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BackgroundBashHandle { + pub pid: u32, + pub sandbox_status: SandboxStatus, + pub output_path: PathBuf, + pub exit_code_path: PathBuf, +} + pub fn execute_bash(input: BashCommandInput) -> io::Result { let cwd = env::current_dir()?; let sandbox_status = sandbox_status_for_input(&input, &cwd); @@ -99,6 +108,43 @@ pub fn execute_bash(input: BashCommandInput) -> io::Result { runtime.block_on(execute_bash_async(input, sandbox_status, cwd)) } +pub fn spawn_background_bash( + input: &BashCommandInput, + output_path: &Path, + exit_code_path: &Path, +) -> io::Result { + let cwd = env::current_dir()?; + let sandbox_status = sandbox_status_for_input(input, &cwd); + if let Some(parent) = output_path.parent() { + std::fs::create_dir_all(parent)?; + } + if let Some(parent) = exit_code_path.parent() { + std::fs::create_dir_all(parent)?; + } + let stdout = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(output_path)?; + let stderr = stdout.try_clone()?; + let wrapped_command = format!( + "({}); __claw_code=$?; printf '%s' \"$__claw_code\" > {}; exit \"$__claw_code\"", + input.command, + shell_quote(&exit_code_path.display().to_string()) + ); + let mut child = prepare_command(&wrapped_command, &cwd, &sandbox_status, false); + let child = child + .stdin(Stdio::null()) + .stdout(Stdio::from(stdout)) + .stderr(Stdio::from(stderr)) + .spawn()?; + Ok(BackgroundBashHandle { + pid: child.id(), + sandbox_status, + output_path: output_path.to_path_buf(), + exit_code_path: exit_code_path.to_path_buf(), + }) +} + async fn execute_bash_async( input: BashCommandInput, sandbox_status: SandboxStatus, @@ -238,6 +284,10 @@ fn prepare_sandbox_dirs(cwd: &std::path::Path) { let _ = std::fs::create_dir_all(cwd.join(".sandbox-tmp")); } +fn shell_quote(value: &str) -> String { + format!("'{}'", value.replace('\'', "'\"'\"'")) +} + #[cfg(test)] mod tests { use super::{execute_bash, BashCommandInput}; diff --git a/rust/crates/runtime/src/lib.rs b/rust/crates/runtime/src/lib.rs index 3025791..b0b6089 100644 --- a/rust/crates/runtime/src/lib.rs +++ b/rust/crates/runtime/src/lib.rs @@ -21,6 +21,7 @@ mod permissions; pub mod plugin_lifecycle; mod policy_engine; mod prompt; +pub mod runtime_task_store; pub mod recovery_recipes; mod remote; pub mod sandbox; @@ -29,14 +30,20 @@ pub mod session_control; mod sse; pub mod stale_branch; pub mod summary_compression; +pub mod task_list_store; pub mod task_packet; pub mod task_registry; +pub mod teamwork_store; pub mod team_cron_registry; pub mod trust_resolver; mod usage; +pub mod workflow_state; pub mod worker_boot; -pub use bash::{execute_bash, BashCommandInput, BashCommandOutput}; +pub use bash::{ + execute_bash, spawn_background_bash, BackgroundBashHandle, BashCommandInput, + BashCommandOutput, +}; pub use bootstrap::{BootstrapPhase, BootstrapPlan}; pub use compact::{ compact_session, estimate_session_tokens, format_compact_summary, @@ -110,6 +117,10 @@ pub use prompt::{ load_system_prompt, prepend_bullets, ContextFile, ProjectContext, PromptBuildError, SystemPromptBuilder, FRONTIER_MODEL_NAME, SYSTEM_PROMPT_DYNAMIC_BOUNDARY, }; +pub use runtime_task_store::{ + background_output_paths, RuntimeTaskKind, RuntimeTaskOutput, RuntimeTaskRecord, + RuntimeTaskStatus, RuntimeTaskStore, +}; pub use recovery_recipes::{ attempt_recovery, recipe_for, EscalationPolicy, FailureScenario, RecoveryContext, RecoveryEvent, RecoveryRecipe, RecoveryResult, RecoveryStep, @@ -138,10 +149,18 @@ pub use task_packet::{ validate_packet, AcceptanceTest, BranchPolicy, CommitPolicy, RepoConfig, ReportingContract, TaskPacket, TaskPacketValidationError, TaskScope, ValidatedPacket, }; +pub use task_list_store::{TaskListPatch, TaskListRecord, TaskListStatus, TaskListStore}; +pub use teamwork_store::{ + MailboxMessage, MailboxSummary, MessageEnvelope, TeamMemberRecord, TeamRecord, TeamStore, +}; pub use trust_resolver::{TrustConfig, TrustDecision, TrustEvent, TrustPolicy, TrustResolver}; pub use usage::{ format_usd, pricing_for_model, ModelPricing, TokenUsage, UsageCostEstimate, UsageTracker, }; +pub use workflow_state::{ + clear_team_context, current_task_list_id, default_session_identity, load_team_context, + sanitize_state_component, state_root, TeamContext, +}; pub use worker_boot::{ Worker, WorkerEvent, WorkerEventKind, WorkerFailure, WorkerFailureKind, WorkerReadySnapshot, WorkerRegistry, WorkerStatus, diff --git a/rust/crates/runtime/src/runtime_task_store.rs b/rust/crates/runtime/src/runtime_task_store.rs new file mode 100644 index 0000000..03b6671 --- /dev/null +++ b/rust/crates/runtime/src/runtime_task_store.rs @@ -0,0 +1,413 @@ +use std::fs; +use std::io; +use std::path::PathBuf; +use std::process::Command; +use std::sync::{Mutex, OnceLock}; + +use serde::{Deserialize, Serialize}; + +use crate::workflow_state::{now_secs, sanitize_state_component, state_root}; + +fn store_lock() -> &'static Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeTaskKind { + Agent, + Shell, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RuntimeTaskStatus { + Running, + Completed, + Failed, + Stopped, +} + +impl RuntimeTaskStatus { + #[must_use] + pub fn is_terminal(self) -> bool { + matches!(self, Self::Completed | Self::Failed | Self::Stopped) + } +} + +impl std::fmt::Display for RuntimeTaskStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Running => write!(f, "running"), + Self::Completed => write!(f, "completed"), + Self::Failed => write!(f, "failed"), + Self::Stopped => write!(f, "stopped"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeTaskRecord { + pub task_id: String, + pub kind: RuntimeTaskKind, + pub status: RuntimeTaskStatus, + pub description: String, + pub prompt: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub output_file: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_code_file: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub final_result: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exit_code: Option, + #[serde(default)] + pub notified: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pid: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub team_name: Option, + pub created_at: u64, + pub started_at: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub completed_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeTaskOutput { + pub task: RuntimeTaskRecord, + pub output: String, + pub has_output: bool, +} + +#[derive(Debug, Clone, Default)] +pub struct RuntimeTaskStore; + +impl RuntimeTaskStore { + #[must_use] + pub fn new() -> Self { + Self + } + + pub fn create_shell_task( + &self, + description: String, + command: String, + pid: u32, + output_file: PathBuf, + exit_code_file: PathBuf, + team_name: Option, + ) -> io::Result { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let now = now_secs(); + let record = RuntimeTaskRecord { + task_id: make_runtime_task_id("shell"), + kind: RuntimeTaskKind::Shell, + status: RuntimeTaskStatus::Running, + description, + prompt: command, + output_file: Some(output_file.display().to_string()), + exit_code_file: Some(exit_code_file.display().to_string()), + final_result: None, + error: None, + exit_code: None, + notified: false, + pid: Some(pid), + agent_id: None, + agent_name: None, + team_name, + created_at: now, + started_at: now, + completed_at: None, + }; + self.write_locked(&record)?; + Ok(record) + } + + pub fn create_agent_task( + &self, + agent_id: String, + agent_name: String, + description: String, + prompt: String, + output_file: String, + team_name: Option, + ) -> io::Result { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let now = now_secs(); + let record = RuntimeTaskRecord { + task_id: agent_id.clone(), + kind: RuntimeTaskKind::Agent, + status: RuntimeTaskStatus::Running, + description, + prompt, + output_file: Some(output_file), + exit_code_file: None, + final_result: None, + error: None, + exit_code: None, + notified: false, + pid: None, + agent_id: Some(agent_id), + agent_name: Some(agent_name), + team_name, + created_at: now, + started_at: now, + completed_at: None, + }; + self.write_locked(&record)?; + Ok(record) + } + + pub fn get(&self, task_id: &str) -> io::Result> { + let mut record = match fs::read_to_string(task_path(task_id)?) { + Ok(contents) => serde_json::from_str::(&contents) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error), + }; + refresh_record(&mut record)?; + Ok(Some(record)) + } + + pub fn list(&self) -> io::Result> { + let dir = tasks_dir()?; + let entries = match fs::read_dir(dir) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(error), + }; + let mut tasks = Vec::new(); + for entry in entries { + let path = entry?.path(); + if path.extension().and_then(|value| value.to_str()) != Some("json") { + continue; + } + let contents = fs::read_to_string(&path)?; + let mut record = serde_json::from_str::(&contents) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + refresh_record(&mut record)?; + tasks.push(record); + } + tasks.sort_by(|left, right| left.created_at.cmp(&right.created_at)); + Ok(tasks) + } + + pub fn mark_agent_terminal( + &self, + task_id: &str, + status: RuntimeTaskStatus, + final_result: Option, + error: Option, + ) -> io::Result> { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(mut record) = self.get(task_id)? else { + return Ok(None); + }; + record.status = status; + record.final_result = final_result; + record.error = error; + record.completed_at = Some(now_secs()); + self.write_locked(&record)?; + Ok(Some(record)) + } + + pub fn stop(&self, task_id: &str) -> io::Result> { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(mut record) = self.get(task_id)? else { + return Ok(None); + }; + if record.status.is_terminal() { + return Ok(Some(record)); + } + if let Some(pid) = record.pid { + stop_pid(pid)?; + } else if record.kind == RuntimeTaskKind::Agent { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "background agent tasks cannot be stopped yet", + )); + } + record.status = RuntimeTaskStatus::Stopped; + record.completed_at = Some(now_secs()); + self.write_locked(&record)?; + Ok(Some(record)) + } + + pub fn output( + &self, + task_id: &str, + block: bool, + timeout_ms: Option, + ) -> io::Result> { + let start = std::time::Instant::now(); + loop { + let Some(record) = self.get(task_id)? else { + return Ok(None); + }; + if !block || record.status.is_terminal() { + let output = record + .output_file + .as_deref() + .map(fs::read_to_string) + .transpose()? + .unwrap_or_default(); + return Ok(Some(RuntimeTaskOutput { + has_output: !output.trim().is_empty(), + output, + task: record, + })); + } + if let Some(timeout_ms) = timeout_ms { + if start.elapsed() >= std::time::Duration::from_millis(timeout_ms) { + let output = record + .output_file + .as_deref() + .map(fs::read_to_string) + .transpose()? + .unwrap_or_default(); + return Ok(Some(RuntimeTaskOutput { + has_output: !output.trim().is_empty(), + output, + task: record, + })); + } + } + std::thread::sleep(std::time::Duration::from_millis(200)); + } + } + + fn write_locked(&self, record: &RuntimeTaskRecord) -> io::Result<()> { + let path = task_path(&record.task_id)?; + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let payload = serde_json::to_vec_pretty(record) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + fs::write(path, payload) + } +} + +fn tasks_dir() -> io::Result { + Ok(state_root()?.join("runtime-tasks")) +} + +fn task_path(task_id: &str) -> io::Result { + Ok(tasks_dir()?.join(format!( + "{}.json", + sanitize_state_component(task_id) + ))) +} + +fn make_runtime_task_id(prefix: &str) -> String { + format!("{prefix}-{}", now_secs()) +} + +fn refresh_record(record: &mut RuntimeTaskRecord) -> io::Result<()> { + if record.status.is_terminal() || record.kind != RuntimeTaskKind::Shell { + return Ok(()); + } + let Some(exit_code_path) = record.exit_code_file.as_deref() else { + return Ok(()); + }; + if let Ok(contents) = fs::read_to_string(exit_code_path) { + if let Ok(code) = contents.trim().parse::() { + record.exit_code = Some(code); + record.status = if code == 0 { + RuntimeTaskStatus::Completed + } else { + RuntimeTaskStatus::Failed + }; + record.completed_at = Some(now_secs()); + let payload = serde_json::to_vec_pretty(record) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + fs::write(task_path(&record.task_id)?, payload)?; + } + } + Ok(()) +} + +fn stop_pid(pid: u32) -> io::Result<()> { + #[cfg(unix)] + { + let status = Command::new("kill") + .arg("-TERM") + .arg(pid.to_string()) + .status()?; + if status.success() { + Ok(()) + } else { + Err(io::Error::other(format!("failed to stop pid {pid}"))) + } + } + #[cfg(not(unix))] + { + let _ = pid; + Err(io::Error::new( + io::ErrorKind::Unsupported, + "runtime task stopping is only supported on unix", + )) + } +} + +pub fn background_output_paths(task_id: &str) -> io::Result<(PathBuf, PathBuf)> { + let root = state_root()?.join("runtime-tasks").join("outputs"); + fs::create_dir_all(&root)?; + let safe = sanitize_state_component(task_id); + Ok((root.join(format!("{safe}.log")), root.join(format!("{safe}.exit")))) +} + +#[cfg(test)] +mod tests { + use std::fs; + + use super::{background_output_paths, RuntimeTaskStatus, RuntimeTaskStore}; + use crate::test_env_lock; + + #[test] + fn shell_runtime_tasks_refresh_from_exit_file() { + let _lock = test_env_lock(); + let root = std::env::temp_dir().join("runtime-task-store-tests"); + let _ = fs::remove_dir_all(&root); + std::env::set_var("CLAW_WORKER_STATE_ROOT", &root); + + let store = RuntimeTaskStore::new(); + let (output_path, exit_path) = background_output_paths("shell-1").expect("paths"); + fs::write(&output_path, "hello").expect("write output"); + let record = store + .create_shell_task( + "Run tests".to_string(), + "cargo test".to_string(), + 123, + output_path.clone(), + exit_path.clone(), + None, + ) + .expect("create task"); + fs::write(&exit_path, "0").expect("write exit code"); + let output = store + .output(&record.task_id, false, None) + .expect("load output") + .expect("task exists"); + assert_eq!(output.task.status, RuntimeTaskStatus::Completed); + assert_eq!(output.output, "hello"); + + let _ = fs::remove_dir_all(&root); + std::env::remove_var("CLAW_WORKER_STATE_ROOT"); + } +} diff --git a/rust/crates/runtime/src/task_list_store.rs b/rust/crates/runtime/src/task_list_store.rs new file mode 100644 index 0000000..a645f30 --- /dev/null +++ b/rust/crates/runtime/src/task_list_store.rs @@ -0,0 +1,369 @@ +use std::collections::BTreeMap; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, OnceLock}; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::workflow_state::{ + current_task_list_id, now_secs, sanitize_state_component, state_root, +}; + +fn store_lock() -> &'static Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TaskListStatus { + Pending, + InProgress, + Completed, +} + +impl std::fmt::Display for TaskListStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Pending => write!(f, "pending"), + Self::InProgress => write!(f, "in_progress"), + Self::Completed => write!(f, "completed"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct TaskListRecord { + pub id: String, + pub subject: String, + pub description: String, + #[serde(rename = "activeForm", default, skip_serializing_if = "Option::is_none")] + pub active_form: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub owner: Option, + pub status: TaskListStatus, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub blocks: Vec, + #[serde(rename = "blockedBy", default, skip_serializing_if = "Vec::is_empty")] + pub blocked_by: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub metadata: Option>, + #[serde(default)] + pub internal: bool, + #[serde(rename = "createdAt")] + pub created_at: u64, + #[serde(rename = "updatedAt")] + pub updated_at: u64, +} + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct TaskListPatch { + #[serde(default)] + pub subject: Option, + #[serde(default)] + pub description: Option, + #[serde(rename = "activeForm", default)] + pub active_form: Option, + #[serde(default)] + pub status: Option, + #[serde(rename = "addBlocks", default)] + pub add_blocks: Vec, + #[serde(rename = "addBlockedBy", default)] + pub add_blocked_by: Vec, + #[serde(default)] + pub owner: Option, + #[serde(default)] + pub metadata: Option>, + #[serde(default)] + pub internal: Option, +} + +#[derive(Debug, Clone)] +pub struct TaskListStore { + task_list_id: String, +} + +impl TaskListStore { + pub fn current() -> io::Result { + Ok(Self { + task_list_id: current_task_list_id()?, + }) + } + + pub fn for_task_list(task_list_id: impl Into) -> Self { + Self { + task_list_id: sanitize_state_component(&task_list_id.into()), + } + } + + #[must_use] + pub fn task_list_id(&self) -> &str { + &self.task_list_id + } + + fn tasks_dir(&self) -> io::Result { + Ok(state_root()?.join("tasks").join(&self.task_list_id)) + } + + fn task_path(&self, task_id: &str) -> io::Result { + Ok(self.tasks_dir()?.join(format!( + "{}.json", + sanitize_state_component(task_id) + ))) + } + + fn high_water_mark_path(&self) -> io::Result { + Ok(self.tasks_dir()?.join(".highwatermark")) + } + + pub fn create( + &self, + subject: String, + description: String, + active_form: Option, + metadata: Option>, + ) -> io::Result { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let task_dir = self.tasks_dir()?; + fs::create_dir_all(&task_dir)?; + let next_id = self.next_task_id_locked()?; + let now = now_secs(); + let record = TaskListRecord { + id: next_id.clone(), + subject, + description, + active_form, + owner: None, + status: TaskListStatus::Pending, + blocks: Vec::new(), + blocked_by: Vec::new(), + metadata, + internal: false, + created_at: now, + updated_at: now, + }; + self.write_record_locked(&record)?; + Ok(record) + } + + pub fn get(&self, task_id: &str) -> io::Result> { + let path = self.task_path(task_id)?; + read_record(&path) + } + + pub fn list(&self, include_internal: bool) -> io::Result> { + let mut records = Vec::new(); + let dir = self.tasks_dir()?; + let entries = match fs::read_dir(dir) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(records), + Err(error) => return Err(error), + }; + for entry in entries { + let entry = entry?; + let path = entry.path(); + if path.extension().and_then(|value| value.to_str()) != Some("json") { + continue; + } + if let Some(record) = read_record(&path)? { + if include_internal || !record.internal { + records.push(record); + } + } + } + records.sort_by_key(|record| record.id.parse::().unwrap_or(0)); + Ok(records) + } + + pub fn update(&self, task_id: &str, patch: TaskListPatch) -> io::Result> { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(mut existing) = self.get(task_id)? else { + return Ok(None); + }; + if let Some(subject) = patch.subject { + existing.subject = subject; + } + if let Some(description) = patch.description { + existing.description = description; + } + if let Some(active_form) = patch.active_form { + existing.active_form = Some(active_form); + } + if let Some(status) = patch.status { + existing.status = status; + } + if let Some(owner) = patch.owner { + existing.owner = if owner.trim().is_empty() { + None + } else { + Some(owner) + }; + } + if let Some(metadata) = patch.metadata { + existing.metadata = Some(metadata); + } + if let Some(internal) = patch.internal { + existing.internal = internal; + } + merge_unique(&mut existing.blocks, patch.add_blocks); + merge_unique(&mut existing.blocked_by, patch.add_blocked_by); + existing.updated_at = now_secs(); + self.write_record_locked(&existing)?; + Ok(Some(existing)) + } + + pub fn delete(&self, task_id: &str) -> io::Result { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let path = self.task_path(task_id)?; + match fs::remove_file(&path) { + Ok(()) => { + for mut other in self.list(true)? { + let original_blocks = other.blocks.len(); + let original_blocked_by = other.blocked_by.len(); + other.blocks.retain(|value| value != task_id); + other.blocked_by.retain(|value| value != task_id); + if other.blocks.len() != original_blocks + || other.blocked_by.len() != original_blocked_by + { + other.updated_at = now_secs(); + self.write_record_locked(&other)?; + } + } + Ok(true) + } + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(false), + Err(error) => Err(error), + } + } + + pub fn reset(&self) -> io::Result<()> { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let dir = self.tasks_dir()?; + let existing = self.list(true)?; + let max_id = existing + .iter() + .filter_map(|task| task.id.parse::().ok()) + .max() + .unwrap_or(0); + fs::create_dir_all(&dir)?; + for entry in fs::read_dir(&dir)? { + let path = entry?.path(); + if path.extension().and_then(|value| value.to_str()) == Some("json") { + let _ = fs::remove_file(path); + } + } + if max_id > 0 { + fs::write(self.high_water_mark_path()?, max_id.to_string())?; + } + Ok(()) + } + + fn next_task_id_locked(&self) -> io::Result { + let existing_max = self + .list(true)? + .into_iter() + .filter_map(|task| task.id.parse::().ok()) + .max() + .unwrap_or(0); + let high_water_mark = fs::read_to_string(self.high_water_mark_path()?) + .ok() + .and_then(|value| value.trim().parse::().ok()) + .unwrap_or(0); + let next = existing_max.max(high_water_mark) + 1; + fs::write(self.high_water_mark_path()?, next.to_string())?; + Ok(next.to_string()) + } + + fn write_record_locked(&self, record: &TaskListRecord) -> io::Result<()> { + let path = self.task_path(&record.id)?; + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let payload = serde_json::to_vec_pretty(record) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + fs::write(path, payload) + } +} + +fn merge_unique(target: &mut Vec, additions: Vec) { + for value in additions { + if !target.contains(&value) { + target.push(value); + } + } +} + +fn read_record(path: &Path) -> io::Result> { + match fs::read_to_string(path) { + Ok(contents) => serde_json::from_str(&contents) + .map(Some) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error), + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use serde_json::json; + + use super::{TaskListPatch, TaskListStatus, TaskListStore}; + use crate::test_env_lock; + + #[test] + fn create_update_and_delete_task_records() { + let _lock = test_env_lock(); + let root = std::env::temp_dir().join("task-list-store-tests"); + let _ = std::fs::remove_dir_all(&root); + std::env::set_var("CLAW_WORKER_STATE_ROOT", &root); + std::env::set_var("CLAW_WORKER_PROFILE_ID", "makar"); + + let store = TaskListStore::for_task_list("alpha"); + let created = store + .create( + "Investigate".to_string(), + "Check the failing worker".to_string(), + Some("Investigating".to_string()), + Some(BTreeMap::from([("priority".to_string(), json!("high"))])), + ) + .expect("task creates"); + assert_eq!(created.id, "1"); + + let updated = store + .update( + &created.id, + TaskListPatch { + status: Some(TaskListStatus::InProgress), + owner: Some("agent-lead".to_string()), + add_blocked_by: vec!["7".to_string()], + ..TaskListPatch::default() + }, + ) + .expect("task updates") + .expect("task exists"); + assert_eq!(updated.status, TaskListStatus::InProgress); + assert_eq!(updated.owner.as_deref(), Some("agent-lead")); + assert_eq!(updated.blocked_by, vec!["7"]); + + let listed = store.list(false).expect("tasks list"); + assert_eq!(listed.len(), 1); + assert!(store.delete(&created.id).expect("delete succeeds")); + assert!(store.list(false).expect("tasks list").is_empty()); + + let _ = std::fs::remove_dir_all(&root); + std::env::remove_var("CLAW_WORKER_STATE_ROOT"); + std::env::remove_var("CLAW_WORKER_PROFILE_ID"); + } +} diff --git a/rust/crates/runtime/src/teamwork_store.rs b/rust/crates/runtime/src/teamwork_store.rs new file mode 100644 index 0000000..29db602 --- /dev/null +++ b/rust/crates/runtime/src/teamwork_store.rs @@ -0,0 +1,384 @@ +use std::fs; +use std::io; +use std::path::PathBuf; +use std::sync::{Mutex, OnceLock}; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::task_list_store::TaskListStore; +use crate::workflow_state::{ + clear_team_context, now_secs, persist_team_context, sanitize_state_component, state_root, + TeamContext, +}; + +fn store_lock() -> &'static Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct TeamMemberRecord { + pub agent_id: String, + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_type: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub status: Option, + pub joined_at: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct TeamRecord { + pub team_name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_type: Option, + pub lead_agent_id: String, + pub created_at: u64, + pub updated_at: u64, + #[serde(default)] + pub deleted: bool, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub members: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MessageEnvelope { + pub id: String, + pub from: String, + pub to: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub summary: Option, + pub message: Value, + pub timestamp: u64, + #[serde(default)] + pub read: bool, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MailboxMessage { + pub recipient: String, + pub envelope: MessageEnvelope, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MailboxSummary { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub team_name: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub recent_messages: Vec, +} + +#[derive(Debug, Clone, Default)] +pub struct TeamStore; + +impl TeamStore { + #[must_use] + pub fn new() -> Self { + Self + } + + pub fn create_team( + &self, + requested_name: &str, + description: Option, + agent_type: Option, + ) -> io::Result { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let final_name = self.unique_team_name(requested_name)?; + let now = now_secs(); + let record = TeamRecord { + team_name: final_name.clone(), + description: description.clone(), + agent_type: agent_type.clone(), + lead_agent_id: format!("lead@{final_name}"), + created_at: now, + updated_at: now, + deleted: false, + members: vec![TeamMemberRecord { + agent_id: format!("lead@{final_name}"), + name: "lead".to_string(), + agent_type: agent_type.clone(), + model: None, + status: Some("active".to_string()), + joined_at: now, + }], + }; + self.write_team_locked(&record)?; + persist_team_context(&TeamContext { + team_name: record.team_name.clone(), + lead_agent_id: record.lead_agent_id.clone(), + description, + agent_type, + task_list_id: record.team_name.clone(), + created_at: now, + })?; + TaskListStore::for_task_list(record.team_name.clone()).reset()?; + Ok(record) + } + + pub fn delete_team(&self, team_name: &str) -> io::Result> { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(mut record) = self.get_team(team_name)? else { + return Ok(None); + }; + record.deleted = true; + record.updated_at = now_secs(); + self.write_team_locked(&record)?; + if record.team_name == team_name { + let _ = clear_team_context(); + } + Ok(Some(record)) + } + + pub fn current_team(&self) -> io::Result> { + let Some(context) = crate::workflow_state::load_team_context()? else { + return Ok(None); + }; + self.get_team(&context.team_name) + } + + pub fn get_team(&self, team_name: &str) -> io::Result> { + let path = team_path(team_name)?; + match fs::read_to_string(path) { + Ok(contents) => serde_json::from_str(&contents) + .map(Some) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error), + } + } + + pub fn upsert_member( + &self, + team_name: &str, + member: TeamMemberRecord, + ) -> io::Result> { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(mut record) = self.get_team(team_name)? else { + return Ok(None); + }; + if let Some(existing) = record.members.iter_mut().find(|value| value.name == member.name) { + *existing = member; + } else { + record.members.push(member); + } + record.updated_at = now_secs(); + self.write_team_locked(&record)?; + Ok(Some(record)) + } + + pub fn send_message( + &self, + team_name: &str, + from: &str, + to: &str, + summary: Option, + message: Value, + ) -> io::Result> { + let recipients = if to == "*" { + self.get_team(team_name)? + .map(|team| { + team.members + .into_iter() + .filter(|member| member.name != from) + .map(|member| member.name) + .collect::>() + }) + .unwrap_or_default() + } else { + vec![to.to_string()] + }; + let mut written = Vec::new(); + for recipient in recipients { + let envelope = MessageEnvelope { + id: format!("msg-{}-{}", now_secs(), sanitize_state_component(&recipient)), + from: from.to_string(), + to: recipient.clone(), + summary: summary.clone(), + message: message.clone(), + timestamp: now_secs(), + read: false, + }; + self.append_mailbox(team_name, &recipient, &envelope)?; + written.push(MailboxMessage { + recipient, + envelope, + }); + } + Ok(written) + } + + pub fn mailbox_summary(&self, team_name: &str, limit: usize) -> io::Result { + let inbox_root = mailboxes_dir(team_name)?; + let entries = match fs::read_dir(&inbox_root) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + return Ok(MailboxSummary { + team_name: Some(team_name.to_string()), + recent_messages: Vec::new(), + }) + } + Err(error) => return Err(error), + }; + let mut messages = Vec::new(); + for entry in entries { + let path = entry?.path(); + if path.extension().and_then(|value| value.to_str()) != Some("json") { + continue; + } + let recipient = path + .file_stem() + .and_then(|value| value.to_str()) + .unwrap_or("unknown") + .to_string(); + let contents = fs::read_to_string(&path)?; + let inbox = serde_json::from_str::>(&contents) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + for envelope in inbox { + messages.push(MailboxMessage { + recipient: recipient.clone(), + envelope, + }); + } + } + messages.sort_by(|left, right| left.envelope.timestamp.cmp(&right.envelope.timestamp)); + if messages.len() > limit { + messages = messages.split_off(messages.len() - limit); + } + Ok(MailboxSummary { + team_name: Some(team_name.to_string()), + recent_messages: messages, + }) + } + + fn write_team_locked(&self, team: &TeamRecord) -> io::Result<()> { + let path = team_path(&team.team_name)?; + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let payload = serde_json::to_vec_pretty(team) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + fs::write(path, payload) + } + + fn append_mailbox( + &self, + team_name: &str, + recipient: &str, + envelope: &MessageEnvelope, + ) -> io::Result<()> { + let path = mailbox_path(team_name, recipient)?; + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let mut existing = match fs::read_to_string(&path) { + Ok(contents) => serde_json::from_str::>(&contents) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?, + Err(error) if error.kind() == io::ErrorKind::NotFound => Vec::new(), + Err(error) => return Err(error), + }; + existing.push(envelope.clone()); + let payload = serde_json::to_vec_pretty(&existing) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + fs::write(path, payload) + } + + fn unique_team_name(&self, requested_name: &str) -> io::Result { + let requested_name = sanitize_state_component(requested_name); + if self.get_team(&requested_name)?.is_none() { + return Ok(requested_name); + } + for index in 2..=128 { + let candidate = format!("{requested_name}-{index}"); + if self.get_team(&candidate)?.is_none() { + return Ok(candidate); + } + } + Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "failed to allocate a unique team name", + )) + } +} + +fn teams_dir() -> io::Result { + Ok(state_root()?.join("teams")) +} + +fn team_path(team_name: &str) -> io::Result { + Ok(teams_dir()?.join(sanitize_state_component(team_name)).join("config.json")) +} + +fn mailboxes_dir(team_name: &str) -> io::Result { + Ok(state_root()? + .join("mailbox") + .join(sanitize_state_component(team_name))) +} + +fn mailbox_path(team_name: &str, recipient: &str) -> io::Result { + Ok(mailboxes_dir(team_name)?.join(format!( + "{}.json", + sanitize_state_component(recipient) + ))) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::{TeamMemberRecord, TeamStore}; + use crate::test_env_lock; + + #[test] + fn team_creation_assigns_unique_names_and_mailbox_messages_round_trip() { + let _lock = test_env_lock(); + let root = std::env::temp_dir().join("team-store-tests"); + let _ = std::fs::remove_dir_all(&root); + std::env::set_var("CLAW_WORKER_STATE_ROOT", &root); + + let store = TeamStore::new(); + let first = store + .create_team("alpha", Some("Team Alpha".to_string()), Some("researcher".to_string())) + .expect("team creates"); + let second = store + .create_team("alpha", None, None) + .expect("second team creates"); + assert_eq!(first.team_name, "alpha"); + assert_eq!(second.team_name, "alpha-2"); + store + .upsert_member( + &first.team_name, + TeamMemberRecord { + agent_id: "agent-1".to_string(), + name: "alice".to_string(), + agent_type: Some("researcher".to_string()), + model: Some("claude-opus-4-6".to_string()), + status: Some("active".to_string()), + joined_at: 1, + }, + ) + .expect("member upserts"); + let messages = store + .send_message(&first.team_name, "lead", "alice", Some("hello".to_string()), json!("hi")) + .expect("message sends"); + assert_eq!(messages.len(), 1); + let summary = store + .mailbox_summary(&first.team_name, 10) + .expect("summary loads"); + assert_eq!(summary.recent_messages.len(), 1); + + let _ = std::fs::remove_dir_all(&root); + std::env::remove_var("CLAW_WORKER_STATE_ROOT"); + } +} diff --git a/rust/crates/runtime/src/workflow_state.rs b/rust/crates/runtime/src/workflow_state.rs new file mode 100644 index 0000000..375ba1b --- /dev/null +++ b/rust/crates/runtime/src/workflow_state.rs @@ -0,0 +1,176 @@ +use std::env; +use std::fs; +use std::io; +use std::path::{Path, PathBuf}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct TeamContext { + pub team_name: String, + pub lead_agent_id: String, + #[serde(default)] + pub description: Option, + #[serde(default)] + pub agent_type: Option, + pub task_list_id: String, + pub created_at: u64, +} + +#[must_use] +pub fn sanitize_state_component(value: &str) -> String { + let sanitized = value + .trim() + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') { + ch + } else { + '-' + } + }) + .collect::(); + let collapsed = sanitized + .split('-') + .filter(|segment| !segment.is_empty()) + .collect::>() + .join("-"); + if collapsed.is_empty() { + "default".to_string() + } else { + collapsed + } +} + +pub fn state_root() -> io::Result { + if let Some(root) = env::var_os("CLAW_WORKER_STATE_ROOT") + .filter(|value| !value.is_empty()) + .map(PathBuf::from) + { + return Ok(root); + } + if let Some(root) = env::var_os("CLAWD_STATE_ROOT") + .filter(|value| !value.is_empty()) + .map(PathBuf::from) + { + return Ok(root); + } + Ok(env::current_dir()?.join(".clawd-state")) +} + +#[must_use] +pub fn default_session_identity() -> String { + env::var("CLAW_WORKER_PROFILE_ID") + .ok() + .filter(|value| !value.trim().is_empty()) + .or_else(|| { + env::current_dir() + .ok() + .and_then(|cwd| cwd.file_name().map(|value| value.to_string_lossy().to_string())) + }) + .map(|value| sanitize_state_component(&value)) + .unwrap_or_else(|| "default".to_string()) +} + +fn team_context_path(root: &Path) -> PathBuf { + root.join("session").join("team-context.json") +} + +pub fn load_team_context() -> io::Result> { + let path = team_context_path(&state_root()?); + match fs::read_to_string(path) { + Ok(contents) => serde_json::from_str(&contents) + .map(Some) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error), + } +} + +pub fn persist_team_context(context: &TeamContext) -> io::Result<()> { + let root = state_root()?; + let path = team_context_path(&root); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let serialized = serde_json::to_vec_pretty(context) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + fs::write(path, serialized) +} + +pub fn clear_team_context() -> io::Result<()> { + let path = team_context_path(&state_root()?); + match fs::remove_file(path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +pub fn current_task_list_id() -> io::Result { + if let Some(explicit) = env::var_os("CLAW_CODE_TASK_LIST_ID") + .filter(|value| !value.is_empty()) + .map(|value| sanitize_state_component(&value.to_string_lossy())) + { + return Ok(explicit); + } + if let Some(team) = load_team_context()? { + return Ok(sanitize_state_component(&team.task_list_id)); + } + Ok(default_session_identity()) +} + +#[must_use] +pub fn now_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +#[cfg(test)] +mod tests { + use super::{ + clear_team_context, current_task_list_id, default_session_identity, load_team_context, + persist_team_context, sanitize_state_component, state_root, TeamContext, + }; + use crate::test_env_lock; + + #[test] + fn sanitize_state_component_removes_path_chars() { + assert_eq!(sanitize_state_component("../Team Alpha"), "Team-Alpha"); + assert_eq!(sanitize_state_component(""), "default"); + } + + #[test] + fn team_context_round_trips() { + let _lock = test_env_lock(); + let root = std::env::temp_dir().join("workflow-state-roundtrip"); + let _ = std::fs::remove_dir_all(&root); + std::env::set_var("CLAW_WORKER_STATE_ROOT", &root); + std::env::set_var("CLAW_WORKER_PROFILE_ID", "makar"); + + let context = TeamContext { + team_name: "alpha".to_string(), + lead_agent_id: "agent-1".to_string(), + description: Some("test".to_string()), + agent_type: Some("researcher".to_string()), + task_list_id: "alpha".to_string(), + created_at: 1, + }; + persist_team_context(&context).expect("context persists"); + let loaded = load_team_context() + .expect("context loads") + .expect("context should exist"); + assert_eq!(loaded, context); + assert_eq!(current_task_list_id().expect("list id"), "alpha"); + clear_team_context().expect("context clears"); + assert!(load_team_context().expect("loads").is_none()); + assert_eq!(default_session_identity(), "makar"); + assert_eq!(state_root().expect("state root"), root); + let _ = std::fs::remove_dir_all(&root); + std::env::remove_var("CLAW_WORKER_STATE_ROOT"); + std::env::remove_var("CLAW_WORKER_PROFILE_ID"); + } +} diff --git a/rust/crates/tools/.gitignore b/rust/crates/tools/.gitignore index 96da1ea..0ee50b1 100644 --- a/rust/crates/tools/.gitignore +++ b/rust/crates/tools/.gitignore @@ -1 +1,2 @@ .clawd-agents/ +.clawd-state/ diff --git a/rust/crates/tools/src/lane_completion.rs b/rust/crates/tools/src/lane_completion.rs index 2850127..922c672 100644 --- a/rust/crates/tools/src/lane_completion.rs +++ b/rust/crates/tools/src/lane_completion.rs @@ -16,7 +16,7 @@ use runtime::{ use crate::AgentOutput; /// Detects if a lane should be automatically marked as completed. -/// +/// /// Returns `Some(LaneContext)` with `completed = true` if all conditions met, /// `None` if lane should remain active. pub(crate) fn detect_lane_completion( @@ -28,29 +28,29 @@ pub(crate) fn detect_lane_completion( if output.error.is_some() { return None; } - + // Must have finished status if !output.status.eq_ignore_ascii_case("completed") && !output.status.eq_ignore_ascii_case("finished") { return None; } - + // Must have no current blocker if output.current_blocker.is_some() { return None; } - + // Must have green tests if !test_green { return None; } - + // Must have pushed code if !has_pushed { return None; } - + // All conditions met — create completed context Some(LaneContext { lane_id: output.agent_id.clone(), @@ -65,9 +65,7 @@ pub(crate) fn detect_lane_completion( } /// Evaluates policy actions for a completed lane. -pub(crate) fn evaluate_completed_lane( - context: &LaneContext, -) -> Vec { +pub(crate) fn evaluate_completed_lane(context: &LaneContext) -> Vec { let engine = PolicyEngine::new(vec![ PolicyRule::new( "closeout-completed-lane", @@ -85,7 +83,7 @@ pub(crate) fn evaluate_completed_lane( 5, ), ]); - + evaluate(&engine, context) } @@ -97,10 +95,13 @@ mod tests { fn test_output() -> AgentOutput { AgentOutput { agent_id: "test-lane-1".to_string(), + task_id: "test-lane-1".to_string(), name: "Test Agent".to_string(), description: "Test".to_string(), subagent_type: None, model: None, + team_name: None, + is_async: true, status: "Finished".to_string(), output_file: "/tmp/test.output".to_string(), manifest_file: "/tmp/test.manifest".to_string(), @@ -112,53 +113,53 @@ mod tests { error: None, } } - + #[test] fn detects_completion_when_all_conditions_met() { let output = test_output(); let result = detect_lane_completion(&output, true, true); - + assert!(result.is_some()); let context = result.unwrap(); assert!(context.completed); assert_eq!(context.green_level, 3); assert_eq!(context.blocker, LaneBlocker::None); } - + #[test] fn no_completion_when_error_present() { let mut output = test_output(); output.error = Some("Build failed".to_string()); - + let result = detect_lane_completion(&output, true, true); assert!(result.is_none()); } - + #[test] fn no_completion_when_not_finished() { let mut output = test_output(); output.status = "Running".to_string(); - + let result = detect_lane_completion(&output, true, true); assert!(result.is_none()); } - + #[test] fn no_completion_when_tests_not_green() { let output = test_output(); - + let result = detect_lane_completion(&output, false, true); assert!(result.is_none()); } - + #[test] fn no_completion_when_not_pushed() { let output = test_output(); - + let result = detect_lane_completion(&output, true, false); assert!(result.is_none()); } - + #[test] fn evaluate_triggers_closeout_for_completed_lane() { let context = LaneContext { @@ -171,9 +172,9 @@ mod tests { completed: true, reconciled: false, }; - + let actions = evaluate_completed_lane(&context); - + assert!(actions.contains(&PolicyAction::CloseoutLane)); assert!(actions.contains(&PolicyAction::CleanupSession)); } diff --git a/rust/crates/tools/src/lib.rs b/rust/crates/tools/src/lib.rs index 4cec927..54dd769 100644 --- a/rust/crates/tools/src/lib.rs +++ b/rust/crates/tools/src/lib.rs @@ -11,14 +11,17 @@ use api::{ use plugins::PluginTool; use reqwest::blocking::Client; use runtime::{ - check_freshness, edit_file, execute_bash, glob_search, grep_search, load_system_prompt, + background_output_paths, check_freshness, current_task_list_id, + edit_file, execute_bash, glob_search, grep_search, load_system_prompt, spawn_background_bash, lsp_client::LspRegistry, mcp_tool_bridge::McpToolRegistry, permission_enforcer::{EnforcementResult, PermissionEnforcer}, read_file, + runtime_task_store::{RuntimeTaskOutput, RuntimeTaskRecord, RuntimeTaskStore}, summary_compression::compress_summary_text, - task_registry::TaskRegistry, - team_cron_registry::{CronRegistry, TeamRegistry}, + task_list_store::{TaskListPatch, TaskListStatus, TaskListStore}, + team_cron_registry::CronRegistry, + teamwork_store::{TeamMemberRecord, TeamStore}, worker_boot::{WorkerReadySnapshot, WorkerRegistry}, write_file, ApiClient, ApiRequest, AssistantEvent, BashCommandInput, BashCommandOutput, BranchFreshness, ContentBlock, ConversationMessage, ConversationRuntime, GrepSearchInput, @@ -42,10 +45,10 @@ fn global_mcp_registry() -> &'static McpToolRegistry { REGISTRY.get_or_init(McpToolRegistry::new) } -fn global_team_registry() -> &'static TeamRegistry { +fn global_team_store() -> &'static TeamStore { use std::sync::OnceLock; - static REGISTRY: OnceLock = OnceLock::new(); - REGISTRY.get_or_init(TeamRegistry::new) + static REGISTRY: OnceLock = OnceLock::new(); + REGISTRY.get_or_init(TeamStore::new) } fn global_cron_registry() -> &'static CronRegistry { @@ -54,10 +57,10 @@ fn global_cron_registry() -> &'static CronRegistry { REGISTRY.get_or_init(CronRegistry::new) } -fn global_task_registry() -> &'static TaskRegistry { +fn global_runtime_task_store() -> &'static RuntimeTaskStore { use std::sync::OnceLock; - static REGISTRY: OnceLock = OnceLock::new(); - REGISTRY.get_or_init(TaskRegistry::new) + static REGISTRY: OnceLock = OnceLock::new(); + REGISTRY.get_or_init(RuntimeTaskStore::new) } fn global_worker_registry() -> &'static WorkerRegistry { @@ -568,7 +571,7 @@ pub fn mvp_tool_specs() -> Vec { }, ToolSpec { name: "Agent", - description: "Launch a specialized agent task and persist its handoff metadata.", + description: "Launch a specialized subagent or teammate with persisted runtime state.", input_schema: json!({ "type": "object", "properties": { @@ -576,7 +579,9 @@ pub fn mvp_tool_specs() -> Vec { "prompt": { "type": "string" }, "subagent_type": { "type": "string" }, "name": { "type": "string" }, - "model": { "type": "string" } + "model": { "type": "string" }, + "run_in_background": { "type": "boolean" }, + "team_name": { "type": "string" } }, "required": ["description", "prompt"], "additionalProperties": false @@ -743,34 +748,36 @@ pub fn mvp_tool_specs() -> Vec { }, ToolSpec { name: "TaskCreate", - description: "Create a background task that runs in a separate subprocess.", + description: "Create a disk-backed task-list item in the current session or team task list.", input_schema: json!({ "type": "object", "properties": { - "prompt": { "type": "string" }, - "description": { "type": "string" } + "subject": { "type": "string" }, + "description": { "type": "string" }, + "activeForm": { "type": "string" }, + "metadata": { "type": "object" } }, - "required": ["prompt"], + "required": ["subject", "description"], "additionalProperties": false }), - required_permission: PermissionMode::DangerFullAccess, + required_permission: PermissionMode::WorkspaceWrite, }, ToolSpec { name: "TaskGet", - description: "Get the status and details of a background task by ID.", + description: "Get a task-list item by ID from the current session or team task list.", input_schema: json!({ "type": "object", "properties": { - "task_id": { "type": "string" } + "taskId": { "type": "string" } }, - "required": ["task_id"], + "required": ["taskId"], "additionalProperties": false }), required_permission: PermissionMode::ReadOnly, }, ToolSpec { name: "TaskList", - description: "List all background tasks and their current status.", + description: "List task-list items for the current session or active team.", input_schema: json!({ "type": "object", "properties": {}, @@ -780,7 +787,7 @@ pub fn mvp_tool_specs() -> Vec { }, ToolSpec { name: "TaskStop", - description: "Stop a running background task by ID.", + description: "Stop a running background runtime task by ID.", input_schema: json!({ "type": "object", "properties": { @@ -793,175 +800,84 @@ pub fn mvp_tool_specs() -> Vec { }, ToolSpec { name: "TaskUpdate", - description: "Send a message or update to a running background task.", + description: "Apply a TypeScript-style patch to an existing task-list item.", + input_schema: json!({ + "type": "object", + "properties": { + "taskId": { "type": "string" }, + "subject": { "type": "string" }, + "description": { "type": "string" }, + "activeForm": { "type": "string" }, + "status": { "type": "string", "enum": ["pending", "in_progress", "completed", "deleted"] }, + "addBlocks": { "type": "array", "items": { "type": "string" } }, + "addBlockedBy": { "type": "array", "items": { "type": "string" } }, + "owner": { "type": "string" }, + "metadata": { "type": "object" } + }, + "required": ["taskId"], + "additionalProperties": false + }), + required_permission: PermissionMode::WorkspaceWrite, + }, + ToolSpec { + name: "TaskOutput", + description: "Read output from a running or completed background runtime task.", input_schema: json!({ "type": "object", "properties": { "task_id": { "type": "string" }, - "message": { "type": "string" } - }, - "required": ["task_id", "message"], - "additionalProperties": false - }), - required_permission: PermissionMode::DangerFullAccess, - }, - ToolSpec { - name: "TaskOutput", - description: "Retrieve the output produced by a background task.", - input_schema: json!({ - "type": "object", - "properties": { - "task_id": { "type": "string" } + "block": { "type": "boolean" }, + "timeout_ms": { "type": "integer", "minimum": 1 } }, "required": ["task_id"], "additionalProperties": false }), required_permission: PermissionMode::ReadOnly, }, - ToolSpec { - name: "WorkerCreate", - description: "Create a coding worker boot session with trust-gate and prompt-delivery guards.", - input_schema: json!({ - "type": "object", - "properties": { - "cwd": { "type": "string" }, - "trusted_roots": { - "type": "array", - "items": { "type": "string" } - }, - "auto_recover_prompt_misdelivery": { "type": "boolean" } - }, - "required": ["cwd"], - "additionalProperties": false - }), - required_permission: PermissionMode::DangerFullAccess, - }, - ToolSpec { - name: "WorkerGet", - description: "Fetch the current worker boot state, last error, and event history.", - input_schema: json!({ - "type": "object", - "properties": { - "worker_id": { "type": "string" } - }, - "required": ["worker_id"], - "additionalProperties": false - }), - required_permission: PermissionMode::ReadOnly, - }, - ToolSpec { - name: "WorkerObserve", - description: "Feed a terminal snapshot into worker boot detection to resolve trust gates, ready handshakes, and prompt misdelivery.", - input_schema: json!({ - "type": "object", - "properties": { - "worker_id": { "type": "string" }, - "screen_text": { "type": "string" } - }, - "required": ["worker_id", "screen_text"], - "additionalProperties": false - }), - required_permission: PermissionMode::ReadOnly, - }, - ToolSpec { - name: "WorkerResolveTrust", - description: "Resolve a detected trust prompt so worker boot can continue.", - input_schema: json!({ - "type": "object", - "properties": { - "worker_id": { "type": "string" } - }, - "required": ["worker_id"], - "additionalProperties": false - }), - required_permission: PermissionMode::DangerFullAccess, - }, - ToolSpec { - name: "WorkerAwaitReady", - description: "Return the current ready-handshake verdict for a coding worker.", - input_schema: json!({ - "type": "object", - "properties": { - "worker_id": { "type": "string" } - }, - "required": ["worker_id"], - "additionalProperties": false - }), - required_permission: PermissionMode::ReadOnly, - }, - ToolSpec { - name: "WorkerSendPrompt", - description: "Send a task prompt only after the worker reaches ready_for_prompt; can replay a recovered prompt.", - input_schema: json!({ - "type": "object", - "properties": { - "worker_id": { "type": "string" }, - "prompt": { "type": "string" } - }, - "required": ["worker_id"], - "additionalProperties": false - }), - required_permission: PermissionMode::DangerFullAccess, - }, - ToolSpec { - name: "WorkerRestart", - description: "Restart worker boot state after a failed or stale startup.", - input_schema: json!({ - "type": "object", - "properties": { - "worker_id": { "type": "string" } - }, - "required": ["worker_id"], - "additionalProperties": false - }), - required_permission: PermissionMode::DangerFullAccess, - }, - ToolSpec { - name: "WorkerTerminate", - description: "Terminate a worker and mark the lane finished from the control plane.", - input_schema: json!({ - "type": "object", - "properties": { - "worker_id": { "type": "string" } - }, - "required": ["worker_id"], - "additionalProperties": false - }), - required_permission: PermissionMode::DangerFullAccess, - }, ToolSpec { name: "TeamCreate", - description: "Create a team of sub-agents for parallel task execution.", + description: "Create a persistent team context for the current profile worker.", input_schema: json!({ "type": "object", "properties": { - "name": { "type": "string" }, - "tasks": { - "type": "array", - "items": { - "type": "object", - "properties": { - "prompt": { "type": "string" }, - "description": { "type": "string" } - }, - "required": ["prompt"] - } - } + "team_name": { "type": "string" }, + "description": { "type": "string" }, + "agent_type": { "type": "string" } }, - "required": ["name", "tasks"], + "required": ["team_name"], "additionalProperties": false }), required_permission: PermissionMode::DangerFullAccess, }, ToolSpec { name: "TeamDelete", - description: "Delete a team and stop all its running tasks.", + description: "Delete the active team context and mark the team deleted.", input_schema: json!({ "type": "object", "properties": { - "team_id": { "type": "string" } + "team_name": { "type": "string" } }, - "required": ["team_id"], + "required": ["team_name"], + "additionalProperties": false + }), + required_permission: PermissionMode::DangerFullAccess, + }, + ToolSpec { + name: "SendMessage", + description: "Send a direct or broadcast message into the active team mailbox.", + input_schema: json!({ + "type": "object", + "properties": { + "to": { "type": "string" }, + "summary": { "type": "string" }, + "message": { + "oneOf": [ + { "type": "string" }, + { "type": "object" } + ] + } + }, + "required": ["to", "message"], "additionalProperties": false }), required_permission: PermissionMode::DangerFullAccess, @@ -1177,11 +1093,11 @@ fn execute_tool_with_enforcer( from_value::(input).and_then(run_ask_user_question) } "TaskCreate" => from_value::(input).and_then(run_task_create), - "TaskGet" => from_value::(input).and_then(run_task_get), + "TaskGet" => from_value::(input).and_then(run_task_get), "TaskList" => run_task_list(input.clone()), "TaskStop" => from_value::(input).and_then(run_task_stop), "TaskUpdate" => from_value::(input).and_then(run_task_update), - "TaskOutput" => from_value::(input).and_then(run_task_output), + "TaskOutput" => from_value::(input).and_then(run_task_output), "WorkerCreate" => from_value::(input).and_then(run_worker_create), "WorkerGet" => from_value::(input).and_then(run_worker_get), "WorkerObserve" => from_value::(input).and_then(run_worker_observe), @@ -1196,6 +1112,7 @@ fn execute_tool_with_enforcer( "WorkerTerminate" => from_value::(input).and_then(run_worker_terminate), "TeamCreate" => from_value::(input).and_then(run_team_create), "TeamDelete" => from_value::(input).and_then(run_team_delete), + "SendMessage" => from_value::(input).and_then(run_send_message), "CronCreate" => from_value::(input).and_then(run_cron_create), "CronDelete" => from_value::(input).and_then(run_cron_delete), "CronList" => run_cron_list(input.clone()), @@ -1278,95 +1195,114 @@ fn run_ask_user_question(input: AskUserQuestionInput) -> Result #[allow(clippy::needless_pass_by_value)] fn run_task_create(input: TaskCreateInput) -> Result { - let registry = global_task_registry(); - let task = registry.create(&input.prompt, input.description.as_deref()); + let store = TaskListStore::current().map_err(|error| error.to_string())?; + let task = store + .create( + input.subject, + input.description, + input.active_form, + input.metadata, + ) + .map_err(|error| error.to_string())?; to_pretty_json(json!({ - "task_id": task.task_id, + "task": { + "id": task.id, + "subject": task.subject, + }, + "task_list_id": store.task_list_id(), "status": task.status, - "prompt": task.prompt, "description": task.description, "created_at": task.created_at })) } #[allow(clippy::needless_pass_by_value)] -fn run_task_get(input: TaskIdInput) -> Result { - let registry = global_task_registry(); - match registry.get(&input.task_id) { - Some(task) => to_pretty_json(json!({ - "task_id": task.task_id, - "status": task.status, - "prompt": task.prompt, - "description": task.description, - "created_at": task.created_at, - "updated_at": task.updated_at, - "messages": task.messages, - "team_id": task.team_id - })), +fn run_task_get(input: TaskGetInput) -> Result { + let store = TaskListStore::current().map_err(|error| error.to_string())?; + match store.get(&input.task_id).map_err(|error| error.to_string())? { + Some(task) => to_pretty_json(task), None => Err(format!("task not found: {}", input.task_id)), } } fn run_task_list(_input: Value) -> Result { - let registry = global_task_registry(); - let tasks: Vec<_> = registry - .list(None) - .into_iter() - .map(|t| { - json!({ - "task_id": t.task_id, - "status": t.status, - "prompt": t.prompt, - "description": t.description, - "created_at": t.created_at, - "updated_at": t.updated_at, - "team_id": t.team_id - }) - }) - .collect(); + let store = TaskListStore::current().map_err(|error| error.to_string())?; + let tasks = store.list(false).map_err(|error| error.to_string())?; to_pretty_json(json!({ "tasks": tasks, - "count": tasks.len() + "count": tasks.len(), + "task_list_id": store.task_list_id() })) } #[allow(clippy::needless_pass_by_value)] fn run_task_stop(input: TaskIdInput) -> Result { - let registry = global_task_registry(); - match registry.stop(&input.task_id) { - Ok(task) => to_pretty_json(json!({ + let store = global_runtime_task_store(); + match store.stop(&input.task_id) { + Ok(Some(task)) => to_pretty_json(json!({ "task_id": task.task_id, "status": task.status, - "message": "Task stopped" + "message": "Task stop requested" })), - Err(e) => Err(e), + Ok(None) => Err(format!("runtime task not found: {}", input.task_id)), + Err(error) => Err(error.to_string()), + } +} + +fn task_update_status(status: TaskUpdateStatus) -> Option { + match status { + TaskUpdateStatus::Pending => Some(TaskListStatus::Pending), + TaskUpdateStatus::InProgress => Some(TaskListStatus::InProgress), + TaskUpdateStatus::Completed => Some(TaskListStatus::Completed), + TaskUpdateStatus::Deleted => None, } } #[allow(clippy::needless_pass_by_value)] fn run_task_update(input: TaskUpdateInput) -> Result { - let registry = global_task_registry(); - match registry.update(&input.task_id, &input.message) { - Ok(task) => to_pretty_json(json!({ - "task_id": task.task_id, - "status": task.status, - "message_count": task.messages.len(), - "last_message": input.message - })), - Err(e) => Err(e), + let store = TaskListStore::current().map_err(|error| error.to_string())?; + if matches!(input.status, Some(TaskUpdateStatus::Deleted)) { + let deleted = store.delete(&input.task_id).map_err(|error| error.to_string())?; + return to_pretty_json(json!({ + "taskId": input.task_id, + "deleted": deleted + })); + } + let patch = TaskListPatch { + subject: input.subject, + description: input.description, + active_form: input.active_form, + status: input.status.and_then(task_update_status), + add_blocks: input.add_blocks, + add_blocked_by: input.add_blocked_by, + owner: input.owner, + metadata: input.metadata, + internal: None, + }; + match store.update(&input.task_id, patch) { + Ok(Some(task)) => to_pretty_json(task), + Ok(None) => Err(format!("task not found: {}", input.task_id)), + Err(error) => Err(error.to_string()), } } #[allow(clippy::needless_pass_by_value)] -fn run_task_output(input: TaskIdInput) -> Result { - let registry = global_task_registry(); - match registry.output(&input.task_id) { - Ok(output) => to_pretty_json(json!({ - "task_id": input.task_id, +fn run_task_output(input: TaskOutputInput) -> Result { + let store = global_runtime_task_store(); + match store.output(&input.task_id, input.block.unwrap_or(false), input.timeout_ms) { + Ok(Some(RuntimeTaskOutput { + task, + output, + has_output, + })) => to_pretty_json(json!({ + "task_id": task.task_id, + "status": task.status, "output": output, - "has_output": !output.is_empty() + "has_output": has_output, + "task": task, })), - Err(e) => Err(e), + Ok(None) => Err(format!("runtime task not found: {}", input.task_id)), + Err(error) => Err(error.to_string()), } } @@ -1426,36 +1362,113 @@ fn run_worker_terminate(input: WorkerIdInput) -> Result { #[allow(clippy::needless_pass_by_value)] fn run_team_create(input: TeamCreateInput) -> Result { - let task_ids: Vec = input - .tasks - .iter() - .filter_map(|t| t.get("task_id").and_then(|v| v.as_str()).map(str::to_owned)) - .collect(); - let team = global_team_registry().create(&input.name, task_ids); - // Register team assignment on each task - for task_id in &team.task_ids { - let _ = global_task_registry().assign_team(task_id, &team.team_id); - } + let team = global_team_store() + .create_team(&input.team_name, input.description, input.agent_type) + .map_err(|error| error.to_string())?; to_pretty_json(json!({ - "team_id": team.team_id, - "name": team.name, - "task_count": team.task_ids.len(), - "task_ids": team.task_ids, - "status": team.status, + "team_name": team.team_name, + "team_file_path": state_root() + .map_err(|error| error.to_string())? + .join("teams") + .join(sanitize_state_component(&team.team_name)) + .join("config.json") + .display() + .to_string(), + "lead_agent_id": team.lead_agent_id, "created_at": team.created_at })) } #[allow(clippy::needless_pass_by_value)] fn run_team_delete(input: TeamDeleteInput) -> Result { - match global_team_registry().delete(&input.team_id) { - Ok(team) => to_pretty_json(json!({ - "team_id": team.team_id, - "name": team.name, - "status": team.status, + match global_team_store() + .delete_team(&input.team_name) + .map_err(|error| error.to_string())? + { + Some(team) => to_pretty_json(json!({ + "team_name": team.team_name, + "deleted": true, "message": "Team deleted" })), - Err(e) => Err(e), + None => Err(format!("team not found: {}", input.team_name)), + } +} + +#[allow(clippy::needless_pass_by_value)] +fn run_send_message(input: SendMessageInput) -> Result { + let team = global_team_store() + .current_team() + .map_err(|error| error.to_string())? + .ok_or_else(|| String::from("no active team context"))?; + let sender = std::env::var("CLAW_TEAMMATE_NAME") + .ok() + .filter(|value| !value.trim().is_empty()) + .unwrap_or_else(|| "lead".to_string()); + let delivered = global_team_store() + .send_message( + &team.team_name, + &sender, + &input.to, + input.summary, + input.message, + ) + .map_err(|error| error.to_string())?; + to_pretty_json(json!({ + "team_name": team.team_name, + "sender": sender, + "delivered": delivered, + "count": delivered.len() + })) +} + +fn state_root() -> Result { + runtime::state_root().map_err(|error| error.to_string()) +} + +fn sanitize_state_component(value: &str) -> String { + runtime::sanitize_state_component(value) +} + +fn current_team_name() -> Option { + global_team_store() + .current_team() + .ok() + .flatten() + .map(|team| team.team_name) +} + +fn current_task_list_id_string() -> String { + current_task_list_id().unwrap_or_else(|_| "default".to_string()) +} + +fn runtime_task_team_name() -> Option { + current_team_name().or_else(|| { + let task_list_id = current_task_list_id_string(); + if task_list_id == runtime::default_session_identity() { + None + } else { + Some(task_list_id) + } + }) +} + +fn register_agent_member(task: &RuntimeTaskRecord) { + if let (Some(team_name), Some(agent_id), Some(agent_name)) = ( + task.team_name.clone(), + task.agent_id.clone(), + task.agent_name.clone(), + ) { + let _ = global_team_store().upsert_member( + &team_name, + TeamMemberRecord { + agent_id, + name: agent_name, + agent_type: None, + model: None, + status: Some(task.status.to_string()), + joined_at: task.created_at, + }, + ); } } @@ -1699,6 +1712,55 @@ fn run_bash(input: BashCommandInput) -> Result { if let Some(output) = workspace_test_branch_preflight(&input.command) { return serde_json::to_string_pretty(&output).map_err(|error| error.to_string()); } + if input.run_in_background.unwrap_or(false) { + let description = input + .description + .clone() + .filter(|value| !value.trim().is_empty()) + .unwrap_or_else(|| input.command.clone()); + let task_seed = format!( + "shell-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ); + let (output_path, exit_code_path) = background_output_paths(&task_seed) + .map_err(|error: std::io::Error| error.to_string())?; + let handle = spawn_background_bash(&input, &output_path, &exit_code_path) + .map_err(|error| error.to_string())?; + let runtime_task = global_runtime_task_store() + .create_shell_task( + description, + input.command.clone(), + handle.pid, + output_path.to_path_buf(), + exit_code_path.to_path_buf(), + runtime_task_team_name(), + ) + .map_err(|error| error.to_string())?; + let output = BashCommandOutput { + stdout: String::new(), + stderr: String::new(), + raw_output_path: None, + interrupted: false, + is_image: None, + background_task_id: Some(runtime_task.task_id.clone()), + backgrounded_by_user: Some(true), + assistant_auto_backgrounded: Some(false), + dangerously_disable_sandbox: input.dangerously_disable_sandbox, + return_code_interpretation: None, + no_output_expected: Some(false), + structured_content: Some(vec![json!({ + "task_id": runtime_task.task_id, + "output_file": output_path.display().to_string(), + })]), + persisted_output_path: Some(output_path.display().to_string()), + persisted_output_size: Some(0), + sandbox_status: Some(handle.sandbox_status), + }; + return serde_json::to_string_pretty(&output).map_err(|error| error.to_string()); + } serde_json::to_string_pretty(&execute_bash(input).map_err(|error| error.to_string())?) .map_err(|error| error.to_string()) } @@ -1824,27 +1886,25 @@ fn branch_divergence_output( dangerously_disable_sandbox: None, return_code_interpretation: Some("preflight_blocked:branch_divergence".to_string()), no_output_expected: Some(false), - structured_content: Some(vec![ - serde_json::to_value( - LaneEvent::new( - LaneEventName::BranchStaleAgainstMain, - LaneEventStatus::Blocked, - iso8601_now(), - ) - .with_failure_class(LaneFailureClass::BranchDivergence) - .with_detail(stderr.clone()) - .with_data(json!({ - "branch": branch, - "mainRef": main_ref, - "commitsBehind": commits_behind, - "commitsAhead": commits_ahead, - "missingCommits": missing_fixes, - "blockedCommand": command, - "recommendedAction": format!("merge or rebase {main_ref} before workspace tests") - })), + structured_content: Some(vec![serde_json::to_value( + LaneEvent::new( + LaneEventName::BranchStaleAgainstMain, + LaneEventStatus::Blocked, + iso8601_now(), ) - .expect("lane event should serialize"), - ]), + .with_failure_class(LaneFailureClass::BranchDivergence) + .with_detail(stderr.clone()) + .with_data(json!({ + "branch": branch, + "mainRef": main_ref, + "commitsBehind": commits_behind, + "commitsAhead": commits_ahead, + "missingCommits": missing_fixes, + "blockedCommand": command, + "recommendedAction": format!("merge or rebase {main_ref} before workspace tests") + })), + ) + .expect("lane event should serialize")]), persisted_output_path: None, persisted_output_size: None, sandbox_status: None, @@ -2029,6 +2089,10 @@ struct AgentInput { subagent_type: Option, name: Option, model: Option, + #[serde(default)] + run_in_background: Option, + #[serde(default)] + team_name: Option, } #[derive(Debug, Deserialize)] @@ -2130,9 +2194,12 @@ struct AskUserQuestionInput { #[derive(Debug, Deserialize)] struct TaskCreateInput { - prompt: String, + subject: String, + description: String, + #[serde(rename = "activeForm", default)] + active_form: Option, #[serde(default)] - description: Option, + metadata: Option>, } #[derive(Debug, Deserialize)] @@ -2142,8 +2209,33 @@ struct TaskIdInput { #[derive(Debug, Deserialize)] struct TaskUpdateInput { + #[serde(rename = "taskId")] task_id: String, - message: String, + #[serde(default)] + subject: Option, + #[serde(default)] + description: Option, + #[serde(rename = "activeForm", default)] + active_form: Option, + #[serde(default)] + status: Option, + #[serde(rename = "addBlocks", default)] + add_blocks: Vec, + #[serde(rename = "addBlockedBy", default)] + add_blocked_by: Vec, + #[serde(default)] + owner: Option, + #[serde(default)] + metadata: Option>, +} + +#[derive(Debug, Deserialize, Clone, Copy, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +enum TaskUpdateStatus { + Pending, + InProgress, + Completed, + Deleted, } #[derive(Debug, Deserialize)] @@ -2179,13 +2271,39 @@ const fn default_auto_recover_prompt_misdelivery() -> bool { #[derive(Debug, Deserialize)] struct TeamCreateInput { - name: String, - tasks: Vec, + team_name: String, + #[serde(default)] + description: Option, + #[serde(default)] + agent_type: Option, } #[derive(Debug, Deserialize)] struct TeamDeleteInput { - team_id: String, + team_name: String, +} + +#[derive(Debug, Deserialize)] +struct SendMessageInput { + to: String, + #[serde(default)] + summary: Option, + message: Value, +} + +#[derive(Debug, Deserialize)] +struct TaskGetInput { + #[serde(rename = "taskId")] + task_id: String, +} + +#[derive(Debug, Deserialize)] +struct TaskOutputInput { + task_id: String, + #[serde(default)] + block: Option, + #[serde(default)] + timeout_ms: Option, } #[derive(Debug, Deserialize)] @@ -2294,11 +2412,17 @@ struct SkillOutput { struct AgentOutput { #[serde(rename = "agentId")] agent_id: String, + #[serde(rename = "taskId")] + task_id: String, name: String, description: String, #[serde(rename = "subagentType")] subagent_type: Option, model: Option, + #[serde(rename = "teamName", skip_serializing_if = "Option::is_none")] + team_name: Option, + #[serde(rename = "isAsync")] + is_async: bool, status: String, #[serde(rename = "outputFile")] output_file: String, @@ -2988,6 +3112,11 @@ where let manifest_file = output_dir.join(format!("{agent_id}.json")); let normalized_subagent_type = normalize_subagent_type(input.subagent_type.as_deref()); let model = resolve_agent_model(input.model.as_deref()); + let team_name = input + .team_name + .clone() + .filter(|value| !value.trim().is_empty()) + .or_else(current_team_name); let agent_name = input .name .as_deref() @@ -2997,6 +3126,7 @@ where let created_at = iso8601_now(); let system_prompt = build_agent_system_prompt(&normalized_subagent_type)?; let allowed_tools = allowed_tools_for_subagent(&normalized_subagent_type); + let run_in_background = input.run_in_background.unwrap_or(false); let output_contents = format!( "# Agent Task @@ -3016,11 +3146,14 @@ where std::fs::write(&output_file, output_contents).map_err(|error| error.to_string())?; let manifest = AgentOutput { + task_id: agent_id.clone(), agent_id, name: agent_name, description: input.description, subagent_type: Some(normalized_subagent_type), model: Some(model), + team_name: team_name.clone(), + is_async: run_in_background, status: String::from("running"), output_file: output_file.display().to_string(), manifest_file: manifest_file.display().to_string(), @@ -3032,6 +3165,17 @@ where error: None, }; write_agent_manifest(&manifest)?; + let runtime_task = global_runtime_task_store() + .create_agent_task( + manifest.agent_id.clone(), + manifest.name.clone(), + manifest.description.clone(), + input.prompt.clone(), + manifest.output_file.clone(), + team_name.clone(), + ) + .map_err(|error| error.to_string())?; + register_agent_member(&runtime_task); let manifest_for_spawn = manifest.clone(); let job = AgentJob { @@ -3040,13 +3184,38 @@ where system_prompt, allowed_tools, }; - if let Err(error) = spawn_fn(job) { - let error = format!("failed to spawn sub-agent: {error}"); - persist_agent_terminal_state(&manifest, "failed", None, Some(error.clone()))?; - return Err(error); + if run_in_background { + if let Err(error) = spawn_fn(job) { + let error = format!("failed to spawn sub-agent: {error}"); + persist_agent_terminal_state(&manifest, "failed", None, Some(error.clone()))?; + let _ = global_runtime_task_store().mark_agent_terminal( + &manifest.task_id, + runtime::RuntimeTaskStatus::Failed, + None, + Some(error.clone()), + ); + return Err(error); + } + return Ok(manifest); + } + match run_agent_job(&job) { + Ok(()) => { + let updated = std::fs::read_to_string(&manifest.manifest_file) + .ok() + .and_then(|contents| serde_json::from_str::(&contents).ok()) + .unwrap_or(manifest); + Ok(updated) + } + Err(error) => { + let _ = global_runtime_task_store().mark_agent_terminal( + &manifest.task_id, + runtime::RuntimeTaskStatus::Failed, + None, + Some(error.clone()), + ); + Err(error) + } } - - Ok(manifest) } fn spawn_agent_job(job: AgentJob) -> Result<(), String> { @@ -3241,14 +3410,14 @@ fn persist_agent_terminal_state( next_manifest.status = status.to_string(); next_manifest.completed_at = Some(iso8601_now()); next_manifest.current_blocker = blocker.clone(); - next_manifest.error = error; + next_manifest.error = error.clone(); if let Some(blocker) = blocker { - next_manifest.lane_events.push( - LaneEvent::blocked(iso8601_now(), &blocker), - ); - next_manifest.lane_events.push( - LaneEvent::failed(iso8601_now(), &blocker), - ); + next_manifest + .lane_events + .push(LaneEvent::blocked(iso8601_now(), &blocker)); + next_manifest + .lane_events + .push(LaneEvent::failed(iso8601_now(), &blocker)); } else { next_manifest.current_blocker = None; let compressed_detail = result @@ -3258,7 +3427,21 @@ fn persist_agent_terminal_state( .lane_events .push(LaneEvent::finished(iso8601_now(), compressed_detail)); } - write_agent_manifest(&next_manifest) + write_agent_manifest(&next_manifest)?; + let runtime_status = if error.is_some() || status == "failed" { + runtime::RuntimeTaskStatus::Failed + } else if status == "stopped" { + runtime::RuntimeTaskStatus::Stopped + } else { + runtime::RuntimeTaskStatus::Completed + }; + let _ = global_runtime_task_store().mark_agent_terminal( + &manifest.task_id, + runtime_status, + result.map(str::to_owned), + next_manifest.error.clone(), + ); + Ok(()) } fn append_agent_output(path: &str, suffix: &str) -> Result<(), String> { @@ -4980,10 +5163,10 @@ mod tests { assert!(names.contains(&"StructuredOutput")); assert!(names.contains(&"REPL")); assert!(names.contains(&"PowerShell")); - assert!(names.contains(&"WorkerCreate")); - assert!(names.contains(&"WorkerObserve")); - assert!(names.contains(&"WorkerAwaitReady")); - assert!(names.contains(&"WorkerSendPrompt")); + assert!(names.contains(&"TaskCreate")); + assert!(names.contains(&"TeamCreate")); + assert!(names.contains(&"SendMessage")); + assert!(!names.contains(&"WorkerCreate")); } #[test] @@ -5576,7 +5759,9 @@ mod tests { #[test] fn skill_loads_local_skill_prompt() { - let _guard = env_lock().lock().expect("env lock should acquire"); + let _guard = env_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); let home = temp_path("skills-home"); let skill_dir = home.join(".agents").join("skills").join("help"); fs::create_dir_all(&skill_dir).expect("skill dir should exist"); @@ -5681,6 +5866,8 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("ship-audit".to_string()), model: None, + run_in_background: Some(true), + team_name: None, }, move |job| { *captured_for_spawn @@ -5724,7 +5911,8 @@ mod tests { &json!({ "description": "Verify the branch", "prompt": "Check tests.", - "subagent_type": "explorer" + "subagent_type": "explorer", + "run_in_background": true }), ) .expect("Agent should normalize built-in aliases"); @@ -5737,7 +5925,8 @@ mod tests { &json!({ "description": "Review the branch", "prompt": "Inspect diff.", - "name": "Ship Audit!!!" + "name": "Ship Audit!!!", + "run_in_background": true }), ) .expect("Agent should normalize explicit names"); @@ -5761,6 +5950,8 @@ mod tests { subagent_type: Some("Explore".to_string()), name: Some("complete-task".to_string()), model: Some("claude-sonnet-4-6".to_string()), + run_in_background: Some(true), + team_name: None, }, |job| { persist_agent_terminal_state( @@ -5798,6 +5989,8 @@ mod tests { subagent_type: Some("Verification".to_string()), name: Some("fail-task".to_string()), model: None, + run_in_background: Some(true), + team_name: None, }, |job| { persist_agent_terminal_state( @@ -5844,6 +6037,8 @@ mod tests { subagent_type: None, name: Some("spawn-error".to_string()), model: None, + run_in_background: Some(true), + team_name: None, }, |_| Err(String::from("thread creation failed")), ) @@ -5904,7 +6099,10 @@ mod tests { "gateway routing rejected the request", LaneFailureClass::GatewayRouting, ), - ("tool failed: denied tool execution from hook", LaneFailureClass::ToolRuntime), + ( + "tool failed: denied tool execution from hook", + LaneFailureClass::ToolRuntime, + ), ("thread creation failed", LaneFailureClass::Infra), ]; @@ -5927,11 +6125,17 @@ mod tests { (LaneEventName::MergeReady, "lane.merge.ready"), (LaneEventName::Finished, "lane.finished"), (LaneEventName::Failed, "lane.failed"), - (LaneEventName::BranchStaleAgainstMain, "branch.stale_against_main"), + ( + LaneEventName::BranchStaleAgainstMain, + "branch.stale_against_main", + ), ]; for (event, expected) in cases { - assert_eq!(serde_json::to_value(event).expect("serialize lane event"), json!(expected)); + assert_eq!( + serde_json::to_value(event).expect("serialize lane event"), + json!(expected) + ); } } @@ -6213,7 +6417,9 @@ mod tests { .expect("bash background should succeed"); let background_output: serde_json::Value = serde_json::from_str(&background).expect("json"); assert!(background_output["backgroundTaskId"].as_str().is_some()); - assert_eq!(background_output["noOutputExpected"], true); + assert_eq!(background_output["backgroundedByUser"], true); + assert_eq!(background_output["noOutputExpected"], false); + assert!(background_output["persistedOutputPath"].as_str().is_some()); } #[test]