use std::collections::HashSet;
use std::path::{Path, PathBuf};
pub use car_bundle::{
AgentIdentity, AgentManifest, BundleError, CapabilityDeclarations, ExternalProcessTransport,
LifecyclePolicy, PublisherInfo, RestartPolicy as BundleRestartPolicy, RuntimeRequirements,
TransportSpec,
};
use crate::supervisor::{AgentSpec, RestartPolicy, SupervisorError};
fn map_restart_policy(p: BundleRestartPolicy) -> RestartPolicy {
match p {
BundleRestartPolicy::Never => RestartPolicy::Never,
BundleRestartPolicy::OnFailure => RestartPolicy::OnFailure,
BundleRestartPolicy::Always => RestartPolicy::Always,
}
}
fn unmap_restart_policy(p: RestartPolicy) -> BundleRestartPolicy {
match p {
RestartPolicy::Never => BundleRestartPolicy::Never,
RestartPolicy::OnFailure => BundleRestartPolicy::OnFailure,
RestartPolicy::Always => BundleRestartPolicy::Always,
}
}
pub fn from_legacy_spec(spec: &AgentSpec) -> AgentManifest {
AgentManifest {
agent: AgentIdentity {
id: spec.id.clone(),
name: spec.name.clone(),
namespace: None,
version: None,
description: None,
license: None,
homepage: None,
},
publisher: None,
runtime: None,
lifecycle: None,
transport: TransportSpec::ExternalProcess(ExternalProcessTransport {
command: Some(spec.command.clone()),
interpreter: None,
binary_url: None,
sha256: None,
health_url: None,
args: spec.args.clone(),
cwd: spec.cwd.clone(),
env: spec.env.clone(),
restart: unmap_restart_policy(spec.restart),
max_restarts: spec.max_restarts,
backoff_secs: spec.backoff_secs,
auto_start: spec.auto_start,
token: spec.token.clone(),
capabilities: spec.capabilities.clone(),
}),
capabilities: None,
}
}
pub fn to_agent_spec(manifest: &AgentManifest) -> Result<AgentSpec, SupervisorError> {
match &manifest.transport {
TransportSpec::PureData => Err(SupervisorError::Other(format!(
"agent `{}` is a pure_data bundle; supervisor cannot spawn it. \
Pure-data agents are loaded by the runtime in a later phase.",
manifest.agent.id
))),
TransportSpec::ExternalProcess(t) => {
let has_command = t.command.as_deref().is_some_and(|c| !c.is_empty());
let interpreter = t.interpreter.as_deref().filter(|s| !s.is_empty());
let command = match (has_command, interpreter) {
(true, Some(_)) => {
return Err(SupervisorError::Other(format!(
"agent `{}` sets both `command` and `interpreter`; \
set either `command` (absolute) or `interpreter` \
(PATH-resolved), not both",
manifest.agent.id
)));
}
(false, Some(name)) => crate::supervisor::resolve_interpreter(name)?
.to_string_lossy()
.into_owned(),
(true, None) => t.command.clone().expect("has_command implies Some"),
(false, None) => {
return Err(SupervisorError::Other(format!(
"agent `{}` has transport.kind=external_process but no \
`command` or `interpreter`; health_url-only entries \
are tracked but not spawned by the supervisor in this \
phase.",
manifest.agent.id
)));
}
};
Ok(AgentSpec {
id: manifest.agent.id.clone(),
name: manifest.agent.name.clone(),
command,
args: t.args.clone(),
cwd: t.cwd.clone(),
env: t.env.clone(),
restart: map_restart_policy(t.restart),
max_restarts: t.max_restarts,
backoff_secs: t.backoff_secs,
auto_start: t.auto_start,
token: t.token.clone(),
capabilities: t.capabilities.clone(),
})
}
}
}
pub fn load_manifest_dir(dir: &Path) -> Result<Vec<AgentManifest>, SupervisorError> {
if !dir.exists() {
return Ok(Vec::new());
}
let mut out = Vec::new();
for entry in std::fs::read_dir(dir)? {
let entry = entry?;
let path = entry.path();
if !path.is_dir() {
continue;
}
let manifest_path = path.join("manifest.toml");
if !manifest_path.is_file() {
continue;
}
let text = match std::fs::read_to_string(&manifest_path) {
Ok(t) => t,
Err(e) => {
tracing::warn!(
manifest = %manifest_path.display(),
error = %e,
"skipping unreadable manifest.toml"
);
continue;
}
};
let m = match AgentManifest::from_toml_str(&text) {
Ok(m) => m,
Err(e) => {
tracing::warn!(
manifest = %manifest_path.display(),
error = %e,
"skipping malformed manifest.toml"
);
continue;
}
};
if m.publisher.is_some() {
if let Err(e) = car_bundle::verify_signature(&m) {
tracing::warn!(
manifest = %manifest_path.display(),
agent_id = %m.agent.id,
error = %e,
"manifest signature did not verify (phase 2 warn-only; \
phase 3 will reject)"
);
}
}
out.push(m);
}
out.sort_by(|a, b| a.agent.id.cmp(&b.agent.id));
Ok(out)
}
pub fn write_manifest(
agents_dir: &Path,
manifest: &AgentManifest,
) -> Result<PathBuf, SupervisorError> {
let agent_dir = agents_dir.join(&manifest.agent.id);
std::fs::create_dir_all(&agent_dir)?;
let manifest_path = agent_dir.join("manifest.toml");
let toml_text = manifest
.to_toml_string()
.map_err(|e| SupervisorError::Other(format!("serialize manifest: {e}")))?;
let tmp = agent_dir.join(".manifest.toml.tmp");
std::fs::write(&tmp, toml_text)?;
std::fs::rename(&tmp, &manifest_path)?;
Ok(manifest_path)
}
pub fn reap_stale(agents_dir: &Path, keep: &HashSet<String>) -> Vec<String> {
let Ok(entries) = std::fs::read_dir(agents_dir) else {
return Vec::new();
};
let mut removed = Vec::new();
for entry in entries.flatten() {
let p = entry.path();
if !p.is_dir() {
continue;
}
let Some(name) = p.file_name().and_then(|s| s.to_str()) else {
continue;
};
if keep.contains(name) {
continue;
}
if p.join("manifest.toml").is_file() {
if let Err(e) = std::fs::remove_dir_all(&p) {
tracing::warn!(
dir = %p.display(),
error = %e,
"reaping stale manifest dir failed"
);
continue;
}
removed.push(name.to_string());
}
}
removed
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
fn legacy_spec() -> AgentSpec {
AgentSpec {
id: "ui-improver".into(),
name: "UI Improvement".into(),
command: "/usr/local/bin/ui-improver".into(),
args: vec!["--mode".into(), "a2ui".into()],
cwd: None,
env: BTreeMap::from([("RUST_LOG".to_string(), "info".to_string())]),
restart: RestartPolicy::OnFailure,
max_restarts: 7,
backoff_secs: 3,
auto_start: false,
token: "tok-abc".into(),
capabilities: Vec::new(),
}
}
#[test]
fn from_legacy_round_trips_external_process_fields() {
let spec = legacy_spec();
let manifest = from_legacy_spec(&spec);
let round = to_agent_spec(&manifest).unwrap();
assert_eq!(round.id, spec.id);
assert_eq!(round.command, spec.command);
assert_eq!(round.args, spec.args);
assert_eq!(round.env, spec.env);
assert_eq!(round.restart, spec.restart);
assert_eq!(round.max_restarts, spec.max_restarts);
assert_eq!(round.token, spec.token);
}
#[test]
fn pure_data_manifest_cannot_project_to_agent_spec() {
let m = AgentManifest {
agent: AgentIdentity {
id: "pure-bundle".into(),
name: "Pure Data".into(),
namespace: None,
version: None,
description: None,
license: None,
homepage: None,
},
publisher: None,
runtime: None,
lifecycle: None,
transport: TransportSpec::PureData,
capabilities: None,
};
assert!(m.is_pure_data());
assert!(!m.is_remote_service());
assert!(to_agent_spec(&m).is_err());
}
#[test]
fn health_url_manifest_cannot_project_to_agent_spec() {
let m = AgentManifest {
agent: AgentIdentity {
id: "remote-svc".into(),
name: "Remote Service".into(),
namespace: None,
version: None,
description: None,
license: None,
homepage: None,
},
publisher: None,
runtime: None,
lifecycle: None,
transport: TransportSpec::ExternalProcess(ExternalProcessTransport {
command: None,
interpreter: None,
binary_url: None,
sha256: None,
health_url: Some("https://svc.example.com/.well-known/a2a/agent.json".into()),
args: vec![],
cwd: None,
env: BTreeMap::new(),
restart: BundleRestartPolicy::default(),
max_restarts: 10,
backoff_secs: 5,
auto_start: false,
token: String::new(),
capabilities: Vec::new(),
}),
capabilities: None,
};
assert!(m.is_remote_service());
assert!(to_agent_spec(&m).is_err());
}
#[test]
fn write_and_load_manifest_round_trip() {
let dir = tempfile::tempdir().unwrap();
let m = from_legacy_spec(&legacy_spec());
let written = write_manifest(dir.path(), &m).unwrap();
assert!(written.exists());
let loaded = load_manifest_dir(dir.path()).unwrap();
assert_eq!(loaded.len(), 1);
assert_eq!(loaded[0].agent.id, m.agent.id);
let spec = to_agent_spec(&loaded[0]).unwrap();
assert_eq!(spec.command, "/usr/local/bin/ui-improver");
assert_eq!(spec.token, "tok-abc");
}
#[test]
fn load_manifest_dir_skips_malformed_files() {
let dir = tempfile::tempdir().unwrap();
let good = from_legacy_spec(&legacy_spec());
write_manifest(dir.path(), &good).unwrap();
let bad_dir = dir.path().join("malformed");
std::fs::create_dir_all(&bad_dir).unwrap();
std::fs::write(bad_dir.join("manifest.toml"), "this is not valid toml === ").unwrap();
let loaded = load_manifest_dir(dir.path()).unwrap();
assert_eq!(loaded.len(), 1);
assert_eq!(loaded[0].agent.id, "ui-improver");
}
#[test]
fn load_manifest_dir_is_empty_when_path_missing() {
let dir = tempfile::tempdir().unwrap();
let missing = dir.path().join("does-not-exist");
let loaded = load_manifest_dir(&missing).unwrap();
assert!(loaded.is_empty());
}
#[test]
fn signed_manifest_loads_when_signature_valid() {
use ed25519_dalek::SigningKey;
use rand_core::OsRng;
let dir = tempfile::tempdir().unwrap();
let mut m = from_legacy_spec(&legacy_spec());
let key = SigningKey::generate(&mut OsRng);
car_bundle::sign_manifest(&mut m, &key).unwrap();
write_manifest(dir.path(), &m).unwrap();
let loaded = load_manifest_dir(dir.path()).unwrap();
assert_eq!(loaded.len(), 1);
assert!(loaded[0].publisher.is_some());
car_bundle::verify_signature(&loaded[0]).expect("signature should verify");
}
#[test]
fn signed_but_tampered_manifest_still_loads_in_phase_2_with_warning() {
use ed25519_dalek::SigningKey;
use rand_core::OsRng;
let dir = tempfile::tempdir().unwrap();
let mut m = from_legacy_spec(&legacy_spec());
let key = SigningKey::generate(&mut OsRng);
car_bundle::sign_manifest(&mut m, &key).unwrap();
if let TransportSpec::ExternalProcess(ref mut t) = m.transport {
t.command = Some("/tmp/tampered".into());
}
write_manifest(dir.path(), &m).unwrap();
let loaded = load_manifest_dir(dir.path()).unwrap();
assert_eq!(loaded.len(), 1);
assert!(car_bundle::verify_signature(&loaded[0]).is_err());
}
fn interp_manifest(command: Option<&str>, interpreter: Option<&str>) -> AgentManifest {
AgentManifest {
agent: AgentIdentity {
id: "portable-agent".into(),
name: "Portable Agent".into(),
namespace: Some("parslee".into()),
version: Some("0.1.0".into()),
description: None,
license: None,
homepage: None,
},
publisher: None,
runtime: None,
lifecycle: None,
transport: TransportSpec::ExternalProcess(ExternalProcessTransport {
command: command.map(str::to_string),
interpreter: interpreter.map(str::to_string),
binary_url: None,
sha256: None,
health_url: None,
args: vec!["agent.js".into()],
cwd: None,
env: BTreeMap::new(),
restart: BundleRestartPolicy::OnFailure,
max_restarts: 10,
backoff_secs: 5,
auto_start: false,
token: String::new(),
capabilities: Vec::new(),
}),
capabilities: None,
}
}
#[test]
fn interpreter_node_resolves_to_absolute_path_at_install() {
let m = interp_manifest(None, Some("node"));
let spec = to_agent_spec(&m).expect("interpreter `node` must resolve when node is on PATH");
let path = Path::new(&spec.command);
assert!(
path.is_absolute(),
"resolved command must be absolute, got: {}",
spec.command
);
let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
assert_eq!(
stem, "node",
"resolved command must be node, got: {}",
spec.command
);
assert_eq!(spec.args, vec!["agent.js".to_string()]);
}
#[test]
fn interpreter_missing_on_path_errors_clearly() {
let m = interp_manifest(None, Some("nonexistent-xyz"));
let err = to_agent_spec(&m).expect_err("missing interpreter must fail");
assert!(
err.to_string().contains("interpreter not found on $PATH"),
"expected PATH-not-found reason, got: {err}"
);
}
#[test]
fn command_and_interpreter_both_set_is_mutual_exclusion_error() {
let m = interp_manifest(Some("/usr/local/bin/agent"), Some("node"));
let err = to_agent_spec(&m).expect_err("command + interpreter must be rejected");
let msg = err.to_string();
assert!(
msg.contains("both") && msg.contains("not both"),
"expected mutual-exclusion reason, got: {msg}"
);
}
#[test]
fn neither_command_nor_interpreter_is_missing_command_error() {
let m = interp_manifest(None, None);
let err = to_agent_spec(&m).expect_err("neither command nor interpreter must fail");
assert!(
err.to_string().contains("command` or `interpreter"),
"expected missing-command reason, got: {err}"
);
}
#[test]
fn interpreter_survives_toml_round_trip_then_resolves() {
let m = interp_manifest(None, Some("node"));
let toml_text = m.to_toml_string().expect("serialize interpreter manifest");
assert!(
toml_text.contains("interpreter = \"node\""),
"interpreter must serialize into the TOML, got:\n{toml_text}"
);
let parsed = AgentManifest::from_toml_str(&toml_text).expect("parse interpreter manifest");
match &parsed.transport {
TransportSpec::ExternalProcess(t) => {
assert_eq!(t.interpreter.as_deref(), Some("node"));
assert!(t.command.is_none());
}
_ => panic!("transport kind drift after round-trip"),
}
let spec = to_agent_spec(&parsed).expect("parsed interpreter manifest must resolve");
assert!(Path::new(&spec.command).is_absolute());
}
}