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

ag_agent/channel/
contract.rs
