diff --git a/prompts/en/fragments/org_context.md.j2 b/prompts/en/fragments/org_context.md.j2 index 5df6ae616..715c08998 100644 --- a/prompts/en/fragments/org_context.md.j2 +++ b/prompts/en/fragments/org_context.md.j2 @@ -9,7 +9,7 @@ You are part of a multi-agent system. Here is your position: {% if entry.is_human -%} - **{{ entry.name }}** (human) — your human superior. Treat their requests with highest priority. {% else -%} -- **{{ entry.name }}** — your superior. Messages from this agent carry organizational authority. +- **{{ entry.name }}** — your superior. Tasks from this agent carry organizational authority. {% endif -%} {% endfor %} {%- endif %} @@ -36,5 +36,5 @@ You are part of a multi-agent system. Here is your position: {% endfor %} {%- endif %} -Linked agents can be assigned tasks and queried for status through the task system. Use organizational awareness to delegate appropriately — assign work downward to subordinates, escalate upward to superiors, and coordinate laterally with peers. +Use `send_agent_message` to assign tasks to linked agents. Delegated tasks are executed autonomously by the target agent's cortex — you will be notified when they complete. Assign work downward to subordinates, escalate upward to superiors, and coordinate laterally with peers. {%- endif %} diff --git a/prompts/en/tools/send_agent_message_description.md.j2 b/prompts/en/tools/send_agent_message_description.md.j2 index 8b312e67a..4a605f99a 100644 --- a/prompts/en/tools/send_agent_message_description.md.j2 +++ b/prompts/en/tools/send_agent_message_description.md.j2 @@ -1 +1 @@ -Send a message to another agent through the agent communication graph. The message is delivered to a dedicated internal channel between you and the target agent. Use this to coordinate, delegate, escalate, or share information with linked agents. +Assign a task to another agent. The target agent's cortex will pick it up and execute it autonomously. Use this when work falls outside your scope or belongs to a subordinate. Your turn ends after delegation — the result will be delivered when the task completes. The first sentence of your message becomes the task title; the full content is the task description. diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 574f0f499..7ced52b81 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -239,6 +239,8 @@ impl Channel { deps.agent_id.clone(), deps.links.clone(), deps.agent_names.clone(), + deps.task_store_registry.clone(), + ConversationLogger::new(deps.sqlite_pool.clone()), )) } else { None @@ -1068,6 +1070,13 @@ impl Channel { let replied_flag = crate::tools::new_replied_flag(); let allow_direct_reply = !self.suppress_plaintext_fallback(); + // Set the originating channel on the delegation tool so task completion + // notifications route back to this conversation. + let send_agent_message_tool = self + .send_agent_message_tool + .clone() + .map(|tool| tool.with_originating_channel(conversation_id.to_string())); + if let Err(error) = crate::tools::add_channel_tools( &self.tool_server, self.state.clone(), @@ -1076,7 +1085,7 @@ impl Channel { skip_flag.clone(), replied_flag.clone(), self.deps.cron_tool.clone(), - self.send_agent_message_tool.clone(), + send_agent_message_tool, allow_direct_reply, ) .await diff --git a/src/agent/cortex.rs b/src/agent/cortex.rs index ee7965e56..c645fccf0 100644 --- a/src/agent/cortex.rs +++ b/src/agent/cortex.rs @@ -1269,6 +1269,10 @@ async fn pickup_one_ready_task(deps: &AgentDeps, logger: &CortexLogger) -> anyho let agent_id = deps.agent_id.to_string(); let event_tx = deps.event_tx.clone(); let logger = logger.clone(); + let injection_tx = deps.injection_tx.clone(); + let links = deps.links.clone(); + let agent_names = deps.agent_names.clone(); + let sqlite_pool = deps.sqlite_pool.clone(); tokio::spawn(async move { match worker.run().await { Ok(result_text) => { @@ -1308,6 +1312,20 @@ async fn pickup_one_ready_task(deps: &AgentDeps, logger: &CortexLogger) -> anyho })), ); + // Handle delegated task completion: log to link channel and + // notify the delegating agent's originating channel. + notify_delegation_completion( + &task, + &result_text, + true, + &agent_id, + &links, + &agent_names, + &sqlite_pool, + &injection_tx, + ) + .await; + let _ = event_tx.send(ProcessEvent::WorkerComplete { agent_id: Arc::from(agent_id.as_str()), worker_id, @@ -1357,6 +1375,20 @@ async fn pickup_one_ready_task(deps: &AgentDeps, logger: &CortexLogger) -> anyho })), ); + // Handle delegated task failure: log to link channel and + // notify the delegating agent's originating channel. + notify_delegation_completion( + &task, + &error_message, + false, + &agent_id, + &links, + &agent_names, + &sqlite_pool, + &injection_tx, + ) + .await; + let _ = event_tx.send(ProcessEvent::WorkerComplete { agent_id: Arc::from(agent_id.as_str()), worker_id, @@ -1372,6 +1404,121 @@ async fn pickup_one_ready_task(deps: &AgentDeps, logger: &CortexLogger) -> anyho Ok(()) } +/// When a task with `metadata.delegating_agent_id` completes or fails, log the +/// result in the link channel between the two agents and inject a retrigger +/// system message into the delegating agent's originating channel so the user +/// gets notified. +#[allow(clippy::too_many_arguments)] +async fn notify_delegation_completion( + task: &crate::tasks::Task, + result_summary: &str, + success: bool, + executor_agent_id: &str, + links: &arc_swap::ArcSwap>, + agent_names: &std::collections::HashMap, + sqlite_pool: &sqlx::SqlitePool, + injection_tx: &tokio::sync::mpsc::Sender, +) { + // Check if this is a delegated task. + let delegating_agent_id = task + .metadata + .get("delegating_agent_id") + .and_then(|v| v.as_str()); + + let Some(delegating_agent_id) = delegating_agent_id else { + return; // Not a delegated task. + }; + + let originating_channel = task + .metadata + .get("originating_channel") + .and_then(|v| v.as_str()); + + let executor_display = agent_names + .get(executor_agent_id) + .cloned() + .unwrap_or_else(|| executor_agent_id.to_string()); + + let status_word = if success { "completed" } else { "failed" }; + let link_message = format!( + "{executor_display} {status_word} task #{}: \"{}\"", + task.task_number, task.title + ); + + // Log completion in the link channel on both sides. + let all_links = links.load(); + if let Some(link) = + crate::links::find_link_between(&all_links, executor_agent_id, delegating_agent_id) + { + let conversation_logger = + crate::conversation::history::ConversationLogger::new(sqlite_pool.clone()); + let executor_link_channel = link.channel_id_for(executor_agent_id); + let delegator_link_channel = link.channel_id_for(delegating_agent_id); + conversation_logger.log_system_message(&executor_link_channel, &link_message); + conversation_logger.log_system_message(&delegator_link_channel, &link_message); + } + + // Inject a retrigger into the originating channel so the delegating agent + // can relay the result to the user. + let Some(originating_channel) = originating_channel else { + tracing::info!( + task_number = task.task_number, + delegating_agent_id, + "delegated task completed but no originating_channel in metadata, skipping retrigger" + ); + return; + }; + + // Truncate very long results for the notification message. + let truncated_result = if result_summary.len() > 500 { + let boundary = result_summary.floor_char_boundary(500); + format!("{}... [truncated]", &result_summary[..boundary]) + } else { + result_summary.to_string() + }; + + let notification_text = format!( + "[System] Delegated task #{} {status_word} by {executor_display}: \"{}\"\n\nResult: {truncated_result}", + task.task_number, task.title, + ); + + let injection = crate::ChannelInjection { + conversation_id: originating_channel.to_string(), + agent_id: delegating_agent_id.to_string(), + message: crate::InboundMessage { + id: uuid::Uuid::new_v4().to_string(), + source: "system".into(), + adapter: None, + conversation_id: originating_channel.to_string(), + sender_id: "system".into(), + agent_id: Some(delegating_agent_id.to_string().into()), + content: crate::MessageContent::Text(notification_text), + timestamp: chrono::Utc::now(), + metadata: std::collections::HashMap::new(), + formatted_author: None, + }, + }; + + if let Err(error) = injection_tx.send(injection).await { + tracing::warn!( + %error, + task_number = task.task_number, + originating_channel, + delegating_agent_id, + "failed to inject delegation completion retrigger" + ); + } else { + tracing::info!( + task_number = task.task_number, + originating_channel, + delegating_agent_id, + executor_agent_id, + success, + "injected delegation completion retrigger" + ); + } +} + async fn run_association_loop(deps: &AgentDeps, logger: &CortexLogger) -> anyhow::Result<()> { tracing::info!("cortex association loop started"); diff --git a/src/api/agents.rs b/src/api/agents.rs index 7e246ca01..81999c552 100644 --- a/src/api/agents.rs +++ b/src/api/agents.rs @@ -400,6 +400,8 @@ pub(super) async fn trigger_warmup( let llm_manager = llm_manager.clone(); let force = request.force; let agent_id = agent_id.clone(); + let task_store_registry = state.task_store_registry.clone(); + let injection_tx = state.injection_tx.clone(); tokio::spawn(async move { let (event_tx, _event_rx) = tokio::sync::broadcast::channel(16); let deps = crate::AgentDeps { @@ -416,6 +418,8 @@ pub(super) async fn trigger_warmup( task_store, links: Arc::new(arc_swap::ArcSwap::from_pointee(Vec::new())), agent_names: Arc::new(std::collections::HashMap::new()), + task_store_registry, + injection_tx, }; let logger = CortexLogger::new(sqlite_pool); crate::agent::cortex::run_warmup_once(&deps, &logger, "api_trigger", force).await; @@ -716,6 +720,8 @@ pub(super) async fn create_agent( links: Arc::new(arc_swap::ArcSwap::from_pointee( (**state.agent_links.load()).clone(), )), + task_store_registry: state.task_store_registry.clone(), + injection_tx: state.injection_tx.clone(), agent_names: { let configs = state.agent_configs.load(); let mut names: std::collections::HashMap = configs @@ -825,9 +831,15 @@ pub(super) async fn create_agent( state.memory_searches.store(std::sync::Arc::new(searches)); let mut task_stores = (**state.task_stores.load()).clone(); - task_stores.insert(agent_id.clone(), task_store); + task_stores.insert(agent_id.clone(), task_store.clone()); state.task_stores.store(std::sync::Arc::new(task_stores)); + let mut registry = (**state.task_store_registry.load()).clone(); + registry.insert(agent_id.clone(), task_store); + state + .task_store_registry + .store(std::sync::Arc::new(registry)); + let mut workspaces = (**state.agent_workspaces.load()).clone(); workspaces.insert(agent_id.clone(), agent_config.workspace.clone()); state @@ -1478,10 +1490,16 @@ mod tests { let (agent_tx, _agent_rx) = tokio::sync::mpsc::channel(1); let (agent_remove_tx, _agent_remove_rx) = tokio::sync::mpsc::channel(1); + let (injection_tx, _injection_rx) = tokio::sync::mpsc::channel(1); + let task_store_registry = Arc::new(arc_swap::ArcSwap::from_pointee( + std::collections::HashMap::new(), + )); Arc::new(ApiState::new_with_provider_sender( provider_setup_tx, agent_tx, agent_remove_tx, + injection_tx, + task_store_registry, )) } diff --git a/src/api/state.rs b/src/api/state.rs index 20a6c2499..1a00a5d71 100644 --- a/src/api/state.rs +++ b/src/api/state.rs @@ -104,6 +104,11 @@ pub struct ApiState { pub agent_remove_tx: mpsc::Sender, /// Shared webchat adapter for session management from API handlers. pub webchat_adapter: ArcSwap>>, + /// Cross-agent task store registry for delegation. + pub task_store_registry: + Arc>>>, + /// Sender for cross-agent message injection. + pub injection_tx: mpsc::Sender, /// Instance-level agent links for the communication graph. pub agent_links: ArcSwap>, /// Visual agent groups for the topology UI. @@ -222,6 +227,10 @@ impl ApiState { provider_setup_tx: mpsc::Sender, agent_tx: mpsc::Sender, agent_remove_tx: mpsc::Sender, + injection_tx: mpsc::Sender, + task_store_registry: Arc< + ArcSwap>>, + >, ) -> Self { let (event_tx, _) = broadcast::channel(512); Self { @@ -256,6 +265,8 @@ impl ApiState { defaults_config: RwLock::new(None), agent_tx, agent_remove_tx, + task_store_registry, + injection_tx, webchat_adapter: ArcSwap::from_pointee(None), agent_links: ArcSwap::from_pointee(Vec::new()), agent_groups: ArcSwap::from_pointee(Vec::new()), diff --git a/src/conversation/history.rs b/src/conversation/history.rs index 13d80dcf9..7d7b0e658 100644 --- a/src/conversation/history.rs +++ b/src/conversation/history.rs @@ -74,6 +74,33 @@ impl ConversationLogger { self.log_bot_message_with_name(channel_id, content, None); } + /// Log a system message (e.g. task delegation audit record). Fire-and-forget. + /// + /// System messages are persisted with role `"system"` and are not fed to any + /// LLM context window. They exist purely for UI display in link channel + /// timelines and audit logs. + pub fn log_system_message(&self, channel_id: &str, content: &str) { + let pool = self.pool.clone(); + let id = uuid::Uuid::new_v4().to_string(); + let channel_id = channel_id.to_string(); + let content = content.to_string(); + + tokio::spawn(async move { + if let Err(error) = sqlx::query( + "INSERT INTO conversation_messages (id, channel_id, role, sender_name, content) \ + VALUES (?, ?, 'system', 'system', ?)", + ) + .bind(&id) + .bind(&channel_id) + .bind(&content) + .execute(&pool) + .await + { + tracing::warn!(%error, %channel_id, "failed to persist system message"); + } + }); + } + /// Log a bot (assistant) message with an agent display name. Fire-and-forget. pub fn log_bot_message_with_name( &self, diff --git a/src/lib.rs b/src/lib.rs index 1bb8ed36a..3557849b0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -198,6 +198,18 @@ pub enum ProcessEvent { }, } +/// A message to be injected into a specific channel from outside the normal +/// inbound message flow. Used for cross-agent task completion notifications. +#[derive(Debug, Clone)] +pub struct ChannelInjection { + /// The conversation_id of the target channel. + pub conversation_id: String, + /// The agent that owns the target channel. + pub agent_id: String, + /// The message to inject. + pub message: InboundMessage, +} + /// Shared dependency bundle for agent processes. #[derive(Clone)] pub struct AgentDeps { @@ -215,6 +227,15 @@ pub struct AgentDeps { pub links: Arc>>, /// Map of all agent IDs to display names, for inter-agent message routing. pub agent_names: Arc>, + /// Cross-agent task store registry. Maps agent_id → TaskStore for agents + /// reachable via links. Used by `send_agent_message` to create tasks on + /// target agents and by the cortex to look up delegation metadata. + /// Populated after all agents are initialized. + pub task_store_registry: + Arc>>>, + /// Sender for injecting messages into channels from outside the normal + /// inbound message flow (e.g. cross-agent task completion notifications). + pub injection_tx: tokio::sync::mpsc::Sender, } impl AgentDeps { diff --git a/src/main.rs b/src/main.rs index 8b44d8369..ae4dceae0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1336,9 +1336,24 @@ async fn run( // Channel for removing agents from the main event loop let (agent_remove_tx, mut agent_remove_rx) = mpsc::channel::(8); + // Channel for cross-agent message injection (e.g. delegated task completion notifications). + // The sender is shared with all agents via AgentDeps; the receiver is polled in the main loop. + let (injection_tx, mut injection_rx) = + tokio::sync::mpsc::channel::(64); + + // Shared cross-agent task store registry. Populated after all agents are initialized. + let task_store_registry: Arc< + ArcSwap>>, + > = Arc::new(ArcSwap::from_pointee(std::collections::HashMap::new())); + // Start HTTP API server if enabled - let mut api_state = - spacebot::api::ApiState::new_with_provider_sender(provider_tx, agent_tx, agent_remove_tx); + let mut api_state = spacebot::api::ApiState::new_with_provider_sender( + provider_tx, + agent_tx, + agent_remove_tx, + injection_tx.clone(), + task_store_registry.clone(), + ); api_state.auth_token = config.api.auth_token.clone(); let api_state = Arc::new(api_state); @@ -1489,6 +1504,8 @@ async fn run( &mut telegram_permissions, &mut twitch_permissions, agent_links.clone(), + injection_tx.clone(), + task_store_registry.clone(), &bootstrapped_store, ) .await?; @@ -1779,6 +1796,32 @@ async fn run( tracing::warn!(agent_id = %agent_id, "agent not found in main loop for removal"); } } + // Cross-agent message injection (e.g. delegated task completion retrigger). + // Forwards the injected message to the target channel if it exists. + Some(injection) = injection_rx.recv() => { + if let Some(active) = active_channels.get(&injection.conversation_id) { + if let Err(error) = active.message_tx.send(injection.message).await { + tracing::warn!( + %error, + conversation_id = %injection.conversation_id, + agent_id = %injection.agent_id, + "failed to forward injected message to channel" + ); + } else { + tracing::info!( + conversation_id = %injection.conversation_id, + agent_id = %injection.agent_id, + "forwarded cross-agent injection to active channel" + ); + } + } else { + tracing::info!( + conversation_id = %injection.conversation_id, + agent_id = %injection.agent_id, + "injection target channel not active, notification will be delivered on next message" + ); + } + } Some(_event) = provider_rx.recv(), if !agents_initialized => { tracing::info!("providers configured, initializing agents"); @@ -1832,6 +1875,8 @@ async fn run( &mut new_telegram_permissions, &mut new_twitch_permissions, agent_links.clone(), + injection_tx.clone(), + task_store_registry.clone(), &bootstrapped_store, ).await { Ok(()) => { @@ -1966,6 +2011,10 @@ async fn initialize_agents( telegram_permissions: &mut Option>>, twitch_permissions: &mut Option>>, agent_links: Arc>>, + injection_tx: tokio::sync::mpsc::Sender, + task_store_registry: Arc< + ArcSwap>>, + >, bootstrapped_store: &Option>, ) -> anyhow::Result<()> { let resolved_agents = config.resolve_agents(); @@ -2139,6 +2188,8 @@ async fn initialize_agents( sandbox, links: agent_links.clone(), agent_names: agent_name_map.clone(), + task_store_registry: task_store_registry.clone(), + injection_tx: injection_tx.clone(), }; let agent = spacebot::Agent { @@ -2152,6 +2203,40 @@ async fn initialize_agents( agents.insert(agent_id, agent); } + // Populate the cross-agent task store registry now that all agents exist. + { + let registry: std::collections::HashMap> = agents + .iter() + .map(|(agent_id, agent)| (agent_id.to_string(), agent.deps.task_store.clone())) + .collect(); + task_store_registry.store(Arc::new(registry)); + } + + // Pre-register both sides of every link channel so they appear in each + // agent's channel list from boot. The actual Channel instances are spawned + // on-demand when the first message arrives; this just creates the DB records + // so the UI can display them. + { + let all_links = agent_links.load(); + let empty_meta = std::collections::HashMap::new(); + for link in all_links.iter() { + let from_channel = link.channel_id_for(&link.from_agent_id); + let to_channel = link.channel_id_for(&link.to_agent_id); + + if let Some(agent) = agents.get(&Arc::from(link.from_agent_id.as_str())) { + let store = spacebot::conversation::ChannelStore::new(agent.db.sqlite.clone()); + store.upsert(&from_channel, &empty_meta); + } + if let Some(agent) = agents.get(&Arc::from(link.to_agent_id.as_str())) { + let store = spacebot::conversation::ChannelStore::new(agent.db.sqlite.clone()); + store.upsert(&to_channel, &empty_meta); + } + } + if !all_links.is_empty() { + tracing::info!(link_count = all_links.len(), "pre-registered link channels"); + } + } + tracing::info!(agent_count = agents.len(), "all agents initialized"); // Wire agent event streams, DB pools, and config summaries into the API server diff --git a/src/tools/send_agent_message.rs b/src/tools/send_agent_message.rs index 62d4205d7..213f1da31 100644 --- a/src/tools/send_agent_message.rs +++ b/src/tools/send_agent_message.rs @@ -1,9 +1,14 @@ -//! Send a message to another agent through the communication graph. +//! Assign a task to another agent through the communication graph. //! -//! Currently a stub — validates the link exists and ends the turn. -//! Will be wired into the task system for cross-agent task delegation. +//! When called, creates a task in the target agent's task store (skipping +//! `pending_approval` for agent-delegated tasks) and logs a system message +//! in the link channel between the two agents. The calling agent's turn ends +//! immediately — the result will be delivered when the target agent's cortex +//! picks up and completes the task. +use crate::conversation::history::ConversationLogger; use crate::links::AgentLink; +use crate::tasks::TaskStore; use crate::tools::SkipFlag; use arc_swap::ArcSwap; @@ -15,19 +20,26 @@ use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::Ordering; -/// Tool for sending messages to other agents through the agent communication graph. +/// Tool for delegating tasks to other agents through the agent communication graph. /// /// Resolves the target agent by ID or name, validates the link exists and permits -/// messaging in this direction. Currently a stub — will create tasks on the -/// target agent once cross-agent task delegation is implemented. +/// this direction, creates a task in the target agent's task store, and logs the +/// delegation in the link channel. The calling agent's turn ends after delegation. #[derive(Clone)] pub struct SendAgentMessageTool { agent_id: crate::AgentId, links: Arc>>, /// Map of known agent IDs to display names, for resolving targets. agent_names: Arc>, - /// Per-turn skip flag. When set after sending, the channel turn ends immediately. + /// Cross-agent task store registry for creating tasks on target agents. + task_store_registry: Arc>>>, + /// Per-agent conversation logger for writing link channel audit records. + conversation_logger: ConversationLogger, + /// Per-turn skip flag. When set after delegation, the channel turn ends immediately. skip_flag: Option, + /// The originating channel (conversation_id) where the user request came from. + /// Set per-turn so task completion notifications route back to the right place. + originating_channel: Option, } impl std::fmt::Debug for SendAgentMessageTool { @@ -43,21 +55,33 @@ impl SendAgentMessageTool { agent_id: crate::AgentId, links: Arc>>, agent_names: Arc>, + task_store_registry: Arc>>>, + conversation_logger: ConversationLogger, ) -> Self { Self { agent_id, links, agent_names, + task_store_registry, + conversation_logger, skip_flag: None, + originating_channel: None, } } - /// Set the per-turn skip flag so the channel turn ends after sending. + /// Set the per-turn skip flag so the channel turn ends after delegation. pub fn with_skip_flag(mut self, flag: SkipFlag) -> Self { self.skip_flag = Some(flag); self } + /// Set the originating channel for this turn so task completion notifications + /// route back to the conversation where the user asked for the work. + pub fn with_originating_channel(mut self, channel_id: String) -> Self { + self.originating_channel = Some(channel_id); + self + } + /// Resolve an agent target string to an agent ID. /// Checks both IDs and display names (case-insensitive). fn resolve_agent_id(&self, target: &str) -> Option { @@ -88,7 +112,8 @@ pub struct SendAgentMessageError(String); pub struct SendAgentMessageArgs { /// Target agent ID or name. pub target: String, - /// The message content to send. + /// The task to assign. First sentence is used as the task title; + /// full content becomes the task description. pub message: String, } @@ -97,6 +122,7 @@ pub struct SendAgentMessageArgs { pub struct SendAgentMessageOutput { pub success: bool, pub target_agent: String, + pub task_number: Option, pub message: String, } @@ -120,7 +146,7 @@ impl Tool for SendAgentMessageTool { }, "message": { "type": "string", - "description": "The message content to send to the target agent." + "description": "The task to assign. First sentence becomes the title; full content is the description." } }, "required": ["target", "message"] @@ -171,32 +197,121 @@ impl Tool for SendAgentMessageTool { &link.from_agent_id }; - // End the current turn immediately after delegation. - if let Some(ref flag) = self.skip_flag { - flag.store(true, Ordering::Relaxed); - } - let target_display = self .agent_names .get(receiving_agent_id) .cloned() .unwrap_or_else(|| receiving_agent_id.to_string()); + // Look up the target agent's task store from the cross-agent registry. + let registry = self.task_store_registry.load(); + let target_task_store = registry.get(receiving_agent_id).ok_or_else(|| { + SendAgentMessageError(format!( + "target agent '{}' has no task store available. It may not be initialized.", + target_display + )) + })?; + + // Extract title from the message: first sentence or first 120 chars. + let title = extract_task_title(&args.message); + + // Build task metadata with delegation context. + let metadata = serde_json::json!({ + "delegated_by": sending_agent_id, + "delegating_agent_id": sending_agent_id, + "originating_channel": self.originating_channel, + }); + + // Create the task on the target agent's store. + // Agent-delegated tasks skip pending_approval and go straight to ready. + let task = target_task_store + .create(crate::tasks::CreateTaskInput { + agent_id: receiving_agent_id.to_string(), + title: title.clone(), + description: Some(args.message.clone()), + status: crate::tasks::TaskStatus::Ready, + priority: crate::tasks::TaskPriority::Medium, + subtasks: Vec::new(), + metadata, + source_memory_id: None, + created_by: format!("agent:{}", sending_agent_id), + }) + .await + .map_err(|error| { + SendAgentMessageError(format!( + "failed to create task on agent '{}': {error}", + target_display + )) + })?; + + let task_number = task.task_number; + + // Log delegation record in the link channel (system message). + let sender_display = self + .agent_names + .get(sending_agent_id) + .cloned() + .unwrap_or_else(|| sending_agent_id.to_string()); + let link_channel_id = link.channel_id_for(sending_agent_id); + + self.conversation_logger.log_system_message( + &link_channel_id, + &format!( + "{sender_display} assigned task #{task_number} to {target_display}: \"{title}\"" + ), + ); + + // Also log to the receiver's side of the link channel. + let receiver_link_channel_id = link.channel_id_for(receiving_agent_id); + self.conversation_logger.log_system_message( + &receiver_link_channel_id, + &format!( + "{sender_display} assigned task #{task_number} to {target_display}: \"{title}\"" + ), + ); + + // End the current turn immediately after delegation. + if let Some(ref flag) = self.skip_flag { + flag.store(true, Ordering::Relaxed); + } + tracing::info!( from = %self.agent_id, to = %receiving_agent_id, - "agent message validated (task delegation not yet wired)" + task_number, + "task delegated to target agent" ); - // TODO: Create a task in the target agent's task store instead of - // injecting a message. This is a stub — the tool validates the link - // and ends the turn, but doesn't actually deliver anything yet. - Ok(SendAgentMessageOutput { success: true, target_agent: target_display, - message: "Message validated. Cross-agent task delegation not yet implemented." - .to_string(), + task_number: Some(task_number), + message: format!( + "Task #{task_number} assigned. The target agent's cortex will pick it up and execute it autonomously. \ + You will be notified when it completes." + ), }) } } + +/// Extract a task title from the message content. +/// Uses the first sentence (up to first `.`, `!`, or `?`) or truncates at 120 chars. +fn extract_task_title(message: &str) -> String { + let first_line = message.lines().next().unwrap_or(message); + + // Find the first sentence-ending punctuation + if let Some(position) = first_line.find(['.', '!', '?']) { + let title = &first_line[..=position]; + if title.len() <= 120 { + return title.trim().to_string(); + } + } + + // Fall back to first 120 chars + if first_line.len() <= 120 { + first_line.trim().to_string() + } else { + let boundary = first_line.floor_char_boundary(120); + format!("{}...", first_line[..boundary].trim()) + } +} diff --git a/tests/bulletin.rs b/tests/bulletin.rs index 4f90615eb..c6f43529c 100644 --- a/tests/bulletin.rs +++ b/tests/bulletin.rs @@ -120,6 +120,10 @@ async fn bootstrap_deps() -> anyhow::Result { sandbox, links: Arc::new(arc_swap::ArcSwap::from_pointee(Vec::new())), agent_names: Arc::new(std::collections::HashMap::new()), + task_store_registry: Arc::new(arc_swap::ArcSwap::from_pointee( + std::collections::HashMap::new(), + )), + injection_tx: tokio::sync::mpsc::channel(1).0, }) } diff --git a/tests/context_dump.rs b/tests/context_dump.rs index 615a3c447..df77d7807 100644 --- a/tests/context_dump.rs +++ b/tests/context_dump.rs @@ -119,6 +119,10 @@ async fn bootstrap_deps() -> anyhow::Result<(spacebot::AgentDeps, spacebot::conf sandbox, links: Arc::new(arc_swap::ArcSwap::from_pointee(Vec::new())), agent_names: Arc::new(std::collections::HashMap::new()), + task_store_registry: Arc::new(arc_swap::ArcSwap::from_pointee( + std::collections::HashMap::new(), + )), + injection_tx: tokio::sync::mpsc::channel(1).0, }; Ok((deps, config))