agentic_core/storage/models/
item.rs

1//! Conversation history item stored in the database.
2
3use serde_json::Value;
4use std::fmt::Write;
5use tracing::warn;
6
7use super::super::pool::{DbPool, DbResult, DbTransaction};
8use super::super::types::item::{InOutItem, ItemKind, STORED_ITEM_KIND_KEY};
9use crate::types::io::{InputItem, OutputItem};
10use crate::utils::common::{deserialize_from_str_opt, utcnow_str};
11
12const ITEM_COLUMN_COUNT: usize = 5;
13const SEQUENCE_COLUMN_INDEX: usize = 4;
14const MAX_BIND_PARAMETERS: usize = 999;
15const MAX_ITEMS_PER_INSERT: usize = MAX_BIND_PARAMETERS / ITEM_COLUMN_COUNT;
16
17/// Conversation history item stored in the database.
18///
19/// Maps to the `items` table and represents a single message/event
20/// in a conversation timeline.
21#[derive(Debug, Clone, sqlx::FromRow)]
22pub struct Item {
23    /// Unique identifier for this item.
24    pub id: String,
25
26    /// Item data stored as JSON text.
27    /// Deserialized based on context (`message`, `tool_call`, etc.)
28    pub data: String,
29
30    /// Creation timestamp as Unix timestamp in seconds.
31    pub created_at: i64,
32
33    /// Optional conversation ID for grouping items.
34    pub conversation_id: Option<String>,
35
36    /// Optional sequence number within conversation.
37    pub seq: Option<i64>,
38}
39
40impl Item {
41    /// Deserialize data column as `InputItem`.
42    #[must_use]
43    pub fn as_input(&self) -> Option<InputItem> {
44        deserialize_from_str_opt(&self.data)
45    }
46
47    /// Deserialize data column as `OutputItem`.
48    #[must_use]
49    pub fn as_output(&self) -> Option<OutputItem> {
50        deserialize_from_str_opt(&self.data)
51    }
52
53    /// Deserialize data column as either `InputItem` or `OutputItem`.
54    #[must_use]
55    pub fn as_inout(&self) -> Option<InOutItem> {
56        if let Some(kind) = self.stored_item_kind() {
57            match kind {
58                ItemKind::Input => {
59                    if let Some(input) = self.as_input() {
60                        return Some(InOutItem::Input(input));
61                    }
62                }
63                ItemKind::Output => {
64                    if let Some(output) = self.as_output() {
65                        return Some(InOutItem::Output(output));
66                    }
67                }
68            }
69        }
70
71        let output = self.as_output();
72        if output.as_ref().is_some_and(|item| !matches!(item, OutputItem::Unknown)) {
73            return output.map(InOutItem::Output);
74        }
75
76        let input = self.as_input();
77        if input.as_ref().is_some_and(|item| !matches!(item, InputItem::Unknown)) {
78            return input.map(InOutItem::Input);
79        }
80
81        match (input, output) {
82            (Some(input), _) => Some(InOutItem::Input(input)),
83            (_, Some(output)) => Some(InOutItem::Output(output)),
84            _ => {
85                warn!(item_id = %self.id, "unrecognized item type in stored data");
86                None
87            }
88        }
89    }
90
91    fn stored_item_kind(&self) -> Option<ItemKind> {
92        let value = deserialize_from_str_opt::<Value>(&self.data)?;
93        ItemKind::from_stored_str(value.get(STORED_ITEM_KIND_KEY)?.as_str()?)
94    }
95}
96
97fn item_values_clause(row_count: usize, first_bind_index: usize, sequence_from_cte: bool) -> String {
98    let mut clause = String::new();
99    let mut bind_index = first_bind_index;
100
101    for row_index in 0..row_count {
102        if row_index > 0 {
103            clause.push_str(", ");
104        }
105        clause.push('(');
106        for column_index in 0..ITEM_COLUMN_COUNT {
107            if column_index > 0 {
108                clause.push_str(", ");
109            }
110            if sequence_from_cte && column_index == SEQUENCE_COLUMN_INDEX {
111                write!(clause, "(SELECT start + ${bind_index} FROM next_seq)").expect("writing to String cannot fail");
112            } else {
113                write!(clause, "${bind_index}").expect("writing to String cannot fail");
114            }
115            bind_index += 1;
116        }
117        clause.push(')');
118    }
119
120    clause
121}
122
123/// Create items in a transaction with optional conversation context.
124///
125/// If `conversation_id` is provided, the next sequence range is computed in the insert statement so
126/// concurrent `SQLite` writers do not take a stale read snapshot before writing.
127///
128/// # Errors
129/// Returns `DbResult::Err` if the database insertion fails.
130pub async fn create_in_tx(
131    tx: &mut DbTransaction<'_>,
132    items: Vec<(String, String)>,
133    conversation_id: Option<&str>,
134) -> DbResult<Vec<Item>> {
135    if items.is_empty() {
136        return Ok(Vec::new());
137    }
138
139    let mut created = Vec::with_capacity(items.len());
140    for batch in items.chunks(MAX_ITEMS_PER_INSERT) {
141        let mut rows = if let Some(conversation_id) = conversation_id {
142            create_in_tx_with_next_conversation_seq(tx, batch, conversation_id).await?
143        } else {
144            create_in_tx_without_conversation(tx, batch).await?
145        };
146        created.append(&mut rows);
147    }
148    Ok(created)
149}
150
151async fn create_in_tx_without_conversation(
152    tx: &mut DbTransaction<'_>,
153    items: &[(String, String)],
154) -> DbResult<Vec<Item>> {
155    let now = utcnow_str();
156    let values_clause = item_values_clause(items.len(), 1, false);
157    let sql =
158        format!("INSERT INTO items (id, data, created_at, conversation_id, seq) VALUES {values_clause} RETURNING *");
159
160    let mut query = sqlx::query_as::<_, Item>(&sql);
161    for (id, data) in items {
162        query = query.bind(id).bind(data).bind(now).bind(None::<&str>).bind(None::<i64>);
163    }
164
165    query.fetch_all(&mut **tx).await
166}
167
168async fn create_in_tx_with_next_conversation_seq(
169    tx: &mut DbTransaction<'_>,
170    items: &[(String, String)],
171    conversation_id: &str,
172) -> DbResult<Vec<Item>> {
173    let now = utcnow_str();
174    let values_clause = item_values_clause(items.len(), 2, true);
175    let sql = format!(
176        "WITH next_seq AS ( \
177             SELECT COALESCE(MAX(seq), -1) + 1 AS start \
178             FROM items \
179             WHERE conversation_id = $1 \
180         ) \
181         INSERT INTO items (id, data, created_at, conversation_id, seq) \
182         VALUES {values_clause} \
183         RETURNING *"
184    );
185
186    let mut query = sqlx::query_as::<_, Item>(&sql).bind(conversation_id);
187    #[allow(clippy::cast_possible_wrap)]
188    for (idx, (id, data)) in items.iter().enumerate() {
189        query = query
190            .bind(id)
191            .bind(data)
192            .bind(now)
193            .bind(conversation_id)
194            .bind(idx as i64);
195    }
196
197    query.fetch_all(&mut **tx).await
198}
199
200/// Get items by IDs.
201///
202/// # Errors
203/// Returns `DbResult::Err` if the database query fails.
204pub async fn get_items(pool: &DbPool, ids: &[String]) -> DbResult<Vec<Item>> {
205    if ids.is_empty() {
206        return Ok(vec![]);
207    }
208    let mut rows = Vec::with_capacity(ids.len());
209    for batch in ids.chunks(MAX_BIND_PARAMETERS) {
210        let placeholders = (1..=batch.len())
211            .map(|index| format!("${index}"))
212            .collect::<Vec<_>>()
213            .join(", ");
214        let sql = format!("SELECT * FROM items WHERE id IN ({placeholders})");
215        let mut query = sqlx::query_as::<_, Item>(&sql);
216        for id in batch {
217            query = query.bind(id);
218        }
219        rows.extend(query.fetch_all(pool).await?);
220    }
221    Ok(rows)
222}
223
224/// Get items by conversation ID ordered by sequence.
225///
226/// # Errors
227/// Returns `DbResult::Err` if the database query fails.
228pub async fn get_items_by_conversation(pool: &DbPool, conversation_id: &str) -> DbResult<Vec<Item>> {
229    sqlx::query_as::<_, Item>("SELECT * FROM items WHERE conversation_id = $1 ORDER BY seq ASC")
230        .bind(conversation_id)
231        .fetch_all(pool)
232        .await
233}
234
235/// Returns the last stored item sequence for a conversation inside a transaction.
236///
237/// # Errors
238/// Returns `DbResult::Err` if the database query fails.
239pub async fn last_conversation_sequence_in_tx(
240    tx: &mut DbTransaction<'_>,
241    conversation_id: &str,
242) -> DbResult<Option<i64>> {
243    sqlx::query_scalar("SELECT MAX(seq) FROM items WHERE conversation_id = $1")
244        .bind(conversation_id)
245        .fetch_one(&mut **tx)
246        .await
247}
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252    use crate::types::event::MessageStatus;
253    use crate::types::io::{InputItem, OutputItem, ReasoningOutput, ReasoningTextContent};
254
255    #[test]
256    fn item_values_clause_numbers_plain_rows() {
257        assert_eq!(
258            item_values_clause(2, 1, false),
259            "($1, $2, $3, $4, $5), ($6, $7, $8, $9, $10)"
260        );
261    }
262
263    #[test]
264    fn item_values_clause_numbers_conversation_rows_after_cte_bind() {
265        assert_eq!(
266            item_values_clause(2, 2, true),
267            "($2, $3, $4, $5, (SELECT start + $6 FROM next_seq)), \
268             ($7, $8, $9, $10, (SELECT start + $11 FROM next_seq))"
269        );
270    }
271
272    #[tokio::test]
273    async fn item_queries_chunk_above_portable_bind_limit() {
274        let pool = crate::storage::create_pool_with_schema(Some("sqlite://?mode=memory"))
275            .await
276            .expect("create in-memory database");
277        let items = (0..=MAX_BIND_PARAMETERS)
278            .map(|index| (format!("item_{index}"), "{}".to_owned()))
279            .collect::<Vec<_>>();
280        let ids = items.iter().map(|(id, _)| id.clone()).collect::<Vec<_>>();
281        let mut transaction = pool.begin().await.expect("begin transaction");
282        let created = create_in_tx(&mut transaction, items, None)
283            .await
284            .expect("insert item batches");
285        transaction.commit().await.expect("commit item batches");
286        let loaded = get_items(&pool, &ids).await.expect("load item batches");
287
288        assert_eq!(created.len(), MAX_BIND_PARAMETERS + 1);
289        assert_eq!(loaded.len(), MAX_BIND_PARAMETERS + 1);
290    }
291
292    #[tokio::test]
293    async fn conversation_item_batches_keep_contiguous_sequences() {
294        let pool = crate::storage::create_pool_with_schema(Some("sqlite://?mode=memory"))
295            .await
296            .expect("create in-memory database");
297        let conversation_id = "conv_batch";
298        crate::storage::models::conversation::create(&pool, conversation_id)
299            .await
300            .expect("create conversation");
301        let item_count = MAX_ITEMS_PER_INSERT + 1;
302        let items = (0..item_count)
303            .map(|index| (format!("conversation_item_{index}"), "{}".to_owned()))
304            .collect::<Vec<_>>();
305        let mut transaction = pool.begin().await.expect("begin transaction");
306        let created = create_in_tx(&mut transaction, items, Some(conversation_id))
307            .await
308            .expect("insert conversation item batches");
309        transaction.commit().await.expect("commit item batches");
310        let stored = get_items_by_conversation(&pool, conversation_id)
311            .await
312            .expect("load conversation item batches");
313        let expected_sequences = (0..i64::try_from(item_count).expect("item count fits in i64")).collect::<Vec<_>>();
314
315        assert_eq!(
316            created
317                .iter()
318                .map(|item| item.seq.expect("created sequence"))
319                .collect::<Vec<_>>(),
320            expected_sequences
321        );
322        assert_eq!(
323            stored
324                .iter()
325                .map(|item| item.seq.expect("stored sequence"))
326                .collect::<Vec<_>>(),
327            expected_sequences
328        );
329    }
330
331    #[test]
332    fn test_item_basic() {
333        let item = Item {
334            id: "item_123".to_string(),
335            data: r#"{"role":"user","content":"hello"}"#.to_string(),
336            created_at: 1_704_067_200,
337            conversation_id: Some("conv_456".to_string()),
338            seq: Some(1),
339        };
340
341        assert_eq!(item.id, "item_123");
342        assert_eq!(item.conversation_id, Some("conv_456".to_string()));
343        assert_eq!(item.seq, Some(1));
344    }
345
346    #[test]
347    fn test_item_optional_fields() {
348        let item = Item {
349            id: "item_789".to_string(),
350            data: r#"{"role":"assistant"}"#.to_string(),
351            created_at: 1_704_067_200,
352            conversation_id: None,
353            seq: None,
354        };
355
356        assert!(item.conversation_id.is_none());
357        assert!(item.seq.is_none());
358    }
359
360    #[test]
361    fn test_as_inout_uses_stored_kind_for_reasoning_output() {
362        let mut reasoning = ReasoningOutput::new("rs_1");
363        reasoning.content.push(ReasoningTextContent::new("thinking..."));
364        let stored = InOutItem::Output(OutputItem::Reasoning(reasoning));
365        let item = Item {
366            id: "item_reasoning".to_string(),
367            data: String::try_from(&stored).expect("serialization failed"),
368            created_at: 1_704_067_200,
369            conversation_id: None,
370            seq: None,
371        };
372
373        assert!(matches!(
374            item.as_inout(),
375            Some(InOutItem::Output(OutputItem::Reasoning(_)))
376        ));
377    }
378
379    #[test]
380    fn test_legacy_output_message_rehydrates_as_output_before_unknown_input() {
381        let item = Item {
382            id: "item_message".to_string(),
383            data: serde_json::json!({
384                "type": "message",
385                "id": "msg_1",
386                "role": "assistant",
387                "status": "completed",
388                "content": [{"type": "output_text", "text": "hello", "annotations": []}]
389            })
390            .to_string(),
391            created_at: 1_704_067_200,
392            conversation_id: None,
393            seq: None,
394        };
395
396        let stored = item.as_inout().expect("stored item");
397        assert!(matches!(stored, InOutItem::Output(OutputItem::Message(_))));
398
399        let inputs = InOutItem::into_input_items(vec![stored]);
400        assert!(matches!(inputs[0], InputItem::Message(_)));
401    }
402
403    #[test]
404    fn test_namespaced_function_call_rehydrates_without_storage_marker() {
405        let stored = InOutItem::Output(OutputItem::FunctionCall(crate::types::io::FunctionToolCall {
406            id: "fc_1".to_string(),
407            call_id: "call_1".to_string(),
408            name: "run".to_string(),
409            namespace: Some("mcp__shell".to_string()),
410            arguments: "{\"cmd\":\"pwd\"}".to_string(),
411            status: MessageStatus::Completed,
412        }));
413        let item = Item {
414            id: "item_function_call".to_string(),
415            data: String::try_from(&stored).expect("serialization failed"),
416            created_at: 1_704_067_200,
417            conversation_id: None,
418            seq: None,
419        };
420
421        let inputs = InOutItem::into_input_items(vec![item.as_inout().expect("stored item")]);
422        let value = serde_json::to_value(&inputs[0]).expect("input value");
423
424        assert_eq!(value["type"], "function_call");
425        assert_eq!(value["namespace"], "mcp__shell");
426        assert_eq!(value["name"], "run");
427        assert!(value.get(STORED_ITEM_KIND_KEY).is_none());
428
429        println!("namespace round-trip: mcp__shell.run -> storage -> input function_call");
430        println!("storage marker stripped: _agentic_item_kind absent");
431    }
432
433    #[test]
434    fn test_multiple_namespaced_function_calls_rehydrate_without_storage_marker() {
435        let stored_items = [
436            InOutItem::Output(OutputItem::FunctionCall(crate::types::io::FunctionToolCall {
437                id: "fc_1".to_string(),
438                call_id: "call_1".to_string(),
439                name: "run".to_string(),
440                namespace: Some("mcp__shell".to_string()),
441                arguments: "{\"cmd\":\"pwd\"}".to_string(),
442                status: MessageStatus::Completed,
443            })),
444            InOutItem::Output(OutputItem::FunctionCall(crate::types::io::FunctionToolCall {
445                id: "fc_2".to_string(),
446                call_id: "call_2".to_string(),
447                name: "run".to_string(),
448                namespace: Some("mcp__git".to_string()),
449                arguments: "{\"args\":[\"status\",\"--short\"]}".to_string(),
450                status: MessageStatus::Completed,
451            })),
452        ];
453        let rows: Vec<InOutItem> = stored_items
454            .iter()
455            .enumerate()
456            .map(|(idx, stored)| Item {
457                id: format!("item_function_call_{idx}"),
458                data: String::try_from(stored).expect("serialization failed"),
459                created_at: 1_704_067_200,
460                conversation_id: None,
461                seq: Some(idx.try_into().expect("seq")),
462            })
463            .map(|item| item.as_inout().expect("stored item"))
464            .collect();
465
466        let inputs = InOutItem::into_input_items(rows);
467        let values = serde_json::to_value(&inputs).expect("input values");
468
469        assert_eq!(values[0]["type"], "function_call");
470        assert_eq!(values[0]["namespace"], "mcp__shell");
471        assert_eq!(values[0]["name"], "run");
472        assert_eq!(values[0]["call_id"], "call_1");
473        assert!(values[0].get(STORED_ITEM_KIND_KEY).is_none());
474
475        assert_eq!(values[1]["type"], "function_call");
476        assert_eq!(values[1]["namespace"], "mcp__git");
477        assert_eq!(values[1]["name"], "run");
478        assert_eq!(values[1]["call_id"], "call_2");
479        assert!(values[1].get(STORED_ITEM_KIND_KEY).is_none());
480
481        println!("namespace round-trip: mcp__shell.run -> call_1");
482        println!("namespace round-trip: mcp__git.run -> call_2");
483        println!("same tool name preserved under separate namespaces");
484    }
485
486    #[test]
487    fn test_unknown_rehydrated_items_are_omitted() {
488        let stored = InOutItem::Output(OutputItem::Unknown);
489        let item = Item {
490            id: "item_unknown".to_string(),
491            data: String::try_from(&stored).expect("serialization failed"),
492            created_at: 1_704_067_200,
493            conversation_id: None,
494            seq: None,
495        };
496
497        let inputs = InOutItem::into_input_items(vec![item.as_inout().expect("stored item")]);
498
499        assert!(inputs.is_empty());
500    }
501}
agentic_core/storage/models/item.rs

agentic_core/storage/models/
item.rs
