use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
pub use leviath_core::run_meta::{
ContextSnapshot, RegionEntrySnapshot, RegionSnapshot, RunMeta, RunStatus, StageRecord,
StageRunStatus,
};
#[cfg(test)]
pub fn write_context_snapshot(run_id: &str, snap: &ContextSnapshot) -> anyhow::Result<()> {
write_context_snapshot_to(&run_dir(run_id), snap)
}
fn write_private_atomic(path: &std::path::Path, body: &str) -> anyhow::Result<()> {
let tmp = path.with_extension("tmp");
leviath_sys::write_private(&tmp, body.as_bytes())?;
std::fs::rename(&tmp, path)?;
Ok(())
}
#[cfg(test)]
fn write_context_snapshot_to(dir: &std::path::Path, snap: &ContextSnapshot) -> anyhow::Result<()> {
let json = serde_json::to_string_pretty(snap)
.expect("infallible: ContextSnapshot always serializes to JSON");
write_private_atomic(&dir.join("context.json"), &json)
}
pub fn read_context_snapshot(run_id: &str) -> Option<ContextSnapshot> {
let path = run_dir(run_id).join("context.json");
let json = std::fs::read_to_string(&path).ok()?;
serde_json::from_str(&json).ok()
}
pub struct StatCache<T> {
entries: std::collections::HashMap<PathBuf, (std::time::SystemTime, u64, Option<Arc<T>>)>,
}
impl<T> Default for StatCache<T> {
fn default() -> Self {
Self {
entries: std::collections::HashMap::new(),
}
}
}
impl<T> StatCache<T> {
pub fn get_with(
&mut self,
path: &Path,
parse: impl FnOnce(&str) -> Option<T>,
) -> Option<Arc<T>> {
let Ok(meta) = std::fs::metadata(path) else {
self.entries.remove(path);
return None;
};
let stamp = (meta.modified().unwrap_or(std::time::UNIX_EPOCH), meta.len());
if let Some((mtime, len, value)) = self.entries.get(path)
&& (*mtime, *len) == stamp
{
return value.clone();
}
let value = std::fs::read_to_string(path)
.ok()
.and_then(|text| parse(&text))
.map(Arc::new);
self.entries
.insert(path.to_path_buf(), (stamp.0, stamp.1, value.clone()));
value
}
pub fn retain_under(&mut self, keep: &std::collections::HashSet<PathBuf>) {
self.entries.retain(|path, _| {
path.parent()
.is_some_and(|dir| keep.contains(&dir.to_path_buf()))
});
}
}
pub fn read_run_archive(run_id: &str) -> Option<Vec<leviath_core::run_archive::RunRecord>> {
let path = run_dir(run_id).join("run.lvr");
let bytes = std::fs::read(&path).ok()?;
leviath_core::run_archive::read_archive(&mut bytes.as_slice())
.ok()
.map(|(_version, records)| records)
}
pub fn visit_run_records(
run_id: &str,
visit: &mut dyn FnMut(&leviath_core::run_archive::RunRecord) -> std::ops::ControlFlow<()>,
) -> Option<()> {
let path = run_dir(run_id).join("run.lvr");
let file = std::fs::File::open(&path).ok()?;
let mut reader = std::io::BufReader::with_capacity(64 * 1024, file);
leviath_core::run_archive::read_archive_start(&mut reader).ok()?;
while let Ok(Some(record)) = leviath_core::run_archive::read_record(&mut reader) {
if visit(&record).is_break() {
break;
}
}
Some(())
}
pub fn visit_run_archive(
run_id: &str,
visit: &mut dyn FnMut(leviath_core::run_archive::PointRef<'_>) -> std::ops::ControlFlow<()>,
) -> Option<()> {
let path = run_dir(run_id).join("run.lvr");
let file = std::fs::File::open(&path).ok()?;
let mut reader = std::io::BufReader::with_capacity(64 * 1024, file);
leviath_core::run_archive::visit_archive_points(&mut reader, visit).ok()
}
pub fn context_history(run_id: &str) -> Vec<leviath_core::run_archive::RunPoint> {
read_run_archive(run_id)
.map(|records| leviath_core::run_archive::replay_points(&records))
.unwrap_or_default()
.into_iter()
.map(|point| leviath_core::run_archive::RunPoint {
meta: point.meta.redacted(),
..point
})
.collect()
}
fn now_secs() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
fn runs_dir_from(env_override: Option<&str>) -> PathBuf {
if let Some(dir) = env_override {
return PathBuf::from(dir);
}
leviath_core::paths::data_dir()
.unwrap_or_default()
.join("runs")
}
pub fn runs_dir() -> PathBuf {
runs_dir_from(std::env::var("LEVIATH_RUNS_DIR").ok().as_deref())
}
pub fn run_dir(run_id: &str) -> PathBuf {
if !leviath_core::is_safe_path_component(run_id) {
tracing::warn!(run_id = %run_id, "rejected an unsafe run id");
return runs_dir().join("<invalid>");
}
runs_dir().join(run_id)
}
fn dashboard_log_path_from(env_override: Option<&str>) -> PathBuf {
if let Some(path) = env_override {
return PathBuf::from(path);
}
leviath_core::paths::data_dir()
.unwrap_or_default()
.join("dashboard.log")
}
pub fn dashboard_log_path() -> PathBuf {
match std::env::var("LEVIATH_DASHBOARD_LOG_PATH") {
Ok(path) => dashboard_log_path_from(Some(&path)),
Err(_) => dashboard_log_path_from(None),
}
}
pub fn append_dashboard_log(msg: &str) {
append_dashboard_log_to(&dashboard_log_path(), msg);
}
pub fn append_dashboard_log_to(path: &Path, msg: &str) {
append_dashboard_log_capped(path, msg, DASHBOARD_LOG_MAX_BYTES);
}
const DASHBOARD_LOG_MAX_BYTES: u64 = 5 * 1024 * 1024;
fn append_dashboard_log_capped(path: &Path, msg: &str, max_bytes: u64) {
use std::io::Write;
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
roll_log_if_over_cap(path, max_bytes);
if let Ok(mut file) = leviath_sys::open_private_append(path) {
let timestamp = chrono::Local::now().format("%Y-%m-%d %H:%M:%S");
let _ = writeln!(file, "{} {}", timestamp, msg);
}
}
fn rolled_log_path(path: &Path) -> PathBuf {
let mut name = path.as_os_str().to_owned();
name.push(".1");
PathBuf::from(name)
}
fn roll_log_if_over_cap(path: &Path, max_bytes: u64) {
let over = std::fs::metadata(path)
.map(|m| m.len() >= max_bytes)
.unwrap_or(false);
if over {
let _ = std::fs::rename(path, rolled_log_path(path));
}
}
const RUN_ID_ENTROPY_BITS: u32 = 48;
pub fn new_run_id(agent_name: &str) -> String {
use rand::RngExt as _;
let entropy: u64 = rand::rng().random::<u64>() >> (u64::BITS - RUN_ID_ENTROPY_BITS);
let safe_name = agent_name.replace(|c: char| !c.is_ascii_alphanumeric() && c != '-', "-");
format!("{}-{}-{:012x}", safe_name, now_secs(), entropy)
}
pub fn create_run(meta: &RunMeta) -> anyhow::Result<()> {
create_run_in(&run_dir(&meta.run_id), meta)
}
pub(crate) fn create_run_in(dir: &std::path::Path, meta: &RunMeta) -> anyhow::Result<()> {
std::fs::create_dir_all(dir)?;
let _ = leviath_sys::secure_dir_perms(dir);
write_meta_to(dir, meta)
}
pub fn write_meta(meta: &RunMeta) -> anyhow::Result<()> {
write_meta_to(&run_dir(&meta.run_id), meta)
}
pub(crate) fn write_meta_to(dir: &std::path::Path, meta: &RunMeta) -> anyhow::Result<()> {
let json =
serde_json::to_string_pretty(meta).expect("infallible: RunMeta always serializes to JSON");
write_private_atomic(&dir.join("meta.json"), &json)
}
pub fn read_meta(run_id: &str) -> anyhow::Result<RunMeta> {
read_meta_from(&run_dir(run_id))
}
pub fn read_final_output(run_id: &str) -> Option<leviath_core::FinalOutput> {
let meta = read_meta(run_id).ok()?;
let descriptor = meta.final_output?;
let content = std::fs::read_to_string(final_output_path(&run_dir(run_id))).ok()?;
Some(leviath_core::FinalOutput {
content,
format: descriptor.format,
stage: descriptor.stage,
submitted_at: descriptor.submitted_at,
truncated: descriptor.truncated,
artifacts: descriptor.artifacts,
})
}
pub fn final_output_path(dir: &std::path::Path) -> PathBuf {
dir.join(leviath_core::FINAL_OUTPUT_FILE)
}
#[cfg(test)]
pub fn write_final_output(dir: &std::path::Path, content: &str) -> anyhow::Result<()> {
write_private_atomic(&final_output_path(dir), content)
}
pub fn is_terminal_status(status: &RunStatus) -> bool {
matches!(
status,
RunStatus::Complete
| RunStatus::CompleteInteractive
| RunStatus::Error
| RunStatus::Cancelled
)
}
pub const STALE_AFTER_SECS: i64 = 300;
pub fn looks_abandoned(
meta: &RunMeta,
live: Option<&std::collections::HashSet<String>>,
now: i64,
) -> bool {
let Some(live) = live else {
return false; };
if is_terminal_status(&meta.status) || live.contains(&meta.run_id) {
return false;
}
let moved_at = meta.last_progress_at.unwrap_or(meta.updated_at);
now.saturating_sub(moved_at) > STALE_AFTER_SECS
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ForceCancelOutcome {
Terminated,
AlreadyTerminal,
NoSuchRun,
WriteFailed,
}
impl ForceCancelOutcome {
pub fn found_run(&self) -> bool {
!matches!(self, Self::NoSuchRun)
}
}
pub fn force_cancel(run_id: &str) -> ForceCancelOutcome {
force_cancel_in(&run_dir(run_id), now_secs())
}
pub fn force_cancel_in(run_dir: &Path, now: i64) -> ForceCancelOutcome {
force_terminal_in(run_dir, RunStatus::Cancelled, None, now)
}
pub fn force_error_in(run_dir: &Path, message: &str, now: i64) -> ForceCancelOutcome {
force_terminal_in(run_dir, RunStatus::Error, Some(message.to_string()), now)
}
fn force_terminal_in(
run_dir: &Path,
status: RunStatus,
error: Option<String>,
now: i64,
) -> ForceCancelOutcome {
if !run_dir.is_dir() {
return ForceCancelOutcome::NoSuchRun;
}
let run_id = run_dir
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
let terminated = match read_meta_from(run_dir) {
Ok(meta) if is_terminal_status(&meta.status) => return ForceCancelOutcome::AlreadyTerminal,
Ok(meta) => RunMeta {
status,
updated_at: now,
error: error.clone().or(meta.error),
..meta
},
Err(_) => RunMeta {
status,
updated_at: now,
error: Some(
error
.clone()
.unwrap_or_else(|| "run metadata was unreadable; cancelled".to_string()),
),
..RunMeta::new(
run_id.clone(),
run_id,
String::new(),
String::new(),
None,
String::new(),
0,
)
},
};
match write_meta_to(run_dir, &terminated) {
Ok(()) => ForceCancelOutcome::Terminated,
Err(e) => {
let path = run_dir.display().to_string();
tracing::warn!(
run_dir = %path,
error = %e,
"could not force a run to a terminal state on disk"
);
ForceCancelOutcome::WriteFailed
}
}
}
pub(crate) fn read_meta_from(dir: &std::path::Path) -> anyhow::Result<RunMeta> {
let path = dir.join("meta.json");
let json = std::fs::read_to_string(&path)?;
Ok(serde_json::from_str(&json)?)
}
fn list_runs_in_dir(dir: PathBuf) -> Vec<RunMeta> {
if !dir.exists() {
return Vec::new();
}
let mut runs = Vec::new();
if let Ok(entries) = std::fs::read_dir(&dir) {
for entry in entries.filter_map(|e| e.ok()) {
let meta_path = entry.path().join("meta.json");
if let Ok(json) = std::fs::read_to_string(&meta_path)
&& let Ok(meta) = serde_json::from_str::<RunMeta>(&json)
{
runs.push(meta);
}
}
}
runs.sort_by_key(|r| std::cmp::Reverse(r.started_at));
runs
}
pub fn list_runs() -> Vec<RunMeta> {
list_runs_in_dir(runs_dir())
}
pub fn list_runs_cached(cache: &mut StatCache<RunMeta>) -> Vec<Arc<RunMeta>> {
let dir = runs_dir();
let mut runs = Vec::new();
let mut live_dirs = std::collections::HashSet::new();
if let Ok(entries) = std::fs::read_dir(&dir) {
for entry in entries.filter_map(|e| e.ok()) {
live_dirs.insert(entry.path());
let meta_path = entry.path().join("meta.json");
if let Some(meta) = cache.get_with(&meta_path, |json| {
serde_json::from_str::<RunMeta>(json).ok()
}) {
runs.push(meta);
}
}
}
cache.retain_under(&live_dirs);
runs.sort_by_key(|r| std::cmp::Reverse(r.started_at));
runs
}
pub fn read_stages_index_cached(
run_id: &str,
cache: &mut StatCache<Vec<StageRecord>>,
) -> Vec<StageRecord> {
let path = run_dir(run_id).join("stages.json");
cache
.get_with(&path, |json| serde_json::from_str(json).ok())
.map(|records| records.as_ref().clone())
.unwrap_or_default()
}
pub fn read_context_snapshot_cached(
run_id: &str,
cache: &mut StatCache<ContextSnapshot>,
) -> Option<Arc<ContextSnapshot>> {
let path = run_dir(run_id).join("context.json");
cache.get_with(&path, |json| serde_json::from_str(json).ok())
}
pub fn tail_file(path: &std::path::Path, max_bytes: u64) -> String {
use std::io::{Read, Seek, SeekFrom};
let mut file = match std::fs::File::open(path) {
Ok(f) => f,
Err(_) => return String::new(),
};
let file_size = file.metadata().map(|m| m.len()).unwrap_or(0);
if file_size <= max_bytes {
let mut buf = Vec::new();
let _ = file.read_to_end(&mut buf);
return String::from_utf8_lossy(&buf).to_string();
}
let offset = file_size - max_bytes;
let _ = file.seek(SeekFrom::Start(offset));
let mut buf = Vec::new();
let _ = file.read_to_end(&mut buf);
if let Some(nl) = buf.iter().position(|&b| b == b'\n') {
String::from_utf8_lossy(&buf[nl + 1..]).to_string()
} else {
String::from_utf8_lossy(&buf).to_string()
}
}
pub fn stage_dir(run_id: &str, stage_idx: usize) -> PathBuf {
run_dir(run_id).join("stages").join(stage_idx.to_string())
}
#[cfg(test)]
pub fn write_stages_index(run_id: &str, stages: &[StageRecord]) -> anyhow::Result<()> {
write_stages_index_to(&run_dir(run_id), stages)
}
#[cfg(test)]
fn write_stages_index_to(dir: &std::path::Path, stages: &[StageRecord]) -> anyhow::Result<()> {
let json = serde_json::to_string_pretty(&stages)
.expect("infallible: StageRecord slice always serializes to JSON");
write_private_atomic(&dir.join("stages.json"), &json)
}
pub fn read_stages_index(run_id: &str) -> Vec<StageRecord> {
read_stages_index_from(&run_dir(run_id))
}
pub fn read_stages_index_from(dir: &std::path::Path) -> Vec<StageRecord> {
let json = match std::fs::read_to_string(dir.join("stages.json")) {
Ok(j) => j,
Err(_) => return Vec::new(),
};
serde_json::from_str(&json).unwrap_or_default()
}
#[cfg(test)]
fn ensure_stage_dir(run_id: &str, stage_idx: usize) {
let dir = stage_dir(run_id, stage_idx);
let _ = leviath_sys::create_private_dir_all(&dir);
}
#[cfg(test)]
pub fn append_stage_output(run_id: &str, stage_idx: usize, text: &str) {
use std::io::Write;
ensure_stage_dir(run_id, stage_idx);
let path = stage_dir(run_id, stage_idx).join("output.log");
if let Ok(mut file) = leviath_sys::open_private_append(&path) {
let _ = writeln!(file, "{}", text);
}
}
#[cfg(test)]
pub fn append_stage_log(run_id: &str, stage_idx: usize, text: &str) {
use std::io::Write;
ensure_stage_dir(run_id, stage_idx);
let path = stage_dir(run_id, stage_idx).join("logs.log");
if let Ok(mut file) = leviath_sys::open_private_append(&path) {
let _ = writeln!(file, "{}", text);
}
}
#[cfg(test)]
pub fn write_stage_context(
run_id: &str,
stage_idx: usize,
snap: &ContextSnapshot,
) -> anyhow::Result<()> {
ensure_stage_dir(run_id, stage_idx);
write_context_snapshot_to(&stage_dir(run_id, stage_idx), snap)
}
pub fn read_stage_context(run_id: &str, stage_idx: usize) -> Option<ContextSnapshot> {
let path = stage_dir(run_id, stage_idx).join("context.json");
let json = std::fs::read_to_string(&path).ok()?;
serde_json::from_str(&json).ok()
}
pub fn tail_stage_output(run_id: &str, stage_idx: usize, max_bytes: u64) -> String {
tail_file(&stage_dir(run_id, stage_idx).join("output.log"), max_bytes)
}
pub fn tail_stage_log(run_id: &str, stage_idx: usize, max_bytes: u64) -> String {
tail_file(&stage_dir(run_id, stage_idx).join("logs.log"), max_bytes)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StageSelector {
Current,
Index(usize),
All,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LogStream {
Output,
Operational,
}
pub fn tail_run_logs(
run_id: &str,
selector: StageSelector,
stream: LogStream,
max_bytes: u64,
) -> String {
let read = |idx: usize| match stream {
LogStream::Output => tail_stage_output(run_id, idx, max_bytes),
LogStream::Operational => tail_stage_log(run_id, idx, max_bytes),
};
let stages = read_stages_index(run_id);
match selector {
StageSelector::Index(idx) => read(idx),
StageSelector::Current => match stages.len().checked_sub(1) {
Some(last) => read(last),
None => tail_file(&run_dir(run_id).join("output.log"), max_bytes),
},
StageSelector::All => {
let joined = stages
.iter()
.map(|stage| {
format!(
"===== stage {}: {} =====\n{}",
stage.index,
stage.name,
read(stage.index)
)
})
.collect::<Vec<_>>()
.join("\n");
let start = leviath_core::text::floor_char_boundary(
&joined,
joined.len().saturating_sub(max_bytes as usize),
);
joined.split_at(start).1.to_string()
}
}
}
#[cfg(test)]
fn make_runs_base_dir(unique: &str) -> std::path::PathBuf {
use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new();
unique.hash(&mut hasher);
let short = format!("{:x}", hasher.finish() & 0xffff_ffff);
let base_dir = dirs::home_dir()
.unwrap_or_default()
.join(".leviath-test")
.join(format!("rs-{short}"));
let _ = std::fs::create_dir_all(base_dir.join("runs"));
base_dir
}
#[cfg(test)]
fn runs_dir_isolation_vars(
base_dir: &std::path::Path,
) -> [(&'static str, Option<std::ffi::OsString>); 2] {
[
(
"LEVIATH_RUNS_DIR",
Some(base_dir.join("runs").into_os_string()),
),
(
"LEVIATH_DASHBOARD_LOG_PATH",
Some(base_dir.join("dashboard.log").into_os_string()),
),
]
}
#[cfg(test)]
pub(crate) fn with_isolated_runs_dir<R>(unique: &str, f: impl FnOnce(&std::path::Path) -> R) -> R {
let base_dir = make_runs_base_dir(unique);
let result = temp_env::with_vars(runs_dir_isolation_vars(&base_dir), || f(&base_dir));
let _ = std::fs::remove_dir_all(&base_dir);
result
}
#[cfg(test)]
pub(crate) async fn with_isolated_runs_dir_async<R, Fut>(
unique: &str,
f: impl FnOnce(std::path::PathBuf) -> Fut,
) -> R
where
Fut: std::future::Future<Output = R>,
{
let base_dir = make_runs_base_dir(unique);
let result =
temp_env::async_with_vars(runs_dir_isolation_vars(&base_dir), f(base_dir.clone())).await;
let _ = std::fs::remove_dir_all(&base_dir);
result
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn run_dir_refuses_an_unsafe_run_id() {
crate::test_support::with_tracing(|| {
for bad in ["../../etc", "/etc/passwd", "..", "a/b"] {
let dir = run_dir(bad);
let shown = dir.display().to_string();
assert!(dir.ends_with("<invalid>"), "{bad} resolved to {shown}");
assert!(!dir.exists(), "{bad} must not resolve to a real path");
}
assert!(run_dir("run-abc123").ends_with("run-abc123"));
});
}
fn live_on_disk(run_id: &str) -> RunMeta {
let mut meta = RunMeta::new(
run_id.to_string(),
"coder".to_string(),
"/agents/coder".to_string(),
"t".to_string(),
None,
"/w".to_string(),
1,
);
meta.status = RunStatus::Running;
meta.updated_at = 1_000;
meta.last_progress_at = Some(1_000);
meta
}
fn held(ids: &[&str]) -> std::collections::HashSet<String> {
ids.iter().map(|s| (*s).to_string()).collect()
}
#[test]
fn a_run_nothing_is_driving_looks_abandoned() {
let meta = live_on_disk("r1");
assert!(looks_abandoned(
&meta,
Some(&held(&["other"])),
1_000 + STALE_AFTER_SECS + 1
));
}
#[test]
fn no_answer_from_the_daemon_condemns_nothing() {
let meta = live_on_disk("r1");
assert!(!looks_abandoned(
&meta,
None,
1_000 + STALE_AFTER_SECS * 100
));
}
#[test]
fn a_run_the_daemon_is_hosting_is_never_abandoned() {
let meta = live_on_disk("r1");
assert!(!looks_abandoned(
&meta,
Some(&held(&["r1"])),
1_000 + STALE_AFTER_SECS * 100
));
}
#[test]
fn a_slow_run_inside_the_window_is_left_alone() {
let meta = live_on_disk("r1");
assert!(!looks_abandoned(
&meta,
Some(&held(&[])),
1_000 + STALE_AFTER_SECS - 1
));
}
#[test]
fn a_finished_run_is_not_abandoned() {
for status in [
RunStatus::Complete,
RunStatus::CompleteInteractive,
RunStatus::Error,
RunStatus::Cancelled,
] {
let mut meta = live_on_disk("r1");
meta.status = status.clone();
assert!(
!looks_abandoned(&meta, Some(&held(&[])), 1_000 + STALE_AFTER_SECS * 100),
"{status} is finished, not abandoned"
);
}
}
#[test]
fn a_fresh_heartbeat_does_not_rescue_a_run_that_stopped_moving() {
let mut meta = live_on_disk("r1");
let now = 1_000 + STALE_AFTER_SECS * 10;
meta.updated_at = now; meta.last_progress_at = Some(1_000); assert!(looks_abandoned(&meta, Some(&held(&[])), now));
}
#[test]
fn a_run_without_the_stamp_falls_back_to_updated_at() {
let mut meta = live_on_disk("r1");
meta.last_progress_at = None;
meta.updated_at = 1_000;
assert!(looks_abandoned(
&meta,
Some(&held(&[])),
1_000 + STALE_AFTER_SECS + 1
));
meta.updated_at = 1_000 + STALE_AFTER_SECS;
assert!(!looks_abandoned(
&meta,
Some(&held(&[])),
1_000 + STALE_AFTER_SECS + 1
));
}
#[test]
fn write_json_atomic_fs_write_failure() {
let path = std::path::Path::new("/nonexistent/leviath/runstate-cov/out.json");
let result = write_private_atomic(path, "{}");
assert!(result.is_err());
assert!(!path.exists());
}
#[test]
fn run_status_serde_roundtrip() {
for status in [
RunStatus::Starting,
RunStatus::Running,
RunStatus::WaitingInput,
RunStatus::Complete,
RunStatus::CompleteInteractive,
RunStatus::Paused,
RunStatus::Error,
RunStatus::Cancelled,
] {
let json = serde_json::to_string(&status).unwrap();
let back: RunStatus = serde_json::from_str(&json).unwrap();
assert_eq!(status, back);
}
}
#[test]
fn run_status_display() {
assert_eq!(RunStatus::Starting.to_string(), "Starting");
assert_eq!(RunStatus::Running.to_string(), "Running");
assert_eq!(RunStatus::WaitingInput.to_string(), "WaitingInput");
assert_eq!(RunStatus::Complete.to_string(), "Complete");
assert_eq!(
RunStatus::CompleteInteractive.to_string(),
"CompleteInteractive"
);
assert_eq!(RunStatus::Paused.to_string(), "Paused");
assert_eq!(RunStatus::Error.to_string(), "Error");
assert_eq!(RunStatus::Cancelled.to_string(), "Cancelled");
}
#[test]
fn run_status_snake_case_serialization() {
let json = serde_json::to_string(&RunStatus::WaitingInput).unwrap();
assert_eq!(json, "\"waiting_input\"");
let json = serde_json::to_string(&RunStatus::CompleteInteractive).unwrap();
assert_eq!(json, "\"complete_interactive\"");
}
#[test]
fn stage_run_status_serde_roundtrip() {
for status in [
StageRunStatus::Pending,
StageRunStatus::Active,
StageRunStatus::WaitingInput,
StageRunStatus::Complete,
StageRunStatus::Error,
] {
let json = serde_json::to_string(&status).unwrap();
let back: StageRunStatus = serde_json::from_str(&json).unwrap();
assert_eq!(status, back);
}
}
#[test]
fn stage_run_status_display() {
assert_eq!(StageRunStatus::Pending.to_string(), "Pending");
assert_eq!(StageRunStatus::Active.to_string(), "Active");
assert_eq!(StageRunStatus::WaitingInput.to_string(), "WaitingInput");
assert_eq!(StageRunStatus::Complete.to_string(), "Complete");
assert_eq!(StageRunStatus::Error.to_string(), "Error");
}
#[test]
fn run_meta_new_defaults() {
let meta = RunMeta::new(
"run-1".into(),
"agent".into(),
"/path".into(),
"do stuff".into(),
Some("gpt-4".into()),
"/work".into(),
3,
);
assert_eq!(meta.run_id, "run-1");
assert_eq!(meta.agent_name, "agent");
assert_eq!(meta.task, "do stuff");
assert_eq!(meta.model.as_deref(), Some("gpt-4"));
assert_eq!(meta.num_stages, 3);
assert_eq!(meta.status, RunStatus::Starting);
assert_eq!(meta.pid, 0);
assert_eq!(meta.stage_index, 0);
assert!(meta.error.is_none());
assert!(meta.title.is_none());
assert!(meta.metadata.is_empty());
assert!(meta.callback_url.is_none());
assert!(meta.parent_run_id.is_none());
}
#[test]
fn run_meta_serde_roundtrip() {
let meta = RunMeta::new(
"test-run".into(),
"test-agent".into(),
"/agents/test".into(),
"run tests".into(),
None,
"/tmp".into(),
2,
);
let json = serde_json::to_string_pretty(&meta).unwrap();
let back: RunMeta = serde_json::from_str(&json).unwrap();
assert_eq!(back.run_id, "test-run");
assert_eq!(back.agent_name, "test-agent");
assert_eq!(back.num_stages, 2);
assert!(back.model.is_none());
}
#[test]
fn run_meta_touch_updates_timestamp() {
let mut meta = RunMeta::new(
"r".into(),
"a".into(),
"/p".into(),
"t".into(),
None,
"/w".into(),
1,
);
let before = meta.updated_at;
meta.touch();
assert!(meta.updated_at >= before);
}
#[test]
fn run_meta_optional_fields_deserialize() {
let json = serde_json::json!({
"run_id": "r1",
"agent_name": "a",
"agent_path": "/p",
"task": "t",
"model": null,
"pid": 123,
"status": "running",
"current_stage": "init",
"stage_index": 0,
"num_stages": 1,
"iteration": 0,
"prompt_tokens": 0,
"completion_tokens": 0,
"workdir": "/w",
"started_at": 1000,
"updated_at": 1000,
"error": null
});
let meta: RunMeta = serde_json::from_value(json).unwrap();
assert_eq!(meta.cached_tokens, 0);
assert!(meta.title.is_none());
assert!(meta.metadata.is_empty());
assert!(meta.callback_url.is_none());
assert!(meta.parent_run_id.is_none());
assert!(meta.last_progress_at.is_none());
}
#[test]
fn run_meta_without_a_pid_still_loads() {
let json = serde_json::json!({
"run_id": "r1",
"agent_name": "a",
"agent_path": "/p",
"task": "t",
"model": null,
"status": "running",
"current_stage": "init",
"stage_index": 0,
"num_stages": 1,
"iteration": 0,
"prompt_tokens": 0,
"completion_tokens": 0,
"workdir": "/w",
"started_at": 1000,
"updated_at": 1000,
"error": null
});
let meta: RunMeta = serde_json::from_value(json).unwrap();
assert_eq!(meta.pid, 0);
}
#[test]
fn stage_record_new_defaults() {
let rec = StageRecord::new("analyze".into(), 2);
assert_eq!(rec.name, "analyze");
assert_eq!(rec.index, 2);
assert_eq!(rec.status, StageRunStatus::Pending);
assert_eq!(rec.prompt_tokens, 0);
assert_eq!(rec.completion_tokens, 0);
assert_eq!(rec.cached_tokens, 0);
assert!(rec.started_at.is_none());
assert!(rec.ended_at.is_none());
}
#[test]
fn stage_record_serde_roundtrip() {
let mut rec = StageRecord::new("build".into(), 0);
rec.status = StageRunStatus::Complete;
rec.prompt_tokens = 100;
rec.started_at = Some(1000);
rec.ended_at = Some(2000);
let json = serde_json::to_string(&rec).unwrap();
let back: StageRecord = serde_json::from_str(&json).unwrap();
assert_eq!(back.name, "build");
assert_eq!(back.status, StageRunStatus::Complete);
assert_eq!(back.prompt_tokens, 100);
assert_eq!(back.started_at, Some(1000));
}
#[test]
fn region_snapshot_serde_roundtrip() {
let snap = RegionSnapshot {
name: "system".into(),
kind: "pinned".into(),
current_tokens: 100,
max_tokens: 500,
entries: vec![RegionEntrySnapshot {
content: "You are helpful".into(),
tokens: 3,
kind: Default::default(),
metadata: None,
key: None,
taint: Default::default(),
}],
};
let json = serde_json::to_string(&snap).unwrap();
let back: RegionSnapshot = serde_json::from_str(&json).unwrap();
assert_eq!(back.name, "system");
assert_eq!(back.entries.len(), 1);
assert_eq!(back.entries[0].content, "You are helpful");
}
#[test]
fn region_snapshot_empty_entries_omitted() {
let snap = RegionSnapshot {
name: "empty".into(),
kind: "temporary".into(),
current_tokens: 0,
max_tokens: 100,
entries: vec![],
};
let json = serde_json::to_value(&snap).unwrap();
assert!(json.get("entries").is_none());
}
#[test]
fn context_snapshot_serde_roundtrip() {
let snap = ContextSnapshot {
stage_name: "analyze".into(),
total_tokens: 500,
max_tokens: 8192,
regions: vec![RegionSnapshot {
name: "history".into(),
kind: "sliding".into(),
current_tokens: 300,
max_tokens: 2000,
entries: vec![],
}],
};
let json = serde_json::to_string(&snap).unwrap();
let back: ContextSnapshot = serde_json::from_str(&json).unwrap();
assert_eq!(back.stage_name, "analyze");
assert_eq!(back.total_tokens, 500);
assert_eq!(back.regions.len(), 1);
}
#[test]
fn tail_file_nonexistent_returns_empty() {
let path = std::path::Path::new("/tmp/nonexistent-leviath-test-file.txt");
assert_eq!(tail_file(path, 1024), "");
}
#[test]
fn tail_file_small_file_returns_all() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("small.txt");
std::fs::write(&path, "line1\nline2\nline3\n").unwrap();
let result = tail_file(&path, 1024);
assert_eq!(result, "line1\nline2\nline3\n");
}
#[test]
fn tail_file_large_file_returns_tail() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("large.txt");
let content = "abcdefghij\n".repeat(100); std::fs::write(&path, &content).unwrap();
let result = tail_file(&path, 50);
assert!(result.len() <= 50);
assert!(result.ends_with('\n'));
}
#[test]
fn read_final_output_needs_both_the_descriptor_and_the_sidecar() {
with_isolated_runs_dir("read-final-output", |_| {
assert!(read_final_output("no-such-run").is_none());
let meta = RunMeta::new(
"run-silent".to_string(),
"a".to_string(),
"/p".to_string(),
"t".to_string(),
None,
"/w".to_string(),
1,
);
create_run(&meta).expect("run dir");
assert!(read_final_output("run-silent").is_none());
let answer = leviath_core::output::FinalOutput::new(
"the answer",
Some("markdown".to_string()),
"present".to_string(),
42,
);
let mut claimed = RunMeta::new(
"run-claimed".to_string(),
"a".to_string(),
"/p".to_string(),
"t".to_string(),
None,
"/w".to_string(),
1,
);
claimed.final_output = Some(answer.descriptor());
create_run(&claimed).expect("run dir");
assert!(read_final_output("run-claimed").is_none());
write_final_output(&run_dir("run-claimed"), &answer.content).expect("sidecar");
let read = read_final_output("run-claimed").expect("both halves are there");
assert_eq!(read.content, "the answer");
assert_eq!(read.format.as_deref(), Some("markdown"));
assert_eq!(read.stage, "present");
});
}
#[test]
fn new_run_id_contains_agent_name() {
let id = new_run_id("my-agent");
assert!(id.starts_with("my-agent-"));
}
#[test]
fn new_run_id_sanitizes_special_chars() {
let id = new_run_id("agent with spaces!");
assert!(!id.contains(' '));
assert!(!id.contains('!'));
}
#[test]
fn every_minted_run_id_is_a_safe_path_component() {
for name in [
"café",
"日本語",
"agent with spaces!",
"../escape",
"a/b",
"..",
"",
"emoji-🚀-agent",
"Ünïcödé",
] {
let id = new_run_id(name);
assert!(
leviath_core::is_safe_path_component(&id),
"agent {name:?} minted {id:?}, which run_dir resolves to <invalid>"
);
}
}
#[test]
fn new_run_id_is_unique_across_rapid_calls_in_same_second() {
let ids: std::collections::HashSet<String> =
(0..100).map(|_| new_run_id("same-agent")).collect();
assert_eq!(ids.len(), 100);
}
fn split_run_id(id: &str) -> (&str, &str) {
let mut parts = id.rsplitn(3, '-');
let suffix = parts.next().expect("run id has a suffix");
let secs = parts.next().expect("run id has a timestamp");
(secs, suffix)
}
#[test]
fn new_run_id_suffix_is_random_not_derived_from_the_clock() {
let ids: Vec<String> = (0..200).map(|_| new_run_id("same-agent")).collect();
let mut by_second: std::collections::HashMap<&str, Vec<&str>> =
std::collections::HashMap::new();
for id in &ids {
let (secs, suffix) = split_run_id(id);
by_second.entry(secs).or_default().push(suffix);
}
let mut largest = 0;
for (secs, suffixes) in &by_second {
let distinct: std::collections::HashSet<&&str> = suffixes.iter().collect();
assert_eq!(
distinct.len(),
suffixes.len(),
"two runs in second {secs} share a suffix: {suffixes:?}"
);
largest = largest.max(suffixes.len());
}
assert!(
largest > 1,
"expected IDs sharing a second, got {by_second:?}"
);
}
#[test]
fn write_and_read_meta_roundtrip() {
with_isolated_runs_dir("write-and-read-meta-roundtrip", |_d| {
let meta = RunMeta::new(
"test-roundtrip-unit".into(),
"test-agent".into(),
"/agents/test".into(),
"unit test".into(),
Some("model-x".into()),
"/tmp".into(),
2,
);
create_run(&meta).unwrap();
let back = read_meta(&meta.run_id).unwrap();
assert_eq!(back.run_id, "test-roundtrip-unit");
assert_eq!(back.agent_name, "test-agent");
assert_eq!(back.task, "unit test");
assert_eq!(back.model.as_deref(), Some("model-x"));
});
}
#[test]
fn read_meta_returns_err_on_corrupted_json() {
with_isolated_runs_dir("read-meta-returns-err-on-corrupted-json", |_d| {
let run_id = "corrupted-meta-run";
let dir = run_dir(run_id);
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(dir.join("meta.json"), "not valid json").unwrap();
let result = read_meta(run_id);
assert!(result.is_err());
});
}
#[test]
fn write_and_read_stages_index_roundtrip() {
with_isolated_runs_dir("write-and-read-stages-index-roundtrip", |_d| {
let run_id = "test-stages-idx-unit";
let dir = run_dir(run_id);
std::fs::create_dir_all(&dir).unwrap();
let stages = vec![
StageRecord::new("init".into(), 0),
StageRecord::new("process".into(), 1),
];
write_stages_index(run_id, &stages).unwrap();
let back = read_stages_index(run_id);
assert_eq!(back.len(), 2);
assert_eq!(back[0].name, "init");
assert_eq!(back[1].name, "process");
});
}
#[test]
fn read_stages_index_missing_returns_empty() {
let back = read_stages_index("nonexistent-run-12345");
assert!(back.is_empty());
}
#[test]
fn write_and_read_context_snapshot_roundtrip() {
with_isolated_runs_dir("write-and-read-context-snapshot-roundtrip", |_d| {
let run_id = "test-ctx-snap-unit";
let dir = run_dir(run_id);
std::fs::create_dir_all(&dir).unwrap();
let snap = ContextSnapshot {
stage_name: "test".into(),
total_tokens: 42,
max_tokens: 8192,
regions: vec![],
};
write_context_snapshot(run_id, &snap).unwrap();
let back = read_context_snapshot(run_id).unwrap();
assert_eq!(back.stage_name, "test");
assert_eq!(back.total_tokens, 42);
});
}
#[test]
fn read_context_snapshot_missing_returns_none() {
assert!(read_context_snapshot("nonexistent-ctx-run").is_none());
}
#[test]
fn read_run_archive_roundtrips_and_context_history_replays() {
with_isolated_runs_dir("read-run-archive-roundtrip", |_d| {
use leviath_core::run_archive::{self, RunIdentity, RunRecord};
let run_id = "archive-unit";
std::fs::create_dir_all(run_dir(run_id)).unwrap();
let mut buf = Vec::new();
run_archive::write_archive_start(&mut buf, run_archive::RUN_ARCHIVE_VERSION).unwrap();
let meta = RunMeta::new(
run_id.to_string(),
"a".to_string(),
"/p".to_string(),
"t".to_string(),
None,
"/w".to_string(),
1,
);
run_archive::write_record(
&mut buf,
&RunRecord::Header {
identity: RunIdentity {
run_id: run_id.to_string(),
machine_id: "m".to_string(),
world_id: "w".to_string(),
created_at: 0,
},
meta: Box::new(meta),
},
)
.unwrap();
run_archive::write_record(
&mut buf,
&RunRecord::ContextCheckpoint {
snapshot: ContextSnapshot {
stage_name: "plan".to_string(),
total_tokens: 3,
max_tokens: 100,
regions: vec![],
},
at: 1,
},
)
.unwrap();
std::fs::write(run_dir(run_id).join("run.lvr"), &buf).unwrap();
let records = read_run_archive(run_id).expect("archive read");
assert_eq!(records.len(), 2);
let history = context_history(run_id);
assert_eq!(history.len(), 1);
assert_eq!(history[0].context.stage_name, "plan");
let mut streamed_points = Vec::new();
visit_run_archive(run_id, &mut |p| {
streamed_points.push((p.index, p.context.stage_name.to_string()));
std::ops::ControlFlow::Continue(())
})
.expect("streamed replay");
assert_eq!(streamed_points, vec![(0, "plan".to_string())]);
let mut streamed_records = 0usize;
visit_run_records(run_id, &mut |_| {
streamed_records += 1;
std::ops::ControlFlow::Continue(())
})
.expect("streamed records");
assert_eq!(streamed_records, 2);
let mut first_only = 0usize;
visit_run_records(run_id, &mut |_| {
first_only += 1;
std::ops::ControlFlow::Break(())
})
.expect("streamed records with break");
assert_eq!(first_only, 1);
});
}
#[test]
fn stat_cache_parses_once_per_stat_change() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("value.json");
std::fs::write(&path, "41").unwrap();
let mut cache: StatCache<i64> = StatCache::default();
let mut parses = 0;
let get = |cache: &mut StatCache<i64>, path: &std::path::Path, parses: &mut usize| {
cache
.get_with(path, |text| {
*parses += 1;
text.trim().parse().ok()
})
.map(|v| *v)
};
assert_eq!(get(&mut cache, &path, &mut parses), Some(41));
assert_eq!(get(&mut cache, &path, &mut parses), Some(41));
assert_eq!(parses, 1, "the second read came from the cache");
std::fs::write(&path, "1234").unwrap();
assert_eq!(get(&mut cache, &path, &mut parses), Some(1234));
assert_eq!(parses, 2);
std::fs::write(&path, "not a number").unwrap();
assert_eq!(get(&mut cache, &path, &mut parses), None);
assert_eq!(get(&mut cache, &path, &mut parses), None);
assert_eq!(parses, 3, "the bad file was parsed once, not per tick");
std::fs::remove_file(&path).unwrap();
assert_eq!(get(&mut cache, &path, &mut parses), None);
assert_eq!(parses, 3);
}
#[test]
fn stat_cache_retain_under_drops_dead_runs() {
let dir = tempfile::tempdir().unwrap();
let live = dir.path().join("live");
let dead = dir.path().join("dead");
std::fs::create_dir_all(&live).unwrap();
std::fs::create_dir_all(&dead).unwrap();
std::fs::write(live.join("meta.json"), "1").unwrap();
std::fs::write(dead.join("meta.json"), "2").unwrap();
let mut cache: StatCache<i64> = StatCache::default();
cache.get_with(&live.join("meta.json"), |t| t.trim().parse().ok());
cache.get_with(&dead.join("meta.json"), |t| t.trim().parse().ok());
assert_eq!(cache.entries.len(), 2);
let keep: std::collections::HashSet<PathBuf> = [live.clone()].into_iter().collect();
cache.retain_under(&keep);
assert_eq!(cache.entries.len(), 1);
assert!(cache.entries.contains_key(&live.join("meta.json")));
}
#[test]
fn cached_run_readers_match_the_uncached_ones() {
with_isolated_runs_dir("cached-run-readers", |_d| {
let meta = RunMeta::new(
"cached-run".to_string(),
"agent".to_string(),
"/p".to_string(),
"t".to_string(),
None,
"/w".to_string(),
2,
);
create_run(&meta).unwrap();
write_stages_index(
"cached-run",
&[leviath_core::run_meta::StageRecord::new(
"plan".to_string(),
0,
)],
)
.unwrap();
write_context_snapshot(
"cached-run",
&ContextSnapshot {
stage_name: "plan".to_string(),
total_tokens: 3,
max_tokens: 100,
regions: vec![],
},
)
.unwrap();
let mut metas = StatCache::default();
let mut stages = StatCache::default();
let mut contexts = StatCache::default();
let listed = list_runs_cached(&mut metas);
assert_eq!(listed.len(), 1);
assert_eq!(listed[0].run_id, list_runs()[0].run_id);
let cached_stages = read_stages_index_cached("cached-run", &mut stages);
let plain_stages = read_stages_index("cached-run");
assert_eq!(cached_stages.len(), plain_stages.len());
assert_eq!(cached_stages[0].name, plain_stages[0].name);
let cached_ctx =
read_context_snapshot_cached("cached-run", &mut contexts).expect("snapshot cached");
assert_eq!(
*cached_ctx,
read_context_snapshot("cached-run").expect("snapshot read")
);
let again = read_context_snapshot_cached("cached-run", &mut contexts).unwrap();
assert!(Arc::ptr_eq(&cached_ctx, &again));
let mut second = RunMeta::new(
"cached-run-2".to_string(),
"agent".to_string(),
"/p".to_string(),
"t".to_string(),
None,
"/w".to_string(),
1,
);
second.started_at += 100;
create_run(&second).unwrap();
let listed = list_runs_cached(&mut metas);
assert_eq!(listed.len(), 2);
assert_eq!(listed[0].run_id, "cached-run-2", "newest first");
std::fs::create_dir_all(run_dir("garbled-run")).unwrap();
std::fs::write(run_dir("garbled-run").join("meta.json"), "not json {{").unwrap();
assert_eq!(list_runs_cached(&mut metas).len(), 2);
std::fs::remove_dir_all(run_dir("garbled-run")).unwrap();
std::fs::remove_dir_all(run_dir("cached-run")).unwrap();
std::fs::remove_dir_all(run_dir("cached-run-2")).unwrap();
assert!(list_runs_cached(&mut metas).is_empty());
assert!(read_stages_index_cached("cached-run", &mut stages).is_empty());
assert!(read_context_snapshot_cached("cached-run", &mut contexts).is_none());
std::fs::remove_dir_all(runs_dir()).unwrap();
assert!(list_runs_cached(&mut metas).is_empty());
});
}
#[test]
fn streaming_visitors_return_none_when_the_archive_is_missing() {
with_isolated_runs_dir("streaming-visitors-missing", |_d| {
let points_seen = std::cell::Cell::new(0usize);
let mut on_point = |_: leviath_core::run_archive::PointRef<'_>| {
points_seen.set(points_seen.get() + 1);
std::ops::ControlFlow::Continue(())
};
let records_seen = std::cell::Cell::new(0usize);
let mut on_record = |_: &leviath_core::run_archive::RunRecord| {
records_seen.set(records_seen.get() + 1);
std::ops::ControlFlow::Continue(())
};
assert!(visit_run_archive("no-such-run", &mut on_point).is_none());
assert!(visit_run_records("no-such-run", &mut on_record).is_none());
let run_id = "bad-preamble";
std::fs::create_dir_all(run_dir(run_id)).unwrap();
std::fs::write(run_dir(run_id).join("run.lvr"), b"junk").unwrap();
assert!(visit_run_archive(run_id, &mut on_point).is_none());
assert!(visit_run_records(run_id, &mut on_record).is_none());
assert_eq!((points_seen.get(), records_seen.get()), (0, 0));
let real = "streaming-visitors-real";
std::fs::create_dir_all(run_dir(real)).unwrap();
write_minimal_archive(real);
assert!(visit_run_archive(real, &mut on_point).is_some());
assert!(visit_run_records(real, &mut on_record).is_some());
assert_eq!(points_seen.get(), 1);
assert_eq!(records_seen.get(), 2);
});
}
fn write_minimal_archive(run_id: &str) {
use leviath_core::run_archive::{self, RunIdentity, RunRecord};
let mut buf = Vec::new();
run_archive::write_archive_start(&mut buf, run_archive::RUN_ARCHIVE_VERSION).unwrap();
let meta = RunMeta::new(
run_id.to_string(),
"a".to_string(),
"/p".to_string(),
"t".to_string(),
None,
"/w".to_string(),
1,
);
run_archive::write_record(
&mut buf,
&RunRecord::Header {
identity: RunIdentity {
run_id: run_id.to_string(),
machine_id: "m".to_string(),
world_id: "w".to_string(),
created_at: 0,
},
meta: Box::new(meta),
},
)
.unwrap();
run_archive::write_record(
&mut buf,
&RunRecord::ContextCheckpoint {
snapshot: ContextSnapshot {
stage_name: "plan".to_string(),
total_tokens: 3,
max_tokens: 100,
regions: vec![],
},
at: 1,
},
)
.unwrap();
std::fs::write(run_dir(run_id).join("run.lvr"), &buf).unwrap();
}
#[test]
fn context_history_redacts_the_webhook_secret_the_journal_keeps() {
with_isolated_runs_dir("context-history-redacts-secret", |_d| {
use leviath_core::run_archive::{self, RunIdentity, RunRecord};
let run_id = "archive-secret-unit";
std::fs::create_dir_all(run_dir(run_id)).unwrap();
let mut buf = Vec::new();
run_archive::write_archive_start(&mut buf, run_archive::RUN_ARCHIVE_VERSION).unwrap();
let mut meta = RunMeta::new(
run_id.to_string(),
"a".to_string(),
"/p".to_string(),
"t".to_string(),
None,
"/w".to_string(),
1,
);
meta.callback_url = Some("https://example.invalid/hook".to_string());
meta.callback_secret = Some("super-secret-signing-key".to_string());
run_archive::write_record(
&mut buf,
&RunRecord::Header {
identity: RunIdentity {
run_id: run_id.to_string(),
machine_id: "m".to_string(),
world_id: "w".to_string(),
created_at: 0,
},
meta: Box::new(meta),
},
)
.unwrap();
run_archive::write_record(
&mut buf,
&RunRecord::ContextCheckpoint {
snapshot: ContextSnapshot {
stage_name: "plan".to_string(),
total_tokens: 3,
max_tokens: 100,
regions: vec![],
},
at: 1,
},
)
.unwrap();
std::fs::write(run_dir(run_id).join("run.lvr"), &buf).unwrap();
let raw = std::fs::read(run_dir(run_id).join("run.lvr")).unwrap();
assert!(String::from_utf8_lossy(&raw).contains("super-secret-signing-key"));
let history = context_history(run_id);
assert_eq!(history.len(), 1);
assert_eq!(history[0].meta.callback_secret, None);
assert_eq!(
history[0].meta.callback_url.as_deref(),
Some("https://example.invalid/hook")
);
assert_eq!(history[0].context.stage_name, "plan");
});
}
#[test]
fn read_run_archive_missing_or_corrupt_returns_none() {
with_isolated_runs_dir("read-run-archive-corrupt", |_d| {
assert!(read_run_archive("no-such-archive-run").is_none());
assert!(context_history("no-such-archive-run").is_empty());
let run_id = "corrupt-archive-unit";
std::fs::create_dir_all(run_dir(run_id)).unwrap();
std::fs::write(run_dir(run_id).join("run.lvr"), b"not an archive").unwrap();
assert!(read_run_archive(run_id).is_none());
assert!(context_history(run_id).is_empty());
});
}
#[test]
fn stage_dir_path_structure() {
let path = stage_dir("run-abc", 2);
assert!(path.ends_with("stages/2"));
assert!(path.to_str().unwrap().contains("run-abc"));
}
#[test]
fn append_and_tail_stage_output() {
with_isolated_runs_dir("append-and-tail-stage-output", |_d| {
let run_id = "test-stage-output-unit";
append_stage_output(run_id, 0, "line 1");
append_stage_output(run_id, 0, "line 2");
let output = tail_stage_output(run_id, 0, 4096);
assert!(output.contains("line 1"));
assert!(output.contains("line 2"));
});
}
#[test]
fn append_and_tail_stage_log() {
with_isolated_runs_dir("append-and-tail-stage-log", |_d| {
let run_id = "test-stage-log-unit";
append_stage_log(run_id, 0, "event A");
append_stage_log(run_id, 0, "event B");
let log = tail_stage_log(run_id, 0, 4096);
assert!(log.contains("event A"));
assert!(log.contains("event B"));
});
}
#[test]
fn write_and_read_stage_context_roundtrip() {
with_isolated_runs_dir("write-and-read-stage-context-roundtrip", |_d| {
let run_id = "test-stage-ctx-unit";
let snap = ContextSnapshot {
stage_name: "stage-0".into(),
total_tokens: 100,
max_tokens: 4096,
regions: vec![],
};
write_stage_context(run_id, 0, &snap).unwrap();
let back = read_stage_context(run_id, 0).unwrap();
assert_eq!(back.stage_name, "stage-0");
});
}
#[test]
fn read_stage_context_missing_returns_none() {
assert!(read_stage_context("nonexistent-run", 99).is_none());
}
#[test]
fn append_dashboard_log_creates_log_file() {
with_isolated_runs_dir("append-dashboard-log-creates-log-file", |_d| {
append_dashboard_log("coverage-test-message");
assert!(dashboard_log_path().exists());
});
}
#[test]
fn append_dashboard_log_open_failure_is_silently_ignored() {
with_isolated_runs_dir("append-dashboard-log-open-failure", |_d| {
let path = dashboard_log_path();
std::fs::create_dir_all(&path).unwrap();
append_dashboard_log("this should not panic");
assert!(path.is_dir());
});
}
#[test]
fn append_dashboard_log_path_with_no_parent_skips_create_dir_all() {
temp_env::with_var("LEVIATH_DASHBOARD_LOG_PATH", Some("/"), || {
assert!(dashboard_log_path().parent().is_none());
append_dashboard_log("this should not panic even with no parent");
});
}
#[test]
fn dashboard_log_rolls_once_over_cap() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("dashboard.log");
append_dashboard_log_capped(&path, "first line well over the tiny cap", 8);
assert!(path.exists());
assert!(!rolled_log_path(&path).exists());
append_dashboard_log_capped(&path, "second", 8);
let rolled = rolled_log_path(&path);
assert!(rolled.exists(), "previous generation rolled to <name>.1");
assert!(
std::fs::read_to_string(&rolled)
.unwrap()
.contains("first line")
);
let live = std::fs::read_to_string(&path).unwrap();
assert!(live.contains("second"));
assert!(!live.contains("first line"));
}
#[test]
fn dashboard_log_does_not_roll_under_cap() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("dashboard.log");
append_dashboard_log_capped(&path, "a", 1_000_000);
append_dashboard_log_capped(&path, "b", 1_000_000);
assert!(!rolled_log_path(&path).exists());
let live = std::fs::read_to_string(&path).unwrap();
assert!(live.contains("a") && live.contains("b"));
}
#[test]
fn dashboard_log_path_structure() {
temp_env::with_var_unset("LEVIATH_DASHBOARD_LOG_PATH", || {
let path = dashboard_log_path();
assert!(path.to_str().unwrap().contains(".leviath"));
assert!(path.to_str().unwrap().ends_with("dashboard.log"));
});
}
#[test]
fn dashboard_log_path_honors_leviath_home() {
temp_env::with_vars(
[
("LEVIATH_DASHBOARD_LOG_PATH", None),
("LEVIATH_HOME", Some("/custom/home")),
],
|| {
assert_eq!(
dashboard_log_path(),
PathBuf::from("/custom/home/.leviath/dashboard.log")
);
},
);
}
#[test]
fn runs_dir_structure() {
temp_env::with_var_unset("LEVIATH_RUNS_DIR", || {
let path = runs_dir();
assert!(path.to_str().unwrap().contains(".leviath"));
assert!(path.to_str().unwrap().ends_with("runs"));
});
}
#[test]
fn runs_dir_from_uses_override_when_provided() {
let path = runs_dir_from(Some("/custom/leviath/runs"));
assert_eq!(path, PathBuf::from("/custom/leviath/runs"));
}
#[test]
fn runs_dir_from_falls_back_to_home_when_none() {
let path = runs_dir_from(None);
#[cfg(unix)]
assert!(path.ends_with(".leviath/runs"));
#[cfg(windows)]
assert!(path.ends_with(".leviath\\runs"));
}
#[test]
fn runs_dir_follows_leviath_home() {
temp_env::with_vars(
[
("LEVIATH_RUNS_DIR", None::<&str>),
("LEVIATH_HOME", Some("/tmp/leviath-home-runs-test")),
],
|| {
assert_eq!(
runs_dir(),
PathBuf::from("/tmp/leviath-home-runs-test")
.join(".leviath")
.join("runs")
);
},
);
}
#[test]
fn dashboard_log_path_from_uses_override_when_provided() {
let path = dashboard_log_path_from(Some("/custom/leviath/dashboard.log"));
assert_eq!(path, PathBuf::from("/custom/leviath/dashboard.log"));
}
#[test]
fn dashboard_log_path_from_falls_back_to_home_when_none() {
let path = dashboard_log_path_from(None);
#[cfg(unix)]
assert!(path.ends_with(".leviath/dashboard.log"));
#[cfg(windows)]
assert!(path.ends_with(".leviath\\dashboard.log"));
}
#[test]
fn run_dir_contains_run_id() {
let path = run_dir("my-run-123");
assert!(path.to_str().unwrap().contains("my-run-123"));
}
#[test]
fn with_isolated_runs_dir_points_at_temp_dir_and_cleans_up_after() {
let inside = with_isolated_runs_dir("helper-self-test", |base_dir| {
let expected = base_dir.join("runs");
assert_eq!(runs_dir(), expected);
assert!(runs_dir().exists());
assert_eq!(dashboard_log_path(), base_dir.join("dashboard.log"));
expected
});
assert!(!inside.exists());
}
#[test]
fn tail_file_exact_size() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("exact.txt");
std::fs::write(&path, "exactly").unwrap();
let result = tail_file(&path, 7);
assert_eq!(result, "exactly");
}
#[test]
fn tail_file_tail_without_newline_returns_whole_window() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("no_newline.txt");
let content = "a".repeat(100);
std::fs::write(&path, content.as_bytes()).unwrap();
let result = tail_file(&path, 10);
assert_eq!(result, "aaaaaaaaaa");
}
#[test]
fn run_meta_with_metadata() {
let mut meta = RunMeta::new(
"meta-run".into(),
"agent".into(),
"/p".into(),
"task".into(),
None,
"/w".into(),
1,
);
meta.metadata
.insert("key1".to_string(), "value1".to_string());
meta.callback_url = Some("https://example.com/hook".to_string());
meta.parent_run_id = Some("parent-123".to_string());
let json = serde_json::to_string(&meta).unwrap();
let back: RunMeta = serde_json::from_str(&json).unwrap();
assert_eq!(back.metadata.get("key1").unwrap(), "value1");
assert_eq!(
back.callback_url.as_deref(),
Some("https://example.com/hook")
);
assert_eq!(back.parent_run_id.as_deref(), Some("parent-123"));
}
#[test]
fn stage_record_mutation() {
let mut rec = StageRecord::new("test".into(), 0);
rec.status = StageRunStatus::Active;
rec.started_at = Some(1000);
rec.prompt_tokens = 500;
rec.completion_tokens = 200;
rec.cached_tokens = 50;
assert_eq!(rec.status, StageRunStatus::Active);
assert_eq!(rec.started_at, Some(1000));
assert_eq!(rec.prompt_tokens, 500);
assert_eq!(rec.completion_tokens, 200);
assert_eq!(rec.cached_tokens, 50);
rec.status = StageRunStatus::Complete;
rec.ended_at = Some(2000);
assert_eq!(rec.status, StageRunStatus::Complete);
assert_eq!(rec.ended_at, Some(2000));
}
#[test]
fn context_snapshot_with_entries() {
let snap = ContextSnapshot {
stage_name: "main".into(),
total_tokens: 1000,
max_tokens: 8192,
regions: vec![
RegionSnapshot {
name: "system".into(),
kind: "pinned".into(),
current_tokens: 100,
max_tokens: 2000,
entries: vec![
RegionEntrySnapshot {
content: "You are helpful".into(),
tokens: 3,
kind: Default::default(),
metadata: None,
key: None,
taint: Default::default(),
},
RegionEntrySnapshot {
content: "Additional instruction".into(),
tokens: 5,
kind: Default::default(),
metadata: Some(serde_json::json!({"source": "user"})),
key: None,
taint: Default::default(),
},
],
},
RegionSnapshot {
name: "conversation".into(),
kind: "sliding".into(),
current_tokens: 900,
max_tokens: 6000,
entries: vec![],
},
],
};
let json = serde_json::to_string_pretty(&snap).unwrap();
let back: ContextSnapshot = serde_json::from_str(&json).unwrap();
assert_eq!(back.regions.len(), 2);
assert_eq!(back.regions[0].entries.len(), 2);
assert_eq!(back.regions[0].entries[1].tokens, 5);
assert!(back.regions[0].entries[1].metadata.is_some());
}
#[test]
fn region_entry_snapshot_metadata_omitted_when_none() {
let entry = RegionEntrySnapshot {
content: "test".into(),
tokens: 1,
kind: Default::default(),
metadata: None,
key: None,
taint: Default::default(),
};
let json = serde_json::to_value(&entry).unwrap();
assert!(json.get("metadata").is_none());
}
#[test]
fn append_stage_output_multiple_stages() {
with_isolated_runs_dir("append-stage-output-multiple-stages", |_d| {
let run_id = "test-multi-stage-out";
append_stage_output(run_id, 0, "stage 0 output");
append_stage_output(run_id, 1, "stage 1 output");
append_stage_output(run_id, 2, "stage 2 output");
let out0 = tail_stage_output(run_id, 0, 4096);
let out1 = tail_stage_output(run_id, 1, 4096);
let out2 = tail_stage_output(run_id, 2, 4096);
assert!(out0.contains("stage 0 output"));
assert!(out1.contains("stage 1 output"));
assert!(out2.contains("stage 2 output"));
assert!(!out0.contains("stage 1 output"));
});
}
#[test]
fn list_runs_returns_sorted() {
with_isolated_runs_dir("list-runs-returns-sorted", |_d| {
let meta1 = RunMeta::new(
"test-list-run-a".into(),
"agent".into(),
"/p".into(),
"task a".into(),
None,
"/w".into(),
1,
);
let meta2 = RunMeta::new(
"test-list-run-b".into(),
"agent".into(),
"/p".into(),
"task b".into(),
None,
"/w".into(),
1,
);
let _ = create_run(&meta1);
let _ = create_run(&meta2);
let runs = list_runs();
let ids: Vec<&str> = runs.iter().map(|r| r.run_id.as_str()).collect();
assert!(ids.contains(&"test-list-run-a"));
assert!(ids.contains(&"test-list-run-b"));
});
}
#[test]
fn tail_stage_output_nonexistent_returns_empty() {
assert_eq!(tail_stage_output("no-such-run-xyz", 0, 4096), "");
}
#[test]
fn tail_stage_log_nonexistent_returns_empty() {
assert_eq!(tail_stage_log("no-such-run-xyz", 0, 4096), "");
}
#[test]
fn list_runs_in_dir_nonexistent_returns_empty() {
let result = list_runs_in_dir(PathBuf::from("/nonexistent/leviath/runs/coverage-test"));
assert!(result.is_empty());
}
#[test]
fn list_runs_in_dir_empty_dir_returns_empty() {
let dir = tempfile::tempdir().unwrap();
let result = list_runs_in_dir(dir.path().to_path_buf());
assert!(result.is_empty());
}
#[test]
fn list_runs_in_dir_unreadable_dir_returns_empty() {
let dir = tempfile::tempdir().unwrap();
let not_a_dir = dir.path().join("runs-is-a-file");
std::fs::write(¬_a_dir, "not a dir").unwrap();
let result = list_runs_in_dir(not_a_dir);
assert!(result.is_empty());
}
#[test]
fn append_stage_output_open_failure_is_silently_skipped() {
crate::runstate::with_isolated_runs_dir("append_stage_output_open_failure", |_d| {
let run_id = "append-out-openfail";
ensure_stage_dir(run_id, 0);
std::fs::create_dir_all(stage_dir(run_id, 0).join("output.log")).unwrap();
append_stage_output(run_id, 0, "ignored"); });
}
#[test]
fn append_stage_log_open_failure_is_silently_skipped() {
crate::runstate::with_isolated_runs_dir("append_stage_log_open_failure", |_d| {
let run_id = "append-log-openfail";
ensure_stage_dir(run_id, 0);
std::fs::create_dir_all(stage_dir(run_id, 0).join("logs.log")).unwrap();
append_stage_log(run_id, 0, "ignored"); });
}
#[test]
fn runs_dir_with_override_set_returns_override() {
let tmpdir = tempfile::tempdir().unwrap();
temp_env::with_var("LEVIATH_RUNS_DIR", Some(tmpdir.path()), || {
assert_eq!(runs_dir(), tmpdir.path());
});
}
#[test]
fn runs_dir_without_override_falls_back_to_home() {
temp_env::with_var_unset("LEVIATH_RUNS_DIR", || {
let dir = runs_dir();
#[cfg(unix)]
assert!(dir.ends_with(".leviath/runs"));
#[cfg(windows)]
assert!(dir.ends_with(".leviath\\runs"));
});
}
#[test]
fn list_runs_empty_when_runs_dir_missing_or_empty() {
with_isolated_runs_dir("list-runs-empty-when-runs-dir-missing-or-empty", |_d| {
let runs = list_runs();
assert!(runs.is_empty());
});
}
#[test]
fn tail_file_nonexistent_path_returns_empty() {
let path = std::path::Path::new("/nonexistent/path/to/a/file.log");
assert_eq!(tail_file(path, 1024), "");
}
#[test]
fn tail_file_small_file_returns_whole_contents() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("small.log");
std::fs::write(&path, "hello world").unwrap();
assert_eq!(tail_file(&path, 1024), "hello world");
}
#[test]
fn tail_file_large_file_truncates_from_offset() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("big.log");
let content = "a".repeat(100) + "\nTAIL_MARKER\n";
std::fs::write(&path, &content).unwrap();
let tailed = tail_file(&path, 20);
assert!(tailed.contains("TAIL_MARKER"));
assert!(tailed.len() < content.len());
}
#[test]
fn tail_file_directory_path_returns_empty() {
let dir = tempfile::tempdir().unwrap();
assert_eq!(tail_file(dir.path(), 4), "");
}
#[cfg(unix)]
#[test]
fn tail_file_open_permission_denied_returns_empty() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("no-permissions.log");
std::fs::write(&path, "x".repeat(100)).unwrap();
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o000)).unwrap();
assert_eq!(tail_file(&path, 4), "");
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();
}
#[test]
fn write_context_snapshot_to_hermetic() {
let dir = tempfile::tempdir().unwrap();
let snap = ContextSnapshot {
stage_name: "cov-stage".into(),
total_tokens: 42,
max_tokens: 8192,
regions: vec![],
};
write_context_snapshot_to(dir.path(), &snap).unwrap();
let json = std::fs::read_to_string(dir.path().join("context.json")).unwrap();
let back: ContextSnapshot = serde_json::from_str(&json).unwrap();
assert_eq!(back.total_tokens, 42);
}
#[test]
fn write_context_snapshot_to_fails_without_dir() {
let snap = ContextSnapshot {
stage_name: "s".into(),
total_tokens: 1,
max_tokens: 100,
regions: vec![],
};
let nonexistent = std::path::Path::new("/nonexistent-cov-dir-xyzzy-abc");
let result = write_context_snapshot_to(nonexistent, &snap);
assert!(result.is_err());
}
#[test]
fn write_context_snapshot_to_fails_when_rename_target_is_a_dir() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir(dir.path().join("context.json")).unwrap();
let snap = ContextSnapshot {
stage_name: "s".into(),
total_tokens: 1,
max_tokens: 100,
regions: vec![],
};
let result = write_context_snapshot_to(dir.path(), &snap);
assert!(result.is_err());
}
#[test]
fn create_run_in_hermetic() {
let tmpdir = tempfile::tempdir().unwrap();
let run_dir = tmpdir.path().join("cov-run");
let meta = RunMeta::new(
"cov-run".into(),
"cov-agent".into(),
"/agents/cov".into(),
"cov task".into(),
None,
"/tmp".into(),
1,
);
create_run_in(&run_dir, &meta).unwrap();
let back = read_meta_from(&run_dir).unwrap();
assert_eq!(back.run_id, "cov-run");
}
#[test]
fn create_run_in_fails_on_bad_parent() {
let dir = tempfile::tempdir().unwrap();
let not_a_dir = dir.path().join("not-a-directory");
std::fs::write(¬_a_dir, "x").unwrap();
let bad = not_a_dir.join("run");
let meta = RunMeta::new(
"run".into(),
"a".into(),
"/".into(),
"t".into(),
None,
"/tmp".into(),
1,
);
let result = create_run_in(&bad, &meta);
assert!(result.is_err());
}
#[test]
fn write_meta_to_hermetic() {
let tmpdir = tempfile::tempdir().unwrap();
let meta = RunMeta::new(
"cov-write-meta".into(),
"a".into(),
"/".into(),
"t".into(),
None,
"/tmp".into(),
1,
);
write_meta_to(tmpdir.path(), &meta).unwrap();
let back = read_meta_from(tmpdir.path()).unwrap();
assert_eq!(back.run_id, "cov-write-meta");
}
#[test]
fn write_meta_to_fails_without_dir() {
let meta = RunMeta::new(
"cov-no-dir".into(),
"a".into(),
"/".into(),
"t".into(),
None,
"/tmp".into(),
1,
);
let bad = std::path::Path::new("/nonexistent-cov-write-meta-xyzzy");
let result = write_meta_to(bad, &meta);
assert!(result.is_err());
}
#[test]
fn write_meta_to_fails_when_rename_target_is_a_dir() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir(dir.path().join("meta.json")).unwrap();
let meta = RunMeta::new(
"cov-rename-fail".into(),
"a".into(),
"/".into(),
"t".into(),
None,
"/tmp".into(),
1,
);
let result = write_meta_to(dir.path(), &meta);
assert!(result.is_err());
}
#[test]
fn read_meta_from_fails_on_missing_file() {
let tmpdir = tempfile::tempdir().unwrap();
let result = read_meta_from(tmpdir.path());
assert!(result.is_err());
}
#[test]
fn write_stages_index_to_hermetic() {
let tmpdir = tempfile::tempdir().unwrap();
let stages = vec![StageRecord::new("cov-stage".into(), 0)];
write_stages_index_to(tmpdir.path(), &stages).unwrap();
let json = std::fs::read_to_string(tmpdir.path().join("stages.json")).unwrap();
let back: Vec<StageRecord> = serde_json::from_str(&json).unwrap();
assert_eq!(back.len(), 1);
assert_eq!(back[0].name, "cov-stage");
}
#[test]
fn write_stages_index_to_fails_without_dir() {
let stages = vec![StageRecord::new("s".into(), 0)];
let bad = std::path::Path::new("/nonexistent-cov-stages-xyzzy");
let result = write_stages_index_to(bad, &stages);
assert!(result.is_err());
}
#[test]
fn write_stages_index_to_fails_when_rename_target_is_a_dir() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir(dir.path().join("stages.json")).unwrap();
let stages = vec![StageRecord::new("s".into(), 0)];
let result = write_stages_index_to(dir.path(), &stages);
assert!(result.is_err());
}
#[test]
fn list_runs_in_dir_includes_valid_run() {
let tmpdir = tempfile::tempdir().unwrap();
let run_id = "cov-listed-run";
let run_subdir = tmpdir.path().join(run_id);
std::fs::create_dir_all(&run_subdir).unwrap();
let meta = RunMeta::new(
run_id.into(),
"list-agent".into(),
"/agents/list".into(),
"list task".into(),
None,
"/tmp".into(),
1,
);
let json = serde_json::to_string_pretty(&meta).unwrap();
std::fs::write(run_subdir.join("meta.json"), &json).unwrap();
let runs = list_runs_in_dir(tmpdir.path().to_path_buf());
assert!(runs.iter().any(|r| r.run_id == run_id));
}
#[test]
fn list_runs_in_dir_skips_entry_with_corrupted_meta_json() {
let tmpdir = tempfile::tempdir().unwrap();
let good_run_id = "cov-listed-good-run";
let bad_run_id = "cov-listed-corrupted-run";
let good_subdir = tmpdir.path().join(good_run_id);
std::fs::create_dir_all(&good_subdir).unwrap();
let meta = RunMeta::new(
good_run_id.into(),
"list-agent".into(),
"/agents/list".into(),
"list task".into(),
None,
"/tmp".into(),
1,
);
let json = serde_json::to_string_pretty(&meta).unwrap();
std::fs::write(good_subdir.join("meta.json"), &json).unwrap();
let bad_subdir = tmpdir.path().join(bad_run_id);
std::fs::create_dir_all(&bad_subdir).unwrap();
std::fs::write(bad_subdir.join("meta.json"), "not valid json").unwrap();
let no_meta_run_id = "cov-listed-no-meta-run";
std::fs::create_dir_all(tmpdir.path().join(no_meta_run_id)).unwrap();
let runs = list_runs_in_dir(tmpdir.path().to_path_buf());
assert!(runs.iter().any(|r| r.run_id == good_run_id));
assert!(!runs.iter().any(|r| r.run_id == bad_run_id));
assert!(!runs.iter().any(|r| r.run_id == no_meta_run_id));
}
fn run_dir_with(base: &std::path::Path, run_id: &str, status: RunStatus) -> PathBuf {
let dir = base.join(run_id);
let meta = RunMeta {
status,
..RunMeta::new(
run_id.into(),
"a".into(),
"/p".into(),
"t".into(),
None,
"/w".into(),
1,
)
};
create_run_in(&dir, &meta).unwrap();
dir
}
#[test]
fn force_cancel_terminates_every_non_terminal_status() {
let base = tempfile::tempdir().unwrap();
for status in [
RunStatus::Starting,
RunStatus::Running,
RunStatus::WaitingInput,
] {
let dir = run_dir_with(base.path(), &format!("live-{status}"), status.clone());
assert_eq!(force_cancel_in(&dir, 99), ForceCancelOutcome::Terminated);
let meta = read_meta_from(&dir).unwrap();
assert_eq!(meta.status, RunStatus::Cancelled, "{status} is killable");
assert_eq!(meta.updated_at, 99, "the cancel is stamped");
}
}
#[test]
fn force_cancel_leaves_a_finished_run_alone() {
let base = tempfile::tempdir().unwrap();
for status in [
RunStatus::Complete,
RunStatus::CompleteInteractive,
RunStatus::Error,
RunStatus::Cancelled,
] {
let dir = run_dir_with(base.path(), &format!("done-{status}"), status.clone());
assert_eq!(
force_cancel_in(&dir, 99),
ForceCancelOutcome::AlreadyTerminal,
"{status} is already finished"
);
assert_eq!(read_meta_from(&dir).unwrap().status, status);
}
}
#[test]
fn force_cancel_reports_no_such_run_for_a_missing_directory() {
let base = tempfile::tempdir().unwrap();
let outcome = force_cancel_in(&base.path().join("ghost"), 99);
assert_eq!(outcome, ForceCancelOutcome::NoSuchRun);
assert!(!outcome.found_run(), "nothing to cancel");
}
#[test]
fn force_cancel_writes_a_record_over_unreadable_metadata() {
let base = tempfile::tempdir().unwrap();
let dir = base.path().join("corrupt-run");
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(dir.join("meta.json"), "{ not json").unwrap();
assert_eq!(force_cancel_in(&dir, 99), ForceCancelOutcome::Terminated);
let meta = read_meta_from(&dir).expect("now parses");
assert_eq!(meta.status, RunStatus::Cancelled);
assert_eq!(meta.run_id, "corrupt-run", "recovered from the dir name");
assert!(meta.error.is_some(), "records why it was synthesized");
}
#[test]
fn force_cancel_reports_a_write_failure_but_still_found_the_run() {
crate::test_support::with_tracing(|| {
let base = tempfile::tempdir().unwrap();
let dir = base.path().join("blocked-run");
std::fs::create_dir_all(&dir).unwrap();
std::fs::create_dir_all(dir.join("meta.json")).unwrap();
let outcome = force_cancel_in(&dir, 99);
assert_eq!(outcome, ForceCancelOutcome::WriteFailed);
assert!(outcome.found_run());
});
}
#[test]
fn force_error_records_the_failure_over_a_starting_placeholder() {
let base = tempfile::tempdir().unwrap();
let dir = base.path().join("stillborn-run");
let meta = RunMeta::new(
"stillborn-run".to_string(),
"agent".to_string(),
"/no/such/agent.leviath".to_string(),
"t".to_string(),
None,
"/tmp".to_string(),
0,
);
create_run_in(&dir, &meta).unwrap();
assert_eq!(read_meta_from(&dir).unwrap().status, RunStatus::Starting);
assert_eq!(
force_error_in(&dir, "blueprint not found", 99),
ForceCancelOutcome::Terminated
);
let written = read_meta_from(&dir).unwrap();
assert_eq!(written.status, RunStatus::Error);
assert_eq!(written.error.as_deref(), Some("blueprint not found"));
assert_eq!(written.updated_at, 99);
assert_eq!(written.task, "t");
}
#[test]
fn force_error_leaves_a_run_that_already_finished_alone() {
let base = tempfile::tempdir().unwrap();
let dir = base.path().join("done-run");
let mut meta = RunMeta::new(
"done-run".to_string(),
"agent".to_string(),
String::new(),
"t".to_string(),
None,
"/tmp".to_string(),
0,
);
meta.status = RunStatus::Complete;
create_run_in(&dir, &meta).unwrap();
assert_eq!(
force_error_in(&dir, "too late", 99),
ForceCancelOutcome::AlreadyTerminal
);
assert_eq!(read_meta_from(&dir).unwrap().status, RunStatus::Complete);
}
#[test]
fn force_cancel_keeps_an_error_the_run_had_already_recorded() {
let base = tempfile::tempdir().unwrap();
let dir = base.path().join("noisy-run");
let mut meta = RunMeta::new(
"noisy-run".to_string(),
"agent".to_string(),
String::new(),
"t".to_string(),
None,
"/tmp".to_string(),
0,
);
meta.error = Some("a provider hiccup".to_string());
create_run_in(&dir, &meta).unwrap();
assert_eq!(force_cancel_in(&dir, 99), ForceCancelOutcome::Terminated);
let written = read_meta_from(&dir).unwrap();
assert_eq!(written.status, RunStatus::Cancelled);
assert_eq!(written.error.as_deref(), Some("a provider hiccup"));
}
#[test]
fn force_error_writes_its_message_over_unreadable_metadata() {
let base = tempfile::tempdir().unwrap();
let dir = base.path().join("corrupt-stillborn");
std::fs::create_dir_all(&dir).unwrap();
std::fs::write(dir.join("meta.json"), "{ not json").unwrap();
assert_eq!(
force_error_in(&dir, "blueprint not found", 99),
ForceCancelOutcome::Terminated
);
let written = read_meta_from(&dir).expect("now parses");
assert_eq!(written.status, RunStatus::Error);
assert_eq!(written.error.as_deref(), Some("blueprint not found"));
}
#[test]
fn append_dashboard_log_writes_message() {
with_isolated_runs_dir("append-dashboard-log-writes-message", |_d| {
let unique = format!("cov-dashboard-log-{}", std::process::id());
append_dashboard_log(&unique);
let content = std::fs::read_to_string(dashboard_log_path()).unwrap_or_default();
assert!(content.contains(&unique));
});
}
}