quantum_sdk/
agent.rs

1use std::collections::HashMap;
2use std::pin::Pin;
3use std::task::{Context, Poll};
4
5use futures_util::Stream;
6use pin_project_lite::pin_project;
7use serde::{Deserialize, Serialize};
8
9use crate::client::Client;
10use crate::error::Result;
11use crate::session::ContextConfig;
12
13// ---------------------------------------------------------------------------
14// Agent
15// ---------------------------------------------------------------------------
16
17/// Describes a worker agent in a multi-agent run.
18#[derive(Debug, Clone, Serialize, Deserialize, Default)]
19pub struct AgentWorker {
20    /// Worker name.
21    pub name: String,
22
23    /// Model ID for this worker.
24    #[serde(skip_serializing_if = "Option::is_none")]
25    pub model: Option<String>,
26
27    /// Worker tier (e.g. "fast", "thinking").
28    #[serde(skip_serializing_if = "Option::is_none")]
29    pub tier: Option<String>,
30
31    /// Description of this worker's role.
32    #[serde(skip_serializing_if = "Option::is_none")]
33    pub description: Option<String>,
34}
35
36/// Request body for an agent run.
37#[derive(Debug, Clone, Serialize, Default)]
38pub struct AgentRequest {
39    /// Session identifier for continuity across runs.
40    #[serde(skip_serializing_if = "Option::is_none")]
41    pub session_id: Option<String>,
42
43    /// The task for the agent to accomplish.
44    pub task: String,
45
46    /// Model for the conductor agent.
47    #[serde(skip_serializing_if = "Option::is_none")]
48    pub conductor_model: Option<String>,
49
50    /// Worker agents available to the conductor.
51    #[serde(skip_serializing_if = "Option::is_none")]
52    pub workers: Option<Vec<AgentWorker>>,
53
54    /// Maximum number of steps before stopping.
55    #[serde(skip_serializing_if = "Option::is_none")]
56    pub max_steps: Option<i32>,
57
58    /// System prompt for the conductor.
59    #[serde(skip_serializing_if = "Option::is_none")]
60    pub system_prompt: Option<String>,
61
62    /// Context configuration for session management.
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub context_config: Option<crate::session::ContextConfig>,
65}
66
67// ---------------------------------------------------------------------------
68// Mission
69// ---------------------------------------------------------------------------
70
71/// Describes a named worker for a mission (map keyed by name).
72#[derive(Debug, Clone, Serialize, Deserialize, Default)]
73pub struct MissionWorker {
74    /// Model ID for this worker.
75    #[serde(skip_serializing_if = "Option::is_none")]
76    pub model: Option<String>,
77
78    /// Worker tier.
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub tier: Option<String>,
81
82    /// Description of this worker's purpose.
83    #[serde(skip_serializing_if = "Option::is_none")]
84    pub description: Option<String>,
85}
86
87/// Request body for a mission run.
88#[derive(Debug, Clone, Serialize, Default)]
89pub struct MissionRequest {
90    /// The high-level goal for the mission.
91    pub goal: String,
92
93    /// Execution strategy hint.
94    #[serde(skip_serializing_if = "Option::is_none")]
95    pub strategy: Option<String>,
96
97    /// Model for the conductor.
98    #[serde(skip_serializing_if = "Option::is_none")]
99    pub conductor_model: Option<String>,
100
101    /// Named workers (key = worker name).
102    #[serde(skip_serializing_if = "Option::is_none")]
103    pub workers: Option<HashMap<String, MissionWorker>>,
104
105    /// Maximum number of steps.
106    #[serde(skip_serializing_if = "Option::is_none")]
107    pub max_steps: Option<i32>,
108
109    /// System prompt for the conductor.
110    #[serde(skip_serializing_if = "Option::is_none")]
111    pub system_prompt: Option<String>,
112
113    /// Session identifier for continuity.
114    #[serde(skip_serializing_if = "Option::is_none")]
115    pub session_id: Option<String>,
116
117    /// Whether to auto-plan before execution.
118    #[serde(skip_serializing_if = "Option::is_none")]
119    pub auto_plan: Option<bool>,
120
121    /// Context management configuration.
122    #[serde(skip_serializing_if = "Option::is_none")]
123    pub context_config: Option<ContextConfig>,
124
125    /// Model for worker nodes (codegen strategy).
126    #[serde(skip_serializing_if = "Option::is_none")]
127    pub worker_model: Option<String>,
128
129    /// Deployment ID — route worker inference to a managed Vertex endpoint.
130    #[serde(skip_serializing_if = "Option::is_none")]
131    pub deployment_id: Option<String>,
132
133    /// Build command to run after codegen (e.g. "cargo build", "npm run build").
134    #[serde(skip_serializing_if = "Option::is_none")]
135    pub build_command: Option<String>,
136
137    /// Workspace directory for generated files.
138    #[serde(skip_serializing_if = "Option::is_none")]
139    pub workspace_path: Option<String>,
140}
141
142/// Backwards-compatible alias for [`AgentWorker`].
143pub type AgentWorkerConfig = AgentWorker;
144
145/// Backwards-compatible alias for [`MissionWorker`].
146pub type MissionWorkerConfig = MissionWorker;
147
148// ---------------------------------------------------------------------------
149// SSE Stream
150// ---------------------------------------------------------------------------
151
152/// A single event from an agent or mission SSE stream.
153#[derive(Debug, Clone, Deserialize)]
154pub struct AgentStreamEvent {
155    /// Event type (e.g. "step", "thought", "tool_call", "tool_result", "message", "error", "done").
156    #[serde(rename = "type", default)]
157    pub event_type: String,
158
159    /// Raw JSON payload for caller to interpret.
160    #[serde(flatten)]
161    pub data: HashMap<String, serde_json::Value>,
162}
163
164/// A single SSE event from an agent run stream.
165/// Alias for [`AgentStreamEvent`] for backwards compatibility.
166pub type AgentEvent = AgentStreamEvent;
167
168/// A single SSE event from a mission run stream.
169/// Alias for [`AgentStreamEvent`] since both use the same SSE format.
170pub type MissionEvent = AgentStreamEvent;
171
172pin_project! {
173    /// An async stream of [`AgentStreamEvent`]s from an agent or mission SSE response.
174    pub struct AgentStream {
175        #[pin]
176        inner: Pin<Box<dyn Stream<Item = AgentStreamEvent> + Send>>,
177    }
178}
179
180impl Stream for AgentStream {
181    type Item = AgentStreamEvent;
182
183    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
184        self.project().inner.poll_next(cx)
185    }
186}
187
188/// Converts a byte stream into a stream of parsed [`AgentStreamEvent`]s.
189fn sse_to_agent_events<S>(byte_stream: S) -> impl Stream<Item = AgentStreamEvent> + Send
190where
191    S: Stream<Item = std::result::Result<bytes::Bytes, reqwest::Error>> + Send + 'static,
192{
193    let pinned_stream = Box::pin(byte_stream);
194
195    let line_stream = futures_util::stream::unfold(
196        (pinned_stream, String::new()),
197        |(mut stream, mut buffer)| async move {
198            use futures_util::StreamExt;
199            loop {
200                if let Some(newline_pos) = buffer.find('\n') {
201                    let line = buffer[..newline_pos].trim_end_matches('\r').to_string();
202                    buffer = buffer[newline_pos + 1..].to_string();
203                    return Some((line, (stream, buffer)));
204                }
205
206                match stream.next().await {
207                    Some(Ok(chunk)) => {
208                        buffer.push_str(&String::from_utf8_lossy(&chunk));
209                    }
210                    Some(Err(_)) | None => {
211                        if !buffer.is_empty() {
212                            let remaining = std::mem::take(&mut buffer);
213                            return Some((remaining, (stream, buffer)));
214                        }
215                        return None;
216                    }
217                }
218            }
219        },
220    );
221
222    let pinned_lines = Box::pin(line_stream);
223    futures_util::stream::unfold(pinned_lines, |mut lines| async move {
224        use futures_util::StreamExt;
225        loop {
226            let line = lines.next().await?;
227
228            if !line.starts_with("data: ") {
229                continue;
230            }
231            let payload = &line["data: ".len()..];
232
233            if payload == "[DONE]" {
234                let ev = AgentStreamEvent {
235                    event_type: "done".to_string(),
236                    data: HashMap::new(),
237                };
238                return Some((ev, lines));
239            }
240
241            match serde_json::from_str::<AgentStreamEvent>(payload) {
242                Ok(ev) => return Some((ev, lines)),
243                Err(e) => {
244                    let mut data = HashMap::new();
245                    data.insert(
246                        "error".to_string(),
247                        serde_json::Value::String(format!("parse SSE: {e}")),
248                    );
249                    let ev = AgentStreamEvent {
250                        event_type: "error".to_string(),
251                        data,
252                    };
253                    return Some((ev, lines));
254                }
255            }
256        }
257    })
258}
259
260impl Client {
261    /// Starts an agent run and returns an SSE event stream.
262    ///
263    /// The agent orchestrates one or more worker models to accomplish the task,
264    /// streaming progress events as it works.
265    pub async fn agent_run(&self, req: &AgentRequest) -> Result<AgentStream> {
266        let (resp, _meta) = self.post_stream_raw("/qai/v1/agent", req).await?;
267        let byte_stream = resp.bytes_stream();
268        let event_stream = sse_to_agent_events(byte_stream);
269        Ok(AgentStream {
270            inner: Box::pin(event_stream),
271        })
272    }
273
274    /// Starts a mission run and returns an SSE event stream.
275    ///
276    /// Missions are higher-level than agents -- they can auto-plan, assign
277    /// named workers, and manage context across multiple steps.
278    pub async fn mission_run(&self, req: &MissionRequest) -> Result<AgentStream> {
279        let (resp, _meta) = self.post_stream_raw("/qai/v1/missions", req).await?;
280        let byte_stream = resp.bytes_stream();
281        let event_stream = sse_to_agent_events(byte_stream);
282        Ok(AgentStream {
283            inner: Box::pin(event_stream),
284        })
285    }
286}
quantum_sdk/agent.rs

quantum_sdk/
agent.rs
