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            cancelled,
61            None,
62        ))
63    }
64
65    fn config_result(
66        &self,
67        command_id: String,
68    ) -> mj_client::session::BoxFuture<'_, Result<Option<Option<String>>>> {
69        let session_id = self.session_id().to_owned();
70        Box::pin(async move {
71            tokio::task::spawn_blocking(move || {
72                crate::database::load_config_result(&session_id, &command_id)
73            })
74            .await
75            .context("read configuration completion task")?
76        })
77    }
78
79    fn clone_box(&self) -> Box<dyn mj_client::session::SessionHandleBackend> {
80        Box::new(self.clone())
81    }
82
83    fn session_id(&self) -> &str {
84        self.0.session_id()
85    }
86
87    fn view(&self) -> ManagedSessionView {
88        self.0.view()
89    }
90
91    fn is_stopped(&self) -> bool {
92        self.0.is_stopped()
93    }
94
95    fn has_changed(&self) -> Result<bool> {
96        self.0.has_changed()
97    }
98
99    fn changed(&mut self) -> mj_client::session::BoxFuture<'_, Result<ManagedSessionView>> {
100        Box::pin(self.0.changed())
101    }
102
103    fn enqueue_submit(
104        &self,
105        command_id: String,
106        command: RelayCommand,
107    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::PendingRelaySubmit>> {
108        Box::pin(async move {
109            let pending = self.0.enqueue_submit(command_id, command).await?;
110            Ok(mj_client::session::PendingRelaySubmit::new(Box::pin(
111                pending.wait(),
112            )))
113        })
114    }
115
116    fn enqueue_sync(
117        &self,
118    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::PendingRelaySync>> {
119        Box::pin(async move {
120            let pending = self.0.enqueue_sync().await?;
121            Ok(mj_client::session::PendingRelaySync::new(Box::pin(
122                pending.wait(),
123            )))
124        })
125    }
126
127    fn respond_elicitation(
128        &self,
129        elicitation_id: String,
130        response: ElicitationResponse,
131    ) -> mj_client::session::BoxFuture<'_, Result<()>> {
132        Box::pin(self.0.respond_elicitation(elicitation_id, response))
133    }
134
135    fn stop_background_task(
136        &self,
137        background_task_id: String,
138    ) -> mj_client::session::BoxFuture<'_, Result<()>> {
139        Box::pin(self.0.stop_background_task(background_task_id))
140    }
141
142    fn install_github_token(&self, token: String) -> mj_client::session::BoxFuture<'_, Result<()>> {
143        Box::pin(self.0.install_github_token(token))
144    }
145
146    fn reviewer(
147        &self,
148        role: Option<String>,
149        action: ReviewerAction,
150    ) -> mj_client::session::BoxFuture<'_, Result<ReviewerOutcome>> {
151        Box::pin(self.0.reviewer_as(role, action))
152    }
153}
154
155#[derive(Clone)]
156pub(super) struct ClientSessionControl(pub(super) SessionManagerControl);
157
158impl mj_client::session::SessionControlBackend for ClientSessionControl {
159    fn session(
160        &self,
161        session_id: String,
162    ) -> mj_client::session::BoxFuture<'_, Result<mj_client::session::SessionHandle>> {
163        Box::pin(async move { Ok(self.0.session(session_id).await?.client()) })
164    }
165}
166
167impl SessionManagerControl {
168    /// Narrow this controller-owned manager to session lookup for a control surface.
169    pub fn client(&self) -> mj_client::session::SessionControl {
170        mj_client::session::SessionControl::new(ClientSessionControl(self.clone()))
171    }
172
173    pub async fn session(&self, session_id: impl Into<String>) -> Result<ManagedSessionHandle> {
174        let session_id = session_id.into();
175        self.find_session(session_id.clone())
176            .await?
177            .with_context(|| format!("session {session_id} is not managed"))
178    }
179
180    /// The session's handle, or `None` while it has no actor.
181    pub async fn find_session(
182        &self,
183        session_id: impl Into<String>,
184    ) -> Result<Option<ManagedSessionHandle>> {
185        let (reply, response) = oneshot::channel();
186        self.commands
187            .send(ManagerCommand::Session {
188                session_id: session_id.into(),
189                reply,
190            })
191            .await
192            .context("session manager stopped")?;
193        response.await.context("session manager stopped")
194    }
195
196    pub async fn wait_for_session(
197        &self,
198        session_id: &str,
199        timeout: Duration,
200    ) -> Result<ManagedSessionHandle> {
201        tokio::time::timeout(timeout, async {
202            loop {
203                match self.session(session_id.to_owned()).await {
204                    Ok(handle) => return Ok(handle),
205                    Err(error) => {
206                        tracing::trace!(session_id, "waiting for session actor: {error:#}");
207                        tokio::time::sleep(Duration::from_millis(25)).await;
208                    }
209                }
210            }
211        })
212        .await
213        .with_context(|| {
214            format!(
215                "session {session_id} did not become available within {} seconds",
216                timeout.as_secs()
217            )
218        })?
219    }
220}
mj_controller/session_manager/client_backend.rs

mj_controller/session_manager/
client_backend.rs
