mj_controller/daemon/
create.rs

1use super::*;
2
3/// How long creating a parent's report root on its target may take.
4const REPORT_ROOT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
5
6impl RuntimeState {
7    pub(super) async fn start_create_session(
8        self: &Arc<Self>,
9        request: CreateSessionRequest,
10    ) -> Result<RegisteredSession> {
11        self.start_create_session_inner(request, CreateSessionControl::default(), None)
12            .await
13    }
14
15    /// Register and start a child worker on its parent's existing target.
16    pub async fn start_subagent_session(
17        self: &Arc<Self>,
18        request: crate::controller::RegisterSubagentRequest,
19    ) -> Result<mj_core::subagent::SubagentRecord> {
20        let _upgrade_work = crate::upgrade::activity("subagent admission")?;
21        let (relation, needs_provisioning) = blocking(move || {
22            let mut controller = Controller::load()?;
23            if let Some(existing) = crate::database::lookup_subagent_request(
24                &request.parent_session_id,
25                &request.request_key,
26            )? {
27                let needs_provisioning = controller
28                    .state
29                    .sessions
30                    .get(&existing.child_session_id)
31                    .is_some_and(|record| {
32                        record.state == mj_core::state::SessionState::Provisioning
33                    });
34                return Ok((existing, needs_provisioning));
35            }
36            let mut request = request;
37            // The parent's report root is made before the child exists, so the
38            // child's first prompt can name its own directory under it. A
39            // request run again after a restart finds the same root.
40            let executor = CancellableProcessExecutor::with_timeout(REPORT_ROOT_TIMEOUT);
41            request.report_root = Some(
42                controller
43                    .prepare_subagent_report_root(&request.parent_session_id, &executor)
44                    .context("create the sub-agent report directory")?,
45            );
46            controller
47                .register_subagent(request)
48                .map(|relation| (relation, true))
49        })
50        .await?;
51        if !needs_provisioning {
52            return Ok(relation);
53        }
54        let session_id = relation.child_session_id.clone();
55        self.start_or_join_lifecycle_controlled(
56            session_id.clone(),
57            LifecycleKind::Create,
58            None,
59            Some(relation.request_key.clone()),
60            None,
61            move |state, session_id, cancelled| async move {
62                let mut controller = tokio::task::spawn_blocking(Controller::load)
63                    .await
64                    .context("load controller for sub-agent startup")??;
65                // The lifecycle owner now excludes competing resume/close work.
66                // Recovery may already have settled this registration before
67                // admission; replay never reinstalls an existing worker.
68                if controller
69                    .state
70                    .sessions
71                    .get(&session_id)
72                    .is_none_or(|record| record.state != mj_core::state::SessionState::Provisioning)
73                {
74                    return Ok(DaemonLifecycleResult::Done);
75                }
76                let executor = DaemonStageReportingExecutor::new(
77                    CancellableProcessExecutor::new(cancelled),
78                    state,
79                    session_id.clone(),
80                );
81                controller
82                    .provision_subagent_session_controlled(&session_id, &executor)
83                    .await?;
84                Ok(DaemonLifecycleResult::Done)
85            },
86        )?;
87        self.reload_controller().await?;
88        Ok(relation)
89    }
90
91    pub async fn start_create_session_controlled(
92        self: &Arc<Self>,
93        request: CreateSessionRequest,
94        control: CreateSessionControl,
95        publication: tokio::sync::oneshot::Receiver<std::result::Result<(), String>>,
96    ) -> Result<RegisteredSession> {
97        self.start_create_session_inner(request, control, Some(publication))
98            .await
99    }
100
101    pub(super) async fn start_create_session_inner(
102        self: &Arc<Self>,
103        request: CreateSessionRequest,
104        control: CreateSessionControl,
105        publication: Option<tokio::sync::oneshot::Receiver<std::result::Result<(), String>>>,
106    ) -> Result<RegisteredSession> {
107        let mut request = request;
108        let parent = request.profile_id.clone();
109        let supplied = request.subagents.clone();
110        let (config, policy) = blocking(move || {
111            let controller = Controller::load()?;
112            let profile = controller
113                .config
114                .enabled_profile(&parent)
115                .context("parent profile unavailable")?;
116            let policy = supplied.unwrap_or_else(|| profile.subagents.clone());
117            Ok((controller.config, policy))
118        })
119        .await?;
120        crate::controller::profile_config::validate_session_subagent_policy(
121            &config,
122            &request.profile_id,
123            &policy,
124        )
125        .await?;
126        // Discovery is restartable preparation, not admitted lifecycle work.
127        let _upgrade_work = crate::upgrade::activity("session admission")?;
128        request.subagents = Some(policy);
129        let path_cancelled = control.cancelled.clone();
130        let registered = blocking(move || {
131            let mut controller = Controller::load()?;
132            let path_executor = crate::targets::CancellableProcessExecutor::new(path_cancelled)
133                .with_deadline(Duration::from_secs(30));
134            let project_directory = request
135                .project_directory
136                .as_deref()
137                .map(|path| {
138                    controller.resolve_project_directory(
139                        &request.target_template_id,
140                        path,
141                        &path_executor,
142                    )
143                })
144                .transpose()?;
145            let session_id = controller.register_session_with_resources(
146                &request.profile_id,
147                &request.bundle_id,
148                &request.target_template_id,
149                request.title,
150                SessionLaunchOptions {
151                    create_managed_worktree: request.create_managed_worktree,
152                    at: request.at,
153                    branch: request.branch,
154                    base: request.base,
155                    subagents: request.subagents,
156                    initial_prompt: request.initial_prompt,
157                    workspace_id: request.workspace_id,
158                    additional_mounts: request.additional_mounts,
159                    resource_allocation: request.resource_allocation,
160                    project_directory,
161                    session_title_override: request.session_title_override,
162                },
163            )?;
164            // Every surface creates sessions through here, so the dashboard,
165            // the phone, `mj new`, and `mj acp` all leave a default pair
166            // behind for the next caller that names none. A preference that
167            // cannot be written does not undo a session that was created.
168            if let Err(error) = mj_core::go::GoPreferences::remember_first_pair(
169                &mj_core::go::GoPreferences::path(),
170                &request.profile_id,
171                &request.target_template_id,
172            ) {
173                tracing::warn!(%error, "could not save the default profile and target");
174            }
175            let session = controller
176                .state
177                .sessions
178                .get(&session_id)
179                .expect("newly registered session exists")
180                .clone();
181            let remembered_container_size = controller
182                .config
183                .targets
184                .get(&request.target_template_id)
185                .and_then(mj_core::config::container_size_host)
186                .and_then(|host| {
187                    controller
188                        .state
189                        .container_sizes
190                        .get(host)
191                        .copied()
192                        .map(|size| (host.to_owned(), size))
193                });
194            Ok(RegisteredSession {
195                session,
196                remembered_container_size,
197            })
198        })
199        .await?;
200        let session_id = registered.session.id.clone();
201        self.start_or_join_lifecycle_controlled(
202            session_id,
203            LifecycleKind::Create,
204            None,
205            None,
206            Some(control.clone()),
207            move |state, session_id, cancelled| async move {
208                let mut controller = tokio::task::spawn_blocking(Controller::load)
209                    .await
210                    .context("load controller for daemon create task")??;
211                let publication_error = if let Some(publication) = publication {
212                    let published = tokio::select! {
213                        result = publication => result.context("session publication owner stopped")
214                            .and_then(|result| result.map_err(anyhow::Error::msg)),
215                        () = async {
216                            while !cancelled.load(Ordering::Acquire) {
217                                tokio::time::sleep(Duration::from_millis(25)).await;
218                            }
219                        } => Err(anyhow!("session creation cancelled before publication")),
220                    };
221                    published.err()
222                } else {
223                    None
224                };
225                if publication_error.is_some() {
226                    control.request_cancel();
227                }
228                let executor = DaemonStageReportingExecutor::new(
229                    CancellableProcessExecutor::new(cancelled),
230                    state,
231                    session_id.clone(),
232                );
233                let provision = controller
234                    .provision_session_controlled_with_commit(&session_id, &executor, || {
235                        ensure!(
236                            control.grant_commit(),
237                            "session creation cancelled before commit"
238                        );
239                        Ok(())
240                    })
241                    .await;
242                if let Some(error) = publication_error {
243                    return match provision {
244                        Ok(()) => Err(error),
245                        Err(rollback) => {
246                            Err(error.context(format!("discard unpublished session: {rollback:#}")))
247                        }
248                    };
249                }
250                provision?;
251                Ok(DaemonLifecycleResult::Done)
252            },
253        )?;
254        self.reload_controller().await?;
255        Ok(registered)
256    }
257
258    pub async fn wait_create_session(&self, session_id: &str) -> Result<()> {
259        let result = {
260            let lifecycle_owner = self.owner();
261            let lifecycle = &lifecycle_owner.lifecycle;
262            let active = lifecycle
263                .get(session_id)
264                .with_context(|| format!("no create operation exists for session {session_id}"))?;
265            ensure!(
266                active.kind == LifecycleKind::Create,
267                "session {session_id} is no longer being created"
268            );
269            active.result.clone()
270        };
271        let channel = result.clone();
272        let outcome = Self::wait_lifecycle_result(result).await;
273        self.remove_completed_lifecycle(&channel);
274        match outcome? {
275            DaemonLifecycleResult::Done => Ok(()),
276            DaemonLifecycleResult::Move(_) | DaemonLifecycleResult::Park(_) => {
277                unreachable!("cleanup cannot return a move outcome")
278            }
279            DaemonLifecycleResult::DeferredCleanup => {
280                unreachable!("session creation cannot schedule target cleanup")
281            }
282        }
283    }
284}
285
286#[cfg(test)]
287mod delegation_replay_tests {
288    use crate::controller::test_support::{IsolatedTest, test_name};
289
290    #[tokio::test]
291    async fn multi_model_creation_is_refused_before_registration_or_provisioning() {
292        const CHILD: &str = "MJ_TEST_MULTI_MODEL_CREATION_REFUSAL";
293        if std::env::var_os(CHILD).is_none() {
294            let root = tempfile::tempdir().unwrap();
295            IsolatedTest::new(test_name(
296                module_path!(),
297                "multi_model_creation_is_refused_before_registration_or_provisioning",
298            ))
299            .env(CHILD, "1")
300            .env("MJ_INSTANCE", "multi-model-creation-refusal")
301            .isolated_store(root.path())
302            .run();
303            return;
304        }
305        let _writer = crate::database::install_isolated_test_writer();
306        let mut config = mj_core::config::Config::default();
307        config.profiles.insert(
308            "codex".into(),
309            mj_core::config::HarnessProfile {
310                enabled: true,
311                kind: mj_core::config::HarnessKind::Codex,
312                home: mj_core::config::data_dir().join("missing-profile-home"),
313                environment: Default::default(),
314                context_window_bytes: None,
315                guardian_review_model: None,
316                subagents: mj_core::subagent::SubagentPolicy::Native,
317            },
318        );
319        config.save().unwrap();
320        let workspace = crate::database::create_workspace("legacy delegation").unwrap();
321        let mut legacy = crate::daemon::tests::runtime_test_session(
322            "legacy-parent",
323            &workspace.id,
324            mj_core::state::SessionState::Running,
325        );
326        legacy.subagents = Some(mj_core::subagent::SubagentPolicy::AllModels);
327        crate::database::save_session(&legacy).unwrap();
328        let runtime = crate::daemon::tests::test_runtime_state();
329        let request = mj_client::daemon::CreateSessionRequest {
330            create_managed_worktree: None,
331            at: None,
332            branch: None,
333            base: None,
334            subagents: Some(mj_core::subagent::SubagentPolicy::AllModels),
335            initial_prompt: None,
336            workspace_id: workspace.id,
337            profile_id: "codex".into(),
338            bundle_id: "missing-bundle".into(),
339            project_directory: None,
340            target_template_id: "missing-target".into(),
341            additional_mounts: Vec::new(),
342            resource_allocation: None,
343            title: "must not register".into(),
344            session_title_override: None,
345        };
346        let error = runtime
347            .start_create_session(request)
348            .await
349            .expect_err("a refusal");
350        let refusal = mj_core::refusal::Refusal::of(&error).expect("a request refusal");
351        assert!(refusal.message().contains("no longer available"));
352        assert!(runtime.active_lifecycles().is_empty());
353        let state = crate::database::load_state().unwrap();
354        assert_eq!(state.sessions.len(), 1);
355        assert_eq!(state.sessions[&legacy.id], legacy);
356    }
357
358    #[tokio::test]
359    async fn replayed_spawn_does_not_reprovision_an_existing_child() {
360        const CHILD: &str = "MJ_TEST_SPAWN_REPLAY";
361        if std::env::var_os(CHILD).is_none() {
362            let root = tempfile::tempdir().unwrap();
363            IsolatedTest::new(test_name(
364                module_path!(),
365                "replayed_spawn_does_not_reprovision_an_existing_child",
366            ))
367            .env(CHILD, "1")
368            .env("MJ_INSTANCE", "concurrency-sweep-spawn")
369            .isolated_store(root.path())
370            .run();
371            return;
372        }
373        let _writer = crate::database::install_isolated_test_writer();
374        let workspace = crate::database::create_workspace("spawn replay").unwrap();
375        let parent = crate::daemon::tests::runtime_test_session(
376            "parent",
377            &workspace.id,
378            mj_core::state::SessionState::Running,
379        );
380        crate::database::save_session(&parent).unwrap();
381        for (index, status) in [
382            mj_core::state::SessionState::Running,
383            mj_core::state::SessionState::Parked,
384            mj_core::state::SessionState::Stopped,
385        ]
386        .into_iter()
387        .enumerate()
388        {
389            let child_id = format!("child-{index}");
390            let child =
391                crate::daemon::tests::runtime_test_session(&child_id, &workspace.id, status);
392            let relation = crate::daemon::tests::runtime_test_subagent(&child_id, "parent");
393            crate::database::save_subagent_session(&child, &relation).unwrap();
394            let runtime = crate::daemon::tests::test_runtime_state();
395            let replayed = runtime
396                .start_subagent_session(crate::controller::RegisterSubagentRequest {
397                    parent_session_id: "parent".into(),
398                    task_name: "must reuse original".into(),
399                    profile_id: "missing-profile".into(),
400                    model: None,
401                    effort: None,
402                    working_directory: Default::default(),
403                    initial_prompt: "must not resend".into(),
404                    request_key: relation.request_key.clone(),
405                    report_root: None,
406                })
407                .await
408                .unwrap();
409            assert_eq!(replayed, relation);
410            assert!(runtime.active_lifecycles().is_empty());
411            assert_eq!(
412                crate::database::load_session_record(&child_id)
413                    .unwrap()
414                    .unwrap()
415                    .state,
416                status
417            );
418        }
419    }
420}
mj_controller/daemon/create.rs

mj_controller/daemon/
create.rs
