systemprompt_agent/services/a2a_server/processing/message/
mod.rs

1//! Message processing for the A2A server.
2//!
3//! [`MessageProcessor`] owns the repositories and services needed to handle an
4//! inbound message and persist the resulting task. [`StreamProcessor`] drives
5//! the streaming execution pipeline, reporting progress as [`StreamEvent`]s
6//! over an mpsc channel. Both the streaming and non-streaming entry points live
7//! in the submodules.
8
9mod message_handler;
10mod persistence;
11mod stream_processor;
12
13pub use stream_processor::StreamProcessor;
14
15use crate::services::shared::{AgentServiceError, Result};
16use std::sync::Arc;
17use tokio::sync::mpsc;
18
19use crate::models::AgentRuntimeInfo;
20use crate::models::a2a::{Artifact, Message, Task};
21use systemprompt_models::{AiProvider, CallToolResult, ToolCall};
22
23#[derive(Debug)]
24pub enum StreamEvent {
25    Text(String),
26    ToolCallStarted(ToolCall),
27    ToolResult {
28        call_id: String,
29        result: CallToolResult,
30    },
31    ExecutionStepUpdate {
32        step: crate::models::ExecutionStep,
33    },
34    Complete {
35        full_text: String,
36        artifacts: Vec<Artifact>,
37    },
38    Error(String),
39}
40use crate::repository::context::ContextRepository;
41use crate::repository::execution::ExecutionStepRepository;
42use crate::repository::task::TaskRepository;
43use crate::services::{ContextService, SkillService};
44use systemprompt_database::DbPool;
45use systemprompt_identifiers::TaskId;
46use systemprompt_models::RequestContext;
47
48#[derive(Debug)]
49pub struct PersistCompletedTaskOnProcessorParams<'a> {
50    pub task: &'a Task,
51    pub user_message: &'a Message,
52    pub agent_message: &'a Message,
53    pub context: &'a RequestContext,
54    pub agent_name: &'a str,
55    pub artifacts_already_published: bool,
56}
57
58#[derive(Debug)]
59pub struct ProcessMessageStreamParams<'a> {
60    pub a2a_message: &'a Message,
61    pub agent_runtime: &'a AgentRuntimeInfo,
62    pub agent_name: &'a str,
63    pub context: &'a RequestContext,
64    pub task_id: TaskId,
65}
66
67pub struct MessageProcessor {
68    db_pool: DbPool,
69    ai_service: Arc<dyn AiProvider>,
70    task_repo: TaskRepository,
71    context_repo: ContextRepository,
72    context_service: ContextService,
73    skill_service: Arc<SkillService>,
74    execution_step_repo: Arc<ExecutionStepRepository>,
75}
76
77impl std::fmt::Debug for MessageProcessor {
78    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79        f.debug_struct("MessageProcessor")
80            .field("ai_service", &"<Arc<dyn AiProvider>>")
81            .finish()
82    }
83}
84
85impl MessageProcessor {
86    pub fn new(db_pool: &DbPool, ai_service: Arc<dyn AiProvider>) -> Result<Self> {
87        let task_repo = TaskRepository::new(db_pool)?;
88        let context_repo = ContextRepository::new(db_pool)?;
89        let context_service = ContextService::new(db_pool)?;
90        let execution_step_repo = Arc::new(ExecutionStepRepository::new(db_pool)?);
91        let skill_service = Arc::new(
92            SkillService::new()?.with_execution_step_repo(Arc::clone(&execution_step_repo)),
93        );
94
95        Ok(Self {
96            db_pool: Arc::clone(db_pool),
97            ai_service,
98            task_repo,
99            context_repo,
100            context_service,
101            skill_service,
102            execution_step_repo,
103        })
104    }
105
106    pub async fn load_agent_runtime(&self, agent_name: &str) -> Result<AgentRuntimeInfo> {
107        use crate::services::registry::AgentRegistry;
108
109        let registry = AgentRegistry::new()?;
110        let agent_config = registry
111            .get_agent(agent_name)
112            .await
113            .map_err(|_e| AgentServiceError::Internal("Agent not found".to_owned()))?;
114
115        Ok(agent_config.into())
116    }
117
118    pub async fn persist_completed_task(
119        &self,
120        params: PersistCompletedTaskOnProcessorParams<'_>,
121    ) -> Result<Task> {
122        persistence::persist_completed_task(persistence::PersistCompletedTaskParams {
123            task: params.task,
124            user_message: params.user_message,
125            agent_message: params.agent_message,
126            context: params.context,
127            task_repo: &self.task_repo,
128            db_pool: &self.db_pool,
129            artifacts_already_published: params.artifacts_already_published,
130        })
131        .await
132    }
133
134    pub async fn process_message_stream(
135        &self,
136        params: ProcessMessageStreamParams<'_>,
137    ) -> Result<mpsc::Receiver<StreamEvent>> {
138        let stream_processor = StreamProcessor {
139            ai_service: Arc::clone(&self.ai_service),
140            context_service: self.context_service.clone(),
141            skill_service: Arc::clone(&self.skill_service),
142            execution_step_repo: Arc::clone(&self.execution_step_repo),
143        };
144
145        stream_processor.process_message_stream(params).await
146    }
147}
systemprompt_agent/services/a2a_server/processing/message/mod.rs

systemprompt_agent/services/a2a_server/processing/message/
mod.rs
