deepstrike_sdk/runtime/
session_log.rs

1use std::collections::HashMap;
2use std::path::{Path, PathBuf};
3
4use async_trait::async_trait;
5use deepstrike_core::runtime::event_log::{Primitive, primitive_for_kind};
6use deepstrike_core::runtime::session::SessionEvent;
7use serde::{Deserialize, Serialize};
8use tokio::fs;
9use tokio::io::AsyncWriteExt;
10
11#[derive(Debug, Clone, Serialize, Deserialize)]
12pub struct SessionEntry {
13    pub seq: u64,
14    pub event: SessionEvent,
15}
16
17#[async_trait]
18pub trait SessionLog: Send + Sync {
19    async fn append(&self, session_id: &str, event: SessionEvent) -> Result<u64, std::io::Error>;
20    async fn read(
21        &self,
22        session_id: &str,
23        from_seq: u64,
24        primitive_filter: Option<Primitive>,
25    ) -> Result<Vec<SessionEntry>, std::io::Error>;
26    async fn latest_seq(&self, session_id: &str) -> Result<i64, std::io::Error>;
27}
28
29pub struct InMemorySessionLog {
30    store: tokio::sync::Mutex<HashMap<String, Vec<SessionEntry>>>,
31}
32
33impl Default for InMemorySessionLog {
34    fn default() -> Self {
35        Self {
36            store: tokio::sync::Mutex::new(HashMap::new()),
37        }
38    }
39}
40
41impl InMemorySessionLog {
42    pub fn new() -> Self {
43        Self::default()
44    }
45}
46
47#[async_trait]
48impl SessionLog for InMemorySessionLog {
49    async fn append(&self, session_id: &str, event: SessionEvent) -> Result<u64, std::io::Error> {
50        let mut store = self.store.lock().await;
51        let entries = store.entry(session_id.to_string()).or_default();
52        let seq = entries.len() as u64;
53        entries.push(SessionEntry { seq, event });
54        Ok(seq)
55    }
56
57    async fn read(
58        &self,
59        session_id: &str,
60        from_seq: u64,
61        primitive_filter: Option<Primitive>,
62    ) -> Result<Vec<SessionEntry>, std::io::Error> {
63        let store = self.store.lock().await;
64        Ok(store
65            .get(session_id)
66            .map(|entries| {
67                entries
68                    .iter()
69                    .filter(|e| {
70                        if e.seq < from_seq {
71                            return false;
72                        }
73                        if let Some(pf) = primitive_filter {
74                            if primitive_for_kind(e.event.kind_str()) != pf {
75                                return false;
76                            }
77                        }
78                        true
79                    })
80                    .cloned()
81                    .collect()
82            })
83            .unwrap_or_default())
84    }
85
86    async fn latest_seq(&self, session_id: &str) -> Result<i64, std::io::Error> {
87        let store = self.store.lock().await;
88        Ok(store
89            .get(session_id)
90            .map(|e| e.len() as i64 - 1)
91            .unwrap_or(-1))
92    }
93}
94
95/// Single-writer per session. Not safe for concurrent writers across processes.
96pub struct FileSessionLog {
97    dir: PathBuf,
98    seq_counters: tokio::sync::Mutex<HashMap<String, u64>>,
99}
100
101impl FileSessionLog {
102    pub fn new(dir: impl AsRef<Path>) -> Self {
103        Self {
104            dir: dir.as_ref().to_path_buf(),
105            seq_counters: tokio::sync::Mutex::new(HashMap::new()),
106        }
107    }
108
109    fn path(&self, session_id: &str) -> PathBuf {
110        self.dir.join(format!("{session_id}.jsonl"))
111    }
112
113    async fn next_seq(&self, session_id: &str) -> Result<u64, std::io::Error> {
114        let mut counters = self.seq_counters.lock().await;
115        if !counters.contains_key(session_id) {
116            let existing = self.read(session_id, 0, None).await?;
117            counters.insert(session_id.to_string(), existing.len() as u64);
118        }
119        let seq = counters.get(session_id).copied().unwrap_or(0);
120        counters.insert(session_id.to_string(), seq + 1);
121        Ok(seq)
122    }
123}
124
125#[async_trait]
126impl SessionLog for FileSessionLog {
127    async fn append(&self, session_id: &str, event: SessionEvent) -> Result<u64, std::io::Error> {
128        fs::create_dir_all(&self.dir).await?;
129        let seq = self.next_seq(session_id).await?;
130        let line = serde_json::json!({ "seq": seq, "event": event });
131        let mut file = fs::OpenOptions::new()
132            .create(true)
133            .append(true)
134            .open(self.path(session_id))
135            .await?;
136        file.write_all(format!("{line}\n").as_bytes()).await?;
137        Ok(seq)
138    }
139
140    async fn read(
141        &self,
142        session_id: &str,
143        from_seq: u64,
144        primitive_filter: Option<Primitive>,
145    ) -> Result<Vec<SessionEntry>, std::io::Error> {
146        let path = self.path(session_id);
147        if !path.exists() {
148            return Ok(Vec::new());
149        }
150        let content = fs::read_to_string(path).await?;
151        let mut results = Vec::new();
152        for line in content.lines() {
153            if line.trim().is_empty() {
154                continue;
155            }
156            let raw: serde_json::Value = serde_json::from_str(line)?;
157            let seq = raw["seq"].as_u64().unwrap_or(0);
158            if seq < from_seq {
159                continue;
160            }
161            let event: SessionEvent = serde_json::from_value(raw["event"].clone())?;
162            if let Some(pf) = primitive_filter {
163                if primitive_for_kind(event.kind_str()) != pf {
164                    continue;
165                }
166            }
167            results.push(SessionEntry { seq, event });
168        }
169        Ok(results)
170    }
171
172    async fn latest_seq(&self, session_id: &str) -> Result<i64, std::io::Error> {
173        let entries = self.read(session_id, 0, None).await?;
174        Ok(entries.len() as i64 - 1)
175    }
176}
deepstrike_sdk/runtime/session_log.rs

deepstrike_sdk/runtime/
session_log.rs
