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}