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    /// Use the exact successor's Root capability for a response CAS. External
18    /// adapters without the named writer reject a Root handoff before consume.
19    fn bind_response_writer(
20        &self,
21        owner: Option<bamboo_domain::RootActorRuntimeWrite>,
22    ) -> Result<Option<std::sync::Arc<dyn SessionAccess>>, SessionSaveError> {
23        if owner.is_some() {
24            return Err(SessionSaveError::StorageError(
25                "Root response writer is unsupported".into(),
26            ));
27        }
28        Ok(None)
29    }
30
31    /// Load a session by ID (from cache or storage).
32    async fn load_session(&self, id: &str) -> Result<Option<Session>, SessionLoadError>;
33
34    /// Load an existing session or create a new one with the given model.
35    async fn load_or_create(&self, id: &str, model: &str) -> Result<Session, SessionLoadError>;
36
37    /// Load a session, merging memory and storage using a preference heuristic.
38    ///
39    /// Prefers storage when it has a pending question or newer `updated_at`.
40    async fn load_merged(&self, id: &str) -> Result<Option<Session>, SessionLoadError>;
41
42    /// Save a session to persistent storage only.
43    ///
44    /// Implementations may merge concurrent UI edits to
45    /// title/title_generated/pinned/title_version
46    /// from disk back into `session` (which is why this takes `&mut`).
47    async fn save_session(&self, session: &mut Session) -> Result<(), SessionSaveError>;
48
49    /// Save a session to persistent storage and update the in-memory cache.
50    ///
51    /// Implementations may merge concurrent UI edits to
52    /// title/title_generated/pinned/title_version
53    /// from disk back into `session` (which is why this takes `&mut`).
54    async fn save_and_cache(&self, session: &mut Session) -> Result<(), SessionSaveError>;
55
56    /// Inspect the authoritative snapshot used by a pending-response
57    /// transaction, without mutating it. Canonical repositories override this
58    /// so durable consumed-response state wins over a stale cache candidate.
59    async fn inspect_for_response(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
60        self.load_merged(id).await
61    }
62
63    /// Compare/mutate/persist the latest session for a pending response.
64    /// Implementations backed by a canonical per-session lock override this so
65    /// the whole operation is atomic. The default preserves compatibility for
66    /// lightweight/in-memory adapters; the respond use case also serializes
67    /// its entrypoints so those adapters cannot double-consume concurrently.
68    async fn mutate_for_response(
69        &self,
70        id: &str,
71        mutate: Box<
72            dyn for<'session> FnOnce(&'session mut Session) -> Result<(), RespondError>
73                + Send
74                + 'static,
75        >,
76    ) -> Result<Option<Session>, RespondError> {
77        let Some(mut session) = self.load_merged(id).await? else {
78            return Ok(None);
79        };
80        mutate(&mut session)?;
81        self.save_and_cache(&mut session).await?;
82        Ok(Some(session))
83    }
84}
85
86/// The framework-owned [`SessionRepository`] is the canonical `SessionAccess`.
87/// Server `AppState` delegates to its `session_repo`; SDK / in-process callers
88/// can use a `SessionRepository` directly as a `SessionAccess`.
89#[async_trait]
90impl SessionAccess for SessionRepository {
91    fn bind_response_writer(
92        &self,
93        owner: Option<bamboo_domain::RootActorRuntimeWrite>,
94    ) -> Result<Option<std::sync::Arc<dyn SessionAccess>>, SessionSaveError> {
95        owner
96            .map(|owner| {
97                self.bind_root_response_writer(owner)
98                    .map(|bound| std::sync::Arc::new(bound) as std::sync::Arc<dyn SessionAccess>)
99                    .map_err(|error| SessionSaveError::StorageError(error.to_string()))
100            })
101            .transpose()
102    }
103
104    async fn load_session(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
105        // Historical contract: absence is an error, not Ok(None).
106        match SessionRepository::load(self, id).await {
107            Some(session) => Ok(Some(session)),
108            None => Err(SessionLoadError::NotFound(id.to_string())),
109        }
110    }
111
112    async fn load_or_create(&self, id: &str, model: &str) -> Result<Session, SessionLoadError> {
113        Ok(SessionRepository::load_or_create(self, id, model).await)
114    }
115
116    async fn load_merged(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
117        SessionRepository::load_merged_checked(self, id)
118            .await
119            .map_err(|error| SessionLoadError::StorageError(error.to_string()))
120    }
121
122    async fn save_session(&self, session: &mut Session) -> Result<(), SessionSaveError> {
123        // Storage-only persist (no cache write), matching the trait contract.
124        self.persistence()
125            .merge_save_runtime(session)
126            .await
127            .map_err(|e| SessionSaveError::StorageError(e.to_string()))
128    }
129
130    async fn save_and_cache(&self, session: &mut Session) -> Result<(), SessionSaveError> {
131        SessionRepository::save_and_cache(self, session).await;
132        Ok(())
133    }
134
135    async fn inspect_for_response(&self, id: &str) -> Result<Option<Session>, SessionLoadError> {
136        let cache = self.cache().clone();
137        let session_id = id.to_string();
138        self.persistence()
139            .inspect_runtime_session_for_response(id, move || {
140                crate::read_cached_session(&cache, &session_id)
141            })
142            .await
143            .map_err(|error| SessionLoadError::StorageError(error.to_string()))
144    }
145
146    async fn mutate_for_response(
147        &self,
148        id: &str,
149        mutate: Box<
150            dyn for<'session> FnOnce(&'session mut Session) -> Result<(), RespondError>
151                + Send
152                + 'static,
153        >,
154    ) -> Result<Option<Session>, RespondError> {
155        let cache_for_load = self.cache().clone();
156        let publish_cache = self.cache().clone();
157        let session_id = id.to_string();
158        match self
159            .persistence()
160            .mutate_runtime_session_and_publish(
161                id,
162                move || crate::read_cached_session(&cache_for_load, &session_id),
163                move |session| mutate(session),
164                move |saved| {
165                    publish_cache.insert(
166                        saved.id.clone(),
167                        std::sync::Arc::new(crate::SessionSnapshot::new(saved.clone())),
168                    );
169                },
170            )
171            .await
172        {
173            Ok(Ok(session)) => Ok(session),
174            Ok(Err(error)) => Err(error),
175            Err(error) => Err(RespondError::SaveFailed(SessionSaveError::StorageError(
176                error.to_string(),
177            ))),
178        }
179    }
180}
bamboo_engine/session_app/repository.rs

bamboo_engine/session_app/
repository.rs
