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

1//! Streaming execution pipeline for inbound messages.
2//!
3//! [`StreamProcessor`] extracts content from an A2A message (including
4//! supported file parts decoded into [`AiContentPart`]s), then spawns the
5//! strategy-driven pipeline that streams text, tool, and completion events back
6//! to the caller.
7//!
8//! Copyright (c) systemprompt.io — Business Source License 1.1.
9//! See <https://systemprompt.io> for licensing details.
10
11mod helpers;
12mod processing;
13
14use base64::Engine;
15use std::sync::Arc;
16
17use crate::models::a2a::{FilePart, Message, Part};
18use crate::repository::execution::ExecutionStepRepository;
19use crate::services::{ContextService, SkillService};
20use systemprompt_models::{
21    AiContentPart, AiProvider, is_supported_audio, is_supported_image, is_supported_text,
22    is_supported_video,
23};
24
25#[expect(
26    missing_debug_implementations,
27    reason = "params struct holds non-Debug references"
28)]
29pub struct StreamProcessor {
30    pub ai_service: Arc<dyn AiProvider>,
31    pub context_service: ContextService,
32    pub skill_service: Arc<SkillService>,
33    pub execution_step_repo: Arc<ExecutionStepRepository>,
34}
35
36impl StreamProcessor {
37    pub fn extract_message_content(message: &Message) -> (String, Vec<AiContentPart>) {
38        let mut text_content = String::new();
39        let mut content_parts = Vec::new();
40
41        for part in &message.parts {
42            match part {
43                Part::Text(text_part) => {
44                    if text_content.is_empty() {
45                        text_content.clone_from(&text_part.text);
46                    }
47                    content_parts.push(AiContentPart::text(&text_part.text));
48                },
49                Part::File(file_part) => {
50                    if let Some(content_part) = Self::file_to_content_part(file_part) {
51                        content_parts.push(content_part);
52                    }
53                },
54                Part::Data(_) => {},
55            }
56        }
57
58        (text_content, content_parts)
59    }
60
61    fn file_to_content_part(file_part: &FilePart) -> Option<AiContentPart> {
62        let mime_type = file_part.file.mime_type.as_deref()?;
63        let file_name = file_part.file.name.as_deref().unwrap_or("unnamed");
64
65        let bytes = file_part.file.bytes.as_deref()?;
66
67        if is_supported_image(mime_type) {
68            return Some(AiContentPart::image(mime_type, bytes));
69        }
70
71        if is_supported_audio(mime_type) {
72            return Some(AiContentPart::audio(mime_type, bytes));
73        }
74
75        if is_supported_video(mime_type) {
76            return Some(AiContentPart::video(mime_type, bytes));
77        }
78
79        if is_supported_text(mime_type) {
80            return Self::decode_text_file(bytes, file_name, mime_type);
81        }
82
83        tracing::warn!(
84            file_name = %file_name,
85            mime_type = %mime_type,
86            "Unsupported file type - file will not be sent to AI"
87        );
88        None
89    }
90
91    fn decode_text_file(bytes: &str, file_name: &str, mime_type: &str) -> Option<AiContentPart> {
92        let decoded = base64::engine::general_purpose::STANDARD
93            .decode(bytes)
94            .map_err(|e| {
95                tracing::warn!(
96                    file_name = %file_name,
97                    mime_type = %mime_type,
98                    error = %e,
99                    "Failed to decode base64 text file"
100                );
101                e
102            })
103            .ok()?;
104
105        let text_content = String::from_utf8(decoded)
106            .map_err(|e| {
107                tracing::warn!(
108                    file_name = %file_name,
109                    mime_type = %mime_type,
110                    error = %e,
111                    "Failed to decode text file as UTF-8"
112                );
113                e
114            })
115            .ok()?;
116
117        let formatted = format!("[File: {file_name} ({mime_type})]\n{text_content}");
118        Some(AiContentPart::text(formatted))
119    }
120}
systemprompt_agent/services/a2a_server/processing/message/stream_processor/mod.rs

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