use anyhow::{Context, Result, anyhow, bail};
use bollard::container::{
Config, CreateContainerOptions, ListContainersOptions, LogOutput, LogsOptions,
NetworkingConfig, RemoveContainerOptions, StartContainerOptions, StopContainerOptions,
};
use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults};
use bollard::image::CreateImageOptions;
use bollard::models::{EndpointSettings, HealthConfig, HostConfig};
use bollard::network::{CreateNetworkOptions, ListNetworksOptions};
use bollard::{API_DEFAULT_VERSION, Docker};
use futures_util::StreamExt;
use runner_core::executor::{DoctorCheck, DoctorReport, Executor, ExecutorCapabilities};
use runner_core::{
artifacts,
config::{RunnerConfig, platform_from_env, validate_platform},
journal::Journal,
policy, state, workspace,
};
use runner_protocol::{
CommandSpec, FailureInfo, FailureKind, JobResult, JobSpec, LogChunk, SandboxResult,
ServiceSpec, WorkflowSpec, validate_feedback_file,
};
use std::sync::atomic::{AtomicU64, Ordering};
use std::{
collections::HashMap,
fs,
path::{Component, Path, PathBuf},
sync::{Arc, Mutex},
time::Instant,
};
#[derive(Debug, thiserror::Error)]
#[error("Docker {code} during {phase}: {message}")]
pub struct DockerFailure {
pub code: &'static str,
pub phase: &'static str,
pub message: String,
pub retryable: bool,
}
impl DockerFailure {
fn image_pull(error: impl std::fmt::Display) -> Self {
Self {
code: "image_pull_failed",
phase: "image_pull",
message: sanitize_docker_error(error),
retryable: true,
}
}
pub fn failure_info(&self) -> FailureInfo {
FailureInfo {
kind: FailureKind::Infrastructure,
code: self.code.into(),
message: format!("{} (phase: {})", self.message, self.phase),
}
}
}
fn sanitize_docker_error(error: impl std::fmt::Display) -> String {
let message = error.to_string();
let mut value = serde_json::Value::String(message);
redact_diagnostic_value(&mut value);
value
.as_str()
.unwrap_or("Docker operation failed")
.chars()
.take(500)
.collect()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ExecutionTermination {
Cancelled,
TimedOut,
}
#[derive(Debug)]
struct ExecResult {
status: i64,
stdout: Vec<u8>,
stderr: Vec<u8>,
truncated: bool,
termination: Option<ExecutionTermination>,
}
fn append_bounded(buffer: &mut Vec<u8>, data: &[u8], limit: u64) -> bool {
let remaining = limit.saturating_sub(buffer.len() as u64) as usize;
let amount = data.len().min(remaining);
buffer.extend_from_slice(&data[..amount]);
amount < data.len()
}
fn redact_diagnostic_value(value: &mut serde_json::Value) {
match value {
serde_json::Value::Object(object) => {
for (key, value) in object {
let normalized = key.to_ascii_lowercase();
if normalized == "command" {
*value = serde_json::json!(["<redacted: command may contain credentials>"]);
} else if normalized == "secrets" {
if let serde_json::Value::Array(secrets) = value {
for secret in secrets.iter_mut() {
if let Some(value) = secret.get_mut("value") {
*value = serde_json::Value::String("<redacted>".into());
}
}
}
redact_diagnostic_value(value);
} else if normalized.contains("token")
|| normalized.contains("password")
|| normalized.contains("secret")
|| normalized.contains("authorization")
|| normalized.ends_with("apikey")
|| normalized.ends_with("api_key")
|| normalized.ends_with("_key")
{
*value = serde_json::Value::String("<redacted>".into());
} else {
redact_diagnostic_value(value);
}
}
}
serde_json::Value::Array(values) => {
for value in values {
redact_diagnostic_value(value);
}
}
_ => {}
}
}
fn redact_diagnostic_bytes(spec: &JobSpec, bytes: &[u8]) -> Vec<u8> {
const MAX_DIAGNOSTIC_BYTES: usize = 64 * 1024;
let mut text = String::from_utf8_lossy(bytes).into_owned();
let mut secrets: Vec<String> = Vec::new();
if let runner_protocol::WorkspaceSpec::Git { token, .. } = &spec.workspace {
secrets.push(token.clone());
}
for secret in &spec.secrets {
secrets.push(secret.value.clone());
}
for name in &spec.environment_from_runner {
if let Ok(value) = std::env::var(name) {
secrets.push(value);
}
}
for secret in secrets {
if !secret.is_empty() {
text = text.replace(&secret, "<redacted>");
}
}
let mut output = text.into_bytes();
if output.len() > MAX_DIAGNOSTIC_BYTES {
output.truncate(MAX_DIAGNOSTIC_BYTES);
output.extend_from_slice(b"\n[diagnostic output truncated]\n");
}
output
}
fn write_failure_diagnostics(
workspace: &Path,
spec: &JobSpec,
phase: Option<&str>,
stdout: &[u8],
stderr: &[u8],
) -> Result<()> {
let dir = workspace.join(".runner/diagnostics");
fs::create_dir_all(&dir)?;
let mut job = serde_json::to_value(spec)?;
redact_diagnostic_value(&mut job);
fs::write(dir.join("job.json"), serde_json::to_vec_pretty(&job)?)?;
fs::write(
dir.join("meta.json"),
serde_json::to_vec_pretty(&serde_json::json!({
"job_id": spec.id,
"attempt": spec.attempt,
"image": spec.image,
"network": spec.network.mode,
"phase": phase,
}))?,
)?;
fs::write(
dir.join("stdout.log"),
redact_diagnostic_bytes(spec, stdout),
)?;
fs::write(
dir.join("stderr.log"),
redact_diagnostic_bytes(spec, stderr),
)?;
Ok(())
}
use tokio::sync::mpsc::{Sender, error::TrySendError};
use tracing::{info, warn};
use uuid::Uuid;
fn runtime_tmpfs() -> HashMap<String, String> {
HashMap::from([
("/tmp".into(), "rw,noexec,nosuid,size=256m".into()),
(
"/home/opencode/.opencode".into(),
"rw,nosuid,nodev,uid=10001,gid=10001,mode=0700,size=64m".into(),
),
(
"/home/opencode/.config/opencode".into(),
"rw,nosuid,nodev,uid=10001,gid=10001,mode=0700,size=512m".into(),
),
(
"/home/opencode/.local/share/opencode".into(),
"rw,nosuid,nodev,uid=10001,gid=10001,mode=0700,size=1g".into(),
),
(
"/home/opencode/.local/state".into(),
"rw,nosuid,nodev,uid=10001,gid=10001,mode=0700,size=32m".into(),
),
(
"/home/opencode/.cache/npm".into(),
"rw,nosuid,nodev,uid=10001,gid=10001,mode=0700,size=1g".into(),
),
(
"/home/opencode/.cache/opencode".into(),
"rw,nosuid,nodev,uid=10001,gid=10001,mode=0700,size=128m".into(),
),
(
"/home/opencode/.pub-cache".into(),
"rw,nosuid,nodev,uid=10001,gid=10001,mode=0700,size=1g".into(),
),
])
}
fn command_with_opencode_retry(command: &[String]) -> Vec<String> {
if command.len() != 3
|| command[0] != "bash"
|| command[1] != "-lc"
|| !command[2].contains("opencode run ")
|| !command[2].contains(".ai-kodu-runner/results/opencode.json")
{
return command.to_vec();
}
let script = command[2].replacen(
"opencode run ",
"opencode run --print-logs --log-level INFO ",
1,
);
let wrapped = format!(
r#"set +e
(
{script}
)
runner_rc=$?
if [ "$runner_rc" -ne 0 ] && grep -Fq 'Unexpected server error' .ai-kodu-runner/results/opencode.json 2>/dev/null; then
echo 'OpenCode reported an unexpected error; retrying once in 2 seconds' >&2
sleep 2
(
{script}
)
runner_rc=$?
fi
exit "$runner_rc""#
);
vec![command[0].clone(), command[1].clone(), wrapped]
}
fn headless_opencode_environment(mut env: Vec<String>, command: &[String]) -> Vec<String> {
if !command.iter().any(|arg| arg.contains("opencode run ")) {
return env;
}
env.push(r#"OPENCODE_CONFIG_CONTENT={"agent":{"title":{"disable":true}}}"#.into());
env.push("OPENCODE_DISABLE_MODELS_FETCH=true".into());
env
}
fn safe_feedback_path(root: &Path, feedback_file: &str) -> Result<PathBuf> {
let relative = validate_feedback_file(feedback_file)?;
let root = root
.canonicalize()
.context("canonicalize feedback workspace")?;
let mut current = root.clone();
if let Some(parent) = relative.parent() {
for component in parent
.components()
.filter(|component| !matches!(component, Component::CurDir))
{
current.push(component.as_os_str());
match fs::symlink_metadata(¤t) {
Ok(metadata) if metadata.file_type().is_symlink() => {
bail!("feedback_file parent must not be a symlink")
}
Ok(metadata) if !metadata.is_dir() => {
bail!("feedback_file parent is not a directory")
}
Ok(_) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
fs::create_dir(¤t)?;
}
Err(error) => return Err(error.into()),
}
}
}
let target = root.join(&relative);
if let Ok(metadata) = fs::symlink_metadata(&target)
&& metadata.file_type().is_symlink()
{
bail!("feedback_file must not be a symlink")
}
let parent = target
.parent()
.context("feedback_file has no parent directory")?
.canonicalize()?;
if !parent.starts_with(&root) {
bail!("feedback_file escapes workspace")
}
Ok(target)
}
pub struct DockerExecutor {
config: RunnerConfig,
docker: Docker,
log_sender: Option<Sender<LogChunk>>,
log_sequence: AtomicU64,
dropped_log_chunks: AtomicU64,
log_context: Arc<Mutex<Option<(String, u32, String)>>>,
}
impl DockerExecutor {
pub fn new(config: RunnerConfig) -> Result<Self> {
Self::with_log_sender(config, None)
}
pub fn with_log_sender(
config: RunnerConfig,
log_sender: Option<Sender<LogChunk>>,
) -> Result<Self> {
validate_platform(&config.docker.platform)?;
let docker =
Docker::connect_with_local_defaults().context("connect to Docker Engine/Desktop")?;
Ok(Self {
config,
docker,
log_sender,
log_sequence: AtomicU64::new(0),
dropped_log_chunks: AtomicU64::new(0),
log_context: Arc::new(Mutex::new(None)),
})
}
fn begin_log_stream(&self, spec: &JobSpec) {
self.log_sequence.store(0, Ordering::Relaxed);
self.dropped_log_chunks.store(0, Ordering::Relaxed);
if let Ok(mut context) = self.log_context.lock() {
*context = Some((spec.id.clone(), spec.attempt, "running".into()));
}
}
fn set_log_phase(&self, phase: &str) {
if let Ok(mut context) = self.log_context.lock()
&& let Some((_, _, current_phase)) = context.as_mut()
{
*current_phase = phase.into();
}
}
fn emit_log(&self, stream: &str, bytes: &[u8]) {
let Some(sender) = &self.log_sender else {
return;
};
let Ok(context) = self.log_context.lock() else {
return;
};
let Some((_, attempt, phase)) = context.as_ref() else {
return;
};
for chunk in bytes.chunks(64 * 1024) {
let sequence = self.log_sequence.fetch_add(1, Ordering::Relaxed) + 1;
let chunk = LogChunk {
attempt: *attempt,
sequence,
stream: stream.to_owned(),
phase: phase.clone(),
level: Some(if stream == "stderr" { "warn" } else { "info" }.into()),
message: String::from_utf8_lossy(chunk).into_owned(),
};
match sender.try_send(chunk) {
Ok(()) | Err(TrySendError::Closed(_)) => {}
Err(TrySendError::Full(_)) => {
let dropped = self.dropped_log_chunks.fetch_add(1, Ordering::Relaxed) + 1;
if dropped == 1 || dropped.is_power_of_two() {
warn!(dropped, "job log queue is full; dropping chunks");
}
}
}
}
}
async fn connect() -> Result<Docker> {
Docker::connect_with_local_defaults().context("connect to Docker Engine/Desktop")
}
pub async fn doctor() -> Result<()> {
let d = Self::connect().await?;
d.ping().await.context("Docker ping")?;
let v = d.version().await?;
let test = format!("ai-kodu-runner-doctor-{}", Uuid::new_v4());
let id = d
.create_container(
Some(CreateContainerOptions::<String> {
name: test.clone(),
platform: Some(platform_from_env()),
}),
Config::<String> {
image: Some("alpine:3.20".into()),
cmd: Some(vec!["true".into()]),
..Default::default()
},
)
.await?
.id;
d.remove_container(
&id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await?;
println!(
"host_os: {}\narchitecture: {}\ndocker_api: {}\nexecutor: docker\ncapabilities: network, artifacts, streaming_logs, cancellation\ntest_container: ok",
std::env::consts::OS,
std::env::consts::ARCH,
v.api_version
.map(|x| format!("{x:?}"))
.unwrap_or_else(|| format!("{API_DEFAULT_VERSION:?}"))
);
Ok(())
}
async fn pull(
&self,
image: &str,
cancellation: &tokio_util::sync::CancellationToken,
deadline: tokio::time::Instant,
) -> Result<Option<ExecutionTermination>> {
if cancellation.is_cancelled() {
return Ok(Some(ExecutionTermination::Cancelled));
}
if tokio::time::Instant::now() >= deadline {
return Ok(Some(ExecutionTermination::TimedOut));
}
if self.config.docker.pull_policy.as_deref() == Some("always") {
let mut s = self.docker.create_image(
Some(CreateImageOptions {
from_image: image,
platform: self.config.docker.platform.as_str(),
..Default::default()
}),
None,
None,
);
loop {
let item = tokio::select! {
_ = cancellation.cancelled() => {
return Ok(Some(ExecutionTermination::Cancelled));
}
_ = tokio::time::sleep_until(deadline) => {
return Ok(Some(ExecutionTermination::TimedOut));
}
item = s.next() => item,
};
let Some(item) = item else {
break;
};
item.map_err(DockerFailure::image_pull)?;
}
}
Ok(None)
}
async fn start_services(
&self,
services: &[ServiceSpec],
network_name: &str,
job_id: &str,
resources: &runner_protocol::Resources,
cancellation: &tokio_util::sync::CancellationToken,
deadline: tokio::time::Instant,
) -> Result<Vec<String>> {
let mut ids = Vec::new();
for service in services {
if service.name.trim().is_empty()
|| service.name.contains('/')
|| service.name.contains(':')
{
return Err(anyhow!("invalid service name: {}", service.name));
}
if service.image.trim().is_empty() || service.image.contains(char::is_whitespace) {
return Err(anyhow!("invalid service image for {}", service.name));
}
if let Err(error) = self
.pull(&service.image, cancellation, deadline)
.await
.and_then(|termination| {
termination.map_or(Ok(()), |reason| {
Err(anyhow!(match reason {
ExecutionTermination::Cancelled => "execution cancelled",
ExecutionTermination::TimedOut => "execution timed out",
}))
})
})
{
self.remove_containers(&ids).await;
return Err(error);
}
let id = match self
.docker
.create_container(
Some(CreateContainerOptions::<String> {
name: format!("ai-kodu-runner-service-{}-{}", service.name, Uuid::new_v4()),
platform: Some(self.config.docker.platform.clone()),
}),
Config {
image: Some(service.image.clone()),
cmd: (!service.command.is_empty()).then(|| service.command.clone()),
env: Some(
service
.environment
.iter()
.map(|(k, v)| format!("{k}={v}"))
.collect(),
),
networking_config: Some(NetworkingConfig {
endpoints_config: HashMap::from([(
network_name.to_string(),
EndpointSettings {
aliases: Some(vec![
service
.alias
.clone()
.unwrap_or_else(|| service.name.clone()),
]),
..Default::default()
},
)]),
}),
healthcheck: service.healthcheck.as_ref().map(|h| HealthConfig {
test: Some(
std::iter::once("CMD".to_string())
.chain(h.command.clone())
.collect(),
),
..Default::default()
}),
labels: Some(HashMap::from([
("ai-kodu-runner.managed".into(), "true".into()),
("ai-kodu-runner.runner_id".into(), self.config.runner_id()),
("ai-kodu-runner.job_id".into(), job_id.to_string()),
("ai-kodu-runner.service".into(), service.name.clone()),
])),
host_config: Some(HostConfig {
memory: Some(resources.memory_mb * 1024 * 1024),
nano_cpus: Some((resources.cpu * 1_000_000_000.0) as i64),
pids_limit: Some(resources.pids),
security_opt: Some(vec!["no-new-privileges:true".into()]),
auto_remove: Some(false),
..Default::default()
}),
..Default::default()
},
)
.await
{
Ok(container) => container.id,
Err(error) => {
self.remove_containers(&ids).await;
return Err(error.into());
}
};
if let Err(error) = self
.docker
.start_container(&id, None::<StartContainerOptions<String>>)
.await
{
let _ = self.remove_container(&id).await;
self.remove_containers(&ids).await;
return Err(error.into());
}
ids.push(id);
}
for (service, id) in services.iter().zip(&ids) {
if let Some(healthcheck) = &service.healthcheck {
let health_deadline = tokio::time::Instant::now()
+ std::time::Duration::from_secs(healthcheck.timeout_seconds);
loop {
let result = self
.exec_command(
id,
&CommandSpec {
command: healthcheck.command.clone(),
working_directory: None,
},
cancellation,
deadline.min(health_deadline),
)
.await?;
if let Some(termination) = result.termination {
self.remove_containers(&ids).await;
return Err(anyhow!(match termination {
ExecutionTermination::Cancelled => "execution cancelled",
ExecutionTermination::TimedOut => "execution timed out",
}));
}
let status = result.status;
let output_stdout = result.stdout;
let output_stderr = result.stderr;
if status == 0 {
break;
}
if tokio::time::Instant::now() >= health_deadline {
self.remove_containers(&ids).await;
return Err(anyhow!(
"service {} did not become healthy: stdout: {}\nstderr: {}",
service.name,
String::from_utf8_lossy(&output_stdout),
String::from_utf8_lossy(&output_stderr)
));
}
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
}
}
}
Ok(ids)
}
async fn remove_container(&self, id: &str) -> Result<()> {
self.docker
.remove_container(
id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await?;
Ok(())
}
async fn remove_containers(&self, ids: &[String]) {
for id in ids {
let _ = self.remove_container(id).await;
}
}
}
#[cfg(test)]
#[allow(clippy::items_after_test_module)]
mod tests {
use super::{
DockerFailure, command_with_opencode_retry, headless_opencode_environment,
redact_diagnostic_bytes, redact_diagnostic_value, runtime_tmpfs, safe_feedback_path,
};
use runner_protocol::JobSpec;
#[test]
fn docker_failure_exposes_stable_code_and_safe_failure_info() {
let failure = DockerFailure::image_pull("registry token=secret");
assert_eq!(failure.code, "image_pull_failed");
assert_eq!(failure.phase, "image_pull");
let info = failure.failure_info();
assert_eq!(info.kind, runner_protocol::FailureKind::Infrastructure);
assert_eq!(info.code, "image_pull_failed");
}
#[test]
fn runtime_tmpfs_includes_writable_opencode_cache() {
let mounts = runtime_tmpfs();
let options = mounts
.get("/home/opencode/.cache/opencode")
.expect("OpenCode cache must be writable with a read-only root filesystem");
assert!(options.contains("rw"));
assert!(options.contains("uid=10001"));
assert!(options.contains("gid=10001"));
assert!(options.contains("mode=0700"));
}
#[test]
fn runtime_tmpfs_includes_all_opencode_persistent_state() {
let mounts = runtime_tmpfs();
let options = mounts
.get("/home/opencode/.local/share/opencode")
.expect("OpenCode logs, snapshots, and future state must be writable");
assert!(options.contains("rw"));
assert!(options.contains("uid=10001"));
assert!(options.contains("gid=10001"));
assert!(options.contains("size=1g"));
assert!(!mounts.contains_key("/home/opencode/.local/share/opencode/log"));
}
#[test]
fn runtime_tmpfs_gives_provider_install_enough_space() {
let mounts = runtime_tmpfs();
let options = mounts
.get("/home/opencode/.config/opencode")
.expect("OpenCode config directory must be writable");
assert!(options.contains("size=512m"));
}
#[test]
fn wraps_generated_opencode_command_with_transient_server_retry() {
let command = vec![
"bash".into(),
"-lc".into(),
"opencode run task | tee .ai-kodu-runner/results/opencode.json; exit ${PIPESTATUS[0]}"
.into(),
];
let wrapped = command_with_opencode_retry(&command);
assert_eq!(wrapped.len(), 3);
assert!(wrapped[2].contains("Unexpected server error"));
assert_eq!(
wrapped[2]
.matches("opencode run --print-logs --log-level INFO task")
.count(),
2
);
}
#[test]
fn leaves_non_opencode_commands_unchanged() {
let command = vec!["bash".into(), "-lc".into(), "flutter test".into()];
assert_eq!(command_with_opencode_retry(&command), command);
}
#[test]
fn disables_title_agent_only_for_headless_opencode() {
let opencode = vec!["bash".into(), "-lc".into(), "opencode run task".into()];
let flutter = vec!["bash".into(), "-lc".into(), "flutter test".into()];
let opencode_env = headless_opencode_environment(Vec::new(), &opencode);
let flutter_env = headless_opencode_environment(Vec::new(), &flutter);
assert!(
opencode_env
.iter()
.any(|value| value.contains(r#""title":{"disable":true}"#))
);
assert!(
opencode_env
.iter()
.any(|value| value == "OPENCODE_DISABLE_MODELS_FETCH=true")
);
assert!(flutter_env.is_empty());
}
#[test]
fn redacts_nested_diagnostic_credentials() {
let mut value = serde_json::json!({
"workspace": { "token": "git-token", "username": "bot" },
"provider": { "apiKey": "api-key" },
"secrets": [{ "name": "MODEL_KEY", "value": "secret-value" }],
"workflow": { "agent": { "command": ["curl", "--header", "secret-value"] } }
});
redact_diagnostic_value(&mut value);
assert_eq!(value["workspace"]["token"], "<redacted>");
assert_eq!(value["provider"]["apiKey"], "<redacted>");
assert_eq!(value["secrets"][0]["value"], "<redacted>");
assert_eq!(
value["workflow"]["agent"]["command"][0],
"<redacted: command may contain credentials>"
);
assert_eq!(value["workspace"]["username"], "bot");
}
#[test]
fn diagnostic_output_redacts_known_secrets_and_is_bounded() {
let spec = JobSpec::from_json(
r#"{
"api_version":"ai-kodu-runner.dev/v1alpha1","id":"diag","attempt":0,
"executor":"docker","image":"example@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"command":["true"],"working_directory":"/workspace",
"workspace":{"kind":"local","path":"."},
"resources":{"cpu":1.0,"memory_mb":128,"pids":32,"timeout_seconds":60},
"network":{"mode":"none"},
"secrets":[{"name":"MODEL_KEY","value":"super-secret"}]
}"#,
)
.unwrap();
let output = redact_diagnostic_bytes(&spec, b"before super-secret after");
assert_eq!(
String::from_utf8(output).unwrap(),
"before <redacted> after"
);
let output = redact_diagnostic_bytes(&spec, &vec![b'x'; 70 * 1024]);
assert!(output.len() < 70 * 1024);
assert!(String::from_utf8_lossy(&output).contains("diagnostic output truncated"));
}
#[test]
fn feedback_path_is_contained_and_creates_safe_parent() {
let root = tempfile::tempdir().unwrap();
let feedback = safe_feedback_path(root.path(), "/workspace/.runner/feedback.md").unwrap();
assert_eq!(
feedback,
root.path()
.canonicalize()
.unwrap()
.join(".runner/feedback.md")
);
assert!(root.path().join(".runner").is_dir());
}
#[cfg(unix)]
#[test]
fn feedback_path_rejects_symlink_parent() {
let root = tempfile::tempdir().unwrap();
let outside = tempfile::tempdir().unwrap();
std::os::unix::fs::symlink(outside.path(), root.path().join(".runner")).unwrap();
assert!(safe_feedback_path(root.path(), "/workspace/.runner/feedback.md").is_err());
}
}
#[async_trait::async_trait]
impl Executor for DockerExecutor {
fn capabilities(&self) -> ExecutorCapabilities {
ExecutorCapabilities::new(["artifacts", "cancellation", "network", "streaming_logs"])
}
async fn doctor(&self) -> Result<DoctorReport> {
self.docker.ping().await.context("Docker ping")?;
let version = self.docker.version().await?;
let test_name = format!("ai-kodu-runner-doctor-{}", Uuid::new_v4());
let test_id = self
.docker
.create_container(
Some(CreateContainerOptions::<String> {
name: test_name,
platform: Some(self.config.docker.platform.clone()),
}),
Config::<String> {
image: Some("alpine:3.20".into()),
cmd: Some(vec!["true".into()]),
..Default::default()
},
)
.await?
.id;
self.docker
.remove_container(
&test_id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await?;
Ok(DoctorReport {
executor: "docker".into(),
healthy: true,
capabilities: self.capabilities(),
checks: vec![
DoctorCheck {
name: "docker_ping".into(),
healthy: true,
message: "Docker Engine responded to ping".into(),
},
DoctorCheck {
name: "docker_version".into(),
healthy: true,
message: version
.version
.unwrap_or_else(|| "Docker version unavailable".into()),
},
DoctorCheck {
name: "test_container".into(),
healthy: true,
message: "Docker can create and remove a test container".into(),
},
],
})
}
async fn run(
&self,
spec: JobSpec,
cancel: Option<tokio_util::sync::CancellationToken>,
) -> Result<JobResult> {
let cancellation = cancel.unwrap_or_default();
self.begin_log_stream(&spec);
self.set_log_phase(if spec.workflow.is_some() {
"prepare"
} else {
"command"
});
info!(
job_id = %spec.id,
image = %spec.image,
network = %spec.network.mode,
platform = %self.config.docker.platform,
"job execution started"
);
let resources = policy::validate(&spec, &self.config, false)?;
let deadline =
tokio::time::Instant::now() + std::time::Duration::from_secs(resources.timeout_seconds);
if let Some(workflow) = spec.workflow.clone() {
return self
.run_workflow(spec, workflow, resources, cancellation, deadline)
.await;
}
let journal = Journal::open(&self.config.work_dir.join("runner.db"))?;
journal.transition(&spec.id, spec.attempt, state::State::Received)?;
journal.transition(&spec.id, spec.attempt, state::State::Preparing)?;
let prepared =
workspace::prepare(&spec.workspace, &self.config, &cancellation, deadline).await?;
if let Some(termination) = self.pull(&spec.image, &cancellation, deadline).await? {
return Err(anyhow!(match termination {
ExecutionTermination::Cancelled => "execution cancelled",
ExecutionTermination::TimedOut => "execution timed out",
}));
}
let network_name = format!("ai-kodu-runner-net-{}", Uuid::new_v4());
let network = if spec.network.mode == "bridge" {
Some(
self.docker
.create_network(CreateNetworkOptions {
name: network_name.clone(),
check_duplicate: true,
driver: "bridge".into(),
internal: false,
attachable: false,
labels: HashMap::from([
("ai-kodu-runner.managed".into(), "true".into()),
("ai-kodu-runner.runner_id".into(), self.config.runner_id()),
("ai-kodu-runner.job_id".into(), spec.id.clone()),
]),
..Default::default()
})
.await?
.id,
)
} else {
None
};
let name = format!("ai-kodu-runner-job-{}", Uuid::new_v4());
let env = headless_opencode_environment(
policy::environment_for_job(&spec, &self.config)?,
&spec.command,
);
let labels = HashMap::from([
("ai-kodu-runner.managed".into(), "true".into()),
("ai-kodu-runner.runner_id".into(), self.config.runner_id()),
("ai-kodu-runner.job_id".into(), spec.id.clone()),
("ai-kodu-runner.attempt".into(), spec.attempt.to_string()),
(
"ai-kodu-runner.expires_at".into(),
(chrono::Utc::now() + chrono::Duration::seconds(resources.timeout_seconds as i64))
.to_rfc3339(),
),
]);
let host = HostConfig {
binds: Some(vec![format!(
"{}:/workspace",
prepared.dir.path().display()
)]),
network_mode: if network.is_some() {
Some(network_name.clone())
} else {
Some("none".into())
},
memory: Some(resources.memory_mb * 1024 * 1024),
nano_cpus: Some((resources.cpu * 1_000_000_000.0) as i64),
pids_limit: Some(resources.pids),
cap_drop: Some(vec!["ALL".into()]),
security_opt: Some(vec!["no-new-privileges:true".into()]),
readonly_rootfs: Some(!spec.writable_rootfs),
tmpfs: Some(runtime_tmpfs()),
auto_remove: Some(false),
..Default::default()
};
let id = match self
.docker
.create_container(
Some(CreateContainerOptions {
name,
platform: Some(self.config.docker.platform.clone()),
}),
Config {
image: Some(spec.image.clone()),
cmd: Some(command_with_opencode_retry(&spec.command)),
working_dir: Some(spec.working_directory.clone()),
env: Some(env),
host_config: Some(host),
labels: Some(labels),
tty: Some(false),
open_stdin: Some(false),
..Default::default()
},
)
.await
{
Ok(container) => container.id,
Err(error) => {
if let Some(network) = &network {
let _ = self.docker.remove_network(network).await;
}
return Err(error.into());
}
};
journal.transition(&spec.id, spec.attempt, state::State::Running)?;
let started = Instant::now();
if let Err(error) = self
.docker
.start_container(&id, None::<StartContainerOptions<String>>)
.await
{
let _ = self
.docker
.remove_container(
&id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await;
if let Some(network) = &network {
let _ = self.docker.remove_network(network).await;
}
return Err(error.into());
}
let mut logs = self.docker.logs(
&id,
Some(LogsOptions::<String> {
follow: true,
stdout: true,
stderr: true,
since: 0,
until: 0,
timestamps: false,
tail: "all".into(),
}),
);
let mut log_bytes = 0u64;
let mut truncated = false;
let mut stdout = Vec::new();
let mut stderr = Vec::new();
let mut cancellation_requested = false;
let mut timeout_requested = false;
loop {
let item = tokio::select! {
_ = cancellation.cancelled() => {
cancellation_requested = true;
let _ = self
.docker
.stop_container(
&id,
Some(StopContainerOptions { t: 5 }),
)
.await;
break;
}
_ = tokio::time::sleep_until(deadline) => {
timeout_requested = true;
let _ = self
.docker
.stop_container(
&id,
Some(StopContainerOptions { t: 5 }),
)
.await;
break;
}
item = logs.next() => item,
};
let Some(item) = item else {
break;
};
match item? {
LogOutput::StdOut { message } => {
if log_bytes < self.config.limits.max_log_bytes {
let remaining = self.config.limits.max_log_bytes - log_bytes;
let n = message.len().min(remaining as usize);
print!("{}", String::from_utf8_lossy(&message[..n]));
stdout.extend_from_slice(&message[..n]);
self.emit_log("stdout", &message[..n]);
log_bytes += n as u64;
if n < message.len() {
truncated = true;
}
} else {
truncated = true;
}
}
LogOutput::StdErr { message } => {
if log_bytes < self.config.limits.max_log_bytes {
let remaining = self.config.limits.max_log_bytes - log_bytes;
let n = message.len().min(remaining as usize);
eprint!("{}", String::from_utf8_lossy(&message[..n]));
stderr.extend_from_slice(&message[..n]);
self.emit_log("stderr", &message[..n]);
log_bytes += n as u64;
if n < message.len() {
truncated = true;
}
} else {
truncated = true;
}
}
_ => {}
}
}
let status = match self.docker.inspect_container(&id, None).await {
Ok(container) => container
.state
.and_then(|state| state.exit_code)
.unwrap_or(-1),
Err(e) => {
let _ = self
.docker
.remove_container(
&id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await;
if let Some(n) = &network {
let _ = self.docker.remove_network(n).await;
}
return Err(anyhow!("Docker container inspect failed for {id}: {e}"));
}
};
journal.transition(&spec.id, spec.attempt, state::State::Collecting)?;
let cancelled = cancellation_requested || cancellation.is_cancelled();
let timed_out =
!cancelled && (timeout_requested || tokio::time::Instant::now() >= deadline);
let mut final_exit_code = status;
let final_status = if cancelled {
"cancelled"
} else if timed_out {
"timed_out"
} else if status == 0 {
"completed"
} else {
"failed"
};
let final_status = if final_status == "completed" {
match workspace::publish_git(
&spec.workspace,
prepared.dir.path(),
&cancellation,
deadline,
)
.await
{
Ok(()) => final_status,
Err(error) => {
warn!(job_id=%spec.id, error=%error, "git publish failed");
stderr.extend_from_slice(format!("git publish failed: {error}\n").as_bytes());
final_exit_code = 1;
"failed"
}
}
} else {
final_status
};
let mut artifact_patterns = spec.artifacts.clone();
if final_status != "completed" {
write_failure_diagnostics(
prepared.dir.path(),
&spec,
Some("command"),
&stdout,
&stderr,
)?;
artifact_patterns.push(".runner/diagnostics/**".into());
}
let export_dir = (!artifact_patterns.is_empty())
.then(|| artifacts::destination(&self.config.work_dir, &spec.id, spec.attempt));
let files = match &export_dir {
Some(dir) => artifacts::export(
prepared.dir.path(),
&artifact_patterns,
dir,
self.config.limits.max_artifact_bytes,
self.config.limits.max_artifact_files,
)?
.unwrap_or_default(),
None => Vec::new(),
};
journal.transition(&spec.id, spec.attempt, state::State::from_str(final_status))?;
journal.transition(&spec.id, spec.attempt, state::State::Destroying)?;
let image_id = self
.docker
.inspect_container(&id, None)
.await
.ok()
.and_then(|x| x.image);
let _ = self
.docker
.remove_container(
&id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await;
if let Some(n) = network {
let _ = self.docker.remove_network(&n).await;
}
journal.transition(&spec.id, spec.attempt, state::State::Destroyed)?;
info!(job_id=%spec.id, status=%final_status, exit_code=final_exit_code,"job finished");
Ok(JobResult {
job_id: spec.id,
attempt: spec.attempt,
status: final_status.into(),
exit_code: (!cancelled && !timed_out).then_some(final_exit_code),
started_at: chrono::Utc::now().to_rfc3339(),
finished_at: chrono::Utc::now().to_rfc3339(),
duration_ms: started.elapsed().as_millis(),
log_truncated: truncated || self.dropped_log_chunks.load(Ordering::Relaxed) > 0,
stdout: String::from_utf8_lossy(&stdout).into_owned(),
stderr: String::from_utf8_lossy(&stderr).into_owned(),
error_summary: runner_protocol::error_summary(
final_status,
&String::from_utf8_lossy(&stdout),
&String::from_utf8_lossy(&stderr),
),
failure: (final_status != "completed").then(|| FailureInfo {
kind: if final_status == "cancelled" {
FailureKind::Cancellation
} else if final_status == "timed_out" {
FailureKind::Timeout
} else {
FailureKind::Execution
},
code: match final_status {
"cancelled" => "cancelled",
"timed_out" => "timeout",
_ => "command_failed",
}
.into(),
message: match final_status {
"cancelled" => "execution cancelled",
"timed_out" => "execution timed out",
_ => "Docker command failed",
}
.into(),
}),
failed_phase: (final_status != "completed").then(|| "command".into()),
artifacts: files,
artifact_dir: export_dir.map(|path| path.to_string_lossy().into_owned()),
sandbox: SandboxResult {
executor: "docker".into(),
container_id: id,
image_id,
},
})
}
async fn cleanup(&self) -> Result<()> {
let filters = HashMap::from([(
"label".to_string(),
vec![
"ai-kodu-runner.managed=true".to_string(),
format!("ai-kodu-runner.runner_id={}", self.config.runner_id()),
],
)]);
let items = self
.docker
.list_containers(Some(ListContainersOptions {
all: true,
filters,
..Default::default()
}))
.await?;
let mut cleanup_error = None;
let mut active_resources = false;
for container in items {
if container.state.as_deref() == Some("running") {
active_resources = true;
continue;
}
if let Some(id) = container.id
&& let Err(error) = self
.docker
.remove_container(
&id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await
{
cleanup_error.get_or_insert(error.to_string());
}
}
let network_filters = HashMap::from([(
"label".to_string(),
vec![
"ai-kodu-runner.managed=true".to_string(),
format!("ai-kodu-runner.runner_id={}", self.config.runner_id()),
],
)]);
let networks = self
.docker
.list_networks(Some(ListNetworksOptions {
filters: network_filters,
}))
.await?;
for network in networks {
if network
.containers
.as_ref()
.is_some_and(|containers| !containers.is_empty())
{
active_resources = true;
continue;
}
if let Some(id) = network.id
&& let Err(error) = self.docker.remove_network(&id).await
{
cleanup_error.get_or_insert(error.to_string());
}
}
let journal = Journal::open(&self.config.work_dir.join("runner.db"))?;
if let Some(error) = cleanup_error {
bail!("cleanup incomplete: {error}")
}
if !active_resources {
for (id, attempt) in journal.unfinished()? {
journal.transition(&id, attempt, state::State::Destroying)?;
journal.transition(&id, attempt, state::State::Destroyed)?;
}
}
Ok(())
}
}
impl DockerExecutor {
async fn exec_command(
&self,
container_id: &str,
command: &CommandSpec,
cancellation: &tokio_util::sync::CancellationToken,
deadline: tokio::time::Instant,
) -> Result<ExecResult> {
let exec = tokio::select! {
_ = cancellation.cancelled() => {
let _ = self.stop_container(container_id).await;
return Ok(ExecResult {
status: -1,
stdout: Vec::new(),
stderr: Vec::new(),
truncated: false,
termination: Some(ExecutionTermination::Cancelled),
});
}
_ = tokio::time::sleep_until(deadline) => {
let _ = self.stop_container(container_id).await;
return Ok(ExecResult {
status: -1,
stdout: Vec::new(),
stderr: Vec::new(),
truncated: false,
termination: Some(ExecutionTermination::TimedOut),
});
}
result = self.docker.create_exec(
container_id,
CreateExecOptions::<String> {
attach_stdout: Some(true),
attach_stderr: Some(true),
cmd: Some(command.command.clone()),
working_dir: command.working_directory.clone(),
..Default::default()
},
) => result?,
};
let mut stdout = Vec::new();
let mut stderr = Vec::new();
let mut truncated = false;
let start = tokio::select! {
_ = cancellation.cancelled() => {
let _ = self.stop_container(container_id).await;
return Ok(ExecResult {
status: -1,
stdout,
stderr,
truncated,
termination: Some(ExecutionTermination::Cancelled),
});
}
_ = tokio::time::sleep_until(deadline) => {
let _ = self.stop_container(container_id).await;
return Ok(ExecResult {
status: -1,
stdout,
stderr,
truncated,
termination: Some(ExecutionTermination::TimedOut),
});
}
result = self.docker.start_exec(
&exec.id,
Some(StartExecOptions {
detach: false,
tty: false,
output_capacity: None,
}),
) => result?,
};
match start {
StartExecResults::Attached { mut output, .. } => loop {
let item = tokio::select! {
_ = cancellation.cancelled() => {
let _ = self.stop_container(container_id).await;
return Ok(ExecResult {
status: -1,
stdout,
stderr,
truncated,
termination: Some(ExecutionTermination::Cancelled),
});
}
_ = tokio::time::sleep_until(deadline) => {
let _ = self.stop_container(container_id).await;
return Ok(ExecResult {
status: -1,
stdout,
stderr,
truncated,
termination: Some(ExecutionTermination::TimedOut),
});
}
item = output.next() => item,
};
let Some(item) = item else {
break;
};
match item? {
LogOutput::StdOut { message } => {
print!("{}", String::from_utf8_lossy(&message));
self.emit_log("stdout", &message);
truncated |=
append_bounded(&mut stdout, &message, self.config.limits.max_log_bytes);
}
LogOutput::StdErr { message } => {
eprint!("{}", String::from_utf8_lossy(&message));
self.emit_log("stderr", &message);
truncated |=
append_bounded(&mut stderr, &message, self.config.limits.max_log_bytes);
}
_ => {}
}
},
StartExecResults::Detached => {
return Err(anyhow!("workflow exec unexpectedly detached"));
}
}
let status = self
.docker
.inspect_exec(&exec.id)
.await?
.exit_code
.unwrap_or(-1);
Ok(ExecResult {
status,
stdout,
stderr,
truncated,
termination: None,
})
}
async fn stop_container(&self, id: &str) -> Result<()> {
self.docker
.stop_container(id, Some(StopContainerOptions { t: 5 }))
.await?;
Ok(())
}
async fn run_workflow(
&self,
spec: JobSpec,
workflow: WorkflowSpec,
resources: runner_protocol::Resources,
cancellation: tokio_util::sync::CancellationToken,
deadline: tokio::time::Instant,
) -> Result<JobResult> {
let journal = Journal::open(&self.config.work_dir.join("runner.db"))?;
journal.transition(&spec.id, spec.attempt, state::State::Received)?;
journal.transition(&spec.id, spec.attempt, state::State::Preparing)?;
let prepared =
workspace::prepare(&spec.workspace, &self.config, &cancellation, deadline).await?;
if let Some(termination) = self.pull(&spec.image, &cancellation, deadline).await? {
return Err(anyhow!(match termination {
ExecutionTermination::Cancelled => "execution cancelled",
ExecutionTermination::TimedOut => "execution timed out",
}));
}
let network_name = format!("ai-kodu-runner-net-{}", Uuid::new_v4());
let network = if spec.network.mode == "bridge" {
Some(
self.docker
.create_network(CreateNetworkOptions {
name: network_name.clone(),
check_duplicate: true,
driver: "bridge".into(),
internal: false,
attachable: false,
labels: HashMap::from([
("ai-kodu-runner.managed".into(), "true".into()),
("ai-kodu-runner.runner_id".into(), self.config.runner_id()),
("ai-kodu-runner.job_id".into(), spec.id.clone()),
]),
..Default::default()
})
.await?
.id,
)
} else {
None
};
let env = policy::environment_for_job(&spec, &self.config)?;
let service_ids = if let Some(network_name) = network.as_deref() {
match self
.start_services(
&workflow.services,
network_name,
&spec.id,
&resources,
&cancellation,
deadline,
)
.await
{
Ok(ids) => ids,
Err(error) => {
if let Some(n) = &network {
let _ = self.docker.remove_network(n).await;
}
return Err(error);
}
}
} else if workflow.services.is_empty() {
Vec::new()
} else {
if let Some(n) = &network {
let _ = self.docker.remove_network(n).await;
}
return Err(anyhow!("workflow services require bridge networking"));
};
let host = HostConfig {
binds: Some(vec![format!(
"{}:/workspace",
prepared.dir.path().display()
)]),
network_mode: if network.is_some() {
Some(network_name)
} else {
Some("none".into())
},
memory: Some(resources.memory_mb * 1024 * 1024),
nano_cpus: Some((resources.cpu * 1_000_000_000.0) as i64),
pids_limit: Some(resources.pids),
cap_drop: Some(vec!["ALL".into()]),
security_opt: Some(vec!["no-new-privileges:true".into()]),
readonly_rootfs: Some(!spec.writable_rootfs),
tmpfs: Some(runtime_tmpfs()),
auto_remove: Some(false),
..Default::default()
};
let id = match self
.docker
.create_container(
Some(CreateContainerOptions::<String> {
name: format!("ai-kodu-runner-job-{}", Uuid::new_v4()),
platform: Some(self.config.docker.platform.clone()),
}),
Config::<String> {
image: Some(spec.image.clone()),
cmd: Some(vec!["sleep".into(), "infinity".into()]),
env: Some(env),
host_config: Some(host),
labels: Some(HashMap::from([
("ai-kodu-runner.managed".into(), "true".into()),
("ai-kodu-runner.runner_id".into(), self.config.runner_id()),
("ai-kodu-runner.job_id".into(), spec.id.clone()),
("ai-kodu-runner.attempt".into(), spec.attempt.to_string()),
])),
..Default::default()
},
)
.await
{
Ok(container) => container.id,
Err(error) => {
self.remove_containers(&service_ids).await;
if let Some(network) = &network {
let _ = self.docker.remove_network(network).await;
}
return Err(error.into());
}
};
if let Err(error) = self
.docker
.start_container(&id, None::<StartContainerOptions<String>>)
.await
{
let _ = self.remove_container(&id).await;
self.remove_containers(&service_ids).await;
if let Some(network) = &network {
let _ = self.docker.remove_network(network).await;
}
return Err(error.into());
}
let feedback = safe_feedback_path(prepared.dir.path(), &workflow.feedback_file)?;
std::fs::write(&feedback, "No verifier feedback yet.\n")?;
journal.transition(&spec.id, spec.attempt, state::State::Running)?;
let started = Instant::now();
let mut final_status = "failed";
let mut setup_ok = true;
let mut failed_phase: Option<String> = None;
let mut stdout = Vec::new();
let mut stderr = Vec::new();
let mut truncated = false;
let mut termination = None;
for setup in &workflow.setup {
if cancellation.is_cancelled() || tokio::time::Instant::now() >= deadline {
termination = Some(if cancellation.is_cancelled() {
ExecutionTermination::Cancelled
} else {
ExecutionTermination::TimedOut
});
failed_phase = Some("setup".into());
break;
}
self.set_log_phase("setup");
let result = self
.exec_command(&id, setup, &cancellation, deadline)
.await?;
truncated |= result.truncated;
truncated |= append_bounded(
&mut stdout,
&result.stdout,
self.config.limits.max_log_bytes,
);
truncated |= append_bounded(
&mut stderr,
&result.stderr,
self.config.limits.max_log_bytes,
);
if let Some(reason) = result.termination {
termination = Some(reason);
failed_phase = Some("setup".into());
break;
}
if result.status != 0 {
setup_ok = false;
failed_phase = Some("setup".into());
}
}
if termination.is_none() && !setup_ok {
std::fs::write(&feedback, "Workflow setup command failed.\n")?;
}
if termination.is_none()
&& !cancellation.is_cancelled()
&& setup_ok
&& let Some(initialize) = &workflow.initialize
{
self.set_log_phase("initialize");
info!(job_id=%spec.id, "workflow agent context initialization started");
let result = self
.exec_command(&id, initialize, &cancellation, deadline)
.await?;
truncated |= result.truncated;
truncated |= append_bounded(
&mut stdout,
&result.stdout,
self.config.limits.max_log_bytes,
);
truncated |= append_bounded(
&mut stderr,
&result.stderr,
self.config.limits.max_log_bytes,
);
if let Some(reason) = result.termination {
termination = Some(reason);
failed_phase = Some("initialize".into());
} else if result.status != 0 {
setup_ok = false;
failed_phase = Some("initialize".into());
std::fs::write(
&feedback,
format!(
"Agent context initialization failed with exit code {}.\n{}",
result.status,
String::from_utf8_lossy(&result.stdout)
),
)?;
}
}
for iteration in 1..=workflow.max_iterations {
if !setup_ok || cancellation.is_cancelled() || termination.is_some() {
break;
}
info!(job_id=%spec.id, iteration, "workflow agent started");
self.set_log_phase("agent");
let agent = self
.exec_command(&id, &workflow.agent, &cancellation, deadline)
.await?;
truncated |= agent.truncated;
truncated |=
append_bounded(&mut stdout, &agent.stdout, self.config.limits.max_log_bytes);
truncated |=
append_bounded(&mut stderr, &agent.stderr, self.config.limits.max_log_bytes);
if let Some(reason) = agent.termination {
termination = Some(reason);
failed_phase = Some("agent".into());
break;
}
let mut all_passed = agent.status == 0;
if agent.status != 0 {
failed_phase = Some("agent".into());
}
let mut report = format!("Iteration {iteration} agent exit code: {}\n", agent.status);
for verifier in &workflow.verifiers {
self.set_log_phase("verifier");
let command = CommandSpec {
command: verifier.command.clone(),
working_directory: verifier.working_directory.clone(),
};
let result = self
.exec_command(&id, &command, &cancellation, deadline)
.await?;
truncated |= result.truncated;
truncated |= append_bounded(
&mut stdout,
&result.stdout,
self.config.limits.max_log_bytes,
);
truncated |= append_bounded(
&mut stderr,
&result.stderr,
self.config.limits.max_log_bytes,
);
if let Some(reason) = result.termination {
termination = Some(reason);
failed_phase = Some("verifier".into());
break;
}
report.push_str(&format!(
"\nVerifier {} exit code: {}\nstdout:\n{}\nstderr:\n{}\n",
verifier.name,
result.status,
String::from_utf8_lossy(&result.stdout),
String::from_utf8_lossy(&result.stderr)
));
if verifier.required && result.status != 0 {
all_passed = false;
failed_phase = Some("verifier".into());
}
}
for verifier in &self.config.security.mandatory_verifiers {
if termination.is_some() {
break;
}
self.set_log_phase("verifier");
let result = self
.exec_command(&id, verifier, &cancellation, deadline)
.await?;
truncated |= result.truncated;
truncated |= append_bounded(
&mut stdout,
&result.stdout,
self.config.limits.max_log_bytes,
);
truncated |= append_bounded(
&mut stderr,
&result.stderr,
self.config.limits.max_log_bytes,
);
if let Some(reason) = result.termination {
termination = Some(reason);
failed_phase = Some("verifier".into());
break;
}
report.push_str(&format!(
"\nMandatory verifier {:?} exit code: {}\nstdout:\n{}\nstderr:\n{}\n",
verifier.command,
result.status,
String::from_utf8_lossy(&result.stdout),
String::from_utf8_lossy(&result.stderr)
));
if result.status != 0 {
all_passed = false;
failed_phase = Some("verifier".into());
}
}
if termination.is_none()
&& all_passed
&& let Some(publish) = &workflow.publish
{
self.set_log_phase("publish");
let result = self
.exec_command(&id, publish, &cancellation, deadline)
.await?;
truncated |= result.truncated;
truncated |= append_bounded(
&mut stdout,
&result.stdout,
self.config.limits.max_log_bytes,
);
truncated |= append_bounded(
&mut stderr,
&result.stderr,
self.config.limits.max_log_bytes,
);
if let Some(reason) = result.termination {
termination = Some(reason);
failed_phase = Some("publish".into());
} else {
all_passed = result.status == 0;
}
if !all_passed && termination.is_none() {
failed_phase = Some("publish".into());
}
}
if termination.is_none() && all_passed {
final_status = "completed";
break;
}
std::fs::write(&feedback, report)?;
if iteration == workflow.max_iterations {
break;
}
info!(job_id=%spec.id, next_iteration=iteration + 1, "verifier feedback written");
}
if termination.is_none() && cancellation.is_cancelled() {
termination = Some(ExecutionTermination::Cancelled);
}
if termination.is_none() && tokio::time::Instant::now() >= deadline {
termination = Some(ExecutionTermination::TimedOut);
}
if let Some(reason) = termination {
final_status = match reason {
ExecutionTermination::Cancelled => "cancelled",
ExecutionTermination::TimedOut => "timed_out",
};
failed_phase = Some(final_status.into());
let _ = self
.docker
.stop_container(&id, Some(StopContainerOptions { t: 5 }))
.await;
}
journal.transition(&spec.id, spec.attempt, state::State::Collecting)?;
let mut artifact_patterns = spec.artifacts.clone();
if final_status != "completed" {
write_failure_diagnostics(
prepared.dir.path(),
&spec,
failed_phase.as_deref(),
&stdout,
&stderr,
)?;
artifact_patterns.push(".runner/diagnostics/**".into());
}
let export_dir = (!artifact_patterns.is_empty())
.then(|| artifacts::destination(&self.config.work_dir, &spec.id, spec.attempt));
let files = match &export_dir {
Some(dir) => artifacts::export(
prepared.dir.path(),
&artifact_patterns,
dir,
self.config.limits.max_artifact_bytes,
self.config.limits.max_artifact_files,
)?
.unwrap_or_default(),
None => Vec::new(),
};
journal.transition(&spec.id, spec.attempt, state::State::from_str(final_status))?;
journal.transition(&spec.id, spec.attempt, state::State::Destroying)?;
let image_id = self
.docker
.inspect_container(&id, None)
.await
.ok()
.and_then(|x| x.image);
let _ = self
.docker
.remove_container(
&id,
Some(RemoveContainerOptions {
force: true,
..Default::default()
}),
)
.await;
self.remove_containers(&service_ids).await;
if let Some(n) = network {
let _ = self.docker.remove_network(&n).await;
}
journal.transition(&spec.id, spec.attempt, state::State::Destroyed)?;
Ok(JobResult {
job_id: spec.id,
attempt: spec.attempt,
status: final_status.into(),
exit_code: (final_status != "cancelled" && final_status != "timed_out")
.then_some(if final_status == "completed" { 0 } else { 1 }),
started_at: chrono::Utc::now().to_rfc3339(),
finished_at: chrono::Utc::now().to_rfc3339(),
duration_ms: started.elapsed().as_millis(),
log_truncated: truncated || self.dropped_log_chunks.load(Ordering::Relaxed) > 0,
stdout: String::from_utf8_lossy(&stdout).into_owned(),
stderr: String::from_utf8_lossy(&stderr).into_owned(),
error_summary: runner_protocol::error_summary(
final_status,
&String::from_utf8_lossy(&stdout),
&String::from_utf8_lossy(&stderr),
),
failure: (final_status != "completed").then(|| FailureInfo {
kind: match final_status {
"cancelled" => FailureKind::Cancellation,
"timed_out" => FailureKind::Timeout,
_ => FailureKind::Execution,
},
code: match final_status {
"cancelled" => "cancelled",
"timed_out" => "timeout",
_ => "workflow_failed",
}
.into(),
message: match final_status {
"cancelled" => "execution cancelled",
"timed_out" => "execution timed out",
_ => "workflow failed",
}
.into(),
}),
failed_phase: (final_status != "completed")
.then(|| failed_phase.unwrap_or_else(|| "workflow".into())),
artifacts: files,
artifact_dir: export_dir.map(|path| path.to_string_lossy().into_owned()),
sandbox: SandboxResult {
executor: "docker".into(),
container_id: id,
image_id,
},
})
}
}