everruns_engine/execution/
input.rs

1//! InputAtom - Atom for recording user input and starting a turn
2//!
3//! This atom is the entry point for a turn. It:
4//! 1. Retrieves the user message from the message store
5//! 2. Returns the message for further processing
6//!
7//! Note: The input.message event is emitted by the API when the message is stored,
8//! not by this atom. This atom simply retrieves the already-stored message.
9
10use serde::{Deserialize, Serialize};
11
12use super::ExecutionContext;
13use crate::error::{AgentLoopError, Result};
14use crate::message::Message;
15use crate::message_retriever::MessageRetriever;
16
17// ============================================================================
18// Input and Output Types
19// ============================================================================
20
21/// Input for InputAtom
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub struct InputAtomInput {
24    /// Atom execution context
25    pub context: ExecutionContext,
26}
27
28/// Result of the InputAtom
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct InputAtomResult {
31    /// The user message that triggered this turn
32    pub message: Message,
33}
34
35// ============================================================================
36// InputAtom
37// ============================================================================
38
39/// Atom that retrieves user input for a turn
40///
41/// This atom:
42/// 1. Retrieves the user message from the message retriever using input_message_id
43/// 2. Returns the message for downstream processing
44///
45/// The message is expected to already be stored by the API layer (which emits input.message).
46/// This atom just retrieves it and prepares for the turn.
47pub struct InputAtom<M>
48where
49    M: MessageRetriever,
50{
51    message_retriever: M,
52}
53
54impl<M> InputAtom<M>
55where
56    M: MessageRetriever + Send + Sync,
57{
58    /// Create a new InputAtom
59    pub fn new(message_retriever: M) -> Self {
60        Self { message_retriever }
61    }
62    /// Stable phase name used by logs and durable activity adapters.
63    pub fn name(&self) -> &'static str {
64        "input"
65    }
66
67    /// Retrieve the turn's already-persisted input message.
68    pub async fn execute(&self, input: InputAtomInput) -> Result<InputAtomResult> {
69        let InputAtomInput { context } = input;
70
71        tracing::debug!(
72            session_id = %context.session_id,
73            turn_id = %context.turn_id,
74            input_message_id = %context.input_message_id,
75            exec_id = %context.exec_id,
76            "InputAtom: retrieving user message"
77        );
78
79        // Retrieve the user message from the retriever
80        let message = self
81            .message_retriever
82            .get(context.session_id, context.input_message_id)
83            .await?
84            .ok_or_else(|| {
85                AgentLoopError::store(format!(
86                    "User message not found: {}",
87                    context.input_message_id
88                ))
89            })?;
90
91        tracing::info!(
92            session_id = %context.session_id,
93            turn_id = %context.turn_id,
94            message_id = %message.id,
95            "InputAtom: turn started with user message"
96        );
97
98        Ok(InputAtomResult { message })
99    }
100}
101
102// ============================================================================
103// Tests
104// ============================================================================
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109    use crate::message_retriever::InputMessage;
110    use crate::test_fixtures::TestMessageRetriever;
111    use crate::typed_id::{MessageId, SessionId, TurnId};
112
113    #[tokio::test]
114    async fn test_input_atom_retrieves_message() {
115        let retriever = TestMessageRetriever::new();
116        let session_id = SessionId::new();
117        let turn_id = TurnId::new();
118
119        // Add a user message to the retriever
120        let user_message = retriever
121            .add(session_id, InputMessage::user("Hello, world!"))
122            .await
123            .unwrap();
124
125        let context = ExecutionContext::new(session_id, turn_id, user_message.id);
126        let atom = InputAtom::new(retriever);
127
128        let result = atom.execute(InputAtomInput { context }).await.unwrap();
129
130        assert_eq!(result.message.id, user_message.id);
131        assert_eq!(result.message.text(), Some("Hello, world!"));
132    }
133
134    #[tokio::test]
135    async fn test_input_atom_not_found() {
136        let retriever = TestMessageRetriever::new();
137        let session_id = SessionId::new();
138        let turn_id = TurnId::new();
139        let missing_id = MessageId::new();
140
141        let context = ExecutionContext::new(session_id, turn_id, missing_id);
142        let atom = InputAtom::new(retriever);
143
144        let result = atom.execute(InputAtomInput { context }).await;
145
146        assert!(result.is_err());
147    }
148}
everruns_engine/execution/input.rs

everruns_engine/execution/
input.rs
