use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use async_trait::async_trait;
use serde_json::Value;
use talon_container::{ContainerRuntime, PalaceRunSpec, build_palace_run_args};
use talon_core::config::TalonConfig;
use talon_http::{
CommandPalaceRunner, DisabledPalaceRunner, HttpConfig, PalaceError, PalaceRunner,
};
use tokio::sync::Semaphore;
pub const PALACE_NAME_PREFIX: &str = "talon-palace";
const RUNTIME_ENV_PASSTHROUGH: &[&str] = &[
"HOME",
"XDG_RUNTIME_DIR",
"DOCKER_HOST",
"DOCKER_CONTEXT",
"DOCKER_CONFIG",
"CONTAINER_HOST",
"CONTAINER_CONNECTION",
];
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PalaceContainerConfig {
pub runtime_bin: PathBuf,
pub image: String,
pub project_root: PathBuf,
pub network: Option<String>,
pub timeout: Duration,
pub max_output_bytes: usize,
pub max_concurrency: usize,
}
pub struct ContainerPalaceRunner {
config: PalaceContainerConfig,
permits: Arc<Semaphore>,
seq: AtomicU64,
env: Vec<(OsString, OsString)>,
cleanup: Option<Arc<ContainerRuntime>>,
}
impl std::fmt::Debug for ContainerPalaceRunner {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ContainerPalaceRunner")
.field("config", &self.config)
.field("cleanup", &self.cleanup.is_some())
.finish_non_exhaustive()
}
}
impl ContainerPalaceRunner {
pub fn new(config: PalaceContainerConfig) -> talon_core::Result<Self> {
if !config.runtime_bin.is_absolute() {
return Err(talon_core::Error::Config(format!(
"container runtime path must be absolute: {}",
config.runtime_bin.display()
)));
}
let runner = Self {
permits: Arc::new(Semaphore::new(config.max_concurrency.max(1))),
seq: AtomicU64::new(0),
env: RUNTIME_ENV_PASSTHROUGH
.iter()
.filter_map(|k| std::env::var_os(k).map(|v| (OsString::from(k), v)))
.collect(),
cleanup: None,
config,
};
runner.args_for(&runner.next_name())?;
Ok(runner)
}
pub fn with_cleanup(mut self, runtime: Arc<ContainerRuntime>) -> Self {
self.cleanup = Some(runtime);
self
}
pub fn config(&self) -> &PalaceContainerConfig {
&self.config
}
pub fn next_name(&self) -> String {
let millis = chrono::Utc::now().timestamp_millis();
let seq = self.seq.fetch_add(1, Ordering::Relaxed);
format!("{PALACE_NAME_PREFIX}-{millis}-{seq}")
}
pub fn args_for(&self, name: &str) -> talon_core::Result<Vec<String>> {
build_palace_run_args(&PalaceRunSpec {
image: self.config.image.clone(),
project_root: self.config.project_root.clone(),
name: name.to_string(),
network: self.config.network.clone(),
})
}
fn command(&self, args: Vec<String>) -> CommandPalaceRunner {
let mut runner =
CommandPalaceRunner::new(self.config.runtime_bin.clone(), args, self.config.timeout)
.with_max_output(self.config.max_output_bytes)
.with_max_concurrency(1);
for (k, v) in &self.env {
runner = runner.with_env(k.clone(), v.clone());
}
runner
}
}
struct KillOnDrop {
runtime: Option<Arc<ContainerRuntime>>,
name: String,
}
impl KillOnDrop {
fn disarm(&mut self) {
self.runtime = None;
}
}
impl Drop for KillOnDrop {
fn drop(&mut self) {
let Some(runtime) = self.runtime.take() else {
return;
};
let Ok(handle) = tokio::runtime::Handle::try_current() else {
return;
};
let name = std::mem::take(&mut self.name);
handle.spawn(async move {
match runtime.kill(&name).await {
Ok(()) => tracing::info!(container = %name, "killed abandoned palace container"),
Err(e) => tracing::debug!(container = %name, error = %e, "palace container kill"),
}
});
}
}
#[async_trait]
impl PalaceRunner for ContainerPalaceRunner {
async fn call(&self, request: Value) -> Result<Value, PalaceError> {
let permit =
tokio::time::timeout(self.config.timeout, self.permits.clone().acquire_owned())
.await
.map_err(|_| PalaceError::Busy)?
.map_err(|_| PalaceError::Busy)?;
let name = self.next_name();
let args = self.args_for(&name).map_err(|e| {
tracing::error!(error = %e, "cannot build palace container arguments");
PalaceError::Spawn("palace container configuration is invalid".into())
})?;
let mut guard = KillOnDrop {
runtime: self.cleanup.clone(),
name: name.clone(),
};
let result = self.command(args).call(request).await;
match &result {
Ok(_) | Err(PalaceError::Spawn(_)) => guard.disarm(),
Err(e) => {
tracing::warn!(container = %name, error = %e, "palace call failed; killing its container")
}
}
drop(guard);
drop(permit);
result
}
fn mode(&self) -> &'static str {
"container"
}
}
pub fn host_palace_runner(
config: &TalonConfig,
http: &HttpConfig,
runtime: &Arc<ContainerRuntime>,
network: Option<String>,
) -> Arc<dyn PalaceRunner> {
if http.palace_host_exec {
tracing::warn!(
"TALON_PALACE_HOST_EXEC=unsafe is ignored: MemPalace calls always run in a hardened \
container (host execution would unpickle agent-writable data on the host)"
);
}
let Some(runtime_bin) = absolute_runtime_bin(runtime.bin()) else {
tracing::warn!(
bin = runtime.bin(),
"container runtime path cannot be resolved; MemPalace endpoints disabled"
);
return Arc::new(DisabledPalaceRunner);
};
let project_root = match std::fs::canonicalize(&config.project_root) {
Ok(p) => p,
Err(e) => {
tracing::warn!(error = %e, "cannot resolve the project root; MemPalace endpoints disabled");
return Arc::new(DisabledPalaceRunner);
}
};
let built = ContainerPalaceRunner::new(PalaceContainerConfig {
runtime_bin,
image: config.container_image.clone(),
project_root,
network,
timeout: http.palace_timeout,
max_output_bytes: http.palace_max_output_bytes,
max_concurrency: http.palace_max_concurrency,
});
match built {
Ok(runner) => {
tracing::info!(
image = %runner.config().image,
network = runner.config().network.as_deref().unwrap_or("default"),
"MemPalace calls run in hardened one-shot containers"
);
Arc::new(runner.with_cleanup(runtime.clone()))
}
Err(e) => {
tracing::warn!(error = %e, "MemPalace endpoints disabled (palace container cannot be configured)");
Arc::new(DisabledPalaceRunner)
}
}
}
pub fn absolute_runtime_bin(bin: &str) -> Option<PathBuf> {
let path = Path::new(bin);
if path.is_absolute() {
return Some(path.to_path_buf());
}
talon_container::runtime::resolve_executable(bin, std::env::var_os("PATH"))
.filter(|p| p.is_absolute())
}
#[cfg(test)]
mod tests {
use std::os::unix::fs::PermissionsExt;
use serde_json::json;
use super::*;
struct Fx {
_tmp: tempfile::TempDir,
root: PathBuf,
runtime: PathBuf,
}
fn fixture(fake_runtime_body: &str) -> Fx {
let tmp = tempfile::tempdir().unwrap();
let root = std::fs::canonicalize(tmp.path()).unwrap();
std::fs::create_dir_all(root.join("mempalace-data")).unwrap();
std::fs::create_dir_all(root.join("mempalace")).unwrap();
std::fs::write(root.join("mempalace/palace-call.sh"), "#!/bin/sh\n").unwrap();
let runtime = root.join("fake-docker");
std::fs::write(&runtime, format!("#!/bin/sh\n{fake_runtime_body}\n")).unwrap();
let mut perm = std::fs::metadata(&runtime).unwrap().permissions();
perm.set_mode(0o755);
std::fs::set_permissions(&runtime, perm).unwrap();
Fx {
_tmp: tmp,
root,
runtime,
}
}
fn config(fx: &Fx) -> PalaceContainerConfig {
PalaceContainerConfig {
runtime_bin: fx.runtime.clone(),
image: "talon-agent:latest".into(),
project_root: fx.root.clone(),
network: Some("talon-agents".into()),
timeout: Duration::from_secs(10),
max_output_bytes: 64 * 1024,
max_concurrency: 2,
}
}
const ECHO_ARGS: &str = r#"read -r line; printf '{"argv":"%s","request":%s}\n' "$*" "$line""#;
#[test]
fn construction_validates_and_builds_hardened_args() {
let fx = fixture(ECHO_ARGS);
let runner = ContainerPalaceRunner::new(config(&fx)).unwrap();
assert_eq!(runner.mode(), "container");
let name = runner.next_name();
assert!(name.starts_with("talon-palace-"));
let args = runner.args_for(&name).unwrap();
let joined = args.join(" ");
assert_eq!(args[0], "run");
assert!(joined.contains(&format!("--name {name}")));
assert!(joined.contains("--network talon-agents"));
assert!(joined.contains("--cap-drop=ALL"));
assert!(joined.contains("--read-only"));
assert!(joined.contains("--user 61008:61008"));
assert_eq!(args.last().map(String::as_str), Some("talon-agent:latest"));
assert_ne!(runner.next_name(), runner.next_name());
}
#[test]
fn construction_refuses_bad_configuration() {
let fx = fixture(ECHO_ARGS);
let mut relative = config(&fx);
relative.runtime_bin = PathBuf::from("docker");
assert!(ContainerPalaceRunner::new(relative).is_err());
let mut bad_image = config(&fx);
bad_image.image = "--privileged".into();
assert!(ContainerPalaceRunner::new(bad_image).is_err());
let mut bad_net = config(&fx);
bad_net.network = Some("host".into());
assert!(ContainerPalaceRunner::new(bad_net).is_err());
std::fs::remove_file(fx.root.join("mempalace/palace-call.sh")).unwrap();
assert!(ContainerPalaceRunner::new(config(&fx)).is_err());
}
#[tokio::test]
async fn every_call_runs_the_runtime_with_a_fresh_container_name() {
let fx = fixture(ECHO_ARGS);
let runner = ContainerPalaceRunner::new(config(&fx)).unwrap();
let first = runner.call(json!({"op": "status"})).await.unwrap();
let second = runner.call(json!({"op": "status"})).await.unwrap();
assert_eq!(first["request"]["op"], "status");
let argv1 = first["argv"].as_str().unwrap();
let argv2 = second["argv"].as_str().unwrap();
assert!(argv1.starts_with("run "), "{argv1}");
assert!(argv1.contains("--name talon-palace-"));
assert!(argv1.contains("--entrypoint /workspace/extra/mempalace/palace-call.sh"));
assert_ne!(argv1, argv2, "container names must differ per call");
}
#[tokio::test]
async fn runtime_failures_and_timeouts_are_reported() {
let fx = fixture("cat >/dev/null; exit 3");
let runner = ContainerPalaceRunner::new(config(&fx)).unwrap();
assert!(matches!(
runner.call(json!({"op": "status"})).await,
Err(PalaceError::Exit(_))
));
let fx = fixture("exec sleep 5");
let mut cfg = config(&fx);
cfg.timeout = Duration::from_millis(200);
let runner = ContainerPalaceRunner::new(cfg).unwrap();
assert!(matches!(
runner.call(json!({"op": "status"})).await,
Err(PalaceError::Timeout(_))
));
}
#[test]
fn host_wiring_uses_the_container_runner_or_disables_the_palace() {
let fx = fixture(ECHO_ARGS);
let config = TalonConfig::from_env(
&fx.root,
talon_core::env::EnvFile::from_str_content("TALON_PALACE_HOST_EXEC=unsafe\n"),
);
let mut http = HttpConfig::new(&fx.root);
http.palace_host_exec = true;
let runtime = Arc::new(ContainerRuntime::new(fx.runtime.to_string_lossy()).unwrap());
let runner = host_palace_runner(&config, &http, &runtime, Some("talon-agents".into()));
assert_eq!(runner.mode(), "container");
std::fs::remove_dir_all(fx.root.join("mempalace")).unwrap();
let runner = host_palace_runner(&config, &http, &runtime, None);
assert_eq!(runner.mode(), DisabledPalaceRunner.mode());
}
#[test]
fn runtime_bin_resolution() {
assert_eq!(
absolute_runtime_bin("/usr/bin/docker"),
Some(PathBuf::from("/usr/bin/docker"))
);
if let Some(sh) = absolute_runtime_bin("sh") {
assert!(sh.is_absolute());
}
assert_eq!(absolute_runtime_bin("definitely-not-a-runtime-xyz"), None);
}
}