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

ag_agent/channel/
contract.rs
