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                    review: request.review,
157                    initial_prompt: request.initial_prompt,
158                    workspace_id: request.workspace_id,
159                    additional_mounts: request.additional_mounts,
160                    resource_allocation: request.resource_allocation,
161                    project_directory,
162                    session_title_override: request.session_title_override,
163                },
164            )?;
165            // Every surface creates sessions through here, so the dashboard,
166            // the phone, `mj new`, and `mj acp` all leave a default pair
167            // behind for the next caller that names none. A preference that
168            // cannot be written does not undo a session that was created.
169            if let Err(error) = mj_core::go::GoPreferences::remember_first_pair(
170                &mj_core::go::GoPreferences::path(),
171                &request.profile_id,
172                &request.target_template_id,
173            ) {
174                tracing::warn!(%error, "could not save the default profile and target");
175            }
176            let session = controller
177                .state
178                .sessions
179                .get(&session_id)
180                .expect("newly registered session exists")
181                .clone();
182            let remembered_container_size = controller
183                .config
184                .targets
185                .get(&request.target_template_id)
186                .and_then(mj_core::config::container_size_host)
187                .and_then(|host| {
188                    controller
189                        .state
190                        .container_sizes
191                        .get(host)
192                        .copied()
193                        .map(|size| (host.to_owned(), size))
194                });
195            Ok(RegisteredSession {
196                session,
197                remembered_container_size,
198            })
199        })
200        .await?;
201        let session_id = registered.session.id.clone();
202        self.start_or_join_lifecycle_controlled(
203            session_id,
204            LifecycleKind::Create,
205            None,
206            None,
207            Some(control.clone()),
208            move |state, session_id, cancelled| async move {
209                let mut controller = tokio::task::spawn_blocking(Controller::load)
210                    .await
211                    .context("load controller for daemon create task")??;
212                let publication_error = if let Some(publication) = publication {
213                    let published = tokio::select! {
214                        result = publication => result.context("session publication owner stopped")
215                            .and_then(|result| result.map_err(anyhow::Error::msg)),
216                        () = async {
217                            while !cancelled.load(Ordering::Acquire) {
218                                tokio::time::sleep(Duration::from_millis(25)).await;
219                            }
220                        } => Err(anyhow!("session creation cancelled before publication")),
221                    };
222                    published.err()
223                } else {
224                    None
225                };
226                if publication_error.is_some() {
227                    control.request_cancel();
228                }
229                let executor = DaemonStageReportingExecutor::new(
230                    CancellableProcessExecutor::new(cancelled),
231                    state,
232                    session_id.clone(),
233                );
234                let provision = controller
235                    .provision_session_controlled_with_commit(&session_id, &executor, || {
236                        ensure!(
237                            control.grant_commit(),
238                            "session creation cancelled before commit"
239                        );
240                        Ok(())
241                    })
242                    .await;
243                if let Some(error) = publication_error {
244                    return match provision {
245                        Ok(()) => Err(error),
246                        Err(rollback) => {
247                            Err(error.context(format!("discard unpublished session: {rollback:#}")))
248                        }
249                    };
250                }
251                provision?;
252                Ok(DaemonLifecycleResult::Done)
253            },
254        )?;
255        self.reload_controller().await?;
256        Ok(registered)
257    }
258
259    pub async fn wait_create_session(&self, session_id: &str) -> Result<()> {
260        let result = {
261            let lifecycle_owner = self.owner();
262            let lifecycle = &lifecycle_owner.lifecycle;
263            let active = lifecycle
264                .get(session_id)
265                .with_context(|| format!("no create operation exists for session {session_id}"))?;
266            ensure!(
267                active.kind == LifecycleKind::Create,
268                "session {session_id} is no longer being created"
269            );
270            active.result.clone()
271        };
272        let channel = result.clone();
273        let outcome = Self::wait_lifecycle_result(result).await;
274        self.remove_completed_lifecycle(&channel);
275        match outcome? {
276            DaemonLifecycleResult::Done => Ok(()),
277            DaemonLifecycleResult::Move(_) | DaemonLifecycleResult::Park(_) => {
278                unreachable!("cleanup cannot return a move outcome")
279            }
280            DaemonLifecycleResult::DeferredCleanup => {
281                unreachable!("session creation cannot schedule target cleanup")
282            }
283        }
284    }
285}
286
287#[cfg(test)]
288mod delegation_replay_tests {
289    use crate::controller::test_support::{IsolatedTest, test_name};
290
291    #[tokio::test]
292    async fn multi_model_creation_is_refused_before_registration_or_provisioning() {
293        const CHILD: &str = "MJ_TEST_MULTI_MODEL_CREATION_REFUSAL";
294        if std::env::var_os(CHILD).is_none() {
295            let root = tempfile::tempdir().unwrap();
296            IsolatedTest::new(test_name(
297                module_path!(),
298                "multi_model_creation_is_refused_before_registration_or_provisioning",
299            ))
300            .env(CHILD, "1")
301            .env("MJ_INSTANCE", "multi-model-creation-refusal")
302            .isolated_store(root.path())
303            .run();
304            return;
305        }
306        let _writer = crate::database::install_isolated_test_writer();
307        let mut config = mj_core::config::Config::default();
308        config.profiles.insert(
309            "codex".into(),
310            mj_core::config::HarnessProfile {
311                enabled: true,
312                kind: mj_core::config::HarnessKind::Codex,
313                home: mj_core::config::data_dir().join("missing-profile-home"),
314                environment: Default::default(),
315                context_window_bytes: None,
316                guardian_review_model: None,
317                subagents: mj_core::subagent::SubagentPolicy::Native,
318            },
319        );
320        config.save().unwrap();
321        let workspace = crate::database::create_workspace("legacy delegation").unwrap();
322        let mut legacy = crate::daemon::tests::runtime_test_session(
323            "legacy-parent",
324            &workspace.id,
325            mj_core::state::SessionState::Running,
326        );
327        legacy.subagents = Some(mj_core::subagent::SubagentPolicy::AllModels);
328        crate::database::save_session(&legacy).unwrap();
329        let runtime = crate::daemon::tests::test_runtime_state();
330        let request = mj_client::daemon::CreateSessionRequest {
331            create_managed_worktree: None,
332            at: None,
333            branch: None,
334            base: None,
335            subagents: Some(mj_core::subagent::SubagentPolicy::AllModels),
336            review: None,
337            initial_prompt: None,
338            workspace_id: workspace.id,
339            profile_id: "codex".into(),
340            bundle_id: "missing-bundle".into(),
341            project_directory: None,
342            target_template_id: "missing-target".into(),
343            additional_mounts: Vec::new(),
344            resource_allocation: None,
345            title: "must not register".into(),
346            session_title_override: None,
347        };
348        let error = runtime
349            .start_create_session(request)
350            .await
351            .expect_err("a refusal");
352        let refusal = mj_core::refusal::Refusal::of(&error).expect("a request refusal");
353        assert!(refusal.message().contains("no longer available"));
354        assert!(runtime.active_lifecycles().is_empty());
355        let state = crate::database::load_state().unwrap();
356        assert_eq!(state.sessions.len(), 1);
357        assert_eq!(state.sessions[&legacy.id], legacy);
358    }
359
360    #[tokio::test]
361    async fn replayed_spawn_does_not_reprovision_an_existing_child() {
362        const CHILD: &str = "MJ_TEST_SPAWN_REPLAY";
363        if std::env::var_os(CHILD).is_none() {
364            let root = tempfile::tempdir().unwrap();
365            IsolatedTest::new(test_name(
366                module_path!(),
367                "replayed_spawn_does_not_reprovision_an_existing_child",
368            ))
369            .env(CHILD, "1")
370            .env("MJ_INSTANCE", "concurrency-sweep-spawn")
371            .isolated_store(root.path())
372            .run();
373            return;
374        }
375        let _writer = crate::database::install_isolated_test_writer();
376        let workspace = crate::database::create_workspace("spawn replay").unwrap();
377        let parent = crate::daemon::tests::runtime_test_session(
378            "parent",
379            &workspace.id,
380            mj_core::state::SessionState::Running,
381        );
382        crate::database::save_session(&parent).unwrap();
383        for (index, status) in [
384            mj_core::state::SessionState::Running,
385            mj_core::state::SessionState::Parked,
386            mj_core::state::SessionState::Stopped,
387        ]
388        .into_iter()
389        .enumerate()
390        {
391            let child_id = format!("child-{index}");
392            let child =
393                crate::daemon::tests::runtime_test_session(&child_id, &workspace.id, status);
394            let relation = crate::daemon::tests::runtime_test_subagent(&child_id, "parent");
395            crate::database::save_subagent_session(&child, &relation).unwrap();
396            let runtime = crate::daemon::tests::test_runtime_state();
397            let replayed = runtime
398                .start_subagent_session(crate::controller::RegisterSubagentRequest {
399                    parent_session_id: "parent".into(),
400                    task_name: "must reuse original".into(),
401                    profile_id: "missing-profile".into(),
402                    model: None,
403                    effort: None,
404                    working_directory: Default::default(),
405                    initial_prompt: "must not resend".into(),
406                    request_key: relation.request_key.clone(),
407                    report_root: None,
408                })
409                .await
410                .unwrap();
411            assert_eq!(replayed, relation);
412            assert!(runtime.active_lifecycles().is_empty());
413            assert_eq!(
414                crate::database::load_session_record(&child_id)
415                    .unwrap()
416                    .unwrap()
417                    .state,
418                status
419            );
420        }
421    }
422}
mj_controller/daemon/create.rs

mj_controller/daemon/
create.rs
