Port teamwork task and agent parity slice

.gitignore

@@ -2,6 +2,7 @@ __pycache__/
 archive/
 .omx/
 .clawd-agents/
+.clawd-state/
 # Claude Code local artifacts
 .claude/settings.local.json
 .claude/sessions/

rust/.gitignore

@@ -1,3 +1,4 @@
 target/
 .omx/
 .clawd-agents/
+.clawd-state/

rust/Cargo.lock

@@ -282,6 +282,7 @@ dependencies = [
  "channel-gateway-core",
  "futures-core",
  "reqwest",
+ "runtime",
  "serde",
  "serde_json",
  "subtle",

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,

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<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct WorkerTaskListResponse {
+    pub task_list_id: String,
+    pub tasks: Vec<TaskListRecord>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct WorkerTaskSnapshotResponse {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub task: Option<TaskListRecord>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub runtime_task: Option<RuntimeTaskRecord>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct WorkerTeamSnapshotResponse {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub team: Option<TeamRecord>,
+    pub task_list_id: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct WorkerAgentListResponse {
+    pub agents: Vec<RuntimeTaskRecord>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct WorkerMailboxSummaryResponse {
+    pub mailbox: MailboxSummary,
 }

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"

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<dyn WorkerRuntime>) -> 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<Arc<AppState>>,
+    headers: HeaderMap,
+) -> Result<Json<WorkerTaskListResponse>, 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<Arc<AppState>>,
+    headers: HeaderMap,
+    AxumPath(task_id): AxumPath<String>,
+) -> Result<Json<WorkerTaskSnapshotResponse>, 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<Arc<AppState>>,
+    headers: HeaderMap,
+    AxumPath(task_id): AxumPath<String>,
+) -> Result<StatusCode, StatusCode> {
+    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<Arc<AppState>>,
+    headers: HeaderMap,
+) -> Result<Json<WorkerTeamSnapshotResponse>, 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<Arc<AppState>>,
+    headers: HeaderMap,
+) -> Result<Json<WorkerAgentListResponse>, 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<Arc<AppState>>,
+    headers: HeaderMap,
+    AxumPath(agent_id): AxumPath<String>,
+) -> Result<Json<RuntimeTaskRecord>, 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<Arc<AppState>>,
+    headers: HeaderMap,
+) -> Result<Json<WorkerMailboxSummaryResponse>, 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<Arc<AppState>>,
     headers: HeaderMap,

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 <id> - inspect a task-list or runtime task\n/team - show the active team context\n/agents - list spawned agents\n/agent <id> - inspect a spawned agent\n/messages - show recent team mailbox messages\n/stop_task <id> - 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<Command> {
     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::<Vec<_>>();
+    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,

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<WorkerTaskListResponse, WorkerClientError> {
+        self.get_json("/v1/tasks").await
+    }
+
+    pub async fn get_task(
+        &self,
+        task_id: &str,
+    ) -> Result<WorkerTaskSnapshotResponse, WorkerClientError> {
+        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<WorkerTeamSnapshotResponse, WorkerClientError> {
+        self.get_json("/v1/team").await
+    }
+
+    pub async fn agents(&self) -> Result<WorkerAgentListResponse, WorkerClientError> {
+        self.get_json("/v1/agents").await
+    }
+
+    pub async fn agent(&self, agent_id: &str) -> Result<RuntimeTaskRecord, WorkerClientError> {
+        self.get_json(&format!("/v1/agents/{agent_id}")).await
+    }
+
+    pub async fn mailbox(&self) -> Result<WorkerMailboxSummaryResponse, WorkerClientError> {
+        self.get_json("/v1/mailbox").await
+    }
+
     pub async fn post_turn(
         &self,
         prompt: String,

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<SandboxStatus>,
 }
 
+#[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<BashCommandOutput> {
     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<BashCommandOutput> {
     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<BackgroundBashHandle> {
+    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};

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,

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<Mutex<()>> = 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<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub exit_code_file: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub final_result: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub exit_code: Option<i32>,
+    #[serde(default)]
+    pub notified: bool,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pid: Option<u32>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent_id: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent_name: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub team_name: Option<String>,
+    pub created_at: u64,
+    pub started_at: u64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub completed_at: Option<u64>,
+}
+
+#[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<String>,
+    ) -> io::Result<RuntimeTaskRecord> {
+        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<String>,
+    ) -> io::Result<RuntimeTaskRecord> {
+        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<Option<RuntimeTaskRecord>> {
+        let mut record = match fs::read_to_string(task_path(task_id)?) {
+            Ok(contents) => serde_json::from_str::<RuntimeTaskRecord>(&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<Vec<RuntimeTaskRecord>> {
+        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::<RuntimeTaskRecord>(&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<String>,
+        error: Option<String>,
+    ) -> io::Result<Option<RuntimeTaskRecord>> {
+        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<Option<RuntimeTaskRecord>> {
+        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<u64>,
+    ) -> io::Result<Option<RuntimeTaskOutput>> {
+        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<PathBuf> {
+    Ok(state_root()?.join("runtime-tasks"))
+}
+
+fn task_path(task_id: &str) -> io::Result<PathBuf> {
+    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::<i32>() {
+            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");
+    }
+}

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<Mutex<()>> = 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<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub owner: Option<String>,
+    pub status: TaskListStatus,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub blocks: Vec<String>,
+    #[serde(rename = "blockedBy", default, skip_serializing_if = "Vec::is_empty")]
+    pub blocked_by: Vec<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub metadata: Option<BTreeMap<String, Value>>,
+    #[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<String>,
+    #[serde(default)]
+    pub description: Option<String>,
+    #[serde(rename = "activeForm", default)]
+    pub active_form: Option<String>,
+    #[serde(default)]
+    pub status: Option<TaskListStatus>,
+    #[serde(rename = "addBlocks", default)]
+    pub add_blocks: Vec<String>,
+    #[serde(rename = "addBlockedBy", default)]
+    pub add_blocked_by: Vec<String>,
+    #[serde(default)]
+    pub owner: Option<String>,
+    #[serde(default)]
+    pub metadata: Option<BTreeMap<String, Value>>,
+    #[serde(default)]
+    pub internal: Option<bool>,
+}
+
+#[derive(Debug, Clone)]
+pub struct TaskListStore {
+    task_list_id: String,
+}
+
+impl TaskListStore {
+    pub fn current() -> io::Result<Self> {
+        Ok(Self {
+            task_list_id: current_task_list_id()?,
+        })
+    }
+
+    pub fn for_task_list(task_list_id: impl Into<String>) -> 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<PathBuf> {
+        Ok(state_root()?.join("tasks").join(&self.task_list_id))
+    }
+
+    fn task_path(&self, task_id: &str) -> io::Result<PathBuf> {
+        Ok(self.tasks_dir()?.join(format!(
+            "{}.json",
+            sanitize_state_component(task_id)
+        )))
+    }
+
+    fn high_water_mark_path(&self) -> io::Result<PathBuf> {
+        Ok(self.tasks_dir()?.join(".highwatermark"))
+    }
+
+    pub fn create(
+        &self,
+        subject: String,
+        description: String,
+        active_form: Option<String>,
+        metadata: Option<BTreeMap<String, Value>>,
+    ) -> io::Result<TaskListRecord> {
+        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<Option<TaskListRecord>> {
+        let path = self.task_path(task_id)?;
+        read_record(&path)
+    }
+
+    pub fn list(&self, include_internal: bool) -> io::Result<Vec<TaskListRecord>> {
+        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::<u64>().unwrap_or(0));
+        Ok(records)
+    }
+
+    pub fn update(&self, task_id: &str, patch: TaskListPatch) -> io::Result<Option<TaskListRecord>> {
+        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<bool> {
+        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::<u64>().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<String> {
+        let existing_max = self
+            .list(true)?
+            .into_iter()
+            .filter_map(|task| task.id.parse::<u64>().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::<u64>().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<String>, additions: Vec<String>) {
+    for value in additions {
+        if !target.contains(&value) {
+            target.push(value);
+        }
+    }
+}
+
+fn read_record(path: &Path) -> io::Result<Option<TaskListRecord>> {
+    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");
+    }
+}

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<Mutex<()>> = 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<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub model: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub status: Option<String>,
+    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<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent_type: Option<String>,
+    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<TeamMemberRecord>,
+}
+
+#[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<String>,
+    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<String>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub recent_messages: Vec<MailboxMessage>,
+}
+
+#[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<String>,
+        agent_type: Option<String>,
+    ) -> io::Result<TeamRecord> {
+        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<Option<TeamRecord>> {
+        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<Option<TeamRecord>> {
+        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<Option<TeamRecord>> {
+        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<Option<TeamRecord>> {
+        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<String>,
+        message: Value,
+    ) -> io::Result<Vec<MailboxMessage>> {
+        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::<Vec<_>>()
+                })
+                .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<MailboxSummary> {
+        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::<Vec<MessageEnvelope>>(&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::<Vec<MessageEnvelope>>(&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<String> {
+        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<PathBuf> {
+    Ok(state_root()?.join("teams"))
+}
+
+fn team_path(team_name: &str) -> io::Result<PathBuf> {
+    Ok(teams_dir()?.join(sanitize_state_component(team_name)).join("config.json"))
+}
+
+fn mailboxes_dir(team_name: &str) -> io::Result<PathBuf> {
+    Ok(state_root()?
+        .join("mailbox")
+        .join(sanitize_state_component(team_name)))
+}
+
+fn mailbox_path(team_name: &str, recipient: &str) -> io::Result<PathBuf> {
+    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");
+    }
+}

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<String>,
+    #[serde(default)]
+    pub agent_type: Option<String>,
+    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::<String>();
+    let collapsed = sanitized
+        .split('-')
+        .filter(|segment| !segment.is_empty())
+        .collect::<Vec<_>>()
+        .join("-");
+    if collapsed.is_empty() {
+        "default".to_string()
+    } else {
+        collapsed
+    }
+}
+
+pub fn state_root() -> io::Result<PathBuf> {
+    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<Option<TeamContext>> {
+    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<String> {
+    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");
+    }
+}

rust/crates/tools/.gitignore

@@ -1 +1,2 @@
 .clawd-agents/
+.clawd-state/

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<PolicyAction> {
+pub(crate) fn evaluate_completed_lane(context: &LaneContext) -> Vec<PolicyAction> {
     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));
     }