ag_agent/channel/
contract.rs

1//! Shared channel trait and provider turn request/result contracts.
2
3use std::fmt;
4use std::future::Future;
5use std::path::PathBuf;
6use std::pin::Pin;
7use std::sync::Arc;
8
9use ag_protocol::{AgentResponse, ProtocolRequestProfile, TurnPrompt};
10use tokio::sync::mpsc;
11
12use crate::model::agent::ReasoningLevel;
13
14/// Boxed async result used by [`AgentChannel`] trait methods.
15pub type AgentFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>;
16
17/// Live transcript projection used when a provider runtime needs replay text.
18pub trait LiveTranscript: fmt::Debug + Send + Sync {
19    /// Returns the latest replayable transcript text, when any content exists.
20    fn replay_text(&self) -> Option<String>;
21}
22
23/// Turn initiation mode for [`TurnRequest`].
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub enum AgentRequestKind {
26    /// Starts a fresh interactive session turn with no prior context.
27    SessionStart,
28    /// Resumes an interactive session turn.
29    SessionResume,
30    /// Runs one utility prompt with utility protocol requirements.
31    ///
32    /// Callers may route this through an isolated one-shot channel or through
33    /// an existing session channel when the utility work needs provider
34    /// conversation continuity without normal post-turn auto-commit handling.
35    UtilityPrompt,
36    /// Reads provider account metadata without creating an agent turn.
37    AccountRead,
38}
39
40impl AgentRequestKind {
41    /// Returns the protocol request profile derived from this request kind.
42    #[must_use]
43    pub fn protocol_profile(&self) -> ProtocolRequestProfile {
44        match self {
45            Self::SessionStart | Self::SessionResume => ProtocolRequestProfile::SessionTurn,
46            Self::UtilityPrompt | Self::AccountRead => ProtocolRequestProfile::UtilityPrompt,
47        }
48    }
49
50    /// Returns whether this request resumes a prior interactive session turn.
51    #[must_use]
52    pub fn is_resume(&self) -> bool {
53        matches!(self, Self::SessionResume)
54    }
55}
56
57/// Continuation state for one provider-agnostic agent turn.
58///
59/// The concrete representation keeps provider-runtime recovery details out of
60/// [`TurnRequest`] while allowing CLI channels to consume only replay text.
61#[derive(Clone, Debug)]
62pub struct TurnContinuation {
63    kind: TurnContinuationKind,
64}
65
66impl TurnContinuation {
67    /// Creates continuation state for a fresh turn with no prior context.
68    #[must_use]
69    pub fn fresh() -> Self {
70        Self {
71            kind: TurnContinuationKind::Fresh,
72        }
73    }
74
75    /// Creates continuation state for a stateless turn that replays prior text.
76    #[must_use]
77    pub fn replaying(replay_transcript: String) -> Self {
78        Self {
79            kind: TurnContinuationKind::Replay { replay_transcript },
80        }
81    }
82
83    /// Creates continuation state for a provider runtime that may resume a
84    /// native conversation and reconstruct context from a live transcript.
85    #[must_use]
86    pub fn provider(
87        live_transcript: Option<Arc<dyn LiveTranscript>>,
88        persisted_instruction_conversation_id: Option<String>,
89        provider_conversation_id: Option<String>,
90        replay_transcript: Option<String>,
91    ) -> Self {
92        Self {
93            kind: TurnContinuationKind::Provider {
94                live_transcript,
95                persisted_instruction_conversation_id,
96                provider_conversation_id,
97                replay_transcript,
98            },
99        }
100    }
101
102    /// Returns replayable transcript text when this turn carries it.
103    #[must_use]
104    pub fn replay_transcript(&self) -> Option<&str> {
105        match &self.kind {
106            TurnContinuationKind::Fresh => None,
107            TurnContinuationKind::Provider {
108                replay_transcript, ..
109            } => replay_transcript.as_deref(),
110            TurnContinuationKind::Replay { replay_transcript } => Some(replay_transcript.as_str()),
111        }
112    }
113
114    /// Returns the provider-native conversation identifier when available.
115    #[must_use]
116    pub fn provider_conversation_id(&self) -> Option<&str> {
117        match &self.kind {
118            TurnContinuationKind::Provider {
119                provider_conversation_id,
120                ..
121            } => provider_conversation_id.as_deref(),
122            TurnContinuationKind::Fresh | TurnContinuationKind::Replay { .. } => None,
123        }
124    }
125
126    /// Returns the conversation identifier that received the instruction
127    /// bootstrap when available.
128    #[must_use]
129    pub fn persisted_instruction_conversation_id(&self) -> Option<&str> {
130        match &self.kind {
131            TurnContinuationKind::Provider {
132                persisted_instruction_conversation_id,
133                ..
134            } => persisted_instruction_conversation_id.as_deref(),
135            TurnContinuationKind::Fresh | TurnContinuationKind::Replay { .. } => None,
136        }
137    }
138
139    pub(crate) fn into_parts(self) -> TurnContinuationParts {
140        match self.kind {
141            TurnContinuationKind::Fresh => TurnContinuationParts::default(),
142            TurnContinuationKind::Replay { replay_transcript } => TurnContinuationParts {
143                replay_transcript: Some(replay_transcript),
144                ..TurnContinuationParts::default()
145            },
146            TurnContinuationKind::Provider {
147                live_transcript,
148                persisted_instruction_conversation_id,
149                provider_conversation_id,
150                replay_transcript,
151            } => TurnContinuationParts {
152                live_transcript,
153                persisted_instruction_conversation_id,
154                provider_conversation_id,
155                replay_transcript,
156            },
157        }
158    }
159}
160
161#[derive(Clone, Debug)]
162enum TurnContinuationKind {
163    Fresh,
164    Provider {
165        live_transcript: Option<Arc<dyn LiveTranscript>>,
166        persisted_instruction_conversation_id: Option<String>,
167        provider_conversation_id: Option<String>,
168        replay_transcript: Option<String>,
169    },
170    Replay {
171        replay_transcript: String,
172    },
173}
174
175#[derive(Default)]
176pub(crate) struct TurnContinuationParts {
177    pub(crate) live_transcript: Option<Arc<dyn LiveTranscript>>,
178    pub(crate) persisted_instruction_conversation_id: Option<String>,
179    pub(crate) provider_conversation_id: Option<String>,
180    pub(crate) replay_transcript: Option<String>,
181}
182
183/// Input payload for one provider-agnostic agent turn.
184#[derive(Debug, Clone)]
185pub struct TurnRequest {
186    /// Prior context needed to continue this turn.
187    pub continuation: TurnContinuation,
188    /// Session worktree folder where the agent runs.
189    pub folder: PathBuf,
190    /// Main repository checkout that must remain read-only during the turn,
191    /// when Agentty can resolve it.
192    pub main_checkout_root: Option<PathBuf>,
193    /// Provider-specific model identifier.
194    pub model: String,
195    /// Structured prompt payload for the turn.
196    pub prompt: TurnPrompt,
197    /// Reasoning effort preference for the turn.
198    ///
199    /// Ignored by providers/models that do not support reasoning effort.
200    pub reasoning_level: ReasoningLevel,
201    /// Canonical request kind that drives transport behavior and protocol
202    /// semantics for this turn.
203    pub request_kind: AgentRequestKind,
204}
205
206/// Incremental event emitted during one agent turn.
207///
208/// Events are sent through an [`mpsc::UnboundedSender`] as the turn
209/// progresses, enabling transient loader updates without appending partial turn
210/// output into the persisted transcript.
211#[derive(Clone, Debug, PartialEq)]
212pub enum TurnEvent {
213    /// A streamed thinking/planning or tool-status fragment shown in the
214    /// transient loader.
215    ThoughtDelta(String),
216    /// The turn completed successfully with final token counts.
217    Completed {
218        /// Whether the provider reset its context for this turn.
219        context_reset: bool,
220        /// Input token count for the turn.
221        input_tokens: u64,
222        /// Output token count for the turn.
223        output_tokens: u64,
224    },
225    /// The turn failed with an error description.
226    Failed(String),
227    /// A child process PID update.
228    ///
229    /// Sent by CLI channels immediately after spawning the child process
230    /// (`Some(pid)`) and again after the child exits (`None`). Consumers
231    /// update the shared PID slot used by cancellation signals.
232    PidUpdate(Option<u32>),
233}
234
235/// Normalized result returned when one agent turn completes successfully.
236#[derive(Debug)]
237pub struct TurnResult {
238    /// Parsed agent response containing structured protocol messages.
239    pub assistant_message: AgentResponse,
240    /// Whether the provider reset its context to complete this turn.
241    pub context_reset: bool,
242    /// Input token count for the turn.
243    pub input_tokens: u64,
244    /// Output token count for the turn.
245    pub output_tokens: u64,
246    /// Provider-native conversation identifier observed after the turn.
247    ///
248    /// App-server providers return this so the worker can persist it for
249    /// future runtime restarts. CLI channels always return `None`.
250    pub provider_conversation_id: Option<String>,
251}
252
253/// Opaque reference to an active agent session.
254pub struct SessionRef {
255    /// Stable session identifier.
256    pub session_id: String,
257}
258
259/// Input payload for initiating a new agent session.
260pub struct StartSessionRequest {
261    /// Session worktree folder.
262    pub folder: PathBuf,
263    /// Stable session identifier.
264    pub session_id: String,
265}
266
267/// Typed error returned by [`AgentChannel`] operations.
268///
269/// Discriminates failure causes so the app layer can route errors without
270/// parsing formatted messages.
271#[derive(Debug, thiserror::Error)]
272pub enum AgentError {
273    /// An app-server infrastructure failure propagated from a persistent
274    /// provider runtime.
275    #[error(transparent)]
276    AppServer(#[from] crate::app_server::AppServerError),
277
278    /// A CLI backend command or process execution failure.
279    #[error("{0}")]
280    Backend(String),
281
282    /// The user explicitly interrupted the active turn.
283    #[error("{0}")]
284    InterruptedByUser(String),
285
286    /// A subprocess IO error such as a spawn failure or unavailable pipe.
287    #[error("{0}")]
288    Io(String),
289}
290
291/// Provider-agnostic session channel for executing agent turns.
292///
293/// Implementations bridge a specific transport - CLI subprocess or app-server
294/// RPC - to the unified [`TurnEvent`] stream consumed by session workers. The
295/// trait is object-safe so it can be held as `Arc<dyn AgentChannel>`.
296#[cfg_attr(any(test, feature = "test-utils"), mockall::automock)]
297pub trait AgentChannel: Send + Sync {
298    /// Initialises a provider session for the given session identifier.
299    ///
300    /// Implementations that do not maintain persistent sessions return
301    /// immediately with a [`SessionRef`] wrapping the supplied identifier.
302    fn start_session(
303        &self,
304        req: StartSessionRequest,
305    ) -> AgentFuture<Result<SessionRef, AgentError>>;
306
307    /// Executes one prompt turn and streams incremental events to `events`.
308    ///
309    /// Implementations may emit [`TurnEvent::ThoughtDelta`] values for
310    /// transient loader updates. Final transcript output is derived from the
311    /// returned [`TurnResult`] after the turn finishes.
312    ///
313    /// # Errors
314    /// Returns [`AgentError`] when the turn cannot be executed (spawn failure,
315    /// transport error) or is interrupted by a signal.
316    fn run_turn(
317        &self,
318        session_id: String,
319        req: TurnRequest,
320        events: mpsc::UnboundedSender<TurnEvent>,
321    ) -> AgentFuture<Result<TurnResult, AgentError>>;
322
323    /// Tears down the provider session associated with `session_id`.
324    ///
325    /// Implementations that do not maintain persistent sessions treat this as
326    /// a no-op and always return `Ok(())`.
327    fn shutdown_session(&self, session_id: String) -> AgentFuture<Result<(), AgentError>>;
328}
329
330#[cfg(test)]
331mod tests {
332    use super::*;
333
334    #[test]
335    fn test_turn_continuation_fresh_has_no_context() {
336        // Arrange / Act
337        let continuation = TurnContinuation::fresh();
338
339        // Assert
340        assert_eq!(continuation.replay_transcript(), None);
341        assert_eq!(continuation.provider_conversation_id(), None);
342        assert_eq!(continuation.persisted_instruction_conversation_id(), None);
343    }
344
345    #[test]
346    fn test_turn_continuation_replaying_exposes_transcript_only() {
347        // Arrange
348        let continuation = TurnContinuation::replaying("prior turn".to_string());
349
350        // Act
351        let parts = continuation.clone().into_parts();
352
353        // Assert
354        assert_eq!(continuation.replay_transcript(), Some("prior turn"));
355        assert_eq!(continuation.provider_conversation_id(), None);
356        assert!(parts.live_transcript.is_none());
357        assert_eq!(parts.persisted_instruction_conversation_id, None);
358        assert_eq!(parts.provider_conversation_id, None);
359        assert_eq!(parts.replay_transcript.as_deref(), Some("prior turn"));
360    }
361
362    #[test]
363    fn test_turn_continuation_provider_exposes_persisted_context() {
364        // Arrange / Act
365        let continuation = TurnContinuation::provider(
366            None,
367            Some("instruction-1".to_string()),
368            Some("thread-1".to_string()),
369            Some("prior turn".to_string()),
370        );
371
372        // Assert
373        assert_eq!(continuation.replay_transcript(), Some("prior turn"));
374        assert_eq!(continuation.provider_conversation_id(), Some("thread-1"));
375        assert_eq!(
376            continuation.persisted_instruction_conversation_id(),
377            Some("instruction-1")
378        );
379    }
380
381    #[test]
382    /// Ensures session request kinds derive the session-turn protocol
383    /// profile.
384    fn test_agent_request_kind_session_variants_use_session_protocol_profile() {
385        // Arrange
386        let start = AgentRequestKind::SessionStart;
387        let resume = AgentRequestKind::SessionResume;
388
389        // Act
390        let start_profile = start.protocol_profile();
391        let resume_profile = resume.protocol_profile();
392
393        // Assert
394        assert_eq!(start_profile, ProtocolRequestProfile::SessionTurn);
395        assert_eq!(resume_profile, ProtocolRequestProfile::SessionTurn);
396    }
397
398    #[test]
399    /// Ensures utility prompts derive the utility protocol profile.
400    fn test_agent_request_kind_utility_prompt_uses_utility_protocol_profile() {
401        // Arrange
402        let request_kind = AgentRequestKind::UtilityPrompt;
403
404        // Act
405        let protocol_profile = request_kind.protocol_profile();
406
407        // Assert
408        assert_eq!(protocol_profile, ProtocolRequestProfile::UtilityPrompt);
409    }
410
411    #[test]
412    /// Ensures account-read requests are non-session utility requests.
413    fn test_agent_request_kind_account_read_uses_utility_protocol_profile() {
414        // Arrange
415        let request_kind = AgentRequestKind::AccountRead;
416
417        // Act
418        let protocol_profile = request_kind.protocol_profile();
419
420        // Assert
421        assert_eq!(protocol_profile, ProtocolRequestProfile::UtilityPrompt);
422    }
423}
ag_agent/channel/contract.rs

ag_agent/channel/
contract.rs
