use std::path::{Path, PathBuf};
use codewhale_lane::control::{
Availability, ControlContext, ControlDomain, ControlFailure, ControlFailureKind,
ControlOperation, ControlReceipt, ControlSurface, DEFAULT_RUN_LIST_LIMIT, Known, RunListPage,
RunRouteDto, RunSummaryDto, RunUsageDto, UnknownReason, parse_target, redact_path,
sanitize_line,
};
use codewhale_protocol::fleet::{
FleetArtifactKind, FleetReceipt, FleetRun, FleetRunId, FleetRunStatus, FleetWorkerEventPayload,
FleetWorkerStatus,
};
use super::ledger::FleetLedgerState;
use super::manager::{FleetControlError, FleetManager, FleetStatusSnapshot, FleetWorkerInspection};
pub const MAX_STATUS_WORKER_ROWS: usize = 24;
pub const MAX_INSPECTION_ARTIFACT_ROWS: usize = 24;
#[must_use]
pub fn fleet_ledger_path(workspace: &Path) -> PathBuf {
workspace.join(".codewhale").join("fleet.jsonl")
}
#[must_use]
pub fn fleet_control_context(workspace: &Path) -> ControlContext {
ControlContext::probe(None, Some(&fleet_ledger_path(workspace)))
}
#[must_use]
pub fn worker_status_label(status: &FleetWorkerStatus) -> &'static str {
match status {
FleetWorkerStatus::Unknown => "unknown",
FleetWorkerStatus::Online => "online",
FleetWorkerStatus::Busy => "busy",
FleetWorkerStatus::Offline => "offline",
FleetWorkerStatus::Unhealthy => "unhealthy",
FleetWorkerStatus::Draining => "draining",
FleetWorkerStatus::Retired => "retired",
}
}
#[must_use]
pub fn run_status_label(status: &FleetRunStatus) -> &'static str {
match status {
FleetRunStatus::Pending => "pending",
FleetRunStatus::Queued => "queued",
FleetRunStatus::Running => "running",
FleetRunStatus::Paused => "paused",
FleetRunStatus::Completed => "completed",
FleetRunStatus::Failed => "failed",
FleetRunStatus::Cancelled => "cancelled",
}
}
#[must_use]
pub fn artifact_kind_label(kind: &FleetArtifactKind) -> String {
match kind {
FleetArtifactKind::Log => "log".to_string(),
FleetArtifactKind::Patch => "patch".to_string(),
FleetArtifactKind::TestResult => "test_result".to_string(),
FleetArtifactKind::Report => "report".to_string(),
FleetArtifactKind::Checkpoint => "checkpoint".to_string(),
FleetArtifactKind::Receipt => "receipt".to_string(),
FleetArtifactKind::Other(value) => value.clone(),
}
}
#[must_use]
pub fn event_label(payload: &FleetWorkerEventPayload) -> String {
match payload {
FleetWorkerEventPayload::Queued => "queued".to_string(),
FleetWorkerEventPayload::Leased { .. } => "leased".to_string(),
FleetWorkerEventPayload::Starting => "starting".to_string(),
FleetWorkerEventPayload::Running => "running".to_string(),
FleetWorkerEventPayload::ModelWait { model } => model
.as_ref()
.map(|model| format!("model_wait model={model}"))
.unwrap_or_else(|| "model_wait".to_string()),
FleetWorkerEventPayload::RunningTool { tool, call_id } => call_id
.as_ref()
.map(|call_id| format!("running_tool tool={tool} call_id={call_id}"))
.unwrap_or_else(|| format!("running_tool tool={tool}")),
FleetWorkerEventPayload::WorkflowEvent {
workflow_run_id,
event,
} => event
.get("type")
.and_then(serde_json::Value::as_str)
.map(|kind| format!("workflow_event run_id={workflow_run_id} type={kind}"))
.unwrap_or_else(|| format!("workflow_event run_id={workflow_run_id}")),
FleetWorkerEventPayload::Heartbeat { .. } => "heartbeat".to_string(),
FleetWorkerEventPayload::Artifact(artifact) => {
format!("artifact kind={}", artifact_kind_label(&artifact.kind))
}
FleetWorkerEventPayload::Completed { exit_code, summary } => match (exit_code, summary) {
(Some(code), Some(summary)) => format!("completed exit_code={code} {summary}"),
(Some(code), None) => format!("completed exit_code={code}"),
(None, Some(summary)) => format!("completed {summary}"),
(None, None) => "completed".to_string(),
},
FleetWorkerEventPayload::Failed {
reason,
recoverable,
} => format!("failed recoverable={recoverable} reason={reason}"),
FleetWorkerEventPayload::Cancelled { cancelled_by } => cancelled_by
.as_ref()
.map(|by| format!("cancelled by={by}"))
.unwrap_or_else(|| "cancelled".to_string()),
FleetWorkerEventPayload::Interrupted { signal } => signal
.as_ref()
.map(|signal| format!("interrupted signal={signal}"))
.unwrap_or_else(|| "interrupted".to_string()),
FleetWorkerEventPayload::Stale { last_heartbeat_at } => last_heartbeat_at
.as_ref()
.map(|ts| format!("stale last_heartbeat_at={ts}"))
.unwrap_or_else(|| "stale".to_string()),
FleetWorkerEventPayload::Restarted { restart_count } => {
format!("restarted count={restart_count}")
}
FleetWorkerEventPayload::Escalated { channel, alert_id } => alert_id
.as_ref()
.map(|alert_id| format!("escalated channel={channel} alert_id={alert_id}"))
.unwrap_or_else(|| format!("escalated channel={channel}")),
}
}
#[must_use]
pub fn status_lines(status: &FleetStatusSnapshot) -> Vec<String> {
let mut lines = vec![format!(
"fleet: runs={} queued={} running={} completed={} partial={} failed={} restarted={} \
escalated={} transport_failed={} task_failed={} verifier_failed={} cancelled={} stale={}",
status.runs,
status.queued,
status.running,
status.completed,
status.partial,
status.failed,
status.restarted,
status.escalated,
status.transport_failed,
status.task_failed,
status.verifier_failed,
status.cancelled,
status.stale
)];
if !status.workers.is_empty() {
lines.push("workers:".to_string());
for (worker_id, worker_status) in status.workers.iter().take(MAX_STATUS_WORKER_ROWS) {
lines.push(format!(
" {worker_id} {}",
worker_status_label(worker_status)
));
}
let omitted = status.workers.len().saturating_sub(MAX_STATUS_WORKER_ROWS);
if omitted > 0 {
lines.push(format!(
" [{omitted} more worker(s) omitted by the {MAX_STATUS_WORKER_ROWS}-row bound]"
));
}
}
lines
}
#[must_use]
pub fn render_fleet_status_snapshot(status: &FleetStatusSnapshot) -> String {
status_lines(status).join("\n")
}
#[must_use]
pub fn inspection_lines(inspection: &FleetWorkerInspection) -> Vec<String> {
let mut lines = vec![
format!("worker: {}", inspection.worker_id),
format!("status: {}", worker_status_label(&inspection.status)),
];
if let Some(run_id) = &inspection.current_run_id {
lines.push(format!("run: {}", run_id.0));
}
if let Some(task_id) = &inspection.current_task_id {
lines.push(format!("task: {task_id}"));
}
if let Some(objective) = &inspection.objective {
lines.push(format!("objective: {objective}"));
}
if let Some(role) = &inspection.role {
lines.push(format!("role: {role}"));
}
if let Some(host) = &inspection.host {
lines.push(format!("host: {host}"));
}
if let Some(heartbeat) = &inspection.latest_heartbeat_at {
lines.push(format!("heartbeat: {heartbeat}"));
}
if let Some(event) = &inspection.latest_event {
lines.push(format!(
"latest_event: seq={} {}",
event.seq,
event_label(&event.payload)
));
}
if !inspection.artifacts.is_empty() {
lines.push("artifacts:".to_string());
for artifact in inspection
.artifacts
.iter()
.take(MAX_INSPECTION_ARTIFACT_ROWS)
{
lines.push(format!(
" {} {}",
artifact_kind_label(&artifact.kind),
artifact.path.display()
));
}
let omitted = inspection
.artifacts
.len()
.saturating_sub(MAX_INSPECTION_ARTIFACT_ROWS);
if omitted > 0 {
lines.push(format!(" [{omitted} more artifact(s) omitted]"));
}
}
if let Some(receipt) = &inspection.receipt_summary {
lines.push(format!("receipt: {receipt}"));
}
if let Some(error) = &inspection.last_error {
lines.push(format!("last_error: {error}"));
}
if let Some(alert) = &inspection.alert_state {
lines.push(format!("alert: {alert}"));
}
lines
}
#[must_use]
pub fn render_inspection(inspection: &FleetWorkerInspection) -> String {
inspection_lines(inspection).join("\n")
}
#[must_use]
pub fn artifact_lines(inspection: &FleetWorkerInspection) -> Vec<String> {
if inspection.artifacts.is_empty() {
return vec!["artifacts: none".to_string()];
}
let mut lines = vec!["artifacts:".to_string()];
for artifact in inspection
.artifacts
.iter()
.take(MAX_INSPECTION_ARTIFACT_ROWS)
{
let size = artifact
.size_bytes
.map(|size| format!(" size={size}"))
.unwrap_or_default();
let mime = artifact
.mime_type
.as_ref()
.map(|mime| format!(" mime={mime}"))
.unwrap_or_default();
lines.push(format!(
" {} {}{}{}",
artifact_kind_label(&artifact.kind),
artifact.path.display(),
size,
mime
));
}
let omitted = inspection
.artifacts
.len()
.saturating_sub(MAX_INSPECTION_ARTIFACT_ROWS);
if omitted > 0 {
lines.push(format!(" [{omitted} more artifact(s) omitted]"));
}
lines
}
#[must_use]
pub fn render_artifacts(inspection: &FleetWorkerInspection) -> String {
artifact_lines(inspection).join("\n")
}
fn label(run: &FleetRun, key: &str) -> Option<String> {
run.labels.get(key).cloned()
}
fn route_dto(receipt: Option<&FleetReceipt>) -> RunRouteDto {
let Some(route) = receipt.and_then(|receipt| receipt.resolved_route.as_ref()) else {
return RunRouteDto::all_unknown(UnknownReason::NotRecorded);
};
RunRouteDto {
provider_id: Known::Known(route.provider_id.clone()),
provider_exact_id: Known::from_option(route.provider_exact_id.clone()),
model: Known::Known(
route
.canonical_model
.clone()
.unwrap_or_else(|| route.wire_model_id.clone()),
),
requested_reasoning: Known::unknown(),
effective_reasoning: Known::from_option(route.reasoning_effort.clone()),
route_source: Known::Known(route.source.clone()),
}
}
#[must_use]
pub fn fleet_run_summary(run: &FleetRun, receipt: Option<&FleetReceipt>) -> RunSummaryDto {
RunSummaryDto {
domain: ControlDomain::Fleet,
run_id: run.id.0.clone(),
status: run_status_label(&run.status).to_string(),
lifecycle_seq: Known::not_applicable(),
runtime: Known::from_option(label(run, "runtime")),
workflow: Known::from_option(label(run, "workflow")),
fleet: Known::from_option(
label(run, "fleet").or_else(|| (!run.name.trim().is_empty()).then(|| run.name.clone())),
),
issue: Known::from_option(label(run, "issue")),
goal: Known::from_option(label(run, "goal").or_else(|| label(run, "objective"))),
started_at: Known::Known(run.created_at.clone()),
stopped_at: Known::from_option(run.completed_at.clone()),
location: Known::not_applicable(),
branch: Known::not_applicable(),
runtime_session: Known::not_applicable(),
runtime_socket: Known::not_applicable(),
attach: Known::not_applicable(),
log: Known::not_applicable(),
route: route_dto(receipt),
usage: RunUsageDto::all_unknown(UnknownReason::NotRecorded),
}
}
#[must_use]
pub fn fleet_run_page(state: &FleetLedgerState, limit: usize) -> RunListPage {
let mut summaries: Vec<RunSummaryDto> = state
.runs
.values()
.map(|run| {
let status = state
.run_status_overrides
.get(&run.id.0)
.cloned()
.unwrap_or_else(|| run.status.clone());
let receipt = state
.receipts
.values()
.filter(|receipt| receipt.run_id.0 == run.id.0)
.max_by(|a, b| a.completed_at.cmp(&b.completed_at));
let mut summary = fleet_run_summary(run, receipt);
summary.status = run_status_label(&status).to_string();
summary
})
.collect();
summaries.sort_by(|a, b| {
instant_of(&a.started_at)
.cmp(&instant_of(&b.started_at))
.reverse()
.then_with(|| a.run_id.cmp(&b.run_id))
});
RunListPage::bounded(summaries, limit)
}
fn instant_of(value: &Known<String>) -> Option<chrono::DateTime<chrono::Utc>> {
let raw = value.as_known()?;
chrono::DateTime::parse_from_rfc3339(raw)
.ok()
.map(|parsed| parsed.with_timezone(&chrono::Utc))
}
#[must_use]
pub fn execute_fleet_control(
surface: ControlSurface,
workspace: &Path,
operation: ControlOperation,
raw_target: Option<&str>,
) -> ControlReceipt {
let descriptor = operation.descriptor();
let availability = descriptor.availability(surface, fleet_control_context(workspace));
if !availability.is_available() {
return ControlReceipt::unavailable(descriptor, surface, availability);
}
match FleetManager::open(workspace) {
Ok(manager) => execute_fleet_control_with(
surface,
workspace,
fleet_control_context(workspace),
&manager,
operation,
raw_target,
),
Err(err) => ControlReceipt::failed(
descriptor,
surface,
None,
ControlFailure::backend(format!("{err:#}")),
),
}
}
#[must_use]
pub fn execute_fleet_control_with(
surface: ControlSurface,
workspace: &Path,
ctx: ControlContext,
manager: &FleetManager,
operation: ControlOperation,
raw_target: Option<&str>,
) -> ControlReceipt {
let descriptor = operation.descriptor();
if descriptor.domain != ControlDomain::Fleet {
return ControlReceipt::rejected(
descriptor,
surface,
None,
ControlFailure::new(
ControlFailureKind::InvalidTarget,
format!("{} is not a Fleet verb", descriptor.id),
),
);
}
let availability = descriptor.availability(surface, ctx);
if !availability.is_available() {
return ControlReceipt::unavailable(descriptor, surface, availability);
}
let target = match parse_target(descriptor, raw_target) {
Ok(target) => target,
Err(failure) => return ControlReceipt::rejected(descriptor, surface, None, failure),
};
match operation {
ControlOperation::FleetList => match manager.rebuild_state() {
Ok(state) => ControlReceipt::inspected(descriptor, surface, None)
.with_runs(fleet_run_page(&state, DEFAULT_RUN_LIST_LIMIT))
.with_detail([format!(
"ledger: {}",
redact_path(&fleet_ledger_path(workspace))
)]),
Err(err) => ControlReceipt::failed(
descriptor,
surface,
None,
ControlFailure::backend(format!("{err:#}")),
),
},
ControlOperation::FleetStatus => match manager.status() {
Ok(status) => ControlReceipt::inspected(descriptor, surface, None).with_detail(
status_lines(&status).into_iter().chain([format!(
"ledger: {}",
redact_path(&fleet_ledger_path(workspace))
)]),
),
Err(err) => ControlReceipt::failed(
descriptor,
surface,
None,
ControlFailure::backend(format!("{err:#}")),
),
},
ControlOperation::FleetInterrupt => {
let Some(target) = target else {
return ControlReceipt::rejected(
descriptor,
surface,
None,
ControlFailure::invalid_target(format!(
"{} needs an exact worker id",
descriptor.id
)),
);
};
match manager.rebuild_state() {
Ok(state) => {
if !state.workers.contains_key(&target.id) {
return ControlReceipt::rejected(
descriptor,
surface,
Some(target.clone()),
ControlFailure::not_found(format!(
"no fleet worker with id {} in this workspace's ledger",
target.id
)),
);
}
}
Err(err) => {
return ControlReceipt::failed(
descriptor,
surface,
Some(target),
ControlFailure::backend(format!("{err:#}")),
);
}
}
let interrupted = manager.interrupt_worker(&target.id);
match interrupted {
Ok(inspection) => ControlReceipt::transitioned(descriptor, surface, Some(target))
.with_detail(inspection_lines(&inspection)),
Err(err) => {
let message = format!("{err:#}");
let failure = match err.downcast_ref::<FleetControlError>() {
Some(FleetControlError::NoActiveTask { .. }) => {
ControlFailure::conflict(message)
}
Some(FleetControlError::UnknownRun { .. }) => {
ControlFailure::not_found(message)
}
None => ControlFailure::backend(message),
};
ControlReceipt::failed(descriptor, surface, Some(target), failure)
}
}
}
ControlOperation::FleetResume => {
let Some(target) = target else {
return ControlReceipt::rejected(
descriptor,
surface,
None,
ControlFailure::invalid_target(format!(
"{} needs an exact run id",
descriptor.id
)),
);
};
match manager.rebuild_state() {
Ok(state) => {
if !state.runs.contains_key(&target.id) {
return ControlReceipt::rejected(
descriptor,
surface,
Some(target.clone()),
ControlFailure::not_found(
FleetControlError::UnknownRun {
run_id: target.id.clone(),
}
.to_string(),
),
);
}
}
Err(err) => {
return ControlReceipt::failed(
descriptor,
surface,
Some(target),
ControlFailure::backend(format!("{err:#}")),
);
}
}
let resumed = manager.resume_run(&FleetRunId::from(target.id.clone()));
match resumed {
Ok(report) => {
let reconciled = report.reclaimed_stale
+ report.restarted
+ report.failed
+ report.escalated;
let detail = [format!(
"fleet resume: {} reclaimed_stale={} restarted={} failed={} escalated={}",
report.run_id.0,
report.reclaimed_stale,
report.restarted,
report.failed,
report.escalated
)]
.into_iter()
.chain(status_lines(&report.status));
if reconciled == 0 {
ControlReceipt::no_change(descriptor, surface, Some(target))
.with_detail(detail)
} else {
ControlReceipt::transitioned(descriptor, surface, Some(target))
.with_detail(detail)
}
}
Err(err) => ControlReceipt::failed(
descriptor,
surface,
Some(target),
ControlFailure::backend(format!("{err:#}")),
),
}
}
_ => ControlReceipt::unavailable(
descriptor,
surface,
Availability::Unavailable {
reason: codewhale_lane::UnavailableReason::SurfaceNotSupported,
hint: sanitize_line(descriptor.cli_invocation),
},
),
}
}
#[cfg(test)]
mod tests {
use super::*;
use codewhale_lane::{ControlAuthority, LifecycleOutcome, PersistenceScope, UnavailableReason};
use codewhale_protocol::fleet::{FleetResolvedRoute, FleetTaskResult};
use std::collections::BTreeMap;
fn run(id: &str) -> FleetRun {
FleetRun {
id: FleetRunId::from(id.to_string()),
name: "stopship".to_string(),
status: FleetRunStatus::Running,
target: None,
workflow: None,
roles: Vec::new(),
max_workers: Some(2),
task_specs: Vec::new(),
worker_specs: Vec::new(),
labels: BTreeMap::new(),
security_policy: None,
created_at: "2026-07-26T00:00:00Z".to_string(),
updated_at: None,
completed_at: None,
}
}
fn receipt_with_route(run_id: &str) -> FleetReceipt {
FleetReceipt {
run_id: FleetRunId::from(run_id.to_string()),
task_id: "task-1".to_string(),
worker_id: "worker-1".to_string(),
attempt: Some(1),
terminal_seq: Some(9),
completed_at: "2026-07-26T00:01:00Z".to_string(),
result: FleetTaskResult::Pass,
failure_kind: None,
artifacts: Vec::new(),
score: None,
resolved_route: Some(FleetResolvedRoute {
provider_id: "deepseek".to_string(),
provider_exact_id: Some("custom".to_string()),
provider_kind: "deepseek".to_string(),
canonical_model: Some("deepseek-v3".to_string()),
wire_model_id: "deepseek-chat".to_string(),
protocol: "chat_completions".to_string(),
role: Some("implementer".to_string()),
loadout: None,
model_class: None,
model_route: None,
reasoning_effort: Some("high".to_string()),
role_source: None,
loadout_source: None,
model_class_source: None,
model_source: None,
source: "resolver".to_string(),
}),
effective_permissions: None,
}
}
#[test]
fn fleet_run_dto_uses_exact_route_and_types_what_the_ledger_omits() {
let run = run("run-1");
let receipt = receipt_with_route("run-1");
let summary = fleet_run_summary(&run, Some(&receipt));
assert_eq!(summary.domain, ControlDomain::Fleet);
assert_eq!(summary.run_id, "run-1");
assert_eq!(summary.status, "running");
assert_eq!(
summary.route.provider_id,
Known::Known("deepseek".to_string())
);
assert_eq!(
summary.route.provider_exact_id,
Known::Known("custom".to_string()),
"the exact provider-table id must not collapse into the generic id"
);
assert_eq!(summary.route.model, Known::Known("deepseek-v3".to_string()));
assert_eq!(
summary.route.effective_reasoning,
Known::Known("high".to_string())
);
assert_eq!(
summary.route.requested_reasoning.unknown_reason(),
Some(UnknownReason::NotRecorded),
"the ledger records the effective tier only; do not invent the request"
);
assert_eq!(summary.route.reasoning_downgraded(), None);
assert_eq!(
summary.route.route_source,
Known::Known("resolver".to_string())
);
assert_eq!(
summary.usage.total_tokens.unknown_reason(),
Some(UnknownReason::NotRecorded)
);
assert_eq!(
summary.lifecycle_seq.unknown_reason(),
Some(UnknownReason::NotApplicable)
);
}
#[test]
fn a_run_without_a_receipt_reports_every_route_field_unknown() {
let summary = fleet_run_summary(&run("run-2"), None);
for reason in [
summary.route.provider_id.unknown_reason(),
summary.route.model.unknown_reason(),
summary.route.effective_reasoning.unknown_reason(),
summary.route.route_source.unknown_reason(),
] {
assert_eq!(reason, Some(UnknownReason::NotRecorded));
}
}
#[test]
fn fleet_run_pages_are_bounded() {
let mut state = FleetLedgerState::default();
for index in 0..10 {
let id = format!("run-{index:03}");
state.runs.insert(id.clone(), run(&id));
}
let page = fleet_run_page(&state, 3);
assert_eq!(page.total, 10);
assert_eq!(page.runs.len(), 3);
assert_eq!(page.truncated, 7);
}
#[test]
fn an_absent_ledger_is_reported_and_not_created() {
let dir = tempfile::tempdir().unwrap();
for surface in ControlSurface::ALL {
let receipt =
execute_fleet_control(*surface, dir.path(), ControlOperation::FleetStatus, None);
assert_eq!(
receipt.availability.reason(),
Some(UnavailableReason::NoFleetLedger),
"{surface}"
);
assert_eq!(receipt.outcome, LifecycleOutcome::Rejected);
}
assert!(
!fleet_ledger_path(dir.path()).exists(),
"a read verb must not create the durable ledger"
);
}
#[test]
fn durable_fleet_status_is_identical_on_every_surface() {
let dir = tempfile::tempdir().unwrap();
FleetManager::open(dir.path()).unwrap();
let mut rendered = std::collections::BTreeSet::new();
for surface in ControlSurface::ALL {
let receipt =
execute_fleet_control(*surface, dir.path(), ControlOperation::FleetStatus, None);
assert_eq!(receipt.operation_id, "fleet.status");
assert_eq!(receipt.authority, ControlAuthority::Read);
assert_eq!(receipt.persistence, PersistenceScope::FleetLedger);
assert_eq!(receipt.outcome, LifecycleOutcome::Inspected);
assert!(
receipt
.detail
.iter()
.any(|line| line.starts_with("fleet: runs=")),
"the durable ledger snapshot must be the payload"
);
let mut normalized = receipt.clone();
normalized.surface = ControlSurface::Cli;
rendered.insert(normalized.render());
}
assert_eq!(rendered.len(), 1, "surfaces rendered different results");
}
#[test]
fn fleet_restart_is_cli_only_and_says_so_elsewhere() {
let dir = tempfile::tempdir().unwrap();
FleetManager::open(dir.path()).unwrap();
{
let surface = ControlSurface::Slash;
let receipt = execute_fleet_control(
surface,
dir.path(),
ControlOperation::FleetRestart,
Some("worker-1"),
);
assert_eq!(receipt.outcome, LifecycleOutcome::Rejected);
assert_eq!(
receipt.availability.reason(),
Some(UnavailableReason::SurfaceNotSupported)
);
assert!(
receipt
.availability
.hint()
.is_some_and(|hint| hint.contains("codewhale fleet restart"))
);
}
}
#[test]
fn interrupt_requires_an_exact_worker_id() {
let dir = tempfile::tempdir().unwrap();
FleetManager::open(dir.path()).unwrap();
for bad in [None, Some(""), Some("worker one"), Some("../escape")] {
let receipt = execute_fleet_control(
ControlSurface::Slash,
dir.path(),
ControlOperation::FleetInterrupt,
bad,
);
assert_eq!(
receipt.failure.as_ref().map(|failure| failure.kind),
Some(ControlFailureKind::InvalidTarget),
"{bad:?}"
);
}
}
#[test]
fn resuming_an_unknown_run_is_not_found_and_writes_nothing() {
let dir = tempfile::tempdir().unwrap();
FleetManager::open(dir.path()).unwrap();
let ledger = fleet_ledger_path(dir.path());
let before = std::fs::read(&ledger).unwrap();
for surface in ControlSurface::ALL {
let receipt = execute_fleet_control(
*surface,
dir.path(),
ControlOperation::FleetResume,
Some("run-does-not-exist"),
);
assert_eq!(receipt.outcome, LifecycleOutcome::Rejected, "{surface}");
assert_eq!(
receipt.failure.as_ref().map(|failure| failure.kind),
Some(ControlFailureKind::NotFound),
"{surface}"
);
}
assert_eq!(
std::fs::read(&ledger).unwrap(),
before,
"a refused resume must not append to the durable ledger"
);
let state = FleetManager::open(dir.path())
.unwrap()
.rebuild_state()
.unwrap();
assert!(
!state
.run_status_overrides
.contains_key("run-does-not-exist"),
"a caller-supplied id must never become a durable ledger key"
);
}
#[test]
fn runs_sort_by_utc_instant_not_by_rendered_text() {
let mut state = FleetLedgerState::default();
let mut earlier = run("run-offset");
earlier.created_at = "2026-07-26T00:30:00+02:00".to_string();
let mut later = run("run-utc");
later.created_at = "2026-07-25T23:00:00Z".to_string();
state.runs.insert("run-offset".to_string(), earlier);
state.runs.insert("run-utc".to_string(), later);
let page = fleet_run_page(&state, DEFAULT_RUN_LIST_LIMIT);
assert_eq!(
page.runs
.iter()
.map(|run| run.run_id.as_str())
.collect::<Vec<_>>(),
vec!["run-utc", "run-offset"],
"a string compare would have put run-offset first"
);
}
#[test]
fn status_and_inspection_rendering_stay_bounded() {
let mut snapshot = FleetStatusSnapshot::default();
for index in 0..(MAX_STATUS_WORKER_ROWS + 5) {
snapshot
.workers
.insert(format!("worker-{index:03}"), FleetWorkerStatus::Online);
}
let lines = status_lines(&snapshot);
assert_eq!(lines.len(), 1 + 1 + MAX_STATUS_WORKER_ROWS + 1);
assert!(lines.last().unwrap().contains("5 more worker(s) omitted"));
}
}