Skip to main content

capsule_runtime/
delegation_plane.rs

1//! Handing one task to one sub-capsule, and learning what became of it.
2//!
3//! This is the whole of a delegation as the delegating side performs it: present the session's
4//! credential to `mur-roost`, take the approval back, launch the approved artifact as a process
5//! and deliver the task over A2A. Every step is composed here from the capsule's own name, version
6//! and task text — the agent that asked for the delegation supplies those three strings and
7//! nothing else. It never sees the daemon's address, the credential, the approval or the child's
8//! directory.
9//!
10//! **Two methods, and they differ only in when they return.**
11//!
12//! | Method | Returns when | Statuses it can produce | Who holds the child | Its caller |
13//! |---|---|---|---|---|
14//! | [`DelegationPlane::start`] | the child is running and holding its task | `started`, `failed`, `refused` | the caller, through [`StartedDelegation::child`] | the agent-facing `delegate-task` tool |
15//! | [`DelegationPlane::delegate`] | the child's task reaches a terminal state, or [`DelegationOrigin::stop`] reports the delegating task cancelled | `completed`, `timed_out`, `failed`, `refused`, `canceled` | the method, until it returns | a plan's `capsule` step |
16//!
17//! [`DelegationPlane::start`] is the one an agent reaches. It hands the child's handle back, the
18//! delegating task holds it until the outcome is delivered or the task ends the child, and the
19//! outcome arrives at the parent's door as a completion the door hands to that task — which is
20//! what [`crate::delegation`] exists for, and why a plane that was never told its own address
21//! through [`DelegationPlane::reporting_to`] refuses to start anything.
22//!
23//! [`DelegationPlane::delegate`] blocks because its caller has nowhere to be told: a plan step
24//! holds no task loop, no conversation id and no A2A door, so a handle naming a delegation in
25//! flight would be a result it could never collect. It therefore injects a spawner with no
26//! [`crate::delegation::CompletionAddress`] — the child knows which session spawned it and reports
27//! to nobody, because the answer arrives on the connection the step already holds.
28//!
29//! Both are blocking calls, so their caller runs them on a blocking thread and the delegating
30//! capsule's own A2A listener keeps answering throughout.
31//!
32//! **Nothing here formats a token.** The credential is read once, into a request header; the
33//! approval is moved into [`ChildLaunchRequest`] without ever being turned into a string. The
34//! success body of `POST /spawn` carries the approval, so that response value is never
35//! interpolated into a message — a missing approval is reported by naming the field, not by
36//! quoting the body that lacked it.
37
38use std::time::{Duration, Instant};
39
40use serde_json::{json, Value};
41
42use crate::a2a::{
43    send_message_params, user_message, A2aTask, SendMessageResult, TaskState, A2A_PROTOCOL_VERSION,
44    A2A_VERSION_HEADER, RESPONSE_ARTIFACT,
45};
46use crate::child_launch::{
47    launch_child_capsule, ChildLaunchRequest, LaunchedChild, ParentWorkdirSource,
48};
49use crate::delegation::{CompletionAddress, Spawner};
50use crate::errors::RuntimeError;
51use crate::formation::FormationId;
52use crate::http_client::http_json;
53use crate::session_status::SessionDeadline;
54use crate::spawn_credential::{SpawnApproval, SpawnCredential, SPAWN_CREDENTIAL_HEADER};
55use crate::workdir_fs::WorkdirRoot;
56use crate::workdir_snapshot::{SnapshotLimits, WorkdirSnapshot};
57
58/// The deadline a session delegates under when nothing declares one.
59///
60/// **The single delegation bound**, and the only knob on either of the two waits a delegation
61/// still has: [`DelegationPlane::delegate`]'s poll for a terminal task state, and the completion
62/// watcher's observation of a child [`DelegationPlane::start`] launched. Everything else in a
63/// launch is bounded elsewhere and separately — reaching the child by [`crate::child_launch`]'s
64/// own launch timeout, delivering its task by [`SEND_DEADLINE`] below.
65///
66/// The same number `murmur_artifact::LifecycleConfig::default().delegation_deadline_secs` carries,
67/// and the value a caller with no manifest to read passes in; the declared key and
68/// [`DELEGATION_TIMEOUT_ENV`] are two ways of setting this one bound, not two bounds. Ten minutes
69/// is long enough for a sub-capsule that thinks, and short enough that a wedged one is reported
70/// within the hour rather than never.
71pub const DELEGATION_RESULT_TIMEOUT: Duration = Duration::from_secs(600);
72
73/// Environment variable setting [`DELEGATION_RESULT_TIMEOUT`] for this process, in whole seconds,
74/// over both the declared `lifecycle.delegation_deadline_secs` and the default.
75///
76/// For an operator whose sub-capsules legitimately run longer than the capsule declared, and for a
77/// value below it. A value that is not a positive integer is ignored and the declared bound stands.
78pub const DELEGATION_TIMEOUT_ENV: &str = "MURMUR_DELEGATION_TIMEOUT_SECS";
79
80/// The headers of every request to a child's door: the A2A version, and `authorization` as
81/// `Authorization` when there is one.
82fn door_headers(authorization: Option<&str>) -> Vec<(&str, &str)> {
83    std::iter::once((A2A_VERSION_HEADER, A2A_PROTOCOL_VERSION))
84        .chain(authorization.map(|value| ("Authorization", value)))
85        .collect()
86}
87
88/// How long the task delivery is retried while the child's listener comes up.
89///
90/// A child reports its URL when it binds, but the first connection can still land between the
91/// bind and the first accept, so the delivery backs off rather than failing on one refusal.
92const SEND_DEADLINE: Duration = Duration::from_secs(30);
93
94/// How often the child is asked whether its task reached a terminal state.
95const POLL_INTERVAL: Duration = Duration::from_millis(500);
96
97/// Ceiling on the answer text a delegation hands its parent, in a plan `capsule` step's output and
98/// in the report a `delegate-task` task continues with alike.
99///
100/// A child that produced a hundred megabytes must not be able to spend its parent's whole context
101/// by being asked one question. Past this, [`crate::delegation::hand_over_answer`] cuts the text,
102/// marks the cut and keeps the whole answer at `delegations/<delegation_id>.txt` in the session's
103/// runtime directory.
104pub(crate) const MAX_OUTPUT_BYTES: usize = 64 * 1024;
105
106/// What [`bounded`] appends to a text it cut.
107pub(crate) const CUT_MARK: &str = " […]";
108
109/// The three strings an agent supplies for one delegation, and the whole of what it supplies.
110#[derive(Debug, Clone)]
111pub struct DelegationRequest {
112    /// The sub-capsule's name, as it appears in the parent's `capabilities.spawn.allow`.
113    pub capsule: String,
114    /// An exact version. There is no `latest`: `murmur_artifact::RESERVED_VERSIONS` exists to
115    /// refuse that word, so the caller states which artifact it means.
116    pub version: String,
117    /// The task text, delivered to the child as the whole of its first user message.
118    pub task: String,
119}
120
121/// Where a delegation got to, as the delegating caller sees it.
122///
123/// The value vocabulary of the `delegate-task` tool result's `status`. The `delegation` trace
124/// event's `outcome` shares only `failed` and `refused` with it: a delegation that started is
125/// closed from the child's own completion instead, in
126/// [`crate::delegation::DelegationStatus`]'s words. Which of the two plane methods can produce
127/// each variant is stated on it, because a caller that reads a word this method cannot produce is
128/// reading the wrong contract.
129#[derive(Debug, Clone, Copy, PartialEq, Eq)]
130pub enum DelegationStatus {
131    /// The child is running and holding its task, and nothing is known yet about what it will do
132    /// with it. Produced only by [`DelegationPlane::start`]; the outcome is delivered afterwards
133    /// into the task that made the delegation.
134    Started,
135    /// The child's task reached `completed` and its answer is in hand. Produced only by
136    /// [`DelegationPlane::delegate`].
137    Completed,
138    /// A child ran and did not answer: its task ended `failed` or `rejected`, or it could not be
139    /// reached, launched or handed its task after the daemon had approved it. Produced by both
140    /// methods.
141    Failed,
142    /// No terminal state within [`DelegationPlane::result_timeout`]. Produced only by
143    /// [`DelegationPlane::delegate`]: a child [`DelegationPlane::start`] launched is bounded by
144    /// its watcher instead, which ends it and posts a `terminated` completion.
145    TimedOut,
146    /// The daemon refused, so no child was launched and no child directory exists. Produced by
147    /// both methods.
148    Refused,
149    /// The task the delegation was made for was cancelled while the child was running, so the
150    /// child was ended and its `completion.json` records `terminated`. Produced only by
151    /// [`DelegationPlane::delegate`], when [`DelegationOrigin::stop`] reports true.
152    ///
153    /// Never written as a `delegation` line's `outcome`: the cancelled task closes that row
154    /// itself, in the sub-capsule vocabulary, as `terminated`.
155    Canceled,
156}
157
158impl DelegationStatus {
159    pub fn as_str(self) -> &'static str {
160        match self {
161            Self::Started => "started",
162            Self::Completed => "completed",
163            Self::Failed => "failed",
164            Self::TimedOut => "timed_out",
165            Self::Refused => "refused",
166            Self::Canceled => "canceled",
167        }
168    }
169}
170
171/// What one delegation produced.
172///
173/// Carries the child's answer text directly, unlike [`crate::delegation::DelegationOutcome`],
174/// which carries none: this is the return value of a step the plan made on purpose, so the answer
175/// is what it asked for. It reaches the model through the untrusted-content fence like every other
176/// tool result.
177#[derive(Debug, Clone)]
178pub struct DelegationResult {
179    /// The id this delegation is named by, in the `dlg_` id space
180    /// ([`crate::delegation::new_delegation_id`]). Minted by the launcher, one per launched
181    /// child, and empty whenever no child was launched — a refusal, or a process that never
182    /// started. A delegation that was never made names no delegation.
183    pub delegation_id: String,
184    /// The child's own session id, so its trace is findable. Empty when no child ran.
185    pub session_id: String,
186    pub capsule: String,
187    pub version: String,
188    pub status: DelegationStatus,
189    /// The child's answer on [`DelegationStatus::Completed`]; on every other status, why there is
190    /// no answer.
191    pub output: String,
192    /// `delegations/<delegation_id>.txt`, relative to the session's runtime directory: where the
193    /// whole answer is kept, present only when a [`DelegationStatus::Completed`] answer was cut at
194    /// [`MAX_OUTPUT_BYTES`] and the copy was written ([`crate::delegation::hand_over_answer`]).
195    /// `None` for an answer handed over whole, and on every non-`Completed` status.
196    pub result_path: Option<String>,
197    /// Whether [`Self::output`] was cut at [`MAX_OUTPUT_BYTES`].
198    pub truncated: bool,
199    /// The child's work folder, in this session's records directory. The child's trace and
200    /// result are in its own records directory, which this session's commands cannot read.
201    ///
202    /// `Some` only on [`DelegationStatus::Started`], which is the one status where a directory is
203    /// all there is to name.
204    pub child_workdir: Option<String>,
205}
206
207impl DelegationResult {
208    /// A result that names no child, for a delegation that was never made.
209    ///
210    /// Two ways to reach it, and they name the same nothing: the daemon refused, or the approved
211    /// child's process never started. A delegation id is minted by the launcher, so neither has
212    /// one to report.
213    fn unmade(request: &DelegationRequest, status: DelegationStatus, reason: String) -> Self {
214        let (output, truncated) = bounded(reason);
215        Self {
216            delegation_id: String::new(),
217            session_id: String::new(),
218            capsule: request.capsule.clone(),
219            version: request.version.clone(),
220            status,
221            output,
222            result_path: None,
223            truncated,
224            child_workdir: None,
225        }
226    }
227
228    /// A delegation the daemon refused, by the spawn envelope or by one of its own bounds.
229    fn refused(request: &DelegationRequest, reason: String) -> Self {
230        Self::unmade(request, DelegationStatus::Refused, reason)
231    }
232}
233
234/// What [`DelegationPlane::start`] produced: the result the agent reads, and on
235/// [`DelegationStatus::Started`] the handle that ends the child.
236///
237/// Dropping [`Self::child`] ends and reaps a child still running, so a caller that does not keep
238/// it — a `delegate-task` call cancelled while the launch was finishing — leaves nothing behind.
239#[derive(Debug)]
240pub struct StartedDelegation {
241    pub result: DelegationResult,
242    /// `Some` exactly when [`DelegationResult::status`] is [`DelegationStatus::Started`].
243    pub child: Option<LaunchedChild>,
244}
245
246impl StartedDelegation {
247    /// A start that left no child running.
248    fn without_child(result: DelegationResult) -> Self {
249        Self {
250            result,
251            child: None,
252        }
253    }
254}
255
256/// What the parent knows the moment one child is up, handed back while the delegation is still
257/// running so the parent's trace names the child before it can hang, crash or be timed out.
258#[derive(Debug, Clone)]
259pub struct DelegationLaunch {
260    /// The `dlg_` id the launcher minted for this launch.
261    pub delegation_id: String,
262    pub capsule: String,
263    pub version: String,
264    /// The session id the child's runtime minted for itself and reported on its launch line.
265    pub child_session_id: String,
266    /// The child's directory, relative to the parent's accessible workdir.
267    pub child_workdir: String,
268    /// The child's records directory, absolute, as the child reported it: where its own
269    /// `trace.jsonl` and `completion.json` are.
270    pub child_records_dir: String,
271    /// What was copied from this session's work folder into the child's before it started, or
272    /// `None` for a capsule not listed under `capabilities.spawn.workdir_snapshot`.
273    pub workdir_snapshot: Option<WorkdirSnapshot>,
274}
275
276/// A session's `capabilities.spawn.workdir_snapshot`, and the work folder it copies.
277#[derive(Debug, Clone)]
278pub struct WorkdirSnapshotGrant {
279    /// The full names whose every delegation starts in a copy of [`Self::source`].
280    pub capsules: Vec<String>,
281    /// The session's accessible work folder, held.
282    pub source: WorkdirRoot,
283    /// `(st_dev, st_ino)` of the session's runtime and records directories, never copied.
284    pub excluded: Vec<(u64, u64)>,
285}
286
287impl WorkdirSnapshotGrant {
288    /// The grant of a session listing `capsules`, whose work folder is `source` and whose own
289    /// directories — runtime and records — are `own_dirs`. Those lie inside `source` only on an
290    /// `advisory`-capped launch, and are never copied.
291    pub fn new(capsules: Vec<String>, source: WorkdirRoot, own_dirs: &[&WorkdirRoot]) -> Self {
292        let excluded = own_dirs
293            .iter()
294            .filter_map(|dir| dir.identity().ok())
295            .collect();
296        Self {
297            capsules,
298            source,
299            excluded,
300        }
301    }
302}
303
304/// The parent's side of one delegation, beyond the three strings its agent supplied.
305///
306/// Per call rather than per plane: a launch that mints one context id per task has no
307/// launch-scoped conversation to name, so the id is only knowable once a task is running.
308#[derive(Clone, Default)]
309pub struct DelegationOrigin<'a> {
310    /// The conversation the delegation was made from. Empty when the caller has none, which
311    /// injects no handle at all rather than one naming a conversation that does not exist.
312    pub context_id: String,
313    /// What to do with the launch notice, or `None` for a caller that records no
314    /// `delegation_start`. Called on the delegating thread, before the delegation has ended.
315    ///
316    /// A callback rather than a channel because the two callers are not the same shape:
317    /// [`DelegationPlane::start`] runs on a blocking tokio worker with a task free to drain a
318    /// receiver, while `plan::execute` calls [`DelegationPlane::delegate`] directly on a
319    /// scheduler thread with no runtime to await on and nothing else running to drain anything.
320    /// Invoked in place, both surfaces get the record on disk while the child is still in flight,
321    /// which is the whole reason it is a record of its own.
322    pub launched: Option<std::sync::Arc<dyn Fn(DelegationLaunch) + Send + Sync + 'a>>,
323    /// Whether the task this delegation was made for has been cancelled, or `None` for a caller
324    /// that cannot be. Polled, so it must be cheap and must never block.
325    ///
326    /// Read only by [`DelegationPlane::delegate`], which ends the child itself when it reports
327    /// true. [`DelegationPlane::start`] ignores it: its caller races its own cancel against the
328    /// launch, and holds the child's handle from then on.
329    pub stop: Option<std::sync::Arc<dyn Fn() -> bool + Send + Sync + 'a>>,
330}
331
332/// Hand-written because [`DelegationOrigin::launched`] and [`DelegationOrigin::stop`] are trait
333/// objects and cannot derive one.
334/// A callback has no readable identity, so what is printed of it is whether there is one.
335impl std::fmt::Debug for DelegationOrigin<'_> {
336    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
337        f.debug_struct("DelegationOrigin")
338            .field("context_id", &self.context_id)
339            .field("launched", &self.launched.is_some())
340            .field("stop", &self.stop.is_some())
341            .finish()
342    }
343}
344
345/// One session's authority to delegate, and everything a delegation needs beyond the three
346/// strings its caller supplies.
347///
348/// Built only for a session that registered with `mur-roost` — which is exactly a session whose
349/// manifest declares `capabilities.spawn.allow`, or one that was itself spawned. A session
350/// without one holds `None` and its agent was never offered the tool.
351pub struct DelegationPlane {
352    /// The daemon's base URL, from the runtime's own `MURMUR_ROOST_URL`. Not in the model's
353    /// context, and not something the agent can name.
354    roost_url: String,
355    /// This session's credential, read exactly once per delegation into one request header.
356    credential: SpawnCredential,
357    /// The parent's records directory. Every child's work folder is created in its `children/`,
358    /// outside the parent's work folder and read-only to the parent's commands.
359    records_root: WorkdirRoot,
360    /// This plane's bound on the wait for an answer: the declared
361    /// `lifecycle.delegation_deadline_secs`, or [`DELEGATION_TIMEOUT_ENV`] where it names a
362    /// positive number of seconds. Read once at construction so every delegation in a session is
363    /// bounded the same way.
364    result_timeout: Duration,
365    /// The delegating session's own id, written into every child's injected handle and from there
366    /// into that child's `session_start.spawned_by`. Empty for a caller that does not know it,
367    /// which injects no handle.
368    session_id: String,
369    /// Where a child's own manifest is read from, so the parent knows which host variables that
370    /// child declares. The same store the child's runtime will resolve the artifact out of, so
371    /// the manifest read here is the manifest the child will load.
372    registry: std::sync::Arc<dyn murmur_artifact::Registry>,
373    /// The delegating capsule's own `capabilities.env.allow`. Every child's declaration is
374    /// clamped to this list, so a delegation can never widen the set of host variables the
375    /// parent's own manifest declares.
376    parent_env_allow: Vec<String>,
377    /// This capsule's own A2A endpoint, `http://host:port`, where a started delegation's
378    /// completion is posted. Empty until [`DelegationPlane::reporting_to`] names it, and a plane
379    /// that never names one cannot [`DelegationPlane::start`] anything: a delegation whose outcome
380    /// has nowhere to arrive is a way to lose work.
381    own_url: String,
382    /// The delegating session's formation, handed to every child as `MURMUR_FORMATION_ID` so the
383    /// child joins it. `None` for a session nobody placed in a formation, whose children carry
384    /// none either.
385    formation_id: Option<FormationId>,
386    /// The delegating session's deadline, the earliest bound placed on it, handed to every child
387    /// as that child's inherited deadline. `None` until [`DelegationPlane::bounded_by`] names one.
388    deadline: Option<SessionDeadline>,
389    /// Which children start in a copy of this session's work folder. `None` until
390    /// [`DelegationPlane::snapshotting_workdir`] names a grant, which starts every child in an
391    /// empty one.
392    workdir_snapshot: Option<WorkdirSnapshotGrant>,
393}
394
395impl DelegationPlane {
396    /// The plane for a registered session.
397    ///
398    /// `roost_url`, `credential` and `session_id` come from the session's own registration, never
399    /// from anything the agent said. `declared_deadline` is the capsule's own
400    /// `lifecycle.delegation_deadline_secs`; [`DELEGATION_TIMEOUT_ENV`] overrides it where it
401    /// names a positive number of seconds. `registry` is the store a child's manifest is read
402    /// from and `parent_env_allow` the delegating capsule's own `capabilities.env.allow`, which
403    /// together decide the host variables a child is handed. `formation_id` is the session's own
404    /// [`crate::StagedSession::formation_id`], inherited by every child whether or not it is also
405    /// handed a spawner.
406    #[allow(clippy::too_many_arguments)]
407    pub fn new(
408        roost_url: String,
409        credential: SpawnCredential,
410        records_root: WorkdirRoot,
411        session_id: String,
412        declared_deadline: Duration,
413        registry: std::sync::Arc<dyn murmur_artifact::Registry>,
414        parent_env_allow: Vec<String>,
415        formation_id: Option<FormationId>,
416    ) -> Self {
417        Self {
418            roost_url: roost_url.trim_end_matches('/').to_string(),
419            credential,
420            records_root,
421            result_timeout: handed_off_work_deadline(declared_deadline),
422            session_id,
423            registry,
424            parent_env_allow,
425            own_url: String::new(),
426            formation_id,
427            deadline: None,
428            workdir_snapshot: None,
429        }
430    }
431
432    /// Start every child `grant` lists in a copy of its work folder, made before the child
433    /// process exists. A grant listing nothing is no grant.
434    pub fn snapshotting_workdir(mut self, grant: WorkdirSnapshotGrant) -> Self {
435        self.workdir_snapshot = (!grant.capsules.is_empty()).then_some(grant);
436        self
437    }
438
439    /// Bound every child this plane launches by `deadline`, the delegating session's own, so no
440    /// child runs past the moment its parent ends. `None` bounds none.
441    pub fn bounded_by(mut self, deadline: Option<SessionDeadline>) -> Self {
442        self.deadline = deadline;
443        self
444    }
445
446    /// Name this capsule's own A2A endpoint, `http://host:port`, as where a started delegation's
447    /// completion is posted.
448    ///
449    /// Separate from [`DelegationPlane::new`] because only a caller that runs a task loop behind
450    /// that address can act on a completion. A plane built without it still delegates through
451    /// [`DelegationPlane::delegate`], which is answered on the connection it opens; it refuses
452    /// [`DelegationPlane::start`], which is not.
453    pub fn reporting_to(mut self, own_url: String) -> Self {
454        self.own_url = own_url;
455        self
456    }
457
458    /// The formation every child this plane launches joins, or `None` for a session in none.
459    pub fn formation_id(&self) -> Option<&FormationId> {
460        self.formation_id.as_ref()
461    }
462
463    /// The deadline every child this plane launches inherits, or `None` for an unbounded plane.
464    pub fn deadline(&self) -> Option<SessionDeadline> {
465        self.deadline
466    }
467
468    /// This plane's bound on the wait for a child's answer.
469    pub fn result_timeout(&self) -> Duration {
470        self.result_timeout
471    }
472
473    /// The host variables this launch copies into the child: what the child's own manifest
474    /// declares under `capabilities.env.allow`, clamped to what the parent's declares.
475    ///
476    /// The manifest is read narrowly, out of the packed artifact, because a full parse resolves
477    /// the child's `${VAR}` references against this process's environment and would refuse a
478    /// manifest naming a variable the parent has not been given. An artifact that cannot be
479    /// resolved, or whose declaration cannot be read, is an error naming the capsule and version
480    /// rather than an empty list: a child launched without what it declares fails later and less
481    /// legibly.
482    fn child_env_allow(&self, capsule: &str, version: &str) -> Result<Vec<String>, RuntimeError> {
483        let manifest_yaml = self.child_manifest_yaml(capsule, version, "capabilities.env.allow")?;
484        let declared =
485            crate::artifact::extract_declared_env_allow(capsule, version, &manifest_yaml)?;
486        Ok(env_allow_intersection(&declared, &self.parent_env_allow))
487    }
488
489    /// The packed manifest of the child `capsule@version`, out of [`Self::registry`], for a
490    /// narrow read of `key`; an artifact that cannot be resolved is an error naming the capsule,
491    /// the version and `key`.
492    fn child_manifest_yaml(
493        &self,
494        capsule: &str,
495        version: &str,
496        key: &str,
497    ) -> Result<String, RuntimeError> {
498        let name = murmur_artifact::ArtifactName::parse(capsule).map_err(|error| {
499            RuntimeError::Runtime(format!(
500                "cannot read what '{capsule}@{version}' declares under {key}: {error}"
501            ))
502        })?;
503        let resolved = self.registry.resolve(&name, version).map_err(|error| {
504            RuntimeError::Runtime(format!(
505                "cannot read what '{capsule}@{version}' declares under {key}: {error}"
506            ))
507        })?;
508        crate::artifact::extract_manifest_yaml(capsule, version, &resolved.bytes)
509    }
510
511    /// What [`launch_child_capsule`] copies into this child's work folder: this session's work
512    /// folder, bounded by the child's own `capabilities.resources.workdir_max_bytes`, for a
513    /// `capsule` this session lists under `capabilities.spawn.workdir_snapshot` (a full name);
514    /// `None` for every other.
515    fn parent_workdir_source(
516        &self,
517        capsule: &str,
518        version: &str,
519    ) -> Result<Option<ParentWorkdirSource>, RuntimeError> {
520        let Some(grant) = self
521            .workdir_snapshot
522            .as_ref()
523            .filter(|grant| grant.capsules.iter().any(|listed| listed == capsule))
524        else {
525            return Ok(None);
526        };
527        let manifest_yaml =
528            self.child_manifest_yaml(capsule, version, "capabilities.resources.workdir_max_bytes")?;
529        let max_bytes =
530            crate::artifact::extract_declared_workdir_max_bytes(capsule, version, &manifest_yaml)?
531                .unwrap_or(crate::resources::DEFAULT_WORKDIR_MAX_BYTES);
532        Ok(Some(ParentWorkdirSource {
533            root: grant.source.clone(),
534            limits: SnapshotLimits {
535                max_bytes,
536                excluded: grant.excluded.clone(),
537            },
538        }))
539    }
540
541    /// This session's credential, cloned for a caller that delegates through a plane of its own.
542    ///
543    /// The one such caller is `plan::execute`, which builds its own [`DelegationPlane`] per
544    /// `capsule` step. Handing it this token rather than storing a second copy beside the plane
545    /// keeps one registration's authority in one place; the clone is still a
546    /// [`SpawnCredential`], so it cannot be formatted, serialized or traced into anything.
547    pub(crate) fn credential(&self) -> SpawnCredential {
548        self.credential.clone()
549    }
550
551    /// Ask the daemon whether this session may spawn that capsule, and take the approval back.
552    ///
553    /// One request, and the only one: the daemon judges the session the credential names, runs the
554    /// referee, and answers with an approval naming the exact artifact it resolved. It launches
555    /// nothing. `Err` is the referee's own sentence, which both methods return to their caller as
556    /// a [`DelegationStatus::Refused`] result unchanged.
557    ///
558    /// The credential's one reading. Every other route out of [`SpawnCredential`] is closed — no
559    /// `Display`, no `Serialize`, redacted `Debug` — so the token cannot reach a tool result, a
560    /// trace or a workdir file by being formatted somewhere.
561    fn approval_for(&self, request: &DelegationRequest) -> Result<SpawnApproval, String> {
562        let spawn_body = json!({ "name": request.capsule, "version": request.version }).to_string();
563        let permission = match http_json(
564            "POST",
565            &format!("{}/spawn", self.roost_url),
566            Some(&spawn_body),
567            &[(SPAWN_CREDENTIAL_HEADER, self.credential.expose())],
568        ) {
569            Ok(value) => value,
570            Err(error) => return Err(daemon_refusal(&error)),
571        };
572        let Some(approval) = permission.get("approval").and_then(Value::as_str) else {
573            // Names the missing field and never the body: the body of a *successful* spawn
574            // response is where the approval is.
575            return Err("mur-roost spawn response missing approval".to_string());
576        };
577        Ok(SpawnApproval::new(approval.to_string()))
578    }
579
580    /// Compose the one launch this plane makes, from the approval the daemon just gave and the
581    /// address — if any — the child's outcome is to be posted to.
582    ///
583    /// **The only place a [`ChildLaunchRequest`] is built.** Both methods reach a child through
584    /// here, so a field that is wrong is wrong on both paths at once rather than on whichever one
585    /// a later edit forgot; a source-sweep test in this file's `mod tests` holds that to one
586    /// construction site. Everything but `report_to` is the plane's own state and the caller's
587    /// three strings, which is why that is the single parameter: it is the whole of the
588    /// difference between a delegation whose outcome is posted to this capsule's door and one
589    /// whose answer arrives on the connection the caller is already holding.
590    ///
591    /// `completion_deadline` is derived from `report_to` rather than passed: the watcher this
592    /// plane's bound applies to only runs for a spawner naming an address, so two parameters
593    /// would be two ways of stating one decision and a way for them to disagree.
594    ///
595    /// `Err` when the child's own `capabilities.env.allow` cannot be read, which starts no
596    /// process on either path: a child launched without the variables its manifest names dies at
597    /// its own manifest load, inside a process nobody is reading. Called after the daemon has
598    /// approved the spawn, so a refusal is still reported as a refusal rather than as a local
599    /// read failure.
600    fn launch_request(
601        &self,
602        request: &DelegationRequest,
603        origin: &DelegationOrigin<'_>,
604        grant: SpawnApproval,
605        report_to: Option<CompletionAddress>,
606    ) -> Result<ChildLaunchRequest, RuntimeError> {
607        let completion_deadline = report_to.as_ref().map(|_| self.result_timeout);
608        Ok(ChildLaunchRequest {
609            parent_records_root: self.records_root.clone(),
610            capsule_name: request.capsule.clone(),
611            capsule_version: request.version.clone(),
612            grant,
613            // What both manifests declare and nothing else, read from the parent's own store
614            // rather than taken from the daemon's answer: the referee answers whether a spawn may
615            // happen, and this clamp has to hold whatever it answered.
616            child_env_allow: self.child_env_allow(&request.capsule, &request.version)?,
617            roost_url: self.roost_url.clone(),
618            // A caller that knows neither its session nor its conversation injects no handle at
619            // all, rather than one naming a lineage that does not exist. `start` holds both by
620            // the time it reaches here — it refuses outright when either is empty — so this is
621            // unconditional there in everything but form.
622            spawner: (!self.session_id.is_empty() && !origin.context_id.is_empty()).then(|| {
623                Spawner {
624                    session_id: self.session_id.clone(),
625                    context_id: origin.context_id.clone(),
626                    report_to,
627                }
628            }),
629            completion_deadline,
630            // Independent of `spawner`: a launch that names no lineage still joins the formation.
631            formation_id: self.formation_id.clone(),
632            deadline: self.deadline,
633            parent_workdir_snapshot: self
634                .parent_workdir_source(&request.capsule, &request.version)?,
635        })
636    }
637
638    /// Hand the launch back to the caller the moment the child is up and has reported its session
639    /// id, so a child that then hangs, crashes or is ended is already attributable.
640    ///
641    /// An unnamed delegation is not announced. A caller with no session id or no conversation to
642    /// name injects no handle, so the launcher minted no id and there is nothing for the terminal
643    /// `delegation` line — which writes `null` rather than an empty id — to join against.
644    /// `delegation_start.delegation_id` is required to be a `dlg_` id; a launch that has none
645    /// writes no line at all.
646    fn announce(
647        &self,
648        request: &DelegationRequest,
649        origin: &DelegationOrigin<'_>,
650        child: &crate::child_launch::LaunchedChild,
651        delegation_id: &str,
652    ) {
653        if let Some(launched) = origin
654            .launched
655            .as_ref()
656            .filter(|_| !delegation_id.is_empty())
657        {
658            launched(DelegationLaunch {
659                delegation_id: delegation_id.to_string(),
660                capsule: request.capsule.clone(),
661                version: request.version.clone(),
662                child_session_id: child.session_id.clone(),
663                child_workdir: child.workdir.to_string_lossy().into_owned(),
664                child_records_dir: child.records_dir.display().to_string(),
665                workdir_snapshot: child.workdir_snapshot.clone(),
666            });
667        }
668    }
669
670    /// Ask the daemon, launch the approved child, hand it its task, and return the child's handle.
671    ///
672    /// **The agent-facing `delegate-task` tool's method.** It returns as soon as the child is
673    /// running *and holding its task*, so a turn can issue several delegations; what the child
674    /// eventually did is posted to the parent's door afterwards, carrying this delegation's id, by
675    /// the child itself or, for a child that could not speak for itself, by the watcher
676    /// [`launch_child_capsule`] started behind it.
677    ///
678    /// Blocking and never `Err`, like [`DelegationPlane::delegate`], but only three
679    /// [`DelegationStatus`] words can come out of it: `started`, `failed` and `refused`.
680    ///
681    /// "Started" means the child holds its task, not merely that a process exists. Everything
682    /// before that point ends and reaps the child on the way out, because a child that will never
683    /// be given work will never report and would strand a process nothing waits on. A started
684    /// child's handle is returned in [`StartedDelegation::child`]: the child lives as long as that
685    /// handle, or its own run, whichever ends first.
686    pub fn start(
687        &self,
688        request: &DelegationRequest,
689        origin: &DelegationOrigin<'_>,
690    ) -> StartedDelegation {
691        // Refused before the daemon is touched, because there is nothing to ask about: a
692        // delegation started here would run to completion and post its outcome nowhere, which is
693        // a way to lose work rather than a way to do it. No `POST /spawn` is made and no process
694        // is launched.
695        if self.own_url.is_empty() || self.session_id.is_empty() || origin.context_id.is_empty() {
696            return StartedDelegation::without_child(DelegationResult::unmade(
697                request,
698                DelegationStatus::Failed,
699                "this capsule cannot start a delegation: a sub-capsule's outcome is posted to \
700                 this runtime's own endpoint, addressed to its session and conversation, and \
701                 this session does not hold all three — so the outcome would have nowhere to be \
702                 reported"
703                    .to_string(),
704            ));
705        }
706
707        let grant = match self.approval_for(request) {
708            Ok(grant) => grant,
709            Err(reason) => {
710                return StartedDelegation::without_child(DelegationResult::refused(request, reason))
711            }
712        };
713
714        // The production caller of the completion path: naming this capsule's own address is what
715        // starts the watcher behind the child, and with it this plane's bound on the watch.
716        let launch = match self.launch_request(
717            request,
718            origin,
719            grant,
720            Some(CompletionAddress {
721                url: self.own_url.clone(),
722            }),
723        ) {
724            Ok(launch) => launch,
725            Err(error) => {
726                return StartedDelegation::without_child(DelegationResult::unmade(
727                    request,
728                    DelegationStatus::Failed,
729                    error.to_string(),
730                ))
731            }
732        };
733
734        let mut child = match launch_child_capsule(launch) {
735            Ok(child) => child,
736            Err(error) => {
737                return StartedDelegation::without_child(DelegationResult::unmade(
738                    request,
739                    DelegationStatus::Failed,
740                    error.to_string(),
741                ))
742            }
743        };
744        let delegation_id = child.delegation_id.clone().unwrap_or_default();
745        self.announce(request, origin, &child, &delegation_id);
746
747        let child_workdir = child.workdir.to_string_lossy().into_owned();
748        // `child` is dropped on every early return below, which ends and reaps it.
749        let failed = |session_id: &str, output: String| {
750            let (output, truncated) = bounded(output);
751            StartedDelegation::without_child(DelegationResult {
752                delegation_id: delegation_id.clone(),
753                session_id: session_id.to_string(),
754                capsule: request.capsule.clone(),
755                version: request.version.clone(),
756                status: DelegationStatus::Failed,
757                output,
758                result_path: None,
759                truncated,
760                child_workdir: None,
761            })
762        };
763
764        if child.capsule_url.is_empty() {
765            return failed(
766                &child.session_id,
767                format!(
768                    "capsule '{}' bound no address, so it cannot be sent a task; delegation needs \
769                     a capsule that serves A2A",
770                    request.capsule
771                ),
772            );
773        }
774        let capsule_url = child.capsule_url.trim_end_matches('/').to_string();
775        let door_token = child.door_token.clone();
776        match deliver_task(
777            &capsule_url,
778            door_token.as_ref(),
779            &delegation_id,
780            request,
781            &|| false,
782        ) {
783            Ok(task_id) => child.task_id = Some(task_id),
784            Err(reason) => return failed(&child.session_id, reason),
785        }
786
787        let session_id = child.session_id.clone();
788        StartedDelegation {
789            result: DelegationResult {
790                delegation_id,
791                session_id,
792                capsule: request.capsule.clone(),
793                version: request.version.clone(),
794                status: DelegationStatus::Started,
795                // Nothing has been produced yet, and saying so is the point of the status word.
796                output: String::new(),
797                result_path: None,
798                truncated: false,
799                child_workdir: Some(child_workdir),
800            },
801            child: Some(child),
802        }
803    }
804
805    /// Ask the daemon, launch the approved child, deliver the task, and wait for the answer.
806    ///
807    /// **A plan `capsule` step's method, and not the agent-facing one.** A step has no task loop,
808    /// no conversation id and no A2A door of its own, so the only place its answer can arrive is
809    /// the connection this call is already holding. That is why the spawner injected below names
810    /// no [`CompletionAddress`]: there is nothing for a completion to tell, and a watcher posting
811    /// into a session that is not listening would only write an undeliverable record. The
812    /// agent-facing tool calls [`DelegationPlane::start`] instead.
813    ///
814    /// Blocking from end to end, and never `Err`: every way a delegation can end here is one of
815    /// five [`DelegationStatus`] words — `completed`, `timed_out`, `failed`, `refused`, `canceled`
816    /// — because the step maps them all onto a step result, and a refusal is as much a fact about
817    /// the run as an answer is.
818    ///
819    /// `runtime_root` is the delegating session's runtime directory: a `completed` answer is handed
820    /// over through [`crate::delegation::hand_over_answer`], which keeps a cut answer whole there.
821    ///
822    /// **Cancellation.** [`DelegationOrigin::stop`] is asked once the child is announced, between
823    /// delivery attempts, and on every poll. When it reports true this method ends the child,
824    /// records it `terminated` by the launcher in the child's `completion.json` with
825    /// [`crate::delegation::PARENT_ENDED_DETAIL`], and returns [`DelegationStatus::Canceled`]. The
826    /// daemon's approval and the child's launch are not interruptible, so a cancel that lands
827    /// during either is honoured at the first check after them — the same bound a `delegate-task`
828    /// call's launch has.
829    pub fn delegate(
830        &self,
831        request: &DelegationRequest,
832        origin: &DelegationOrigin<'_>,
833        runtime_root: &WorkdirRoot,
834    ) -> DelegationResult {
835        // Step 1: ask whether this session may spawn that capsule.
836        let grant = match self.approval_for(request) {
837            Ok(grant) => grant,
838            Err(reason) => return DelegationResult::refused(request, reason),
839        };
840
841        // Step 2: launch it here, in this runtime, as a process of its own. `child` owns that
842        // process: dropping it — including on every early return below, and on the deadline in
843        // step 4 — terminates and reaps the child rather than leaving it holding a port.
844        //
845        // The spawner this composes is lineage and nothing else. The answer arrives on the
846        // connection this plane opened, so there is nothing for a completion to tell: naming no
847        // address means no watcher and no post — and no completion deadline, since the poll in
848        // step 4 is this method's bound — while the child still learns which session spawned it.
849        //
850        // A launch request that cannot be composed starts no child: a read of the child's own
851        // declaration that fails would otherwise launch a capsule without the variables its
852        // manifest names, which dies at its own manifest load inside a process nobody is reading.
853        let launch = match self.launch_request(request, origin, grant, None) {
854            Ok(launch) => launch,
855            Err(error) => {
856                return DelegationResult::unmade(
857                    request,
858                    DelegationStatus::Failed,
859                    error.to_string(),
860                )
861            }
862        };
863        let launched_at = Instant::now();
864        let mut child = match launch_child_capsule(launch) {
865            Ok(child) => child,
866            // A child whose process never started named no delegation, the same as one the daemon
867            // refused: the id is minted by the launcher, and this launch reached no launcher.
868            Err(error) => {
869                return DelegationResult::unmade(
870                    request,
871                    DelegationStatus::Failed,
872                    error.to_string(),
873                )
874            }
875        };
876        // Adopted, never minted here: one launch, one id, and the same string the child was
877        // injected with and wrote into its own `session_start`.
878        let delegation_id = child.delegation_id.clone().unwrap_or_default();
879
880        self.announce(request, origin, &child, &delegation_id);
881        let stopped = || origin.stop.as_ref().is_some_and(|stop| stop());
882
883        // Every result this call can produce past the launch, built from one place so the
884        // delegation id, the capsule and the version cannot disagree between two of them — and so
885        // [`MAX_OUTPUT_BYTES`] bounds every branch. A child's *failure* message is as much its own
886        // text as its answer is, so it is cut on the same terms; its answer is bounded by
887        // [`crate::delegation::hand_over_answer`] instead, which also keeps it whole.
888        let outcome = |status, session_id: &str, output: String| {
889            let (output, truncated) = bounded(output);
890            DelegationResult {
891                delegation_id: delegation_id.clone(),
892                session_id: session_id.to_string(),
893                capsule: request.capsule.clone(),
894                version: request.version.clone(),
895                status,
896                output,
897                result_path: None,
898                truncated,
899                child_workdir: None,
900            }
901        };
902        // The task this delegation was made for was cancelled: end the child here, where its
903        // handle is, and leave the delegation's row to the task, which closes it after its
904        // `task_canceled`.
905        let canceled = |child: &mut LaunchedChild| {
906            end_for_canceled_task(request, &delegation_id, child, launched_at);
907            outcome(
908                DelegationStatus::Canceled,
909                &child.session_id,
910                format!(
911                    "capsule '{}' was stopped because the task that started it was cancelled",
912                    request.capsule
913                ),
914            )
915        };
916        if stopped() {
917            return canceled(&mut child);
918        }
919
920        if child.capsule_url.is_empty() {
921            return outcome(
922                DelegationStatus::Failed,
923                &child.session_id,
924                format!(
925                    "capsule '{}' bound no address, so it cannot be sent a task; delegation needs \
926                     a capsule that serves A2A",
927                    request.capsule
928                ),
929            );
930        }
931        let capsule_url = child.capsule_url.trim_end_matches('/').to_string();
932        // `Authorization` is present on every call to a child that declares
933        // `network.authentication`.
934        let authorization = child
935            .door_token
936            .as_ref()
937            .map(crate::door_auth::bearer_header);
938        let headers = door_headers(authorization.as_deref());
939
940        // Step 3: deliver the task as the child's first user message.
941        let task_id = match deliver_task(
942            &capsule_url,
943            child.door_token.as_ref(),
944            &delegation_id,
945            request,
946            &stopped,
947        ) {
948            Ok(task_id) => task_id,
949            Err(_) if stopped() => return canceled(&mut child),
950            Err(reason) => return outcome(DelegationStatus::Failed, &child.session_id, reason),
951        };
952
953        // Step 4: wait for a terminal state, bounded. On expiry the parent stops waiting; what
954        // happens to the child is decided below.
955        let poll_deadline = Instant::now() + self.result_timeout;
956        loop {
957            if stopped() {
958                return canceled(&mut child);
959            }
960            if Instant::now() >= poll_deadline {
961                // The wait and the child end together. This caller has nowhere for a later outcome
962                // to arrive — that is why it is waiting on this connection at all — so a child left
963                // running past it would be a process nothing would ever collect from or reap.
964                // `child` is dropped on the way out, which ends and reaps it.
965                return outcome(
966                    DelegationStatus::TimedOut,
967                    &child.session_id,
968                    format!(
969                        "capsule '{}' did not answer within {}s; the delegation was ended and the \
970                         child was stopped",
971                        request.capsule,
972                        self.result_timeout.as_secs()
973                    ),
974                );
975            }
976            std::thread::sleep(POLL_INTERVAL);
977            if stopped() {
978                return canceled(&mut child);
979            }
980
981            let answer = match poll_task(&capsule_url, &headers, &delegation_id, &task_id) {
982                Ok(task) => task,
983                Err(error) => {
984                    return outcome(
985                        DelegationStatus::Failed,
986                        &child.session_id,
987                        format!("capsule '{}' stopped answering: {error}", request.capsule),
988                    )
989                }
990            };
991            let task = match answer
992                .get("result")
993                .map(|result| serde_json::from_value::<A2aTask>(result.clone()))
994            {
995                Some(Ok(task)) => task,
996                Some(Err(error)) => {
997                    return outcome(
998                        DelegationStatus::Failed,
999                        &child.session_id,
1000                        format!(
1001                            "capsule '{}' answered GetTask with no task it could read: {error}",
1002                            request.capsule
1003                        ),
1004                    )
1005                }
1006                None => {
1007                    return outcome(
1008                        DelegationStatus::Failed,
1009                        &child.session_id,
1010                        format!(
1011                            "capsule '{}' answered GetTask with no task: {answer}",
1012                            request.capsule
1013                        ),
1014                    )
1015                }
1016            };
1017            match task.status.state {
1018                TaskState::Submitted | TaskState::Working | TaskState::InputRequired => continue,
1019                // Two places the answer can be, and both are read. A2A carries a completed task's
1020                // output in its artifacts: a murmur door attaches a `response` artifact to a
1021                // completed task that produced a response, and a capsule this runtime did not
1022                // build puts its answer there too. A task that ended with no response text has no
1023                // artifact, so the result file the child's runtime wrote into the directory this
1024                // parent composed for it is read as well.
1025                TaskState::Completed => {
1026                    let carried = task
1027                        .artifact(RESPONSE_ARTIFACT)
1028                        .and_then(|artifact| artifact.first_text())
1029                        .filter(|text| !text.is_empty())
1030                        .map(str::to_string);
1031                    let answer = carried
1032                        .or_else(|| {
1033                            crate::delegation::read_child_result(
1034                                Some(&child.records_root),
1035                                Some(&child.workdir_root),
1036                                Some(&task_id),
1037                            )
1038                            .map(|(_, text)| text)
1039                        })
1040                        .unwrap_or_default();
1041                    let handed =
1042                        crate::delegation::hand_over_answer(answer, &delegation_id, runtime_root);
1043                    let mut result = outcome(
1044                        DelegationStatus::Completed,
1045                        &child.session_id,
1046                        String::new(),
1047                    );
1048                    result.output = handed.text;
1049                    result.truncated = handed.truncated;
1050                    result.result_path = handed.full_path;
1051                    return result;
1052                }
1053                TaskState::Failed | TaskState::Rejected => {
1054                    return outcome(
1055                        DelegationStatus::Failed,
1056                        &child.session_id,
1057                        task.status_text()
1058                            .unwrap_or("the delegated capsule's task failed")
1059                            .to_string(),
1060                    )
1061                }
1062                ref other @ TaskState::Canceled => {
1063                    return outcome(
1064                        DelegationStatus::Failed,
1065                        &child.session_id,
1066                        format!(
1067                            "capsule '{}' reported an unknown task state: {}",
1068                            request.capsule,
1069                            other.as_str()
1070                        ),
1071                    )
1072                }
1073            }
1074        }
1075    }
1076}
1077
1078/// One `GetTask` for the child's task `task_id`, sent with `headers`: the child's answer, or the
1079/// transport error.
1080fn poll_task(
1081    capsule_url: &str,
1082    headers: &[(&str, &str)],
1083    delegation_id: &str,
1084    task_id: &str,
1085) -> Result<Value, String> {
1086    let body = json!({
1087        "jsonrpc": "2.0",
1088        "id": delegation_id,
1089        "method": "GetTask",
1090        "params": { "id": task_id }
1091    })
1092    .to_string();
1093    http_json("POST", capsule_url, Some(&body), headers)
1094}
1095
1096/// Deliver the task as the child's first user message, backing off while its listener comes up.
1097///
1098/// A child reports its URL when it binds, but the first connection can still land between the bind
1099/// and the first accept, so this backs off rather than failing on one refusal. `Ok` carries the id
1100/// the child gave the task it accepted, which is what makes the child's per-task result file
1101/// findable; `Err` is the sentence the delegating caller reports, already naming the capsule.
1102///
1103/// `door_token` is the child's operator token when it declares `network.authentication`. `stop`
1104/// is asked before every retry, and a send it stops returns `Err` at once rather than backing off
1105/// to [`SEND_DEADLINE`].
1106fn deliver_task(
1107    capsule_url: &str,
1108    door_token: Option<&crate::door_auth::DoorToken>,
1109    delegation_id: &str,
1110    request: &DelegationRequest,
1111    stop: &dyn Fn() -> bool,
1112) -> Result<String, String> {
1113    let send_body = json!({
1114        "jsonrpc": "2.0",
1115        "id": delegation_id,
1116        "method": "SendMessage",
1117        "params": send_message_params(user_message(
1118            &format!("msg_{delegation_id}"),
1119            None,
1120            &request.task
1121        ))
1122    })
1123    .to_string();
1124    let authorization = door_token.map(crate::door_auth::bearer_header);
1125    let headers = door_headers(authorization.as_deref());
1126    let send_deadline = Instant::now() + SEND_DEADLINE;
1127    let mut delay = Duration::from_millis(100);
1128    let sent = loop {
1129        match http_json("POST", capsule_url, Some(&send_body), &headers) {
1130            Ok(response) => break response,
1131            Err(error) if stop() => {
1132                return Err(format!(
1133                    "capsule '{}' was not handed its task: the delegating task was cancelled \
1134                     ({error})",
1135                    request.capsule
1136                ))
1137            }
1138            Err(_) if Instant::now() < send_deadline => {
1139                std::thread::sleep(delay);
1140                delay = (delay * 2).min(Duration::from_secs(2));
1141            }
1142            Err(error) => {
1143                return Err(format!(
1144                    "capsule '{}' did not accept a task within {}s: {error}",
1145                    request.capsule,
1146                    SEND_DEADLINE.as_secs()
1147                ))
1148            }
1149        }
1150    };
1151    match sent.get("result").map(SendMessageResult::from_result) {
1152        Some(Ok(SendMessageResult::Task(task))) if !task.id.is_empty() => Ok(task.id),
1153        Some(Ok(SendMessageResult::Message(message))) => Err(format!(
1154            "capsule '{}' answered the delivered task with a message and started no task: \
1155             {message}",
1156            request.capsule
1157        )),
1158        _ => Err(format!(
1159            "capsule '{}' answered the delivered task with no task id",
1160            request.capsule
1161        )),
1162    }
1163}
1164
1165/// End a child [`DelegationPlane::delegate`] holds because the task it was made for was
1166/// cancelled, and record that in the child's own `completion.json`.
1167///
1168/// The record is the one a `delegate-task` child's completion watcher writes when its parent ends
1169/// it on purpose: this launch started no watcher, so the launcher writes it here, after the reap.
1170fn end_for_canceled_task(
1171    request: &DelegationRequest,
1172    delegation_id: &str,
1173    child: &mut LaunchedChild,
1174    launched_at: Instant,
1175) {
1176    if let Err(error) = child.shutdown() {
1177        crate::runtime_err!("[capsule-runtime] delegation {delegation_id}: {error}");
1178    }
1179    crate::delegation::record_terminated(
1180        &child.records_root,
1181        crate::delegation::DelegationOutcome {
1182            delegation_id: delegation_id.to_string(),
1183            capsule_name: request.capsule.clone(),
1184            capsule_version: request.version.clone(),
1185            session_id: child.session_id.clone(),
1186            status: crate::delegation::DelegationStatus::Terminated,
1187            result_path: None,
1188            workdir: child.workdir.display().to_string(),
1189            duration_ms: crate::member_call::elapsed_ms(launched_at),
1190            detail: Some(crate::delegation::PARENT_ENDED_DETAIL.to_string()),
1191            reported_by: crate::delegation::Reporter::Launcher,
1192            delivered: false,
1193            delivery_error: None,
1194        },
1195    );
1196}
1197
1198/// `(output, truncated)`, with `output` cut by [`bound_output`] when it is over
1199/// [`MAX_OUTPUT_BYTES`].
1200pub(crate) fn bounded(output: String) -> (String, bool) {
1201    if output.len() > MAX_OUTPUT_BYTES {
1202        (bound_output(output), true)
1203    } else {
1204        (output, false)
1205    }
1206}
1207
1208/// `output` cut to [`MAX_OUTPUT_BYTES`] at a character boundary, with the cut marked.
1209fn bound_output(output: String) -> String {
1210    let mut end = MAX_OUTPUT_BYTES;
1211    while end > 0 && !output.is_char_boundary(end) {
1212        end -= 1;
1213    }
1214    format!("{}{CUT_MARK}", &output[..end])
1215}
1216
1217/// The host variables a child is handed: the names it declares that its parent declares too.
1218///
1219/// Order follows the child's declaration and no name appears twice, so what the launcher copies
1220/// reads as the child's own list. The clamp is the parent's second, independent statement of the
1221/// `capabilities.env.allow` envelope axis `mur-roost` already referees — a child can never hold a
1222/// variable the capsule that spawned it does not itself declare, whatever a daemon answered.
1223fn env_allow_intersection(child: &[String], parent: &[String]) -> Vec<String> {
1224    let mut allowed: Vec<String> = Vec::new();
1225    for name in child {
1226        if parent.contains(name) && !allowed.contains(name) {
1227            allowed.push(name.clone());
1228        }
1229    }
1230    allowed
1231}
1232
1233/// The bound on work this session hands off: [`DELEGATION_TIMEOUT_ENV`] when it names a positive
1234/// number of seconds, and `declared` — the capsule's own `lifecycle.delegation_deadline_secs` —
1235/// otherwise.
1236///
1237/// The one resolver for the one bound: a delegation's wait for its child, and a `call-member`
1238/// call's wait for the callee's answer ([`crate::member_call`]), both read it.
1239pub(crate) fn handed_off_work_deadline(declared: Duration) -> Duration {
1240    result_timeout_from(
1241        std::env::var(DELEGATION_TIMEOUT_ENV).ok().as_deref(),
1242        declared,
1243    )
1244}
1245
1246/// [`handed_off_work_deadline`]'s rule, without the environment read, so it is testable without
1247/// mutating process-wide state that every other test in this binary shares.
1248///
1249/// `declared` is taken as written, including zero: a capsule that declares `0` means the first
1250/// poll gives up. The override is only honoured for a positive integer, because an operator who
1251/// exported nonsense meant to change nothing rather than to remove the bound.
1252fn result_timeout_from(value: Option<&str>, declared: Duration) -> Duration {
1253    value
1254        .and_then(|value| value.trim().parse::<u64>().ok())
1255        .filter(|secs| *secs > 0)
1256        .map(Duration::from_secs)
1257        .unwrap_or(declared)
1258}
1259
1260/// The daemon's own refusal, lifted out of the transport error it arrived wrapped in.
1261///
1262/// `http_json` reports a non-2xx as its status line, its header block and its body, which is the
1263/// right shape for a log and the wrong one for a model: what an operator has to act on is the
1264/// referee's sentence naming the manifest key and the offending entry. Anything that is not a
1265/// daemon refusal — a connection that was never made, a body that is not the daemon's — is passed
1266/// through unchanged, and carries no token either way because `http_json` never puts a request
1267/// header or body into an error.
1268fn daemon_refusal(error: &str) -> String {
1269    error
1270        .split_once("; body: ")
1271        .and_then(|(_, body)| serde_json::from_str::<Value>(body).ok())
1272        .and_then(|body| {
1273            body.get("error")
1274                .and_then(Value::as_str)
1275                .map(str::to_string)
1276        })
1277        .unwrap_or_else(|| error.to_string())
1278}
1279
1280#[cfg(test)]
1281mod tests {
1282    use super::*;
1283    use std::path::PathBuf;
1284
1285    /// The five words a tool result's `status` can carry.
1286    #[test]
1287    fn every_status_has_one_wire_word() {
1288        assert_eq!(DelegationStatus::Started.as_str(), "started");
1289        assert_eq!(DelegationStatus::Completed.as_str(), "completed");
1290        assert_eq!(DelegationStatus::Failed.as_str(), "failed");
1291        assert_eq!(DelegationStatus::TimedOut.as_str(), "timed_out");
1292        assert_eq!(DelegationStatus::Refused.as_str(), "refused");
1293    }
1294
1295    /// The word `delegate` returns for a child it ended because its task was cancelled. A plan
1296    /// step reads it to leave the delegation's row to the task, so it is never a `delegation`
1297    /// line's `outcome`.
1298    #[test]
1299    fn a_canceled_delegation_has_its_own_word() {
1300        assert_eq!(DelegationStatus::Canceled.as_str(), "canceled");
1301    }
1302
1303    /// A stop predicate is a trait object, so `Debug` says whether there is one.
1304    #[test]
1305    fn an_origin_prints_whether_it_can_be_stopped() {
1306        let stoppable = DelegationOrigin {
1307            stop: Some(std::sync::Arc::new(|| false)),
1308            ..origin()
1309        };
1310        assert!(
1311            format!("{stoppable:?}").contains("stop: true"),
1312            "{stoppable:?}"
1313        );
1314        assert!(format!("{:?}", origin()).contains("stop: false"));
1315    }
1316
1317    fn plane() -> DelegationPlane {
1318        DelegationPlane::new(
1319            // Nothing here is reachable: every case below refuses before a request is built.
1320            "http://127.0.0.1:1".to_string(),
1321            SpawnCredential::new("msc1.test".to_string()),
1322            crate::workdir_fs::WorkdirRoot::open(&PathBuf::from("/tmp")).unwrap(),
1323            "ses_parent".to_string(),
1324            DELEGATION_RESULT_TIMEOUT,
1325            std::sync::Arc::new(murmur_artifact::LocalRegistry::new("/tmp")),
1326            Vec::new(),
1327            None,
1328        )
1329    }
1330
1331    fn request() -> DelegationRequest {
1332        DelegationRequest {
1333            capsule: "local/worker".to_string(),
1334            version: "0.1.0".to_string(),
1335            task: "t".to_string(),
1336        }
1337    }
1338
1339    /// A delegated task carries no origin claim, so the child records it as `event` and a child
1340    /// that does not consent to peer tasks still takes it — while the same door refuses a
1341    /// peer-origin message. Delegation needs no `exports.peer_tasks`.
1342    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
1343    async fn a_delegated_task_reaches_a_child_that_refuses_peer_tasks() {
1344        let (addr, _shutdown, mut task_rx) = crate::identity::serve_test_door(false).await;
1345        let url = format!("http://{addr}");
1346        let delivered = tokio::task::spawn_blocking(move || {
1347            deliver_task(&url, None, "dlg_0000000000000001", &request(), &|| false)
1348        })
1349        .await
1350        .unwrap()
1351        .expect("the child takes the delegated task");
1352        assert!(delivered.starts_with("tsk_"), "{delivered}");
1353        let incoming = task_rx
1354            .recv()
1355            .await
1356            .expect("the task reached the child's loop");
1357        assert_eq!(
1358            incoming.provenance.origin(),
1359            crate::origin::TaskOrigin::Event
1360        );
1361
1362        let refused = crate::outgoing::send_a2a_message(
1363            &addr,
1364            crate::outgoing::OutgoingMessage {
1365                message_id: "msg_peer".to_string(),
1366                context_id: None,
1367                text: "from a peer".to_string(),
1368            },
1369            None,
1370            None,
1371            None,
1372            None,
1373        )
1374        .await
1375        .expect_err("the same door refuses a peer message")
1376        .message;
1377        assert!(refused.contains("403 peer_not_accepted"), "{refused}");
1378    }
1379
1380    /// The task goes to the child as an A2A 1.0 `SendMessage` answered at acceptance, and is polled
1381    /// with `GetTask`, each naming the version beside the child's bearer.
1382    #[test]
1383    fn a_delegation_hands_over_and_polls_with_a2a_1_0_requests() {
1384        use crate::http_client::capture::answer_one;
1385
1386        let (addr, sent) = answer_one(
1387            json!({"jsonrpc": "2.0", "id": 1, "result": {"task": {"id": "tsk_child",
1388                "contextId": "c", "status": {"state": "TASK_STATE_SUBMITTED"}}}})
1389            .to_string(),
1390        );
1391        let token = crate::door_auth::DoorToken::new("tok".to_string());
1392        let task_id = deliver_task(
1393            &format!("http://{addr}"),
1394            Some(&token),
1395            "dlg_0000000000000001",
1396            &request(),
1397            &|| false,
1398        )
1399        .unwrap();
1400        assert_eq!(task_id, "tsk_child");
1401        let sent = sent.join().unwrap();
1402        assert_eq!(sent.method(), "SendMessage");
1403        assert!(sent.returns_immediately(), "{}", sent.body);
1404        assert_eq!(sent.header("a2a-version"), ["1.0"]);
1405        assert_eq!(sent.header("authorization"), ["Bearer tok"]);
1406        let message = sent.message();
1407        assert_eq!(message["role"], "ROLE_USER");
1408        assert_eq!(message["parts"][0]["text"], request().task);
1409        assert_eq!(message["parts"][0]["mediaType"], "text/plain");
1410
1411        let (addr, sent) = answer_one(
1412            json!({"jsonrpc": "2.0", "id": 1, "result": {"id": "tsk_child", "contextId": "c",
1413                "status": {"state": "TASK_STATE_WORKING"}}})
1414            .to_string(),
1415        );
1416        let authorization = crate::door_auth::bearer_header(&token);
1417        poll_task(
1418            &format!("http://{addr}"),
1419            &door_headers(Some(&authorization)),
1420            "dlg_0000000000000001",
1421            "tsk_child",
1422        )
1423        .unwrap();
1424        let sent = sent.join().unwrap();
1425        assert_eq!(sent.method(), "GetTask");
1426        assert_eq!(sent.header("a2a-version"), ["1.0"]);
1427        assert_eq!(sent.header("authorization"), ["Bearer tok"]);
1428    }
1429
1430    /// A plane that was never told this capsule's own address cannot start a delegation: the
1431    /// child would run and post its outcome nowhere. Refused before the daemon is asked, so no
1432    /// child directory and no `POST /spawn` exist to clean up.
1433    #[test]
1434    fn a_start_with_nowhere_to_report_is_refused_before_the_daemon_is_asked() {
1435        let started = plane().start(
1436            &request(),
1437            &DelegationOrigin {
1438                context_id: "ctx_parent".to_string(),
1439                ..DelegationOrigin::default()
1440            },
1441        );
1442        assert!(started.child.is_none());
1443        let result = started.result;
1444
1445        assert_eq!(result.status, DelegationStatus::Failed);
1446        assert!(result.delegation_id.is_empty(), "{result:?}");
1447        assert!(result.child_workdir.is_none(), "{result:?}");
1448        assert!(
1449            result.output.contains("nowhere to be reported"),
1450            "the sentence says why: {}",
1451            result.output
1452        );
1453    }
1454
1455    /// The conversation is half the address a completion is delivered against, so a delegation
1456    /// made from no conversation is refused on the same terms as one made from no address.
1457    #[test]
1458    fn a_start_with_no_conversation_is_refused_too() {
1459        let started = plane()
1460            .reporting_to("http://127.0.0.1:7000".to_string())
1461            .start(&request(), &DelegationOrigin::default());
1462        assert!(started.child.is_none());
1463        let result = started.result;
1464
1465        assert_eq!(result.status, DelegationStatus::Failed);
1466        assert!(
1467            result.output.contains("nowhere to be reported"),
1468            "the sentence says why: {}",
1469            result.output
1470        );
1471    }
1472
1473    /// A referee's refusal reaches the agent as the referee's own sentence, with the HTTP
1474    /// transcript `http_json` wrapped it in removed.
1475    #[test]
1476    fn a_refusal_is_lifted_out_of_its_http_transcript() {
1477        let violation = "capabilities.shell.allow: the child declares 'bash', which its parent \
1478                         does not hold — a spawned capsule can never hold more capability than \
1479                         the capsule that spawned it";
1480        let wrapped = format!(
1481            "HTTP request failed: HTTP/1.1 403 Forbidden\r\ncontent-type: application/json; \
1482             body: {}",
1483            json!({ "error": violation })
1484        );
1485        assert_eq!(daemon_refusal(&wrapped), violation);
1486    }
1487
1488    /// A failure that is not a daemon refusal is reported as it arrived rather than swallowed.
1489    #[test]
1490    fn a_transport_failure_survives_unchanged() {
1491        let refused = "failed to connect to 127.0.0.1:7700: Connection refused (os error 111)";
1492        assert_eq!(daemon_refusal(refused), refused);
1493        let unparseable = "HTTP request failed: HTTP/1.1 500 Internal Server Error; body: nope";
1494        assert_eq!(daemon_refusal(unparseable), unparseable);
1495    }
1496
1497    /// The bound is the constant unless the override names another, and a value that is not a
1498    /// positive number of seconds does not name one.
1499    #[test]
1500    fn the_result_timeout_reads_its_override() {
1501        let declared = Duration::from_secs(30);
1502        assert_eq!(
1503            result_timeout_from(Some(" 20 "), declared),
1504            Duration::from_secs(20)
1505        );
1506        for ignored in [None, Some(""), Some("0"), Some("-5"), Some("later")] {
1507            assert_eq!(
1508                result_timeout_from(ignored, declared),
1509                declared,
1510                "{ignored:?} is not a positive number of seconds, so the declared bound stands"
1511            );
1512        }
1513    }
1514
1515    /// The declared bound is the whole of the fallback, and a capsule that declares `0` means it.
1516    #[test]
1517    fn a_declared_deadline_of_zero_is_a_bound_and_not_an_absence() {
1518        assert_eq!(
1519            result_timeout_from(None, Duration::ZERO),
1520            Duration::ZERO,
1521            "0 gives up at the first poll rather than falling back to the default"
1522        );
1523        assert_eq!(
1524            result_timeout_from(None, DELEGATION_RESULT_TIMEOUT),
1525            DELEGATION_RESULT_TIMEOUT
1526        );
1527    }
1528
1529    /// The ceiling holds on every status, not only on an answer: a child's failure message and a
1530    /// daemon's refusal reach the same model context an answer does.
1531    #[test]
1532    fn the_output_ceiling_bounds_a_refusal_too() {
1533        let request = request();
1534        let result = DelegationResult::refused(&request, "x".repeat(MAX_OUTPUT_BYTES + 4096));
1535        assert!(result.truncated);
1536        assert!(result.output.len() <= MAX_OUTPUT_BYTES + " […]".len());
1537        assert!(result.output.ends_with(" […]"));
1538
1539        let short = DelegationResult::refused(&request, "no".to_string());
1540        assert!(!short.truncated);
1541        assert_eq!(short.output, "no");
1542    }
1543
1544    /// A cut lands on a character boundary rather than splitting a multi-byte character.
1545    #[test]
1546    fn a_cut_answer_stays_valid_utf8() {
1547        let output = "é".repeat(MAX_OUTPUT_BYTES);
1548        let bounded = bound_output(output);
1549        assert!(bounded.ends_with(" […]"));
1550        assert!(bounded.len() <= MAX_OUTPUT_BYTES + " […]".len());
1551    }
1552
1553    /// A child is handed the names both manifests declare, in the order the child declared them.
1554    #[test]
1555    fn a_child_is_handed_only_what_both_manifests_declare() {
1556        let names =
1557            |list: &[&str]| -> Vec<String> { list.iter().map(|name| name.to_string()).collect() };
1558
1559        assert_eq!(
1560            env_allow_intersection(&names(&["A", "B", "C"]), &names(&["B", "C", "D"])),
1561            names(&["B", "C"])
1562        );
1563        assert_eq!(
1564            env_allow_intersection(&names(&["A"]), &names(&[])),
1565            Vec::<String>::new(),
1566            "a parent that declares nothing hands nothing on"
1567        );
1568        assert_eq!(
1569            env_allow_intersection(&names(&[]), &names(&["A"])),
1570            Vec::<String>::new(),
1571            "a child that declares nothing asks for nothing"
1572        );
1573    }
1574
1575    /// The order is the child's, and a name it declared twice is copied once.
1576    #[test]
1577    fn the_intersection_keeps_the_childs_order_and_names_nothing_twice() {
1578        let names =
1579            |list: &[&str]| -> Vec<String> { list.iter().map(|name| name.to_string()).collect() };
1580
1581        assert_eq!(
1582            env_allow_intersection(&names(&["C", "A", "C"]), &names(&["A", "B", "C"])),
1583            names(&["C", "A"])
1584        );
1585    }
1586
1587    /// A capsule the parent's own store cannot resolve fails the read by name.
1588    ///
1589    /// The delegation ends there rather than launching a child without the variables its manifest
1590    /// declares: that child would die at its own manifest load, in a process nobody is reading.
1591    #[test]
1592    fn a_capsule_the_store_cannot_resolve_fails_the_read_by_name() {
1593        let store = tempfile::tempdir().unwrap();
1594        let plane = plane_over(&store, &["MURMUR_TEST_PROVIDER_KEY"]);
1595
1596        let error = plane
1597            .child_env_allow("local/worker", "0.1.0")
1598            .expect_err("an empty store resolves nothing");
1599        let message = error.to_string();
1600        assert!(message.contains("worker"), "{message}");
1601        assert!(message.contains("0.1.0"), "{message}");
1602        assert!(message.contains("capabilities.env.allow"), "{message}");
1603    }
1604
1605    /// A `local/worker@0.1.0` in a store of its own, declaring `capabilities.env.allow: [A, B, C]`.
1606    ///
1607    /// Packed the way `mur publish` packs one — a `murmur.yaml` entry in a `.mur.zip` — because
1608    /// [`DelegationPlane::child_env_allow`] reads the child's declaration out of the archive
1609    /// rather than out of a directory.
1610    fn store_with_worker() -> tempfile::TempDir {
1611        store_with_worker_declaring(
1612            "name: worker\nversion: 0.1.0\ncapabilities:\n  env:\n    allow: [A, B, C]\n",
1613        )
1614    }
1615
1616    /// [`store_with_worker`], with `local/worker@0.1.0`'s packed manifest being `manifest`.
1617    fn store_with_worker_declaring(manifest: &str) -> tempfile::TempDir {
1618        let store = tempfile::tempdir().unwrap();
1619        let mut zipped = std::io::Cursor::new(Vec::<u8>::new());
1620        {
1621            let mut zip = zip::ZipWriter::new(&mut zipped);
1622            zip.start_file(
1623                "murmur.yaml",
1624                zip::write::SimpleFileOptions::default()
1625                    .compression_method(zip::CompressionMethod::Deflated),
1626            )
1627            .unwrap();
1628            std::io::Write::write_all(&mut zip, manifest.as_bytes()).unwrap();
1629            zip.finish().unwrap();
1630        }
1631        murmur_artifact::Registry::publish(
1632            &murmur_artifact::LocalRegistry::new(store.path()),
1633            murmur_artifact::ArtifactMeta {
1634                name: murmur_artifact::ArtifactName::parse("local/worker").unwrap(),
1635                version: "0.1.0".to_string(),
1636                runtime: murmur_artifact::RuntimeType::Wasm,
1637                artifact_runtime: "capsule".to_string(),
1638                description: None,
1639                tags: Vec::new(),
1640                wit_contracts: None,
1641                owner_id: None,
1642                built_at_level: None,
1643            },
1644            &zipped.into_inner(),
1645        )
1646        .unwrap();
1647        store
1648    }
1649
1650    /// A plane over the given store, holding the given `capabilities.env.allow` of its own.
1651    fn plane_over(store: &tempfile::TempDir, parent_env_allow: &[&str]) -> DelegationPlane {
1652        plane_in_formation(store, parent_env_allow, None)
1653    }
1654
1655    /// [`plane_over`] for a session in the given formation.
1656    fn plane_in_formation(
1657        store: &tempfile::TempDir,
1658        parent_env_allow: &[&str],
1659        formation_id: Option<FormationId>,
1660    ) -> DelegationPlane {
1661        DelegationPlane::new(
1662            "http://127.0.0.1:1".to_string(),
1663            SpawnCredential::new("msc1.test".to_string()),
1664            crate::workdir_fs::WorkdirRoot::open(&PathBuf::from("/tmp")).unwrap(),
1665            "ses_parent".to_string(),
1666            DELEGATION_RESULT_TIMEOUT,
1667            std::sync::Arc::new(murmur_artifact::LocalRegistry::new(store.path())),
1668            parent_env_allow.iter().map(|n| n.to_string()).collect(),
1669            formation_id,
1670        )
1671        .reporting_to("http://127.0.0.1:7000".to_string())
1672    }
1673
1674    /// The conversation both paths delegate from.
1675    fn origin() -> DelegationOrigin<'static> {
1676        DelegationOrigin {
1677            context_id: "ctx_parent".to_string(),
1678            ..DelegationOrigin::default()
1679        }
1680    }
1681
1682    /// Where `start` tells a child to report, as it composes it.
1683    fn completion_address() -> CompletionAddress {
1684        CompletionAddress {
1685            url: "http://127.0.0.1:7000".to_string(),
1686        }
1687    }
1688
1689    /// Every launch this crate makes is composed in one place, so a field can never be right on
1690    /// one path and wrong on the other.
1691    ///
1692    /// Reads every `.rs` under `src/`, drops line comments and the trailing `mod tests`, and
1693    /// counts the literal. `child_launch.rs` is excluded: it defines the type and builds one in
1694    /// its own tests, where the launcher rather than a launch path is the unit under test.
1695    #[test]
1696    fn a_launch_request_is_built_in_exactly_one_place() {
1697        let mut found: Vec<(String, usize)> = Vec::new();
1698
1699        for (name, text) in crate::source_scan::crate_sources() {
1700            if name == "child_launch.rs" {
1701                continue;
1702            }
1703            let count = crate::source_scan::production_part(&text)
1704                .lines()
1705                .filter(|line| !line.trim_start().starts_with("//"))
1706                .map(|line| line.matches("ChildLaunchRequest {").count())
1707                .sum::<usize>();
1708            if count > 0 {
1709                found.push((name, count));
1710            }
1711        }
1712
1713        found.sort();
1714        assert_eq!(
1715            found,
1716            vec![("delegation_plane.rs".to_string(), 1)],
1717            "`ChildLaunchRequest` may be built in exactly one place: \
1718`DelegationPlane::launch_request`, which every launch path in this crate goes through so that \
1719what a child is handed cannot differ between them. Found: {found:?}. Route a new launch path \
1720through that constructor rather than composing a second request literal."
1721        );
1722    }
1723
1724    /// A child whose declaration the parent cannot read starts no process, on the path that names
1725    /// a completion address and on the path that does not.
1726    #[test]
1727    fn an_unreadable_child_declaration_composes_no_launch_on_either_path() {
1728        let store = tempfile::tempdir().unwrap();
1729        let plane = plane_over(&store, &["A"]);
1730
1731        for report_to in [Some(completion_address()), None] {
1732            let named = if report_to.is_some() {
1733                "the started path"
1734            } else {
1735                "the plan-step path"
1736            };
1737            let error = plane
1738                .launch_request(
1739                    &request(),
1740                    &origin(),
1741                    SpawnApproval::new("approved".to_string()),
1742                    report_to,
1743                )
1744                .err()
1745                .unwrap_or_else(|| panic!("{named}: an empty store resolves nothing"));
1746
1747            let message = error.to_string();
1748            assert!(message.contains("worker"), "{named}: {message}");
1749            assert!(message.contains("0.1.0"), "{named}: {message}");
1750            assert!(
1751                message.contains("capabilities.env.allow"),
1752                "{named}: {message}"
1753            );
1754        }
1755    }
1756
1757    /// Every launch carries what both manifests declare, whichever path composed it. The two
1758    /// helper tests above pin the intersection itself; this one pins that a launch is handed its
1759    /// result.
1760    #[test]
1761    fn every_launch_carries_what_both_manifests_declare() {
1762        let store = store_with_worker();
1763        let plane = plane_over(&store, &["B", "C", "D"]);
1764
1765        for report_to in [Some(completion_address()), None] {
1766            let named = if report_to.is_some() {
1767                "the started path"
1768            } else {
1769                "the plan-step path"
1770            };
1771            let launch = plane
1772                .launch_request(
1773                    &request(),
1774                    &origin(),
1775                    SpawnApproval::new("approved".to_string()),
1776                    report_to,
1777                )
1778                .unwrap_or_else(|error| panic!("{named}: {error}"));
1779
1780            assert_eq!(
1781                launch.child_env_allow,
1782                vec!["B".to_string(), "C".to_string()],
1783                "{named} hands on what both manifests declare, in the child's order"
1784            );
1785        }
1786    }
1787
1788    /// A capsule listed under `capabilities.spawn.workdir_snapshot` — by its full name, as
1789    /// `delegate-task` spawns it — is launched with this session's work folder to copy, bounded by
1790    /// the child's own `workdir_max_bytes` or the default; every other capsule with none, on both
1791    /// paths.
1792    #[test]
1793    fn workdir_snapshot_is_composed_for_a_listed_capsule_only() {
1794        let work = tempfile::tempdir().unwrap();
1795        let runtime = tempfile::tempdir().unwrap();
1796        let source = crate::workdir_fs::WorkdirRoot::open(work.path()).unwrap();
1797        let runtime_root = crate::workdir_fs::WorkdirRoot::open(runtime.path()).unwrap();
1798        let grant = |capsules: &[&str]| {
1799            WorkdirSnapshotGrant::new(
1800                capsules.iter().map(|name| name.to_string()).collect(),
1801                source.clone(),
1802                &[&runtime_root],
1803            )
1804        };
1805        let compose = |plane: &DelegationPlane, report_to: Option<CompletionAddress>| {
1806            plane
1807                .launch_request(
1808                    &request(),
1809                    &origin(),
1810                    SpawnApproval::new("approved".to_string()),
1811                    report_to,
1812                )
1813                .expect("the store holds the child's declaration")
1814                .parent_workdir_snapshot
1815        };
1816
1817        let defaulted = store_with_worker();
1818        let declaring = store_with_worker_declaring(
1819            "name: worker\nversion: 0.1.0\ncapabilities:\n  resources:\n    \
1820             workdir_max_bytes: 1024\n",
1821        );
1822        for report_to in [Some(completion_address()), None] {
1823            let listed = compose(
1824                &plane_over(&defaulted, &[]).snapshotting_workdir(grant(&["local/worker"])),
1825                report_to.clone(),
1826            )
1827            .expect("a listed capsule is handed the work folder");
1828            assert_eq!(listed.root.path(), work.path());
1829            assert_eq!(
1830                listed.limits.max_bytes,
1831                crate::resources::DEFAULT_WORKDIR_MAX_BYTES
1832            );
1833            assert_eq!(
1834                listed.limits.excluded,
1835                vec![runtime_root.identity().unwrap()]
1836            );
1837
1838            let bounded = compose(
1839                &plane_over(&declaring, &[]).snapshotting_workdir(grant(&["local/worker"])),
1840                report_to.clone(),
1841            )
1842            .unwrap();
1843            assert_eq!(bounded.limits.max_bytes, 1024);
1844
1845            for plane in [
1846                plane_over(&defaulted, &[]),
1847                plane_over(&defaulted, &[]).snapshotting_workdir(grant(&[])),
1848                plane_over(&defaulted, &[]).snapshotting_workdir(grant(&["local/other"])),
1849                plane_over(&defaulted, &[]).snapshotting_workdir(grant(&["worker"])),
1850            ] {
1851                assert!(
1852                    compose(&plane, report_to.clone()).is_none(),
1853                    "{report_to:?}"
1854                );
1855            }
1856        }
1857    }
1858
1859    /// A bounded plane hands its deadline to every launch, on the started path and the plan-step
1860    /// path alike; an unbounded one hands none.
1861    #[test]
1862    fn a_bounded_plane_hands_its_deadline_to_every_launch() {
1863        let store = store_with_worker();
1864        let deadline =
1865            SessionDeadline::after(std::time::Instant::now(), std::time::SystemTime::now(), 60);
1866        let bounded = plane_over(&store, &[]).bounded_by(Some(deadline));
1867        let unbounded = plane_over(&store, &[]);
1868
1869        for report_to in [Some(completion_address()), None] {
1870            let compose = |plane: &DelegationPlane| {
1871                plane
1872                    .launch_request(
1873                        &request(),
1874                        &origin(),
1875                        SpawnApproval::new("approved".to_string()),
1876                        report_to.clone(),
1877                    )
1878                    .expect("the store holds the child's declaration")
1879            };
1880            assert_eq!(compose(&bounded).deadline, Some(deadline), "{report_to:?}");
1881            assert_eq!(compose(&unbounded).deadline, None, "{report_to:?}");
1882        }
1883    }
1884
1885    /// The two paths differ in where the outcome goes and in what bounds the wait for it. Every
1886    /// other thing a child is launched with is the same, because it comes from the same plane.
1887    #[test]
1888    fn the_two_paths_differ_in_the_completion_target_and_its_deadline_alone() {
1889        let store = store_with_worker();
1890        let plane = plane_over(&store, &["B", "C", "D"]);
1891        let compose = |report_to| {
1892            plane
1893                .launch_request(
1894                    &request(),
1895                    &origin(),
1896                    SpawnApproval::new("approved".to_string()),
1897                    report_to,
1898                )
1899                .expect("the store holds the child's declaration")
1900        };
1901
1902        let started = compose(Some(completion_address()));
1903        let stepped = compose(None);
1904
1905        let started_spawner = started
1906            .spawner
1907            .as_ref()
1908            .expect("a named session and context");
1909        assert_eq!(started_spawner.report_to, Some(completion_address()));
1910        assert_eq!(started.completion_deadline, Some(plane.result_timeout()));
1911
1912        let stepped_spawner = stepped
1913            .spawner
1914            .as_ref()
1915            .expect("a named session and context");
1916        assert_eq!(stepped_spawner.report_to, None);
1917        assert_eq!(stepped.completion_deadline, None);
1918
1919        assert_eq!(
1920            started.parent_records_root.path(),
1921            stepped.parent_records_root.path()
1922        );
1923        assert_eq!(started.capsule_name, stepped.capsule_name);
1924        assert_eq!(started.capsule_version, stepped.capsule_version);
1925        assert_eq!(started.grant.expose(), stepped.grant.expose());
1926        assert_eq!(started.child_env_allow, stepped.child_env_allow);
1927        assert_eq!(started.roost_url, stepped.roost_url);
1928        assert_eq!(started_spawner.session_id, stepped_spawner.session_id);
1929        assert_eq!(started_spawner.context_id, stepped_spawner.context_id);
1930    }
1931
1932    /// Every launch joins the delegating session's formation, on both paths and whether or not
1933    /// the launch names a lineage; a session in no formation hands its children none.
1934    #[test]
1935    fn every_launch_carries_the_sessions_formation_id() {
1936        let store = store_with_worker();
1937        let member = FormationId::mint();
1938        for formation_id in [Some(member), None] {
1939            let plane = plane_in_formation(&store, &["B"], formation_id.clone());
1940            assert_eq!(plane.formation_id(), formation_id.as_ref());
1941            let unnamed = DelegationOrigin::default();
1942            for (origin, report_to) in [
1943                (origin(), Some(completion_address())),
1944                (origin(), None),
1945                (unnamed, None),
1946            ] {
1947                let launch = plane
1948                    .launch_request(
1949                        &request(),
1950                        &origin,
1951                        SpawnApproval::new("approved".to_string()),
1952                        report_to,
1953                    )
1954                    .expect("the store holds the child's declaration");
1955                assert_eq!(
1956                    launch.formation_id,
1957                    formation_id,
1958                    "spawner present: {}",
1959                    launch.spawner.is_some()
1960                );
1961            }
1962        }
1963    }
1964
1965    /// A delegated child is never handed its parent's formation channel or callees: the request
1966    /// carries no such field, and the environment built from it holds neither
1967    /// `MURMUR_FORMATION_CHANNEL` nor `MURMUR_FORMATION_PEERS` even when the parent's process
1968    /// carries both and the child allowlists both names. The formation id is still handed on.
1969    #[test]
1970    fn a_launch_request_hands_a_child_no_formation_channel_or_peers() {
1971        let _guard = crate::formation::FORMATION_ENV_LOCK
1972            .lock()
1973            .unwrap_or_else(|poisoned| poisoned.into_inner());
1974        let peers = crate::formation::FORMATION_PEERS_ENV;
1975        let channel = crate::formation_credentials::FORMATION_CHANNEL_ENV;
1976        let store = store_with_worker();
1977        let plane = plane_in_formation(&store, &["B", peers, channel], Some(FormationId::mint()));
1978        let mut launch = plane
1979            .launch_request(
1980                &request(),
1981                &origin(),
1982                SpawnApproval::new("approved".to_string()),
1983                Some(completion_address()),
1984            )
1985            .expect("the store holds the child's declaration");
1986        launch.child_env_allow.push(peers.to_string());
1987        launch.child_env_allow.push(channel.to_string());
1988        std::env::set_var(peers, "coder=http://coder.formation.invalid");
1989        std::env::set_var(channel, "7");
1990        let env = crate::child_launch::child_environment(&launch, None);
1991        std::env::remove_var(peers);
1992        std::env::remove_var(channel);
1993        assert!(
1994            env.iter().all(|(key, _)| key != peers && key != channel),
1995            "{env:?}"
1996        );
1997        assert!(
1998            env.iter()
1999                .any(|(key, _)| key == crate::formation::FORMATION_ID_ENV),
2000            "the formation id is still handed on: {env:?}"
2001        );
2002    }
2003
2004    /// A delegated child is handed neither of its parent's lifelines, even when it allowlists both
2005    /// names and the parent's process carries both: each names a descriptor of the parent's own
2006    /// process. The child holds a spawner lifeline of its own, which the launch hands it beside
2007    /// this environment, and no formation lifeline at all. The formation id is still handed on.
2008    #[test]
2009    fn a_launch_request_hands_a_child_its_own_spawner_lifeline_and_no_formation_one() {
2010        let _guard = crate::formation::FORMATION_ENV_LOCK
2011            .lock()
2012            .unwrap_or_else(|poisoned| poisoned.into_inner());
2013        let formation = crate::lifeline::FORMATION_LIFELINE_ENV;
2014        let spawner = crate::lifeline::SPAWNER_LIFELINE_ENV;
2015        let store = store_with_worker();
2016        let plane = plane_in_formation(
2017            &store,
2018            &["B", formation, spawner],
2019            Some(FormationId::mint()),
2020        );
2021        let mut launch = plane
2022            .launch_request(
2023                &request(),
2024                &origin(),
2025                SpawnApproval::new("approved".to_string()),
2026                Some(completion_address()),
2027            )
2028            .expect("the store holds the child's declaration");
2029        launch.child_env_allow.push(formation.to_string());
2030        launch.child_env_allow.push(spawner.to_string());
2031        std::env::set_var(formation, "7");
2032        std::env::set_var(spawner, "8");
2033        let env = crate::child_launch::child_environment(&launch, None);
2034        std::env::remove_var(formation);
2035        std::env::remove_var(spawner);
2036        assert!(
2037            env.iter()
2038                .all(|(key, _)| key != formation && key != spawner),
2039            "{env:?}"
2040        );
2041        assert!(
2042            env.iter().all(|(_, value)| value != "7" && value != "8"),
2043            "{env:?}"
2044        );
2045        assert!(
2046            env.iter()
2047                .any(|(key, _)| key == crate::formation::FORMATION_ID_ENV),
2048            "the formation id is still handed on: {env:?}"
2049        );
2050    }
2051
2052    /// The trailing slash is taken off once, so no request is built against `//spawn`.
2053    #[test]
2054    fn a_trailing_slash_on_the_daemon_url_is_trimmed_once() {
2055        let plane = DelegationPlane::new(
2056            "http://127.0.0.1:7700/".to_string(),
2057            SpawnCredential::new("msc1.test".to_string()),
2058            crate::workdir_fs::WorkdirRoot::open(&PathBuf::from("/tmp")).unwrap(),
2059            "ses_parent".to_string(),
2060            DELEGATION_RESULT_TIMEOUT,
2061            std::sync::Arc::new(murmur_artifact::LocalRegistry::new("/tmp")),
2062            Vec::new(),
2063            None,
2064        );
2065        assert_eq!(plane.roost_url, "http://127.0.0.1:7700");
2066    }
2067}