orchestral_runtime/api/
agent.rs

1//! API composition surface backed by the same Agent SDK and Controller used by
2//! embedded callers and the CLI.
3
4use std::collections::BTreeMap;
5use std::sync::Arc;
6
7use orchestral_core::agent_connector::AgentSessionActionInvocation;
8use orchestral_core::agent_protocol::spi::AgentRunCatalogEntry;
9use orchestral_core::agent_protocol::wire::{
10    AgentCommandEnvelope, AgentJournalRecord, AgentRunView, AgentSessionId, CommandAck, CommandId,
11    Content, Extensions, ResourceBinding, RunId,
12};
13use tokio::sync::broadcast;
14use tokio::sync::RwLock;
15
16use crate::{AgentClient, AgentControlEvent, AgentController, AgentRunHandle, AgentSdkError};
17
18#[derive(Clone)]
19pub struct AgentApi {
20    controller: Arc<AgentController>,
21    default_resources: Arc<Vec<ResourceBinding>>,
22    default_extensions: Arc<Extensions>,
23    sessions: Arc<RwLock<BTreeMap<AgentSessionId, AgentClient>>>,
24}
25
26impl AgentApi {
27    pub fn new(controller: Arc<AgentController>) -> Self {
28        Self::with_resources(controller, Vec::new())
29    }
30
31    /// Creates an API whose sessions inherit the same immutable resource
32    /// bindings. Transports can therefore create sessions without knowing how
33    /// a concrete Host discovered Skills or other resources.
34    pub fn with_resources(
35        controller: Arc<AgentController>,
36        default_resources: Vec<ResourceBinding>,
37    ) -> Self {
38        Self {
39            controller,
40            default_resources: Arc::new(default_resources),
41            default_extensions: Arc::new(Extensions::new()),
42            sessions: Arc::new(RwLock::new(BTreeMap::new())),
43        }
44    }
45
46    /// Sets Host metadata inherited by sessions subsequently created here.
47    pub fn with_default_extensions(mut self, extensions: Extensions) -> Self {
48        self.default_extensions = Arc::new(extensions);
49        self
50    }
51
52    pub async fn create_session(
53        &self,
54        preferred_id: Option<AgentSessionId>,
55    ) -> Result<AgentSessionId, AgentSdkError> {
56        let session_id = preferred_id.unwrap_or_else(|| {
57            AgentSessionId::new(format!("api-session-{}", uuid::Uuid::new_v4()))
58        });
59        if session_id.is_empty() {
60            return Err(AgentSdkError::InvalidInput(
61                "Agent session id must not be empty".to_owned(),
62            ));
63        }
64        self.sessions
65            .write()
66            .await
67            .entry(session_id.clone())
68            .or_insert_with(|| {
69                AgentClient::new(self.controller.clone(), session_id.clone())
70                    .with_resources(self.default_resources.as_ref().clone())
71                    .with_default_extensions(self.default_extensions.as_ref().clone())
72            });
73        Ok(session_id)
74    }
75
76    pub async fn start_text(
77        &self,
78        session_id: &AgentSessionId,
79        run_id: Option<RunId>,
80        input: impl Into<String>,
81    ) -> Result<AgentRunHandle, AgentSdkError> {
82        let input = input.into();
83        if input.trim().is_empty() {
84            return Err(AgentSdkError::InvalidInput(
85                "Agent input must not be empty".to_owned(),
86            ));
87        }
88        self.start_content(session_id, run_id, vec![Content::text(input)])
89            .await
90    }
91
92    pub async fn start_content(
93        &self,
94        session_id: &AgentSessionId,
95        run_id: Option<RunId>,
96        input: Vec<Content>,
97    ) -> Result<AgentRunHandle, AgentSdkError> {
98        self.start_content_with_extensions(session_id, run_id, input, Extensions::new())
99            .await
100    }
101
102    pub async fn start_content_with_extensions(
103        &self,
104        session_id: &AgentSessionId,
105        run_id: Option<RunId>,
106        input: Vec<Content>,
107        extensions: Extensions,
108    ) -> Result<AgentRunHandle, AgentSdkError> {
109        if input.is_empty() {
110            return Err(AgentSdkError::InvalidInput(
111                "Agent input must not be empty".to_owned(),
112            ));
113        }
114        let client = self.session(session_id).await?;
115        match run_id {
116            Some(run_id) => {
117                client
118                    .start_with_run_id_and_extensions(run_id, input, extensions)
119                    .await
120            }
121            None => {
122                client
123                    .start_with_run_id_and_extensions(
124                        RunId::new(format!("api-session-{}", uuid::Uuid::new_v4())),
125                        input,
126                        extensions,
127                    )
128                    .await
129            }
130        }
131    }
132
133    pub async fn start_session_action(
134        &self,
135        session_id: &AgentSessionId,
136        run_id: RunId,
137        title: impl Into<String>,
138        action: AgentSessionActionInvocation,
139    ) -> Result<AgentRunHandle, AgentSdkError> {
140        self.session(session_id)
141            .await?
142            .start_session_action_with_run_id(run_id, title, action)
143            .await
144    }
145
146    pub async fn inspect(&self, run_id: &RunId) -> Result<AgentRunView, AgentSdkError> {
147        Ok(self.controller.inspect(run_id).await?)
148    }
149
150    pub async fn initial_input(&self, run_id: &RunId) -> Result<Vec<Content>, AgentSdkError> {
151        Ok(self.controller.initial_input(run_id).await?)
152    }
153
154    pub async fn run_extensions(&self, run_id: &RunId) -> Result<Extensions, AgentSdkError> {
155        Ok(self.controller.run_extensions(run_id).await?)
156    }
157
158    pub async fn catalog_runs(&self) -> Result<Vec<AgentRunCatalogEntry>, AgentSdkError> {
159        Ok(self.controller.catalog_runs().await?)
160    }
161
162    pub async fn can_control_run(&self, run_id: &RunId) -> Result<bool, AgentSdkError> {
163        Ok(self.controller.can_control_run(run_id).await?)
164    }
165
166    pub async fn has_run(&self, run_id: &RunId) -> Result<bool, AgentSdkError> {
167        Ok(self.controller.has_run(run_id).await?)
168    }
169
170    pub async fn events(
171        &self,
172        run_id: &RunId,
173        after_run_seq: u64,
174    ) -> Result<Vec<AgentJournalRecord>, AgentSdkError> {
175        Ok(self.controller.events(run_id, after_run_seq).await?)
176    }
177
178    /// Subscribes to best-effort live events. Durable consumers must combine
179    /// this with [`Self::events`] before rendering and after broadcast lag.
180    pub async fn subscribe(
181        &self,
182        run_id: &RunId,
183    ) -> Result<broadcast::Receiver<AgentControlEvent>, AgentSdkError> {
184        Ok(self.controller.subscribe(run_id).await?)
185    }
186
187    /// Looks up a durable command without starting or recovering native work.
188    pub async fn recorded_command(
189        &self,
190        run_id: &RunId,
191        command_id: &CommandId,
192    ) -> Result<Option<AgentCommandEnvelope>, AgentSdkError> {
193        Ok(self.controller.recorded_command(run_id, command_id).await?)
194    }
195
196    pub async fn command(
197        &self,
198        command: AgentCommandEnvelope,
199    ) -> Result<CommandAck, AgentSdkError> {
200        Ok(self.controller.command(command).await?)
201    }
202
203    pub async fn cancel(
204        &self,
205        run_id: &RunId,
206        reason: impl Into<String>,
207    ) -> Result<CommandAck, AgentSdkError> {
208        Ok(self.controller.cancel(run_id, reason).await?)
209    }
210
211    pub async fn recover(&self, run_id: &RunId) -> Result<AgentRunView, AgentSdkError> {
212        Ok(self.controller.recover(run_id).await?)
213    }
214
215    async fn session(&self, session_id: &AgentSessionId) -> Result<AgentClient, AgentSdkError> {
216        self.sessions
217            .read()
218            .await
219            .get(session_id)
220            .cloned()
221            .ok_or_else(|| {
222                AgentSdkError::InvalidInput(format!(
223                    "Agent session does not exist: {}",
224                    session_id.as_str()
225                ))
226            })
227    }
228}
orchestral_runtime/api/agent.rs

orchestral_runtime/api/
agent.rs
