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

ag_agent/channel/
contract.rs
