mj_controller/session_manager/
client_backend.rs

1use super::*;
2
3#[derive(Clone)]
4pub(super) struct ClientSessionHandle(pub(super) ManagedSessionHandle);
5
6impl mj_client::session::SessionHandleBackend for ClientSessionHandle {
7    fn transcript_history(
8        &self,
9        before: Option<mj_core::storage::TranscriptCursor>,
10    ) -> mj_client::session::BoxFuture<'_, Result<mj_core::storage::TranscriptHistoryPage>> {
11        let session_id = self.0.session_id().to_owned();
12        Box::pin(async move {
13            tokio::task::spawn_blocking(move || {
14                crate::database::load_transcript_history(&session_id, before.as_ref(), 128)?
15                    .context("session history is unavailable")
16            })
17            .await
18            .context("earlier conversation history task failed")?
19        })
20    }
21
22    fn search_prompts(
23        &self,
24        bundle_id: String,
25        scope: mj_core::storage::HistoryScope,
26        query: String,
27    ) -> mj_client::session::BoxFuture<'_, Result<Vec<mj_core::storage::PromptHistoryEntry>>> {
28        let session_id = self.0.session_id().to_owned();
29        Box::pin(async move {
30            tokio::task::spawn_blocking(move || {
31                crate::database::search_prompts(&session_id, &bundle_id, scope, &query)
32            })
33            .await
34            .context("history search task")?
35        })
36    }
37    fn review_state(
38        &self,
39    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::ReviewState>> {
40        let session_id = self.0.session_id().to_owned();
41        Box::pin(async move {
42            tokio::task::spawn_blocking(move || {
43                Ok(mj_client::session::ReviewState {
44                    review: crate::database::active_review(&session_id)?,
45                })
46            })
47            .await
48            .context("review restoration task")?
49        })
50    }
51
52    fn resolve_review_settings(
53        &self,
54        cancelled: std::sync::Arc<std::sync::atomic::AtomicBool>,
55    ) -> mj_client::session::BoxFuture<'_, Result<mj_core::review::settings::ResolvedReviewSettings>>
56    {
57        Box::pin(crate::review_selection::resolve(
58            self.0.clone(),
59            None,
60            false,
61            cancelled,
62            None,
63        ))
64    }
65
66    fn config_result(
67        &self,
68        command_id: String,
69    ) -> mj_client::session::BoxFuture<'_, Result<Option<Option<String>>>> {
70        let session_id = self.session_id().to_owned();
71        Box::pin(async move {
72            tokio::task::spawn_blocking(move || {
73                crate::database::load_config_result(&session_id, &command_id)
74            })
75            .await
76            .context("read configuration completion task")?
77        })
78    }
79
80    fn clone_box(&self) -> Box<dyn mj_client::session::SessionHandleBackend> {
81        Box::new(self.clone())
82    }
83
84    fn session_id(&self) -> &str {
85        self.0.session_id()
86    }
87
88    fn view(&self) -> ManagedSessionView {
89        self.0.view()
90    }
91
92    fn is_stopped(&self) -> bool {
93        self.0.is_stopped()
94    }
95
96    fn has_changed(&self) -> Result<bool> {
97        self.0.has_changed()
98    }
99
100    fn changed(&mut self) -> mj_client::session::BoxFuture<'_, Result<ManagedSessionView>> {
101        Box::pin(self.0.changed())
102    }
103
104    fn enqueue_submit(
105        &self,
106        command_id: String,
107        command: RelayCommand,
108    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::PendingRelaySubmit>> {
109        Box::pin(async move {
110            let pending = self.0.enqueue_submit(command_id, command).await?;
111            Ok(mj_client::session::PendingRelaySubmit::new(Box::pin(
112                pending.wait(),
113            )))
114        })
115    }
116
117    fn enqueue_sync(
118        &self,
119    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::PendingRelaySync>> {
120        Box::pin(async move {
121            let pending = self.0.enqueue_sync().await?;
122            Ok(mj_client::session::PendingRelaySync::new(Box::pin(
123                pending.wait(),
124            )))
125        })
126    }
127
128    fn respond_elicitation(
129        &self,
130        elicitation_id: String,
131        response: ElicitationResponse,
132    ) -> mj_client::session::BoxFuture<'_, Result<()>> {
133        Box::pin(self.0.respond_elicitation(elicitation_id, response))
134    }
135
136    fn stop_background_task(
137        &self,
138        background_task_id: String,
139    ) -> mj_client::session::BoxFuture<'_, Result<()>> {
140        Box::pin(self.0.stop_background_task(background_task_id))
141    }
142
143    fn reviewer(
144        &self,
145        role: Option<String>,
146        action: ReviewerAction,
147    ) -> mj_client::session::BoxFuture<'_, Result<ReviewerOutcome>> {
148        Box::pin(self.0.reviewer_as(role, action))
149    }
150}
151
152#[derive(Clone)]
153pub(super) struct ClientSessionControl(pub(super) SessionManagerControl);
154
155impl mj_client::session::SessionControlBackend for ClientSessionControl {
156    fn session(
157        &self,
158        session_id: String,
159    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::SessionHandle>> {
160        Box::pin(async move { Ok(self.0.session(session_id).await?.client()) })
161    }
162}
163
164impl SessionManagerControl {
165    /// Narrow this controller-owned manager to session lookup for a control surface.
166    pub fn client(&self) -> mj_client::session::SessionControl {
167        mj_client::session::SessionControl::new(ClientSessionControl(self.clone()))
168    }
169
170    pub async fn session(&self, session_id: impl Into<String>) -> Result<ManagedSessionHandle> {
171        let session_id = session_id.into();
172        self.find_session(session_id.clone())
173            .await?
174            .with_context(|| format!("session {session_id} is not managed"))
175    }
176
177    /// The session's handle, or `None` while it has no actor.
178    pub async fn find_session(
179        &self,
180        session_id: impl Into<String>,
181    ) -> Result<Option<ManagedSessionHandle>> {
182        let (reply, response) = oneshot::channel();
183        self.commands
184            .send(ManagerCommand::Session {
185                session_id: session_id.into(),
186                reply,
187            })
188            .await
189            .context("session manager stopped")?;
190        response.await.context("session manager stopped")
191    }
192
193    pub async fn wait_for_session(
194        &self,
195        session_id: &str,
196        timeout: Duration,
197    ) -> Result<ManagedSessionHandle> {
198        tokio::time::timeout(timeout, async {
199            loop {
200                match self.session(session_id.to_owned()).await {
201                    Ok(handle) => return Ok(handle),
202                    Err(error) => {
203                        tracing::trace!(session_id, "waiting for session actor: {error:#}");
204                        tokio::time::sleep(Duration::from_millis(25)).await;
205                    }
206                }
207            }
208        })
209        .await
210        .with_context(|| {
211            format!(
212                "session {session_id} did not become available within {} seconds",
213                timeout.as_secs()
214            )
215        })?
216    }
217}
mj_controller/session_manager/client_backend.rs

mj_controller/session_manager/
client_backend.rs
