use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use serde_json::{json, Value};
use crate::child_launch::{
launch_child_capsule, workdir_relative_to, ChildLaunchRequest, LaunchedChild,
};
use crate::delegation::{CompletionAddress, Spawner};
use crate::errors::RuntimeError;
use crate::formation::FormationId;
use crate::http_client::http_json;
use crate::spawn_credential::{SpawnApproval, SpawnCredential, SPAWN_CREDENTIAL_HEADER};
pub const DELEGATION_RESULT_TIMEOUT: Duration = Duration::from_secs(600);
pub const DELEGATION_TIMEOUT_ENV: &str = "MURMUR_DELEGATION_TIMEOUT_SECS";
const SEND_DEADLINE: Duration = Duration::from_secs(30);
const POLL_INTERVAL: Duration = Duration::from_millis(500);
pub(crate) const MAX_OUTPUT_BYTES: usize = 64 * 1024;
#[derive(Debug, Clone)]
pub struct DelegationRequest {
pub capsule: String,
pub version: String,
pub task: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DelegationStatus {
Started,
Completed,
Failed,
TimedOut,
Refused,
Canceled,
}
impl DelegationStatus {
pub fn as_str(self) -> &'static str {
match self {
Self::Started => "started",
Self::Completed => "completed",
Self::Failed => "failed",
Self::TimedOut => "timed_out",
Self::Refused => "refused",
Self::Canceled => "canceled",
}
}
}
#[derive(Debug, Clone)]
pub struct DelegationResult {
pub delegation_id: String,
pub session_id: String,
pub capsule: String,
pub version: String,
pub status: DelegationStatus,
pub output: String,
pub result_path: Option<String>,
pub truncated: bool,
pub child_workdir: Option<String>,
}
impl DelegationResult {
fn unmade(request: &DelegationRequest, status: DelegationStatus, reason: String) -> Self {
let (output, truncated) = bounded(reason);
Self {
delegation_id: String::new(),
session_id: String::new(),
capsule: request.capsule.clone(),
version: request.version.clone(),
status,
output,
result_path: None,
truncated,
child_workdir: None,
}
}
fn refused(request: &DelegationRequest, reason: String) -> Self {
Self::unmade(request, DelegationStatus::Refused, reason)
}
}
#[derive(Debug)]
pub struct StartedDelegation {
pub result: DelegationResult,
pub child: Option<LaunchedChild>,
}
impl StartedDelegation {
fn without_child(result: DelegationResult) -> Self {
Self {
result,
child: None,
}
}
}
#[derive(Debug, Clone)]
pub struct DelegationLaunch {
pub delegation_id: String,
pub capsule: String,
pub version: String,
pub child_session_id: String,
pub child_workdir: String,
}
#[derive(Clone, Default)]
pub struct DelegationOrigin<'a> {
pub context_id: String,
pub launched: Option<std::sync::Arc<dyn Fn(DelegationLaunch) + Send + Sync + 'a>>,
pub stop: Option<std::sync::Arc<dyn Fn() -> bool + Send + Sync + 'a>>,
}
impl std::fmt::Debug for DelegationOrigin<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DelegationOrigin")
.field("context_id", &self.context_id)
.field("launched", &self.launched.is_some())
.field("stop", &self.stop.is_some())
.finish()
}
}
pub struct DelegationPlane {
roost_url: String,
credential: SpawnCredential,
accessible_workdir: PathBuf,
result_timeout: Duration,
session_id: String,
registry: std::sync::Arc<dyn murmur_artifact::Registry>,
parent_env_allow: Vec<String>,
own_url: String,
formation_id: Option<FormationId>,
}
impl DelegationPlane {
#[allow(clippy::too_many_arguments)]
pub fn new(
roost_url: String,
credential: SpawnCredential,
accessible_workdir: PathBuf,
session_id: String,
declared_deadline: Duration,
registry: std::sync::Arc<dyn murmur_artifact::Registry>,
parent_env_allow: Vec<String>,
formation_id: Option<FormationId>,
) -> Self {
Self {
roost_url: roost_url.trim_end_matches('/').to_string(),
credential,
accessible_workdir,
result_timeout: handed_off_work_deadline(declared_deadline),
session_id,
registry,
parent_env_allow,
own_url: String::new(),
formation_id,
}
}
pub fn reporting_to(mut self, own_url: String) -> Self {
self.own_url = own_url;
self
}
pub fn formation_id(&self) -> Option<&FormationId> {
self.formation_id.as_ref()
}
pub fn result_timeout(&self) -> Duration {
self.result_timeout
}
fn child_env_allow(&self, capsule: &str, version: &str) -> Result<Vec<String>, RuntimeError> {
let resolved = self
.registry
.resolve_with_platform(capsule, version, Some(murmur_artifact::current_platform()))
.map_err(|error| {
RuntimeError::Runtime(format!(
"cannot read what '{capsule}@{version}' declares under \
capabilities.env.allow: {error}"
))
})?;
let manifest_yaml =
crate::artifact::extract_manifest_yaml(capsule, version, &resolved.bytes)?;
let declared =
crate::artifact::extract_declared_env_allow(capsule, version, &manifest_yaml)?;
Ok(env_allow_intersection(&declared, &self.parent_env_allow))
}
pub(crate) fn credential(&self) -> SpawnCredential {
self.credential.clone()
}
fn workdir_relative(&self, workdir: &Path) -> String {
workdir_relative_to(workdir, &self.accessible_workdir)
}
fn approval_for(&self, request: &DelegationRequest) -> Result<SpawnApproval, String> {
let spawn_body = json!({ "name": request.capsule, "version": request.version }).to_string();
let permission = match http_json(
"POST",
&format!("{}/spawn", self.roost_url),
Some(&spawn_body),
&[(SPAWN_CREDENTIAL_HEADER, self.credential.expose())],
) {
Ok(value) => value,
Err(error) => return Err(daemon_refusal(&error)),
};
let Some(approval) = permission.get("approval").and_then(Value::as_str) else {
return Err("mur-roost spawn response missing approval".to_string());
};
Ok(SpawnApproval::new(approval.to_string()))
}
fn launch_request(
&self,
request: &DelegationRequest,
origin: &DelegationOrigin<'_>,
grant: SpawnApproval,
report_to: Option<CompletionAddress>,
) -> Result<ChildLaunchRequest, RuntimeError> {
let completion_deadline = report_to.as_ref().map(|_| self.result_timeout);
Ok(ChildLaunchRequest {
parent_accessible_workdir: self.accessible_workdir.clone(),
capsule_name: request.capsule.clone(),
capsule_version: request.version.clone(),
grant,
child_env_allow: self.child_env_allow(&request.capsule, &request.version)?,
roost_url: self.roost_url.clone(),
spawner: (!self.session_id.is_empty() && !origin.context_id.is_empty()).then(|| {
Spawner {
session_id: self.session_id.clone(),
context_id: origin.context_id.clone(),
report_to,
}
}),
completion_deadline,
formation_id: self.formation_id.clone(),
})
}
fn announce(
&self,
request: &DelegationRequest,
origin: &DelegationOrigin<'_>,
child: &crate::child_launch::LaunchedChild,
delegation_id: &str,
) {
if let Some(launched) = origin
.launched
.as_ref()
.filter(|_| !delegation_id.is_empty())
{
launched(DelegationLaunch {
delegation_id: delegation_id.to_string(),
capsule: request.capsule.clone(),
version: request.version.clone(),
child_session_id: child.session_id.clone(),
child_workdir: self.workdir_relative(&child.workdir),
});
}
}
pub fn start(
&self,
request: &DelegationRequest,
origin: &DelegationOrigin<'_>,
) -> StartedDelegation {
if self.own_url.is_empty() || self.session_id.is_empty() || origin.context_id.is_empty() {
return StartedDelegation::without_child(DelegationResult::unmade(
request,
DelegationStatus::Failed,
"this capsule cannot start a delegation: a sub-capsule's outcome is posted to \
this runtime's own endpoint, addressed to its session and conversation, and \
this session does not hold all three — so the outcome would have nowhere to be \
reported"
.to_string(),
));
}
let grant = match self.approval_for(request) {
Ok(grant) => grant,
Err(reason) => {
return StartedDelegation::without_child(DelegationResult::refused(request, reason))
}
};
let launch = match self.launch_request(
request,
origin,
grant,
Some(CompletionAddress {
url: self.own_url.clone(),
}),
) {
Ok(launch) => launch,
Err(error) => {
return StartedDelegation::without_child(DelegationResult::unmade(
request,
DelegationStatus::Failed,
error.to_string(),
))
}
};
let child = match launch_child_capsule(launch) {
Ok(child) => child,
Err(error) => {
return StartedDelegation::without_child(DelegationResult::unmade(
request,
DelegationStatus::Failed,
error.to_string(),
))
}
};
let delegation_id = child.delegation_id.clone().unwrap_or_default();
self.announce(request, origin, &child, &delegation_id);
let child_workdir = self.workdir_relative(&child.workdir);
let failed = |session_id: &str, output: String| {
let (output, truncated) = bounded(output);
StartedDelegation::without_child(DelegationResult {
delegation_id: delegation_id.clone(),
session_id: session_id.to_string(),
capsule: request.capsule.clone(),
version: request.version.clone(),
status: DelegationStatus::Failed,
output,
result_path: None,
truncated,
child_workdir: None,
})
};
if child.capsule_url.is_empty() {
return failed(
&child.session_id,
format!(
"capsule '{}' bound no address, so it cannot be sent a task; delegation needs \
a capsule that serves A2A",
request.capsule
),
);
}
let capsule_url = child.capsule_url.trim_end_matches('/').to_string();
let door_token = child.door_token.clone();
if let Err(reason) = deliver_task(
&capsule_url,
door_token.as_ref(),
&delegation_id,
request,
&|| false,
) {
return failed(&child.session_id, reason);
}
let session_id = child.session_id.clone();
StartedDelegation {
result: DelegationResult {
delegation_id,
session_id,
capsule: request.capsule.clone(),
version: request.version.clone(),
status: DelegationStatus::Started,
output: String::new(),
result_path: None,
truncated: false,
child_workdir: Some(child_workdir),
},
child: Some(child),
}
}
pub fn delegate(
&self,
request: &DelegationRequest,
origin: &DelegationOrigin<'_>,
) -> DelegationResult {
let grant = match self.approval_for(request) {
Ok(grant) => grant,
Err(reason) => return DelegationResult::refused(request, reason),
};
let launch = match self.launch_request(request, origin, grant, None) {
Ok(launch) => launch,
Err(error) => {
return DelegationResult::unmade(
request,
DelegationStatus::Failed,
error.to_string(),
)
}
};
let launched_at = Instant::now();
let mut child = match launch_child_capsule(launch) {
Ok(child) => child,
Err(error) => {
return DelegationResult::unmade(
request,
DelegationStatus::Failed,
error.to_string(),
)
}
};
let delegation_id = child.delegation_id.clone().unwrap_or_default();
self.announce(request, origin, &child, &delegation_id);
let stopped = || origin.stop.as_ref().is_some_and(|stop| stop());
let outcome = |status, session_id: &str, output: String| {
let (output, truncated) = bounded(output);
DelegationResult {
delegation_id: delegation_id.clone(),
session_id: session_id.to_string(),
capsule: request.capsule.clone(),
version: request.version.clone(),
status,
output,
result_path: None,
truncated,
child_workdir: None,
}
};
let canceled = |child: &mut LaunchedChild| {
end_for_canceled_task(request, &delegation_id, child, launched_at);
outcome(
DelegationStatus::Canceled,
&child.session_id,
format!(
"capsule '{}' was stopped because the task that started it was cancelled",
request.capsule
),
)
};
if stopped() {
return canceled(&mut child);
}
if child.capsule_url.is_empty() {
return outcome(
DelegationStatus::Failed,
&child.session_id,
format!(
"capsule '{}' bound no address, so it cannot be sent a task; delegation needs \
a capsule that serves A2A",
request.capsule
),
);
}
let capsule_url = child.capsule_url.trim_end_matches('/').to_string();
let authorization = child
.door_token
.as_ref()
.map(crate::door_auth::bearer_header);
let headers: Vec<(&str, &str)> = authorization
.as_deref()
.map(|value| ("Authorization", value))
.into_iter()
.collect();
let task_id = match deliver_task(
&capsule_url,
child.door_token.as_ref(),
&delegation_id,
request,
&stopped,
) {
Ok(task_id) => task_id,
Err(_) if stopped() => return canceled(&mut child),
Err(reason) => return outcome(DelegationStatus::Failed, &child.session_id, reason),
};
let poll_body = json!({
"jsonrpc": "2.0",
"id": delegation_id,
"method": "tasks/get",
"params": { "id": task_id }
})
.to_string();
let poll_deadline = Instant::now() + self.result_timeout;
loop {
if stopped() {
return canceled(&mut child);
}
if Instant::now() >= poll_deadline {
return outcome(
DelegationStatus::TimedOut,
&child.session_id,
format!(
"capsule '{}' did not answer within {}s; the delegation was ended and the \
child was stopped",
request.capsule,
self.result_timeout.as_secs()
),
);
}
std::thread::sleep(POLL_INTERVAL);
if stopped() {
return canceled(&mut child);
}
let task = match http_json("POST", &capsule_url, Some(&poll_body), &headers) {
Ok(task) => task,
Err(error) => {
return outcome(
DelegationStatus::Failed,
&child.session_id,
format!("capsule '{}' stopped answering: {error}", request.capsule),
)
}
};
match task.pointer("/result/status/state").and_then(Value::as_str) {
Some("submitted" | "working" | "input-required") => continue,
Some("completed") => {
let carried = task
.pointer("/result/artifacts/0/parts/0/text")
.and_then(Value::as_str)
.filter(|text| !text.is_empty())
.map(str::to_string);
let found = read_child_result(&child.workdir, &child.session_id, &task_id);
let mut result = outcome(
DelegationStatus::Completed,
&child.session_id,
carried
.or_else(|| found.as_ref().map(|(_, text)| text.clone()))
.unwrap_or_default(),
);
result.result_path = found.and_then(|(path, _)| {
path.strip_prefix(&self.accessible_workdir)
.ok()
.map(|path| path.to_string_lossy().replace('\\', "/"))
});
return result;
}
Some("failed" | "rejected") => {
return outcome(
DelegationStatus::Failed,
&child.session_id,
task.pointer("/result/status/message/parts/0/text")
.and_then(Value::as_str)
.unwrap_or("the delegated capsule's task failed")
.to_string(),
)
}
other => {
return outcome(
DelegationStatus::Failed,
&child.session_id,
format!(
"capsule '{}' reported an unknown task state: {other:?}",
request.capsule
),
)
}
}
}
}
}
fn deliver_task(
capsule_url: &str,
door_token: Option<&crate::door_auth::DoorToken>,
delegation_id: &str,
request: &DelegationRequest,
stop: &dyn Fn() -> bool,
) -> Result<String, String> {
let send_body = json!({
"jsonrpc": "2.0",
"id": delegation_id,
"method": "message/send",
"params": {
"message": {
"messageId": format!("msg_{delegation_id}"),
"role": "user",
"parts": [{"text": request.task}]
}
}
})
.to_string();
let authorization = door_token.map(crate::door_auth::bearer_header);
let headers: Vec<(&str, &str)> = authorization
.as_deref()
.map(|value| ("Authorization", value))
.into_iter()
.collect();
let send_deadline = Instant::now() + SEND_DEADLINE;
let mut delay = Duration::from_millis(100);
let sent = loop {
match http_json("POST", capsule_url, Some(&send_body), &headers) {
Ok(response) => break response,
Err(error) if stop() => {
return Err(format!(
"capsule '{}' was not handed its task: the delegating task was cancelled \
({error})",
request.capsule
))
}
Err(_) if Instant::now() < send_deadline => {
std::thread::sleep(delay);
delay = (delay * 2).min(Duration::from_secs(2));
}
Err(error) => {
return Err(format!(
"capsule '{}' did not accept a task within {}s: {error}",
request.capsule,
SEND_DEADLINE.as_secs()
))
}
}
};
sent.pointer("/result/id")
.and_then(Value::as_str)
.map(str::to_string)
.ok_or_else(|| {
format!(
"capsule '{}' answered the delivered task with no task id",
request.capsule
)
})
}
fn end_for_canceled_task(
request: &DelegationRequest,
delegation_id: &str,
child: &mut LaunchedChild,
launched_at: Instant,
) {
if let Err(error) = child.shutdown() {
crate::runtime_err!("[capsule-runtime] delegation {delegation_id}: {error}");
}
crate::delegation::record_terminated(
&child.workdir,
crate::delegation::DelegationOutcome {
delegation_id: delegation_id.to_string(),
capsule_name: request.capsule.clone(),
capsule_version: request.version.clone(),
session_id: child.session_id.clone(),
status: crate::delegation::DelegationStatus::Terminated,
result_path: None,
workdir: child.workdir.display().to_string(),
duration_ms: crate::member_call::elapsed_ms(launched_at),
detail: Some(crate::delegation::PARENT_ENDED_DETAIL.to_string()),
reported_by: crate::delegation::Reporter::Launcher,
delivered: false,
delivery_error: None,
},
);
}
fn read_child_result(
child_workdir: &Path,
session_id: &str,
task_id: &str,
) -> Option<(PathBuf, String)> {
crate::delegation::child_result_candidates(child_workdir, session_id, task_id)
.into_iter()
.find_map(|path| std::fs::read_to_string(&path).ok().map(|text| (path, text)))
}
pub(crate) fn bounded(output: String) -> (String, bool) {
if output.len() > MAX_OUTPUT_BYTES {
(bound_output(output), true)
} else {
(output, false)
}
}
fn bound_output(output: String) -> String {
let mut end = MAX_OUTPUT_BYTES;
while end > 0 && !output.is_char_boundary(end) {
end -= 1;
}
format!("{} […]", &output[..end])
}
fn env_allow_intersection(child: &[String], parent: &[String]) -> Vec<String> {
let mut allowed: Vec<String> = Vec::new();
for name in child {
if parent.contains(name) && !allowed.contains(name) {
allowed.push(name.clone());
}
}
allowed
}
pub(crate) fn handed_off_work_deadline(declared: Duration) -> Duration {
result_timeout_from(
std::env::var(DELEGATION_TIMEOUT_ENV).ok().as_deref(),
declared,
)
}
fn result_timeout_from(value: Option<&str>, declared: Duration) -> Duration {
value
.and_then(|value| value.trim().parse::<u64>().ok())
.filter(|secs| *secs > 0)
.map(Duration::from_secs)
.unwrap_or(declared)
}
fn daemon_refusal(error: &str) -> String {
error
.split_once("; body: ")
.and_then(|(_, body)| serde_json::from_str::<Value>(body).ok())
.and_then(|body| {
body.get("error")
.and_then(Value::as_str)
.map(str::to_string)
})
.unwrap_or_else(|| error.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn every_status_has_one_wire_word() {
assert_eq!(DelegationStatus::Started.as_str(), "started");
assert_eq!(DelegationStatus::Completed.as_str(), "completed");
assert_eq!(DelegationStatus::Failed.as_str(), "failed");
assert_eq!(DelegationStatus::TimedOut.as_str(), "timed_out");
assert_eq!(DelegationStatus::Refused.as_str(), "refused");
}
#[test]
fn a_canceled_delegation_has_its_own_word() {
assert_eq!(DelegationStatus::Canceled.as_str(), "canceled");
}
#[test]
fn an_origin_prints_whether_it_can_be_stopped() {
let stoppable = DelegationOrigin {
stop: Some(std::sync::Arc::new(|| false)),
..origin()
};
assert!(
format!("{stoppable:?}").contains("stop: true"),
"{stoppable:?}"
);
assert!(format!("{:?}", origin()).contains("stop: false"));
}
fn plane() -> DelegationPlane {
DelegationPlane::new(
"http://127.0.0.1:1".to_string(),
SpawnCredential::new("msc1.test".to_string()),
PathBuf::from("/tmp"),
"ses_parent".to_string(),
DELEGATION_RESULT_TIMEOUT,
std::sync::Arc::new(murmur_artifact::LocalRegistry::new("/tmp")),
Vec::new(),
None,
)
}
fn request() -> DelegationRequest {
DelegationRequest {
capsule: "worker".to_string(),
version: "0.1.0".to_string(),
task: "t".to_string(),
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_delegated_task_reaches_a_child_that_refuses_peer_tasks() {
let (addr, _shutdown, mut task_rx) = crate::identity::serve_test_door(false).await;
let url = format!("http://{addr}");
let delivered = tokio::task::spawn_blocking(move || {
deliver_task(&url, None, "dlg_0000000000000001", &request(), &|| false)
})
.await
.unwrap()
.expect("the child takes the delegated task");
assert!(delivered.starts_with("tsk_"), "{delivered}");
let incoming = task_rx
.recv()
.await
.expect("the task reached the child's loop");
assert_eq!(
incoming.provenance.origin(),
crate::origin::TaskOrigin::Event
);
let refused = crate::outgoing::send_a2a_message(
&addr,
crate::outgoing::OutgoingMessage {
message_id: "msg_peer".to_string(),
context_id: None,
text: "from a peer".to_string(),
},
None,
None,
None,
)
.await
.expect_err("the same door refuses a peer message");
assert!(refused.contains("403 peer_not_accepted"), "{refused}");
}
#[test]
fn a_start_with_nowhere_to_report_is_refused_before_the_daemon_is_asked() {
let started = plane().start(
&request(),
&DelegationOrigin {
context_id: "ctx_parent".to_string(),
..DelegationOrigin::default()
},
);
assert!(started.child.is_none());
let result = started.result;
assert_eq!(result.status, DelegationStatus::Failed);
assert!(result.delegation_id.is_empty(), "{result:?}");
assert!(result.child_workdir.is_none(), "{result:?}");
assert!(
result.output.contains("nowhere to be reported"),
"the sentence says why: {}",
result.output
);
}
#[test]
fn a_start_with_no_conversation_is_refused_too() {
let started = plane()
.reporting_to("http://127.0.0.1:7000".to_string())
.start(&request(), &DelegationOrigin::default());
assert!(started.child.is_none());
let result = started.result;
assert_eq!(result.status, DelegationStatus::Failed);
assert!(
result.output.contains("nowhere to be reported"),
"the sentence says why: {}",
result.output
);
}
#[test]
fn a_refusal_is_lifted_out_of_its_http_transcript() {
let violation = "capabilities.shell.allow: the child declares 'bash', which its parent \
does not hold — a spawned capsule can never hold more capability than \
the capsule that spawned it";
let wrapped = format!(
"HTTP request failed: HTTP/1.1 403 Forbidden\r\ncontent-type: application/json; \
body: {}",
json!({ "error": violation })
);
assert_eq!(daemon_refusal(&wrapped), violation);
}
#[test]
fn a_transport_failure_survives_unchanged() {
let refused = "failed to connect to 127.0.0.1:7700: Connection refused (os error 111)";
assert_eq!(daemon_refusal(refused), refused);
let unparseable = "HTTP request failed: HTTP/1.1 500 Internal Server Error; body: nope";
assert_eq!(daemon_refusal(unparseable), unparseable);
}
#[test]
fn the_result_timeout_reads_its_override() {
let declared = Duration::from_secs(30);
assert_eq!(
result_timeout_from(Some(" 20 "), declared),
Duration::from_secs(20)
);
for ignored in [None, Some(""), Some("0"), Some("-5"), Some("later")] {
assert_eq!(
result_timeout_from(ignored, declared),
declared,
"{ignored:?} is not a positive number of seconds, so the declared bound stands"
);
}
}
#[test]
fn a_declared_deadline_of_zero_is_a_bound_and_not_an_absence() {
assert_eq!(
result_timeout_from(None, Duration::ZERO),
Duration::ZERO,
"0 gives up at the first poll rather than falling back to the default"
);
assert_eq!(
result_timeout_from(None, DELEGATION_RESULT_TIMEOUT),
DELEGATION_RESULT_TIMEOUT
);
}
#[test]
fn the_output_ceiling_bounds_a_refusal_too() {
let request = request();
let result = DelegationResult::refused(&request, "x".repeat(MAX_OUTPUT_BYTES + 4096));
assert!(result.truncated);
assert!(result.output.len() <= MAX_OUTPUT_BYTES + " […]".len());
assert!(result.output.ends_with(" […]"));
let short = DelegationResult::refused(&request, "no".to_string());
assert!(!short.truncated);
assert_eq!(short.output, "no");
}
#[test]
fn a_cut_answer_stays_valid_utf8() {
let output = "é".repeat(MAX_OUTPUT_BYTES);
let bounded = bound_output(output);
assert!(bounded.ends_with(" […]"));
assert!(bounded.len() <= MAX_OUTPUT_BYTES + " […]".len());
}
#[test]
fn a_child_is_handed_only_what_both_manifests_declare() {
let names =
|list: &[&str]| -> Vec<String> { list.iter().map(|name| name.to_string()).collect() };
assert_eq!(
env_allow_intersection(&names(&["A", "B", "C"]), &names(&["B", "C", "D"])),
names(&["B", "C"])
);
assert_eq!(
env_allow_intersection(&names(&["A"]), &names(&[])),
Vec::<String>::new(),
"a parent that declares nothing hands nothing on"
);
assert_eq!(
env_allow_intersection(&names(&[]), &names(&["A"])),
Vec::<String>::new(),
"a child that declares nothing asks for nothing"
);
}
#[test]
fn the_intersection_keeps_the_childs_order_and_names_nothing_twice() {
let names =
|list: &[&str]| -> Vec<String> { list.iter().map(|name| name.to_string()).collect() };
assert_eq!(
env_allow_intersection(&names(&["C", "A", "C"]), &names(&["A", "B", "C"])),
names(&["C", "A"])
);
}
#[test]
fn a_capsule_the_store_cannot_resolve_fails_the_read_by_name() {
let store = tempfile::tempdir().unwrap();
let plane = plane_over(&store, &["MURMUR_TEST_PROVIDER_KEY"]);
let error = plane
.child_env_allow("worker", "0.1.0")
.expect_err("an empty store resolves nothing");
let message = error.to_string();
assert!(message.contains("worker"), "{message}");
assert!(message.contains("0.1.0"), "{message}");
assert!(message.contains("capabilities.env.allow"), "{message}");
}
fn store_with_worker() -> tempfile::TempDir {
let store = tempfile::tempdir().unwrap();
let mut zipped = std::io::Cursor::new(Vec::<u8>::new());
{
let mut zip = zip::ZipWriter::new(&mut zipped);
zip.start_file(
"murmur.yaml",
zip::write::SimpleFileOptions::default()
.compression_method(zip::CompressionMethod::Deflated),
)
.unwrap();
std::io::Write::write_all(
&mut zip,
b"name: worker\nversion: 0.1.0\ncapabilities:\n env:\n allow: [A, B, C]\n",
)
.unwrap();
zip.finish().unwrap();
}
murmur_artifact::Registry::publish(
&murmur_artifact::LocalRegistry::new(store.path()),
murmur_artifact::ArtifactMeta {
name: "worker".to_string(),
version: "0.1.0".to_string(),
runtime: murmur_artifact::RuntimeType::Wasm,
artifact_runtime: "capsule".to_string(),
platforms: Vec::new(),
description: None,
tags: Vec::new(),
wit_contracts: None,
},
&zipped.into_inner(),
)
.unwrap();
store
}
fn plane_over(store: &tempfile::TempDir, parent_env_allow: &[&str]) -> DelegationPlane {
plane_in_formation(store, parent_env_allow, None)
}
fn plane_in_formation(
store: &tempfile::TempDir,
parent_env_allow: &[&str],
formation_id: Option<FormationId>,
) -> DelegationPlane {
DelegationPlane::new(
"http://127.0.0.1:1".to_string(),
SpawnCredential::new("msc1.test".to_string()),
PathBuf::from("/tmp"),
"ses_parent".to_string(),
DELEGATION_RESULT_TIMEOUT,
std::sync::Arc::new(murmur_artifact::LocalRegistry::new(store.path())),
parent_env_allow.iter().map(|n| n.to_string()).collect(),
formation_id,
)
.reporting_to("http://127.0.0.1:7000".to_string())
}
fn origin() -> DelegationOrigin<'static> {
DelegationOrigin {
context_id: "ctx_parent".to_string(),
..DelegationOrigin::default()
}
}
fn completion_address() -> CompletionAddress {
CompletionAddress {
url: "http://127.0.0.1:7000".to_string(),
}
}
#[test]
fn a_launch_request_is_built_in_exactly_one_place() {
let mut found: Vec<(String, usize)> = Vec::new();
for (name, text) in crate::source_scan::crate_sources() {
if name == "child_launch.rs" {
continue;
}
let count = crate::source_scan::production_part(&text)
.lines()
.filter(|line| !line.trim_start().starts_with("//"))
.map(|line| line.matches("ChildLaunchRequest {").count())
.sum::<usize>();
if count > 0 {
found.push((name, count));
}
}
found.sort();
assert_eq!(
found,
vec![("delegation_plane.rs".to_string(), 1)],
"`ChildLaunchRequest` may be built in exactly one place: \
`DelegationPlane::launch_request`, which every launch path in this crate goes through so that \
what a child is handed cannot differ between them. Found: {found:?}. Route a new launch path \
through that constructor rather than composing a second request literal."
);
}
#[test]
fn an_unreadable_child_declaration_composes_no_launch_on_either_path() {
let store = tempfile::tempdir().unwrap();
let plane = plane_over(&store, &["A"]);
for report_to in [Some(completion_address()), None] {
let named = if report_to.is_some() {
"the started path"
} else {
"the plan-step path"
};
let error = plane
.launch_request(
&request(),
&origin(),
SpawnApproval::new("approved".to_string()),
report_to,
)
.err()
.unwrap_or_else(|| panic!("{named}: an empty store resolves nothing"));
let message = error.to_string();
assert!(message.contains("worker"), "{named}: {message}");
assert!(message.contains("0.1.0"), "{named}: {message}");
assert!(
message.contains("capabilities.env.allow"),
"{named}: {message}"
);
}
}
#[test]
fn every_launch_carries_what_both_manifests_declare() {
let store = store_with_worker();
let plane = plane_over(&store, &["B", "C", "D"]);
for report_to in [Some(completion_address()), None] {
let named = if report_to.is_some() {
"the started path"
} else {
"the plan-step path"
};
let launch = plane
.launch_request(
&request(),
&origin(),
SpawnApproval::new("approved".to_string()),
report_to,
)
.unwrap_or_else(|error| panic!("{named}: {error}"));
assert_eq!(
launch.child_env_allow,
vec!["B".to_string(), "C".to_string()],
"{named} hands on what both manifests declare, in the child's order"
);
}
}
#[test]
fn the_two_paths_differ_in_the_completion_target_and_its_deadline_alone() {
let store = store_with_worker();
let plane = plane_over(&store, &["B", "C", "D"]);
let compose = |report_to| {
plane
.launch_request(
&request(),
&origin(),
SpawnApproval::new("approved".to_string()),
report_to,
)
.expect("the store holds the child's declaration")
};
let started = compose(Some(completion_address()));
let stepped = compose(None);
let started_spawner = started
.spawner
.as_ref()
.expect("a named session and context");
assert_eq!(started_spawner.report_to, Some(completion_address()));
assert_eq!(started.completion_deadline, Some(plane.result_timeout()));
let stepped_spawner = stepped
.spawner
.as_ref()
.expect("a named session and context");
assert_eq!(stepped_spawner.report_to, None);
assert_eq!(stepped.completion_deadline, None);
assert_eq!(
started.parent_accessible_workdir,
stepped.parent_accessible_workdir
);
assert_eq!(started.capsule_name, stepped.capsule_name);
assert_eq!(started.capsule_version, stepped.capsule_version);
assert_eq!(started.grant.expose(), stepped.grant.expose());
assert_eq!(started.child_env_allow, stepped.child_env_allow);
assert_eq!(started.roost_url, stepped.roost_url);
assert_eq!(started_spawner.session_id, stepped_spawner.session_id);
assert_eq!(started_spawner.context_id, stepped_spawner.context_id);
}
#[test]
fn every_launch_carries_the_sessions_formation_id() {
let store = store_with_worker();
let member = FormationId::mint();
for formation_id in [Some(member), None] {
let plane = plane_in_formation(&store, &["B"], formation_id.clone());
assert_eq!(plane.formation_id(), formation_id.as_ref());
let unnamed = DelegationOrigin::default();
for (origin, report_to) in [
(origin(), Some(completion_address())),
(origin(), None),
(unnamed, None),
] {
let launch = plane
.launch_request(
&request(),
&origin,
SpawnApproval::new("approved".to_string()),
report_to,
)
.expect("the store holds the child's declaration");
assert_eq!(
launch.formation_id,
formation_id,
"spawner present: {}",
launch.spawner.is_some()
);
}
}
}
#[test]
fn a_launch_request_hands_a_child_no_formation_channel_or_peers() {
let _guard = crate::formation::FORMATION_ENV_LOCK
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let peers = crate::formation::FORMATION_PEERS_ENV;
let channel = crate::formation_credentials::FORMATION_CHANNEL_ENV;
let store = store_with_worker();
let plane = plane_in_formation(&store, &["B", peers, channel], Some(FormationId::mint()));
let mut launch = plane
.launch_request(
&request(),
&origin(),
SpawnApproval::new("approved".to_string()),
Some(completion_address()),
)
.expect("the store holds the child's declaration");
launch.child_env_allow.push(peers.to_string());
launch.child_env_allow.push(channel.to_string());
std::env::set_var(peers, "coder=http://coder.formation.invalid");
std::env::set_var(channel, "7");
let env = crate::child_launch::child_environment(&launch, None);
std::env::remove_var(peers);
std::env::remove_var(channel);
assert!(
env.iter().all(|(key, _)| key != peers && key != channel),
"{env:?}"
);
assert!(
env.iter()
.any(|(key, _)| key == crate::formation::FORMATION_ID_ENV),
"the formation id is still handed on: {env:?}"
);
}
#[test]
fn a_launch_request_hands_a_child_its_own_spawner_lifeline_and_no_formation_one() {
let _guard = crate::formation::FORMATION_ENV_LOCK
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let formation = crate::lifeline::FORMATION_LIFELINE_ENV;
let spawner = crate::lifeline::SPAWNER_LIFELINE_ENV;
let store = store_with_worker();
let plane = plane_in_formation(
&store,
&["B", formation, spawner],
Some(FormationId::mint()),
);
let mut launch = plane
.launch_request(
&request(),
&origin(),
SpawnApproval::new("approved".to_string()),
Some(completion_address()),
)
.expect("the store holds the child's declaration");
launch.child_env_allow.push(formation.to_string());
launch.child_env_allow.push(spawner.to_string());
std::env::set_var(formation, "7");
std::env::set_var(spawner, "8");
let env = crate::child_launch::child_environment(&launch, None);
std::env::remove_var(formation);
std::env::remove_var(spawner);
assert!(
env.iter()
.all(|(key, _)| key != formation && key != spawner),
"{env:?}"
);
assert!(
env.iter().all(|(_, value)| value != "7" && value != "8"),
"{env:?}"
);
assert!(
env.iter()
.any(|(key, _)| key == crate::formation::FORMATION_ID_ENV),
"the formation id is still handed on: {env:?}"
);
}
#[test]
fn a_trailing_slash_on_the_daemon_url_is_trimmed_once() {
let plane = DelegationPlane::new(
"http://127.0.0.1:7700/".to_string(),
SpawnCredential::new("msc1.test".to_string()),
PathBuf::from("/tmp"),
"ses_parent".to_string(),
DELEGATION_RESULT_TIMEOUT,
std::sync::Arc::new(murmur_artifact::LocalRegistry::new("/tmp")),
Vec::new(),
None,
);
assert_eq!(plane.roost_url, "http://127.0.0.1:7700");
}
}