bamboo_engine/session_app/
repository.rs

1//! Session access trait for decoupling use cases from server infrastructure.
2
3use async_trait::async_trait;
4use bamboo_domain::Session;
5
6use super::errors::{RespondError, SessionLoadError, SessionSaveError};
7use crate::SessionRepository;
8
9/// Trait for loading and persisting sessions.
10///
11/// The canonical implementation is [`SessionRepository`] (the framework-owned
12/// coordinator). The server's `AppState` also implements it by delegating to
13/// its `session_repo`. Use cases depend on this trait rather than concrete
14/// server types.
15#[async_trait]
16pub trait SessionAccess: Send + Sync {
17    /// Load a session by ID (from cache or storage).
18    async fn load_session(&self, id: &str) -> Result<Option<Session>, SessionLoadError>;
19
20    /// Load an existing session or create a new one with the given model.
21    async fn load_or_create(&self, id: &str, model: &str) -> Result<Session, SessionLoadError>;
22
23    /// Load a session, merging memory and storage using a preference heuristic.
24    ///
25    /// Prefers storage when it has a pending question or newer `updated_at`.
26    async fn load_merged(&self, id: &str) -> Result<Option<Session>, SessionLoadError>;
27
28    /// Save a session to persistent storage only.
29    ///
30    /// Implementations may merge concurrent UI edits to
31    /// title/title_generated/pinned/title_version
32    /// from disk back into `session` (which is why this takes `&mut`).
33    async fn save_session(&self, session: &mut Session) -> Result<(), SessionSaveError>;
34
35    /// Save a session to persistent storage and update the in-memory cache.
36    ///
37    /// Implementations may merge concurrent UI edits to
38    /// title/title_generated/pinned/title_version
39    /// from disk back into `session` (which is why this takes `&mut`).
40    async fn save_and_cache(&self, session: &mut Session) -> Result<(), SessionSaveError>;
41
42    /// Inspect the authoritative snapshot used by a pending-response
43    /// transaction, without mutating it. Canonical repositories override this
44    /// so durable consumed-response state wins over a stale cache candidate.
45    async fn inspect_for_response(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
46        self.load_merged(id).await
47    }
48
49    /// Compare/mutate/persist the latest session for a pending response.
50    /// Implementations backed by a canonical per-session lock override this so
51    /// the whole operation is atomic. The default preserves compatibility for
52    /// lightweight/in-memory adapters; the respond use case also serializes
53    /// its entrypoints so those adapters cannot double-consume concurrently.
54    async fn mutate_for_response(
55        &self,
56        id: &str,
57        mutate: Box<
58            dyn for<'session> FnOnce(&'session mut Session) -> Result<(), RespondError>
59                + Send
60                + 'static,
61        >,
62    ) -> Result<Option<Session>, RespondError> {
63        let Some(mut session) = self.load_merged(id).await? else {
64            return Ok(None);
65        };
66        mutate(&mut session)?;
67        self.save_and_cache(&mut session).await?;
68        Ok(Some(session))
69    }
70}
71
72/// The framework-owned [`SessionRepository`] is the canonical `SessionAccess`.
73/// Server `AppState` delegates to its `session_repo`; SDK / in-process callers
74/// can use a `SessionRepository` directly as a `SessionAccess`.
75#[async_trait]
76impl SessionAccess for SessionRepository {
77    async fn load_session(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
78        // Historical contract: absence is an error, not Ok(None).
79        match SessionRepository::load(self, id).await {
80            Some(session) => Ok(Some(session)),
81            None => Err(SessionLoadError::NotFound(id.to_string())),
82        }
83    }
84
85    async fn load_or_create(&self, id: &str, model: &str) -> Result<Session, SessionLoadError> {
86        Ok(SessionRepository::load_or_create(self, id, model).await)
87    }
88
89    async fn load_merged(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
90        SessionRepository::load_merged_checked(self, id)
91            .await
92            .map_err(|error| SessionLoadError::StorageError(error.to_string()))
93    }
94
95    async fn save_session(&self, session: &mut Session) -> Result<(), SessionSaveError> {
96        // Storage-only persist (no cache write), matching the trait contract.
97        self.persistence()
98            .merge_save_runtime(session)
99            .await
100            .map_err(|e| SessionSaveError::StorageError(e.to_string()))
101    }
102
103    async fn save_and_cache(&self, session: &mut Session) -> Result<(), SessionSaveError> {
104        SessionRepository::save_and_cache(self, session).await;
105        Ok(())
106    }
107
108    async fn inspect_for_response(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
109        let cache = self.cache().clone();
110        let session_id = id.to_string();
111        self.persistence()
112            .inspect_runtime_session_for_response(id, move || {
113                crate::read_cached_session(&cache, &session_id)
114            })
115            .await
116            .map_err(|error| SessionLoadError::StorageError(error.to_string()))
117    }
118
119    async fn mutate_for_response(
120        &self,
121        id: &str,
122        mutate: Box<
123            dyn for<'session> FnOnce(&'session mut Session) -> Result<(), RespondError>
124                + Send
125                + 'static,
126        >,
127    ) -> Result<Option<Session>, RespondError> {
128        let cache_for_load = self.cache().clone();
129        let publish_cache = self.cache().clone();
130        let session_id = id.to_string();
131        match self
132            .persistence()
133            .mutate_runtime_session_and_publish(
134                id,
135                move || crate::read_cached_session(&cache_for_load, &session_id),
136                move |session| mutate(session),
137                move |saved| {
138                    publish_cache.insert(
139                        saved.id.clone(),
140                        std::sync::Arc::new(parking_lot::RwLock::new(saved.clone())),
141                    );
142                },
143            )
144            .await
145        {
146            Ok(Ok(session)) => Ok(session),
147            Ok(Err(error)) => Err(error),
148            Err(error) => Err(RespondError::SaveFailed(SessionSaveError::StorageError(
149                error.to_string(),
150            ))),
151        }
152    }
153}
bamboo_engine/session_app/repository.rs

bamboo_engine/session_app/
repository.rs
