use std::collections::BTreeSet;
use serde::Serialize;
use serde_json::{Map, Value};
use crate::client::GatewayApi;
use crate::client::eam::{
DeleteOutcome, EamHistoryItem, EamScheduledTask, EamTaskRecord, ModifyOutcome, ResourceChange,
};
use crate::error::CoreError;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TaskCreateVerdict {
Unguarded,
NeedsYes,
Refused,
}
const REFUSED_TYPES: [&str; 3] = [
"eam_restoreBackup",
"eam_installModules",
"eam_remoteUpgrade",
];
const MUTATING_TYPES: [&str; 7] = [
"eam_restart",
"eam_sendProject",
"eam_sendResource",
"eam_sendTags",
"eam_activateLicense",
"eam_updateLicense",
"eam_unactivateLicense",
];
pub fn task_create_guard(task_type: &str, schedule_mode: &str) -> TaskCreateVerdict {
if REFUSED_TYPES.contains(&task_type) {
return TaskCreateVerdict::Refused;
}
if !schedule_mode.eq("OnDemand") {
return TaskCreateVerdict::NeedsYes;
}
if MUTATING_TYPES.contains(&task_type) || task_type != "eam_backup" {
return TaskCreateVerdict::NeedsYes;
}
TaskCreateVerdict::Unguarded
}
#[derive(Debug, Serialize)]
pub struct EamTaskCreateResult {
pub name: String,
pub task_type: String,
pub schedule_mode: String,
pub definition: Value,
}
#[derive(Debug, Serialize)]
pub struct EamTaskForceResult {
pub task: String,
pub owner: String,
pub dispatched: bool,
pub history: Option<EamHistoryItem>,
pub preview: BlastRadiusPreview,
}
pub fn parse_setting(raw: &str) -> Result<(String, Value), CoreError> {
let Some((key, value)) = raw.split_once('=') else {
return Err(CoreError::InvalidInput {
reason: format!(
"--setting expects K=V (got {raw:?}) — a value that parses as \
bool/int rides typed, anything else stays a string; arrays and \
objects need --definition <PATH>"
),
});
};
if key.is_empty() || value.is_empty() {
return Err(CoreError::InvalidInput {
reason: format!("--setting expects non-empty K and V (got {raw:?})"),
});
}
Ok((key.to_string(), auto_type(value)))
}
fn auto_type(value: &str) -> Value {
if value == "true" {
return Value::Bool(true);
}
if value == "false" {
return Value::Bool(false);
}
if let Ok(int) = value.parse::<i64>() {
return Value::Number(int.into());
}
Value::String(value.to_string())
}
fn deep_merge(base: &mut Value, overlay: &Value) {
match (base, overlay) {
(Value::Object(base_map), Value::Object(overlay_map)) => {
for (key, overlay_value) in overlay_map {
match base_map.get_mut(key) {
Some(base_value @ Value::Object(_)) if overlay_value.is_object() => {
deep_merge(base_value, overlay_value);
}
_ => {
base_map.insert(key.clone(), overlay_value.clone());
}
}
}
}
(base, overlay) => *base = overlay.clone(),
}
}
fn compose_task_definition(
name: &str,
task_type: &str,
targets: &[String],
settings: &[String],
definition: Option<&Value>,
schedule_mode: &str,
) -> Result<Value, CoreError> {
let mut profile = Map::new();
profile.insert("type".to_string(), Value::String(task_type.to_string()));
profile.insert(
"scheduleMode".to_string(),
Value::String(schedule_mode.to_string()),
);
let mut composed_settings = Map::new();
composed_settings.insert(
"targetGateways".to_string(),
if targets.is_empty() {
Value::Array(vec![Value::String("_controller".to_string())])
} else {
Value::Array(targets.iter().map(|t| Value::String(t.clone())).collect())
},
);
composed_settings.insert("targetGroups".to_string(), Value::Array(vec![]));
for raw in settings {
let (key, value) = parse_setting(raw)?;
composed_settings.insert(key, value);
}
let mut settings_value = Value::Object(composed_settings);
if let Some(overlay) = definition {
deep_merge(&mut settings_value, overlay);
}
Ok(serde_json::json!({
"name": name,
"config": {
"profile": Value::Object(profile),
"settings": settings_value,
},
}))
}
pub async fn eam_task_create(
api: &dyn GatewayApi,
name: &str,
task_type: &str,
targets: &[String],
settings: &[String],
definition: Option<&Value>,
schedule_mode: &str,
) -> Result<EamTaskCreateResult, CoreError> {
if let TaskCreateVerdict::Refused = task_create_guard(task_type, schedule_mode) {
return Err(CoreError::EamTaskTypeRefused {
task_type: task_type.to_string(),
});
}
let composed = compose_task_definition(
name,
task_type,
targets,
settings,
definition,
schedule_mode,
)?;
api.eam_task_create(&composed).await?;
Ok(EamTaskCreateResult {
name: name.to_string(),
task_type: task_type.to_string(),
schedule_mode: schedule_mode.to_string(),
definition: composed,
})
}
pub async fn eam_task_force(
api: &dyn GatewayApi,
name: &str,
) -> Result<EamTaskForceResult, CoreError> {
let preview = build_blast_radius(api, "force", name).await?;
let owner = preview.owner.clone().unwrap_or_else(|| "eam".to_string());
api.eam_task_force(&owner, name).await?;
let history = api
.eam_task_history(Some(20), Some(name))
.await
.ok()
.and_then(|page| {
page.items.into_iter().find(|item| {
let forced = format!("{name} (forced)");
item.task_name == name || item.task_name == forced
})
});
Ok(EamTaskForceResult {
task: name.to_string(),
owner,
dispatched: true,
history,
preview,
})
}
#[derive(Debug, Serialize)]
pub struct EamLifecycleResult {
pub task: String,
pub action: String,
pub previous_state: Option<String>,
pub config_suspended: Option<bool>,
pub pending: Option<EamScheduledTask>,
pub fired: bool,
pub reason: Option<String>,
}
pub fn lifecycle_precheck(action: &str, task_name: &str) -> Result<(), CoreError> {
if task_name.trim().is_empty() {
return Err(CoreError::InvalidInput {
reason: format!("eam task {action}: the task name must not be empty/whitespace"),
});
}
Ok(())
}
pub fn suspend_recheck(record: &EamTaskRecord) -> Result<(), CoreError> {
let suspended = record
.config
.get("profile")
.and_then(|profile| profile.get("isSuspended"))
.and_then(Value::as_bool);
if suspended == Some(true) {
return Err(CoreError::InvalidInput {
reason: format!(
"task {:?} is already suspended (config.profile.isSuspended=true) — \
nothing to suspend; `eam task resume` it first",
record.name
),
});
}
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CancelDecision {
Fire,
NoPending,
NotPermitted,
}
pub fn cancel_decision(pending: Option<&EamScheduledTask>) -> CancelDecision {
match pending {
None => CancelDecision::NoPending,
Some(row) if row.can_cancel => CancelDecision::Fire,
Some(_) => CancelDecision::NotPermitted,
}
}
fn current_state_of(record: &EamTaskRecord) -> Option<String> {
record
.scheduled_task_state
.as_ref()
.and_then(|state| state.get("currentState"))
.and_then(Value::as_str)
.map(str::to_string)
}
fn is_suspended_of(record: &EamTaskRecord) -> Option<bool> {
record
.config
.get("profile")
.and_then(|profile| profile.get("isSuspended"))
.and_then(Value::as_bool)
}
async fn pending_row_for(
api: &dyn GatewayApi,
name: &str,
) -> Result<Option<EamScheduledTask>, CoreError> {
Ok(pending_rows_for(api, name).await?.into_iter().next())
}
pub async fn eam_task_suspend(
api: &dyn GatewayApi,
name: &str,
) -> Result<EamLifecycleResult, CoreError> {
lifecycle_precheck("suspend", name)?;
let record = api.eam_task_find(name).await?;
let previous_state = current_state_of(&record);
suspend_recheck(&record)?;
api.eam_task_suspend(name).await?;
let readback = api.eam_task_find(name).await?;
Ok(EamLifecycleResult {
task: name.to_string(),
action: "suspended".to_string(),
previous_state,
config_suspended: is_suspended_of(&readback),
pending: None,
fired: true,
reason: None,
})
}
pub async fn eam_task_resume(
api: &dyn GatewayApi,
name: &str,
) -> Result<EamLifecycleResult, CoreError> {
lifecycle_precheck("resume", name)?;
let record = api.eam_task_find(name).await?;
let previous_state = current_state_of(&record);
api.eam_task_resume(name).await?;
let readback = api.eam_task_find(name).await?;
Ok(EamLifecycleResult {
task: name.to_string(),
action: "resumed".to_string(),
previous_state,
config_suspended: is_suspended_of(&readback),
pending: None,
fired: true,
reason: None,
})
}
pub async fn eam_task_cancel(
api: &dyn GatewayApi,
name: &str,
) -> Result<EamLifecycleResult, CoreError> {
lifecycle_precheck("cancel", name)?;
let record = api.eam_task_find(name).await?;
let previous_state = current_state_of(&record);
let pending = pending_row_for(api, name).await?;
match cancel_decision(pending.as_ref()) {
CancelDecision::Fire => {
api.eam_task_cancel(name).await?;
let after = pending_row_for(api, name).await?;
Ok(EamLifecycleResult {
task: name.to_string(),
action: "cancelled".to_string(),
previous_state,
config_suspended: None,
pending: after,
fired: true,
reason: None,
})
}
CancelDecision::NoPending => Ok(EamLifecycleResult {
task: name.to_string(),
action: "cancelled".to_string(),
previous_state,
config_suspended: None,
pending: None,
fired: false,
reason: Some("no pending execution".to_string()),
}),
CancelDecision::NotPermitted => Ok(EamLifecycleResult {
task: name.to_string(),
action: "cancelled".to_string(),
previous_state,
config_suspended: None,
pending,
fired: false,
reason: Some(
"the gateway reports canCancel=false for the pending execution".to_string(),
),
}),
}
}
#[derive(Debug, Clone, Default)]
pub struct TaskChange {
pub enabled: Option<bool>,
pub description: Option<String>,
pub schedule_mode: Option<String>,
pub settings_overlay: Option<Value>,
}
#[derive(Debug, Serialize)]
pub struct EamModifyResult {
pub task: String,
pub changed: Vec<String>,
pub definition: Value,
pub put_outcome: Option<ModifyOutcome>,
pub readback: Value,
}
#[derive(Debug, Serialize)]
pub struct EamDeleteResult {
pub task: String,
pub deleted: bool,
pub changes: Vec<ResourceChange>,
pub affected: Vec<String>,
}
fn apply_task_change(body: &mut Value, change: &TaskChange) -> Vec<String> {
let mut changed = Vec::new();
let TaskChange {
enabled,
description,
schedule_mode,
settings_overlay,
} = change;
if let Some(enabled) = enabled {
*slot(body, "enabled") = Value::Bool(*enabled);
changed.push("enabled".to_string());
}
if let Some(description) = description {
*slot(body, "description") = Value::String(description.clone());
changed.push("description".to_string());
}
if let Some(schedule_mode) = schedule_mode {
let profile = slot(slot(body, "config"), "profile");
if !profile.is_object() {
*profile = Value::Object(Map::new());
}
*slot(profile, "scheduleMode") = Value::String(schedule_mode.clone());
changed.push("config.profile.scheduleMode".to_string());
}
if let Some(overlay) = settings_overlay {
let settings = slot(slot(body, "config"), "settings");
deep_merge(settings, overlay);
changed.push("config.settings".to_string());
}
changed
}
fn slot<'a>(parent: &'a mut Value, key: &str) -> &'a mut Value {
if !parent.is_object() {
*parent = Value::Object(Map::new());
}
parent
.as_object_mut()
.expect("just ensured an object")
.entry(key.to_string())
.or_insert(Value::Null)
}
fn record_to_value(record: &EamTaskRecord) -> Result<Value, CoreError> {
let mut value = serde_json::to_value(record).map_err(|err| {
CoreError::Internal(format!(
"task record failed to serialize for the clone: {err}"
))
})?;
if value.get("scheduledTaskState") == Some(&Value::Null)
&& let Some(map) = value.as_object_mut()
{
map.remove("scheduledTaskState");
}
Ok(value)
}
async fn reclassify_stale_signature(
api: &dyn GatewayApi,
name: &str,
sent_signature: &str,
err: CoreError,
) -> CoreError {
let stale = matches!(
api.eam_task_find(name).await,
Ok(fresh) if fresh.signature.as_deref().is_some_and(|sig| sig != sent_signature)
);
if stale {
return CoreError::InvalidInput {
reason: format!(
"definition {name:?} changed concurrently (signature mismatch on write) — \
the gateway answers mismatches with a 500 and leaves the resource untouched; \
re-run to apply against the current signature"
),
};
}
err
}
pub async fn eam_task_modify(
api: &dyn GatewayApi,
name: &str,
change: TaskChange,
) -> Result<EamModifyResult, CoreError> {
lifecycle_precheck("modify", name)?;
let TaskChange {
enabled,
description,
schedule_mode,
settings_overlay,
} = &change;
if enabled.is_none()
&& description.is_none()
&& schedule_mode.is_none()
&& settings_overlay.is_none()
{
return Err(CoreError::InvalidInput {
reason: format!(
"eam task modify {name:?}: no targeted keys — a modify must change \
something (enabled / description / schedule-mode / settings overlay)"
),
});
}
let record = api.eam_task_find(name).await?;
let signature = record
.signature
.clone()
.ok_or_else(|| CoreError::InvalidInput {
reason: format!(
"the found record for {name:?} carries no mutation signature — modify \
requires it (list-shape records don't carry one; re-find)"
),
})?;
let mut body = record_to_value(&record)?;
let changed = apply_task_change(&mut body, &change);
let definition = body.clone();
let put_outcome = match api.eam_task_modify(&body).await {
Ok(outcome) => outcome,
Err(err) => {
return Err(reclassify_stale_signature(api, name, &signature, err).await);
}
};
let readback = match api.eam_task_find(name).await {
Ok(fresh) => record_to_value(&fresh)?,
Err(_) => Value::Null,
};
Ok(EamModifyResult {
task: name.to_string(),
changed,
definition,
put_outcome,
readback,
})
}
fn affected_resources(outcome: &DeleteOutcome) -> Vec<String> {
let mut names: Vec<String> = outcome
.changes
.iter()
.map(|change| change.name.clone())
.collect();
if let Some(references) = &outcome.references {
for reference in references {
match reference {
Value::String(name) => names.push(name.clone()),
Value::Object(map) => {
if let Some(Value::String(name)) = map.get("name") {
names.push(name.clone());
}
}
_ => {}
}
}
}
let mut seen = BTreeSet::new();
names
.into_iter()
.filter(|name| seen.insert(name.clone()))
.collect()
}
pub async fn eam_task_delete(
api: &dyn GatewayApi,
name: &str,
) -> Result<EamDeleteResult, CoreError> {
lifecycle_precheck("delete", name)?;
let record = api.eam_task_find(name).await?;
let signature = record
.signature
.clone()
.ok_or_else(|| CoreError::InvalidInput {
reason: format!(
"the found record for {name:?} carries no mutation signature — delete \
is signature-keyed (list-shape records don't carry one; re-find)"
),
})?;
let outcome = match api.eam_task_delete(name, &signature, false).await {
Ok(outcome) if outcome.success => outcome,
Ok(demand) => {
let _ = demand;
match api.eam_task_delete(name, &signature, true).await {
Ok(retry) => retry,
Err(err) => {
return Err(reclassify_stale_signature(api, name, &signature, err).await);
}
}
}
Err(err) => {
return Err(reclassify_stale_signature(api, name, &signature, err).await);
}
};
Ok(EamDeleteResult {
task: name.to_string(),
deleted: outcome.success,
changes: outcome.changes.clone(),
affected: affected_resources(&outcome),
})
}
#[derive(Debug, Serialize)]
pub struct BlastRadiusPreview {
pub task: String,
pub task_type: Option<String>,
pub schedule_mode: Option<String>,
pub state: Option<String>,
pub owner: Option<String>,
pub config_suspended: Option<bool>,
pub target_gateways: Vec<String>,
pub pending_executions: Vec<EamScheduledTask>,
pub controller_impact: String,
pub verb: String,
}
fn agents_fragment(count: usize) -> String {
format!("{count} agent{}", if count == 1 { "" } else { "s" })
}
fn controller_impact(
verb: &str,
task: &str,
task_type: Option<&str>,
agents: usize,
pending: usize,
) -> String {
let type_note = task_type.map(|t| format!(" ({t})")).unwrap_or_default();
let agents = agents_fragment(agents);
match verb {
"suspend" => format!(
"suspends task {task}{type_note} — future scheduled dispatches to {agents} stop until resumed"
),
"resume" => format!(
"resumes task {task}{type_note} — scheduled dispatches to {agents} can fire again"
),
"cancel" if pending > 0 => format!(
"cancels the pending execution of task {task}{type_note} — {pending} queued dispatch{} to {agents}",
if pending == 1 { "" } else { "es" }
),
"cancel" => format!("task {task}{type_note} has no pending execution to cancel"),
"force" => format!("dispatches task {task}{type_note} now to {agents}"),
"modify" => format!(
"rewrites the definition of task {task}{type_note} — dispatch behavior to {agents} follows the new body"
),
"delete" => {
format!("deletes task {task}{type_note} permanently — dispatches to {agents} stop")
}
other => format!("examines task {task}{type_note} for {other} — targets {agents}"),
}
}
fn compose_blast_radius(
record: &EamTaskRecord,
pending_executions: Vec<EamScheduledTask>,
verb: &str,
) -> BlastRadiusPreview {
let pending_executions: Vec<EamScheduledTask> = pending_executions
.into_iter()
.filter(|row| row.name == record.name)
.collect();
let profile = record.config.get("profile");
let task_type = profile
.and_then(|p| p.get("type"))
.and_then(Value::as_str)
.map(str::to_string);
let schedule_mode = profile
.and_then(|p| p.get("scheduleMode"))
.and_then(Value::as_str)
.map(str::to_string);
let config_suspended = profile
.and_then(|p| p.get("isSuspended"))
.and_then(Value::as_bool);
let state = current_state_of(record);
let owner = record
.scheduled_task_state
.as_ref()
.and_then(|s| s.get("details"))
.and_then(|d| d.get("owner"))
.and_then(Value::as_str)
.map(str::to_string);
let target_gateways: Vec<String> = record
.config
.get("settings")
.and_then(|settings| settings.get("targetGateways"))
.and_then(Value::as_array)
.map(|gateways| {
gateways
.iter()
.filter_map(Value::as_str)
.map(str::to_string)
.collect()
})
.unwrap_or_default();
let controller_impact = controller_impact(
verb,
&record.name,
task_type.as_deref(),
target_gateways.len(),
pending_executions.len(),
);
BlastRadiusPreview {
task: record.name.clone(),
task_type,
schedule_mode,
state,
owner,
config_suspended,
target_gateways,
pending_executions,
controller_impact,
verb: verb.to_string(),
}
}
async fn pending_rows_for(
api: &dyn GatewayApi,
name: &str,
) -> Result<Vec<EamScheduledTask>, CoreError> {
let mut rows = Vec::new();
for running in [false, true] {
rows.extend(
api.eam_tasks_scheduled(running)
.await?
.into_iter()
.filter(|row| row.name == name),
);
}
Ok(rows)
}
pub async fn build_blast_radius(
api: &dyn GatewayApi,
verb: &str,
task_name: &str,
) -> Result<BlastRadiusPreview, CoreError> {
let record = api.eam_task_find(task_name).await?;
let pending_executions = pending_rows_for(api, task_name).await?;
Ok(compose_blast_radius(&record, pending_executions, verb))
}
pub fn render_preview_line(preview: &BlastRadiusPreview) -> String {
format!(
"{verb} {task}: {impact} targets: [{targets}] pending: {pending}",
verb = preview.verb,
task = preview.task,
impact = preview.controller_impact,
targets = preview.target_gateways.join(", "),
pending = preview.pending_executions.len(),
)
}
#[derive(Debug, Serialize)]
pub struct EamHistoryResult {
pub items: Vec<EamHistoryItem>,
pub count: usize,
}
#[derive(Debug, Serialize)]
pub struct EamTaskSummary {
pub name: String,
pub task_type: Option<String>,
pub schedule_mode: Option<String>,
pub current_state: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct EamTasksResult {
pub tasks: Vec<EamTaskSummary>,
}
#[derive(Debug, Serialize)]
pub struct EamTaskDetailResult {
pub name: String,
pub definition: serde_json::Value,
pub state: serde_json::Value,
}
pub async fn eam_history(
api: &dyn GatewayApi,
limit: Option<u32>,
search: Option<&str>,
) -> Result<EamHistoryResult, CoreError> {
let page = api.eam_task_history(limit, search).await?;
Ok(EamHistoryResult {
count: page.items.len(),
items: page.items,
})
}
pub async fn eam_tasks(api: &dyn GatewayApi) -> Result<EamTasksResult, CoreError> {
let page = api.eam_task_definitions().await?;
Ok(EamTasksResult {
tasks: page.items.iter().map(summary_from).collect(),
})
}
pub async fn eam_task_detail(
api: &dyn GatewayApi,
name: &str,
) -> Result<EamTaskDetailResult, CoreError> {
let record = api.eam_task_find(name).await?;
Ok(EamTaskDetailResult {
name: record.name.clone(),
definition: serde_json::to_value(&record).unwrap_or(serde_json::Value::Null),
state: record
.scheduled_task_state
.clone()
.unwrap_or(serde_json::Value::Null),
})
}
fn summary_from(record: &EamTaskRecord) -> EamTaskSummary {
let profile = record.config.get("profile");
EamTaskSummary {
name: record.name.clone(),
task_type: profile
.and_then(|p| p.get("type"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
schedule_mode: profile
.and_then(|p| p.get("scheduleMode"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
current_state: record
.scheduled_task_state
.as_ref()
.and_then(|state| state.get("currentState"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
}
}
#[cfg(test)]
mod tests {
use super::{
CancelDecision, EamLifecycleResult, EamTaskRecord, TaskChange, TaskCreateVerdict,
affected_resources, apply_task_change, auto_type, cancel_decision, compose_blast_radius,
compose_task_definition, deep_merge, lifecycle_precheck, parse_setting,
render_preview_line, summary_from, suspend_recheck, task_create_guard,
};
use crate::client::eam::{DeleteOutcome, EamScheduledTask};
#[test]
fn guard_ladder_is_exhaustive_over_the_taxonomy() {
assert_eq!(
task_create_guard("eam_backup", "OnDemand"),
TaskCreateVerdict::Unguarded
);
for refused in [
"eam_restoreBackup",
"eam_installModules",
"eam_remoteUpgrade",
] {
assert_eq!(
task_create_guard(refused, "OnDemand"),
TaskCreateVerdict::Refused,
"{refused} refuses even OnDemand"
);
assert_eq!(
task_create_guard(refused, "Scheduled"),
TaskCreateVerdict::Refused,
"{refused} refuses under any schedule"
);
}
for mutating in [
"eam_restart",
"eam_sendProject",
"eam_sendResource",
"eam_sendTags",
"eam_activateLicense",
"eam_updateLicense",
"eam_unactivateLicense",
] {
assert_eq!(
task_create_guard(mutating, "OnDemand"),
TaskCreateVerdict::NeedsYes,
"{mutating} needs --yes"
);
}
for mode in ["Immediate", "Scheduled", "AtTime", "AtDelay", "weird-mode"] {
assert_eq!(
task_create_guard("eam_backup", mode),
TaskCreateVerdict::NeedsYes,
"scheduleMode {mode} arms the task"
);
}
assert_eq!(
task_create_guard("eam_unknownFutureType", "OnDemand"),
TaskCreateVerdict::NeedsYes
);
}
#[test]
fn setting_parsing_auto_types_scalars() {
assert_eq!(
parse_setting("concurrentBackups=2").unwrap(),
("concurrentBackups".to_string(), serde_json::json!(2))
);
assert_eq!(
parse_setting("forceBackups=true").unwrap(),
("forceBackups".to_string(), serde_json::json!(true))
);
assert_eq!(
parse_setting("forceBackups=false").unwrap(),
("forceBackups".to_string(), serde_json::json!(false))
);
assert_eq!(
parse_setting("n=-7").unwrap(),
("n".to_string(), serde_json::json!(-7))
);
assert_eq!(
parse_setting("note=hello world").unwrap(),
("note".to_string(), serde_json::json!("hello world"))
);
assert_eq!(
parse_setting("v=1.5").unwrap(),
("v".to_string(), serde_json::json!("1.5")),
"floats are NOT auto-typed (the tags-write rule: bool/int only)"
);
let err = parse_setting("noequalsign").expect_err("refuses");
assert_eq!(err.exit_code(), 2);
assert_eq!(err.code(), "invalid_input");
assert!(err.to_string().contains("--definition"));
assert!(parse_setting("=v").is_err(), "empty key refuses");
assert!(parse_setting("k=").is_err(), "empty value refuses");
}
#[test]
fn auto_type_covers_bool_int_string() {
assert_eq!(auto_type("true"), serde_json::json!(true));
assert_eq!(auto_type("false"), serde_json::json!(false));
assert_eq!(auto_type("42"), serde_json::json!(42));
assert_eq!(auto_type("text"), serde_json::json!("text"));
assert_eq!(auto_type("True"), serde_json::json!("True"), "case matters");
}
#[test]
fn deep_merge_merges_objects_replaces_arrays() {
let mut base = serde_json::json!({
"type": "eam_backup",
"scheduleMode": "OnDemand",
"targetGateways": ["gw-a"],
"settingsNested": {"a": 1, "b": {"x": 1}}
});
deep_merge(
&mut base,
&serde_json::json!({
"targetGateways": ["gw-b", "gw-c"],
"targetGroups": [],
"concurrentBackups": 2,
"forceBackups": true,
"settingsNested": {"b": {"y": 2}}
}),
);
assert_eq!(
base,
serde_json::json!({
"type": "eam_backup",
"scheduleMode": "OnDemand",
"targetGateways": ["gw-b", "gw-c"],
"targetGroups": [],
"concurrentBackups": 2,
"forceBackups": true,
"settingsNested": {"a": 1, "b": {"x": 1, "y": 2}}
})
);
}
#[test]
fn composition_splits_profile_and_settings_the_live_shape() {
let bare =
compose_task_definition("uat-backup-demo", "eam_backup", &[], &[], None, "OnDemand")
.expect("bare composition");
assert_eq!(bare["name"], serde_json::json!("uat-backup-demo"));
assert_eq!(
bare["config"]["profile"],
serde_json::json!({"type": "eam_backup", "scheduleMode": "OnDemand"}),
"profile carries type + scheduleMode ONLY (isSuspended is server-owned)"
);
assert_eq!(
bare["config"]["settings"],
serde_json::json!({"targetGateways": ["_controller"], "targetGroups": []})
);
let targeted = compose_task_definition(
"nightly-backup",
"eam_backup",
&["gw-a".to_string()],
&[
"concurrentBackups=2".to_string(),
"forceBackups=true".to_string(),
],
None,
"OnDemand",
)
.expect("targeted composition");
assert_eq!(
targeted["config"]["settings"]["targetGateways"],
serde_json::json!(["gw-a"])
);
assert_eq!(
targeted["config"]["settings"]["concurrentBackups"],
serde_json::json!(2),
"K=V lands in config.SETTINGS"
);
assert!(
targeted["config"]["profile"]
.get("concurrentBackups")
.is_none(),
"profile carries NO settings keys"
);
assert!(
targeted["config"]["profile"]
.get("targetGateways")
.is_none(),
"targetGateways lives in settings, not profile"
);
let overlayed = compose_task_definition(
"t3",
"eam_backup",
&["gw-a".to_string()],
&[],
Some(&serde_json::json!({
"targetGateways": ["gw-b", "gw-c"],
"concurrentBackups": 5
})),
"OnDemand",
)
.expect("overlay composition");
assert_eq!(
overlayed["config"]["settings"]["targetGateways"],
serde_json::json!(["gw-b", "gw-c"]),
"the overlay's array REPLACES the composed default"
);
assert_eq!(overlayed["config"]["settings"]["concurrentBackups"], 5);
assert_eq!(
overlayed["config"]["settings"]["targetGroups"],
serde_json::json!([]),
"composed keys the overlay omits survive the merge"
);
}
#[test]
fn summary_projects_the_agent_stable_keys() {
let record: EamTaskRecord = serde_json::from_value(serde_json::json!({
"name": "nightly-backup",
"config": {"profile": {"type": "eam_backup", "scheduleMode": "OnDemand"}},
"scheduledTaskState": {"currentState": "IDLE", "details": {"owner": "eam"}}
}))
.expect("record parses");
let summary = summary_from(&record);
assert_eq!(summary.name, "nightly-backup");
assert_eq!(summary.task_type.as_deref(), Some("eam_backup"));
assert_eq!(summary.schedule_mode.as_deref(), Some("OnDemand"));
assert_eq!(summary.current_state.as_deref(), Some("IDLE"));
let bare: EamTaskRecord = serde_json::from_value(serde_json::json!({
"name": "bare"
}))
.expect("bare record parses");
let summary = summary_from(&bare);
assert_eq!(summary.task_type, None);
assert_eq!(summary.schedule_mode, None);
assert_eq!(summary.current_state, None);
}
#[test]
fn lifecycle_result_serializes_all_keys_always() {
let fired = EamLifecycleResult {
task: "nightly-backup".to_string(),
action: "suspended".to_string(),
previous_state: Some("Scheduled".to_string()),
config_suspended: Some(true),
pending: None,
fired: true,
reason: None,
};
let json = serde_json::to_value(&fired).expect("serializes");
let map = json.as_object().expect("object shape");
for key in [
"task",
"action",
"previous_state",
"config_suspended",
"pending",
"fired",
"reason",
] {
assert!(map.contains_key(key), "key {key} always rides");
}
assert_eq!(json["pending"], serde_json::Value::Null);
assert_eq!(json["reason"], serde_json::Value::Null);
assert_eq!(json["config_suspended"], serde_json::json!(true));
let noop = EamLifecycleResult {
task: "t".to_string(),
action: "cancelled".to_string(),
previous_state: None,
config_suspended: None,
pending: None,
fired: false,
reason: Some("no pending execution".to_string()),
};
let json = serde_json::to_value(&noop).expect("serializes");
assert_eq!(json["fired"], serde_json::json!(false));
assert_eq!(json["reason"], serde_json::json!("no pending execution"));
assert_eq!(
json["previous_state"],
serde_json::Value::Null,
"no state on a find that carried none — null, not omitted"
);
}
#[test]
fn lifecycle_precheck_refuses_empty_and_whitespace_names() {
for name in ["", " ", "\t\n"] {
for action in ["suspend", "resume", "cancel"] {
let err =
lifecycle_precheck(action, name).expect_err("empty/whitespace names refuse");
assert_eq!(err.exit_code(), 2, "usage class");
assert_eq!(err.code(), "invalid_input");
let message = err.to_string();
assert!(
message.contains(action),
"the refusal names the verb: {message}"
);
}
}
lifecycle_precheck("suspend", "nightly-backup").expect("real names pass");
lifecycle_precheck("cancel", " x ")
.expect("trimmed-nonempty passes (the gateway owns identifier rules)");
}
#[test]
fn suspend_recheck_refuses_only_already_suspended() {
let record_of = |is_suspended: serde_json::Value| -> EamTaskRecord {
serde_json::from_value(serde_json::json!({
"name": "nightly-backup",
"config": {"profile": {"isSuspended": is_suspended}}
}))
.expect("record parses")
};
let err = suspend_recheck(&record_of(serde_json::json!(true)))
.expect_err("already-suspended refuses pre-write");
assert_eq!(err.exit_code(), 2);
assert_eq!(err.code(), "invalid_input");
let message = err.to_string();
assert!(
message.contains("nightly-backup") && message.contains("already suspended"),
"the refusal names the task + state: {message}"
);
suspend_recheck(&record_of(serde_json::json!(false)))
.expect("false fires (the normal case)");
suspend_recheck(&record_of(serde_json::Value::Null))
.expect("absent flag fires (no invented rules)");
suspend_recheck(&record_of(serde_json::json!("weird")))
.expect("unparseable flag fires (the gateway's 500 is the honest answer)");
}
#[test]
fn cancel_decision_branches_mirror_the_captures() {
assert_eq!(
cancel_decision(None),
CancelDecision::NoPending,
"nothing pending → the honest no-op, no doomed POST"
);
let row_of = |can_cancel: bool| -> EamScheduledTask {
serde_json::from_value(serde_json::json!({
"name": "nightly-backup",
"owner": "eam",
"type": "Collect Backup",
"execStart": null,
"message": "",
"repeats": true,
"canPause": true,
"canResume": false,
"canCancel": can_cancel,
"taskState": "Scheduled",
"isForced": false,
"isRunning": false,
"progress": 0.0
}))
.expect("the captured row shape parses")
};
assert_eq!(
cancel_decision(Some(&row_of(true))),
CancelDecision::Fire,
"the captured Scheduled cell (canCancel: true) fires"
);
assert_eq!(
cancel_decision(Some(&row_of(false))),
CancelDecision::NotPermitted,
"the gateway's own canCancel=false is reported, not overridden"
);
}
fn full_record_fixture() -> serde_json::Value {
serde_json::json!({
"type": "com.inductiveautomation.eam/eam-tasks",
"name": "ign-p10-scratch-sched",
"description": "scratch",
"enabled": true,
"version": 1,
"collection": "core",
"collections": ["core"],
"signature": "e5ac8bee3a6ba85e40923c0e02d29507600c57519eb8e4d78bd8c258197fe9c6",
"config": {
"profile": {
"type": "eam_backup",
"isSuspended": false,
"scheduleMode": "Scheduled",
"scheduleDetails": "0/30 * * * * ?"
},
"settings": {
"targetGateways": ["_controller"],
"targetGroups": [],
"concurrentBackups": 0,
"forceBackups": false
}
},
"data": ["config.json"],
"attributes": {"uuid": "c1aa2b52-46ad-46ea-962b-9d2498f35db1", "enabled": true},
"metrics": {},
"healthchecks": {"scheduledTaskState": {"currentState": "Scheduled"}}
})
}
#[test]
fn modify_put_body_preserves_the_fixture_record_except_targeted_keys() {
let mut body = full_record_fixture();
let changed = apply_task_change(
&mut body,
&TaskChange {
description: Some("rewritten note".to_string()),
..Default::default()
},
);
assert_eq!(changed, vec!["description"]);
let fixture = full_record_fixture();
assert_eq!(
body["config"]["settings"], fixture["config"]["settings"],
"config.settings rides VERBATIM — omitting/reshaping it is the 422 trap"
);
assert_eq!(
body["signature"], fixture["signature"],
"the ORIGINAL signature"
);
assert_eq!(body["collection"], fixture["collection"]);
assert_eq!(body["config"]["profile"], fixture["config"]["profile"]);
assert_eq!(
body["data"], fixture["data"],
"unknown round-trip keys survive"
);
assert_eq!(body["attributes"], fixture["attributes"]);
assert_eq!(body["description"], serde_json::json!("rewritten note"));
let mut body = full_record_fixture();
let changed = apply_task_change(
&mut body,
&TaskChange {
enabled: Some(false),
settings_overlay: Some(serde_json::json!({"concurrentBackups": 4})),
..Default::default()
},
);
assert_eq!(changed, vec!["enabled", "config.settings"]);
assert_eq!(body["enabled"], serde_json::json!(false));
let mut expected_settings = fixture["config"]["settings"].clone();
expected_settings["concurrentBackups"] = serde_json::json!(4);
assert_eq!(body["config"]["settings"], expected_settings);
assert_eq!(body["signature"], fixture["signature"]);
let mut body = full_record_fixture();
let changed = apply_task_change(
&mut body,
&TaskChange {
schedule_mode: Some("OnDemand".to_string()),
..Default::default()
},
);
assert_eq!(changed, vec!["config.profile.scheduleMode"]);
assert_eq!(
body["config"]["profile"]["scheduleMode"],
serde_json::json!("OnDemand")
);
assert_eq!(
body["config"]["profile"]["scheduleDetails"],
fixture["config"]["profile"]["scheduleDetails"],
"scheduleDetails rides the clone (this change's scope is the mode key)"
);
assert_eq!(body["config"]["settings"], fixture["config"]["settings"]);
}
#[test]
fn modify_result_serializes_all_keys_always() {
let result = super::EamModifyResult {
task: "t".to_string(),
changed: vec!["enabled".to_string()],
definition: serde_json::json!({"name": "t"}),
put_outcome: None,
readback: serde_json::Value::Null,
};
let json = serde_json::to_value(&result).expect("serializes");
for key in ["task", "changed", "definition", "put_outcome", "readback"] {
assert!(
json.as_object().unwrap().contains_key(key),
"{key} always rides"
);
}
assert_eq!(json["put_outcome"], serde_json::Value::Null);
}
#[test]
fn delete_result_extracts_affected_names_from_changes_and_references() {
let success: DeleteOutcome = serde_json::from_value(serde_json::json!({
"success": true,
"changes": [
{
"name": "ign-p10-scratch-sched",
"type": "com.inductiveautomation.eam/eam-tasks",
"collection": "core",
"newSignature": "ec961ee921c63b18013094870ed2664331e965c4770fdf84bfe136e0b4164244"
}
],
"problem": null,
"references": []
}))
.expect("the captured success body parses");
assert_eq!(
affected_resources(&success),
vec!["ign-p10-scratch-sched".to_string()],
"a lone-resource delete touches exactly the deleted resource"
);
let demanded: DeleteOutcome = serde_json::from_value(serde_json::json!({
"success": false,
"changes": [
{"name": "task-a", "type": "com.inductiveautomation.eam/eam-tasks", "collection": "core", "newSignature": "x"}
],
"problem": null,
"references": ["agent-b", {"name": "task-c"}, {"shapeless": true}, 42, "task-a"]
}))
.expect("the lenient shape parses");
assert_eq!(
affected_resources(&demanded),
vec![
"task-a".to_string(),
"agent-b".to_string(),
"task-c".to_string()
],
"strings + name-keyed objects ride; shapeless elements skipped; dedup holds"
);
}
fn scheduled_row(name: &str) -> EamScheduledTask {
serde_json::from_value(serde_json::json!({
"name": name,
"owner": "eam",
"type": "Collect Backup",
"execStart": null,
"message": "",
"repeats": true,
"canPause": true,
"canResume": false,
"canCancel": true,
"taskState": "Scheduled",
"isForced": false,
"isRunning": false,
"progress": 0.0
}))
.expect("the captured row shape parses")
}
#[test]
fn preview_composes_over_fixture_find_and_scheduled() {
let record: EamTaskRecord = serde_json::from_value(serde_json::json!({
"name": "ign-p10-scratch-sched",
"config": {
"profile": {"type": "eam_backup", "isSuspended": false, "scheduleMode": "Scheduled"},
"settings": {"targetGateways": ["gw-a", "gw-b"], "targetGroups": []}
},
"signature": "sig",
"scheduledTaskState": {"currentState": "Scheduled", "details": {"owner": "eam"}}
}))
.expect("fixture record parses");
let pending = vec![
scheduled_row("ign-p10-scratch-sched"),
scheduled_row("some-other-task"),
];
let preview = compose_blast_radius(&record, pending, "suspend");
assert_eq!(preview.task, "ign-p10-scratch-sched");
assert_eq!(preview.task_type.as_deref(), Some("eam_backup"));
assert_eq!(preview.schedule_mode.as_deref(), Some("Scheduled"));
assert_eq!(preview.state.as_deref(), Some("Scheduled"));
assert_eq!(preview.owner.as_deref(), Some("eam"));
assert_eq!(preview.config_suspended, Some(false));
assert_eq!(
preview.target_gateways,
vec!["gw-a".to_string(), "gw-b".to_string()],
"the AGENTS the write touches"
);
assert_eq!(
preview.pending_executions.len(),
1,
"rows for other tasks are filtered out"
);
assert!(preview.pending_executions[0].can_cancel);
assert_eq!(preview.verb, "suspend");
let impact = &preview.controller_impact;
assert!(
impact.contains("ign-p10-scratch-sched")
&& impact.contains("eam_backup")
&& impact.contains("2 agents")
&& impact.contains("stop until resumed"),
"the impact names task + type + agent count + consequence: {impact}"
);
let json = serde_json::to_value(&preview).expect("serializes");
for key in [
"task",
"task_type",
"schedule_mode",
"state",
"owner",
"config_suspended",
"target_gateways",
"pending_executions",
"controller_impact",
"verb",
] {
assert!(
json.as_object().unwrap().contains_key(key),
"{key} always rides"
);
}
}
#[test]
fn preview_tolerates_empty_targets_and_pending() {
let record: EamTaskRecord = serde_json::from_value(serde_json::json!({
"name": "bare",
"config": {}
}))
.expect("bare record parses");
let preview = compose_blast_radius(&record, Vec::new(), "cancel");
assert_eq!(preview.target_gateways, Vec::<String>::new());
assert_eq!(preview.pending_executions, Vec::<EamScheduledTask>::new());
assert_eq!(preview.task_type, None);
assert_eq!(preview.owner, None);
assert!(
preview.controller_impact.contains("bare")
&& preview.controller_impact.contains("no pending execution"),
"the cancel impact names the empty case factually: {}",
preview.controller_impact
);
}
#[test]
fn preview_impacts_differ_per_verb() {
let record: EamTaskRecord = serde_json::from_value(serde_json::json!({
"name": "nightly-backup",
"config": {
"profile": {"type": "eam_backup", "scheduleMode": "Scheduled"},
"settings": {"targetGateways": ["gw-a"]}
},
"scheduledTaskState": {"currentState": "Scheduled", "details": {"owner": "eam"}}
}))
.expect("fixture record parses");
let pending = vec![scheduled_row("nightly-backup")];
let mut impacts = Vec::new();
for verb in ["suspend", "resume", "cancel", "force", "modify", "delete"] {
let preview = compose_blast_radius(&record, pending.clone(), verb);
let impact = preview.controller_impact.clone();
assert!(
impact.contains("nightly-backup") && impact.contains("1 agent"),
"{verb}'s impact names the task + agent count: {impact}"
);
impacts.push(impact);
}
let distinct: std::collections::BTreeSet<&String> = impacts.iter().collect();
assert_eq!(
distinct.len(),
impacts.len(),
"each verb's factual sentence is distinct: {impacts:?}"
);
let cancel_with = compose_blast_radius(&record, pending.clone(), "cancel");
assert!(cancel_with.controller_impact.contains("1 queued dispatch"));
let cancel_without = compose_blast_radius(&record, Vec::new(), "cancel");
assert!(
cancel_without
.controller_impact
.contains("no pending execution")
);
let odd = compose_blast_radius(&record, Vec::new(), "teleport");
assert!(odd.controller_impact.contains("teleport"));
}
#[test]
fn render_preview_line_contains_task_agents_verb_and_pending() {
let record: EamTaskRecord = serde_json::from_value(serde_json::json!({
"name": "ign-p10-scratch",
"config": {
"profile": {"type": "eam_backup", "scheduleMode": "OnDemand"},
"settings": {"targetGateways": ["_controller"]}
}
}))
.expect("fixture record parses");
let preview = compose_blast_radius(
&record,
vec![
scheduled_row("ign-p10-scratch"),
scheduled_row("ign-p10-scratch"),
],
"delete",
);
assert_eq!(
preview.pending_executions.len(),
2,
"same-name rows from both segments both count"
);
let line = render_preview_line(&preview);
assert_eq!(
line,
"delete ign-p10-scratch: deletes task ign-p10-scratch (eam_backup) permanently \
— dispatches to 1 agent stop targets: [_controller] pending: 2"
);
assert!(line.contains("ign-p10-scratch"));
assert!(line.contains("1 agent"));
let bare = compose_blast_radius(
&serde_json::from_value::<EamTaskRecord>(serde_json::json!({
"name": "bare", "config": {}
}))
.expect("bare parses"),
Vec::new(),
"resume",
);
assert_eq!(
render_preview_line(&bare),
"resume bare: resumes task bare — scheduled dispatches to 0 agents can fire again targets: [] pending: 0"
);
}
}