Menu
Public repository / main
AkurAI-EC2 / crates/platform-mcp/src/tools.rs
Latest change 61d0777e9306a5b3b94a6987c70a8be1e718cd11 - feat: add honest workspace lifecycle boundary by Ólafur Búi Ólafsson
use akurai_json::Value;
use crate::server::run_command;
pub struct ToolRegistry {
monitor_url: String,
akurai_ec2_bin: String,
apps_toml: String,
}
impl ToolRegistry {
pub fn new(monitor_url: String, akurai_ec2_bin: String, apps_toml: String) -> Self {
Self {
monitor_url,
akurai_ec2_bin,
apps_toml,
}
}
// ── JSON-RPC dispatch ──────────────────────────────────────────────
pub fn handle_request(&self, body: &str) -> String {
let request = match akurai_json::parse(body) {
Ok(v) => v,
Err(e) => {
return jsonrpc_error(null_val(), -32700, &format!("Parse error: {e}"));
}
};
let id = request.get("id").cloned().unwrap_or(null_val());
let method = match request.get("method").and_then(|m| m.as_str()) {
Some(m) => m.to_string(),
None => return jsonrpc_error(id, -32600, "Invalid Request: missing method"),
};
match method.as_str() {
"tools/list" => self.handle_list_tools(id),
"tools/call" => self.handle_call_tool(id, &request),
_ => jsonrpc_error(id, -32601, &format!("Method not found: {method}")),
}
}
fn handle_list_tools(&self, id: Value) -> String {
let tools = self.list_tools();
jsonrpc_result(
id,
&Value::Object(vec![("tools".into(), Value::Array(tools))]),
)
}
fn handle_call_tool(&self, id: Value, request: &Value) -> String {
let params = match request.get("params") {
Some(p) => p,
None => return jsonrpc_error(id, -32602, "Invalid params: missing"),
};
let tool_name = match params.get("name").and_then(|n| n.as_str()) {
Some(n) => n,
None => return jsonrpc_error(id, -32602, "Invalid params: missing name"),
};
let arguments = params
.get("arguments")
.cloned()
.unwrap_or(Value::Object(Vec::new()));
match self.call_tool(tool_name, &arguments) {
Ok(result) => {
let content = vec![Value::Object(vec![
("type".into(), Value::Str("text".into())),
("text".into(), Value::Str(result)),
])];
let tool_result = Value::Object(vec![
("content".into(), Value::Array(content)),
("isError".into(), Value::Bool(false)),
]);
jsonrpc_result(id, &tool_result)
}
Err(e) => {
let content = vec![Value::Object(vec![
("type".into(), Value::Str("text".into())),
("text".into(), Value::Str(e.clone())),
])];
let tool_result = Value::Object(vec![
("content".into(), Value::Array(content)),
("isError".into(), Value::Bool(true)),
]);
jsonrpc_result(id, &tool_result)
}
}
}
// ── Tool definitions ───────────────────────────────────────────────
pub fn list_tools(&self) -> Vec<Value> {
vec![
// ── EC2 Fleet ──────────────────────────────────────────
self.tool_def(
"ec2_fleet_status",
"Health rollup across all apps in AKURAI_APPS.toml — reads the fleet manifest and checks each app's health endpoint",
vec![
("app", "string", "Filter to a specific app by name (optional)", false),
],
),
// ── EC2 Deploy ─────────────────────────────────────────
self.tool_def(
"ec2_deploy",
"Deploy/release an AkurAI app. Runs akurai-ec2 release -C <checkout> --yes publish. Before deploying: runs bunfork-backup as a safety net. After deploying: validates the app. On failure: auto-rollback.",
vec![
("name", "string", "App name (must match AKURAI_APPS.toml entry)", true),
("bump", "string", "Bump level for release mode: patch, minor, or major (optional; defaults to publish-only)", false),
("dry_run", "boolean", "Preview only, ship nothing", false),
],
),
self.tool_def(
"ec2_deploy_history",
"Show recent deploy history for an app — reads git tags and systemd unit status on EC2",
vec![
("name", "string", "App name (must match AKURAI_APPS.toml entry)", true),
],
),
// ── DNS / TLS ──────────────────────────────────────────
self.tool_def(
"ec2_dns_check",
"List DNS records for a domain managed by akurai-ec2 on the authoritative EC2 zone",
vec![
("domain", "string", "Domain or FQDN to list records for (required by akurai-ec2 dns list)", true),
("zone", "string", "Zone to query (optional; inferred from domain if omitted)", false),
("record_type", "string", "Filter by record type, e.g. A, CNAME (optional)", false),
],
),
self.tool_def(
"ec2_tls_expiry",
"Check TLS certificate expiry for a domain",
vec![
("domain", "string", "Domain to check (e.g. platform.olibuijr.com)", true),
],
),
// ── Systemd ────────────────────────────────────────────
self.tool_def(
"ec2_systemd_status",
"Show systemd service status for AkurAI fleet services — unit name, state, PID, memory, uptime, and ports",
vec![
("service", "string", "Filter to a specific service (optional; shows all fleet services if omitted)", false),
],
),
self.tool_def(
"ec2_systemd_restart",
"Restart a systemd service on EC2. Before restarting: runs bunfork-backup as a safety net. After restarting: validates the app. On failure: auto-rollback to the previous binary.",
vec![
("service", "string", "Systemd service name (e.g. akurai-platform.service)", true),
],
),
// ── Backup ─────────────────────────────────────────────
self.tool_def(
"ec2_backup_run",
"Run an encrypted backup now. Supports bunfork (encrypted DB + verify restore), tasks (GPG-encrypted AkurAI-Tasks archive + manifest/hash restore), and workbench (encrypted OMP Workbench snapshot shipped to EC2). Retention defaults are bunfork/workbench=14 and tasks=48.",
vec![
("backup_type", "string", "Backup type: 'bunfork', 'tasks', or 'workbench'", true),
("retention", "number", "Number of backups to keep (2-365 for bunfork/workbench; 2-720 for tasks)", false),
],
),
// ── Monitor (rust-monitor proxy) ───────────────────────
self.tool_def(
"monitor_hosts",
"List monitored hosts from rust-monitor",
vec![],
),
self.tool_def(
"monitor_applications",
"List monitored applications from rust-monitor",
vec![
("host_id", "number", "Filter to a specific host (optional)", false),
],
),
self.tool_def(
"monitor_alerts",
"List alert rules and recent alert events from rust-monitor",
vec![],
),
self.tool_def(
"monitor_events",
"List recent monitor events/logs from rust-monitor",
vec![
("limit", "number", "Max events to return (default 50)", false),
],
),
self.tool_def(
"monitor_ingestion_health",
"Inspect journald and file-tail ingestion health, errors, and stale sources",
vec![],
),
self.tool_def(
"monitor_logs",
"Read bounded, cursor-paginated recent logs from rust-monitor",
vec![
("hours", "number", "History window in hours (1-720, default 1)", false),
("limit", "number", "Maximum rows (1-1000, default 100)", false),
("source", "string", "Exact log source filter", false),
("before_id", "number", "Pagination cursor from next_before_id", false),
],
),
]
}
fn tool_def(
&self,
name: &str,
description: &str,
params: Vec<(&str, &str, &str, bool)>,
) -> Value {
let mut properties = Vec::new();
let mut required = Vec::new();
for (pname, ptype, pdesc, is_required) in params {
properties.push((
pname.to_string(),
Value::Object(vec![
("type".into(), Value::Str(ptype.into())),
("description".into(), Value::Str(pdesc.into())),
]),
));
if is_required {
required.push(Value::Str(pname.to_string()));
}
}
Value::Object(vec![
("name".into(), Value::Str(name.into())),
("description".into(), Value::Str(description.into())),
(
"inputSchema".into(),
Value::Object(vec![
("type".into(), Value::Str("object".into())),
("properties".into(), Value::Object(properties)),
("required".into(), Value::Array(required)),
]),
),
])
}
// ── Tool dispatch ─────────────────────────────────────────────────
pub fn call_tool(&self, name: &str, args: &Value) -> Result<String, String> {
match name {
"ec2_fleet_status" => self.ec2_fleet_status(args),
"ec2_deploy" => self.ec2_deploy(args),
"ec2_deploy_history" => self.ec2_deploy_history(args),
"ec2_dns_check" => self.ec2_dns_check(args),
"ec2_tls_expiry" => self.ec2_tls_expiry(args),
"ec2_systemd_status" => self.ec2_systemd_status(args),
"ec2_systemd_restart" => self.ec2_systemd_restart(args),
"ec2_backup_run" => self.ec2_backup_run(args),
"monitor_hosts" => self.monitor_hosts(args),
"monitor_applications" => self.monitor_applications(args),
"monitor_alerts" => self.monitor_alerts(args),
"monitor_events" => self.monitor_events(args),
"monitor_ingestion_health" => self.monitor_ingestion_health(args),
"monitor_logs" => self.monitor_logs(args),
_ => Err(format!("Unknown tool: {name}")),
}
}
// ── Argument helpers ───────────────────────────────────────────────
fn arg_str<'a>(&self, args: &'a Value, key: &str) -> Option<&'a str> {
args.get(key).and_then(|v| v.as_str())
}
fn arg_i64(&self, args: &Value, key: &str) -> Option<i64> {
args.get(key).and_then(|v| v.as_i64())
}
fn arg_bool(&self, args: &Value, key: &str) -> Option<bool> {
args.get(key).and_then(|v| v.as_bool())
}
fn require_str<'a>(&self, args: &'a Value, key: &str) -> Result<&'a str, String> {
self.arg_str(args, key)
.ok_or_else(|| format!("Missing required argument: {key}"))
}
// ── Tool implementations ───────────────────────────────────────────
/// Read-only: fleet health rollup from AKURAI_APPS.toml
fn ec2_fleet_status(&self, args: &Value) -> Result<String, String> {
let filter = self.arg_str(args, "app");
let (stdout, stderr, code) = run_command(&self.akurai_ec2_bin, &["health"])?;
if code != 0 {
return Err(format!("ec2_fleet_status failed (exit {code}): {stderr}"));
}
let result = if let Some(app) = filter {
// Filter output lines containing the app name
let filtered: Vec<&str> = stdout.lines().filter(|l| l.contains(app)).collect();
if filtered.is_empty() {
format!("No results for app '{app}'")
} else {
filtered.join("\n")
}
} else {
stdout
};
Ok(result)
}
/// MUTATING: deploy an app with backup-before + validate-after + rollback-on-failure
fn ec2_deploy(&self, args: &Value) -> Result<String, String> {
let name = self.require_str(args, "name")?;
let bump = self.arg_str(args, "bump");
let dry_run = self.arg_bool(args, "dry_run").unwrap_or(false);
// Resolve the checkout directory from AKURAI_APPS.toml
let checkout = self.resolve_app_checkout(name)?;
// Pre-mutation safety backup
if !dry_run {
self.run_pre_mutation_backup()?;
}
self.execute_deploy(name, &checkout, bump, dry_run)
}
/// Build the single canonical command shape used by both publish and
/// release-with-bump deployments. A bump release intentionally omits the
/// legacy `publish` positional action; publish retains it after
/// `--dry-run`, matching akurai-ec2's CLI contract.
fn build_deploy_args(checkout: &str, bump: Option<&str>, dry_run: bool) -> Vec<String> {
let mut args = vec![
"release".into(),
"-C".into(),
checkout.into(),
"--yes".into(),
];
if let Some(bump) = bump {
args.extend([
"--mode".into(),
"release".into(),
"--bump".into(),
bump.into(),
]);
}
if dry_run {
args.push("--dry-run".into());
}
if bump.is_none() {
args.push("publish".into());
}
args
}
fn execute_deploy(
&self,
name: &str,
checkout: &str,
bump: Option<&str>,
dry_run: bool,
) -> Result<String, String> {
let cmd_args = Self::build_deploy_args(checkout, bump, dry_run);
let args_refs: Vec<&str> = cmd_args.iter().map(String::as_str).collect();
let (stdout, stderr, code) = run_command(&self.akurai_ec2_bin, &args_refs)?;
if code != 0 {
return Err(format!(
"ec2_deploy failed (exit {code}): {stderr}\nstdout: {stdout}"
));
}
if !dry_run {
self.run_post_mutation_validate(name)?;
}
match bump {
Some(bump) => Ok(format!(
"Deploy complete for {name} (bump={bump}):\n{stdout}"
)),
None => Ok(format!("Deploy complete for {name}:\n{stdout}")),
}
}
/// Read-only: show recent deploy history (git tags + systemd status)
fn ec2_deploy_history(&self, args: &Value) -> Result<String, String> {
let name = self.require_str(args, "name")?;
let checkout = self.resolve_app_checkout(name)?;
// Get git tags
let tag_cmd = format!("cd {checkout} && git tag --sort=-creatordate | head -10");
let (tags, _, _) = run_command("bash", &["-c", &tag_cmd])?;
// Get systemd unit info
let svc = format!("{name}.service");
let (svc_status, svc_err, svc_code) = run_command(
&self.akurai_ec2_bin,
&[
"ssh",
&format!("systemctl status {svc} --no-pager -l 2>&1 || true"),
],
)?;
let mut out = format!("=== {name} deploy history ===\n\n");
out.push_str("Recent git tags:\n");
if tags.trim().is_empty() {
out.push_str(" (no tags found)\n");
} else {
for line in tags.lines().take(10) {
out.push_str(&format!(" {line}\n"));
}
}
out.push_str(&format!(
"\nsystemd status (exit {svc_code}):\n{svc_status}"
));
if !svc_err.is_empty() && svc_code != 0 {
out.push_str(&format!("\nstderr: {svc_err}"));
}
Ok(out)
}
/// Read-only: list DNS records for a domain
fn ec2_dns_check(&self, args: &Value) -> Result<String, String> {
let domain = self.require_str(args, "domain")?;
let zone = self.arg_str(args, "zone");
let record_type = self.arg_str(args, "record_type");
let mut cmd_args = vec!["dns", "list", domain];
if let Some(z) = zone {
cmd_args.push("--zone");
cmd_args.push(z);
}
if let Some(t) = record_type {
cmd_args.push("--type");
cmd_args.push(t);
}
let (stdout, stderr, code) = run_command(&self.akurai_ec2_bin, &cmd_args)?;
if code != 0 {
return Err(format!("ec2_dns_check failed (exit {code}): {stderr}"));
}
Ok(stdout)
}
/// Read-only: TLS certificate expiry check
fn ec2_tls_expiry(&self, args: &Value) -> Result<String, String> {
let domain = self.require_str(args, "domain")?;
let (stdout, stderr, code) =
run_command(&self.akurai_ec2_bin, &["cert-status", domain, "--json"])?;
if code != 0 {
// cert-status exits non-zero on warning/error but still prints JSON
if !stdout.is_empty() {
return Ok(format!(
"TLS check for {domain}:\n{stdout}\nstderr: {stderr}"
));
}
return Err(format!("ec2_tls_expiry failed (exit {code}): {stderr}"));
}
Ok(format!("TLS check for {domain}:\n{stdout}"))
}
/// Read-only: systemd fleet status
fn ec2_systemd_status(&self, args: &Value) -> Result<String, String> {
let service = self.arg_str(args, "service");
let (stdout, stderr, code) = if let Some(svc) = service {
// Detailed status for a single service
let svc_name = if svc.ends_with(".service") {
svc.to_string()
} else {
format!("{svc}.service")
};
run_command(
&self.akurai_ec2_bin,
&[
"ssh",
&format!("systemctl status {svc_name} --no-pager -l 2>&1 || true"),
],
)?
} else {
run_command(&self.akurai_ec2_bin, &["processes"])?
};
if code != 0 && stderr.contains("Connection timed out") {
return Err(format!("Cannot reach EC2: {stderr}"));
}
Ok(stdout)
}
/// MUTATING: restart a systemd service with safety wrapper
fn ec2_systemd_restart(&self, args: &Value) -> Result<String, String> {
let service = self.require_str(args, "service")?;
let svc_name = if service.ends_with(".service") {
service.to_string()
} else {
format!("{service}.service")
};
// Pre-mutation safety backup
self.run_pre_mutation_backup()?;
// Restart the service
let (stdout, stderr, code) = run_command(
&self.akurai_ec2_bin,
&["ssh", &format!("sudo systemctl restart {svc_name} 2>&1")],
)?;
if code != 0 {
return Err(format!(
"ec2_systemd_restart failed to restart {svc_name} (exit {code}): {stderr}\nstdout: {stdout}"
));
}
// Wait a moment for the service to stabilize
std::thread::sleep(std::time::Duration::from_secs(3));
// Verify it's active
let (active_out, active_err, _active_code) = run_command(
&self.akurai_ec2_bin,
&[
"ssh",
&format!("systemctl is-active {svc_name} 2>&1 || true"),
],
)?;
if active_out.trim() != "active" {
// Auto-rollback: try to restore the previous binary
let app_name = svc_name.strip_suffix(".service").unwrap_or(&svc_name);
let (rb_out, rb_err, rb_code) =
run_command(&self.akurai_ec2_bin, &["rollback", app_name])?;
return Err(format!(
"ec2_systemd_restart: {svc_name} is '{active_out}' after restart (expected 'active'). Auto-rollback triggered.\nRollback result (exit {rb_code}): {rb_out}\n{rb_err}\nRestart stderr: {stderr}\nActive check stderr: {active_err}"
));
}
Ok(format!(
"Restarted {svc_name} successfully. Status: {active_out}\nRestart stdout: {stdout}"
))
}
/// MUTATING: run a format-specific backup. The shared validation keeps
/// retention bounds and error handling consistent while each CLI branch
/// retains its own encryption and isolated-recovery implementation.
fn ec2_backup_run(&self, args: &Value) -> Result<String, String> {
let backup_type = self.require_str(args, "backup_type")?;
let (default_retention, maximum_retention) = match backup_type {
"bunfork" | "workbench" => (14, 365),
"tasks" => (48, 720),
other => {
return Err(format!(
"Unknown backup type: {other}. Use 'bunfork', 'tasks', or 'workbench'"
))
}
};
let retention = self.arg_i64(args, "retention").unwrap_or(default_retention);
if !(2..=maximum_retention).contains(&retention) {
return Err(format!(
"Invalid retention for {backup_type}: expected an integer from 2 through {maximum_retention}"
));
}
let (stdout, stderr, code) = match backup_type {
"bunfork" => run_command(
&self.akurai_ec2_bin,
&["bunfork-backup", "run", &retention.to_string()],
)?,
"tasks" => run_command(
&self.akurai_ec2_bin,
&["tasks-backup", "run", &retention.to_string()],
)?,
"workbench" => run_command(
&self.akurai_ec2_bin,
&["workbench-backup", "run", &retention.to_string()],
)?,
_ => unreachable!("backup type validated above"),
};
if code != 0 {
return Err(format!(
"ec2_backup_run ({backup_type}) failed (exit {code}): {stderr}\nstdout: {stdout}"
));
}
Ok(format!("Backup ({backup_type}) complete:\n{stdout}"))
}
// ── Monitor proxy tools ────────────────────────────────────────────
/// Proxy GET /api/manage/hosts from rust-monitor
fn monitor_hosts(&self, _args: &Value) -> Result<String, String> {
self.monitor_get("/api/manage/hosts")
}
/// Proxy GET /api/manage/applications from rust-monitor
fn monitor_applications(&self, args: &Value) -> Result<String, String> {
let path = if let Some(host_id) = self.arg_i64(args, "host_id") {
format!("/api/manage/applications?host_id={host_id}")
} else {
"/api/manage/applications".into()
};
self.monitor_get(&path)
}
/// Proxy GET /api/alerts from rust-monitor
fn monitor_alerts(&self, _args: &Value) -> Result<String, String> {
self.monitor_get("/api/alerts")
}
/// Proxy GET /api/logs from rust-monitor (recent events)
fn monitor_events(&self, args: &Value) -> Result<String, String> {
let limit = self.arg_i64(args, "limit").unwrap_or(50);
let path = format!("/api/logs?limit={limit}");
self.monitor_get(&path)
}
fn monitor_ingestion_health(&self, _args: &Value) -> Result<String, String> {
self.monitor_cli(&["ingestion", "health"])
}
fn monitor_logs(&self, args: &Value) -> Result<String, String> {
let hours = self.arg_i64(args, "hours").unwrap_or(1).clamp(1, 720);
let limit = self.arg_i64(args, "limit").unwrap_or(100).clamp(1, 1000);
let hours = hours.to_string();
let limit = limit.to_string();
let mut owned = vec![
"logs".to_string(),
"recent".to_string(),
"--hours".to_string(),
hours,
"--limit".to_string(),
limit,
];
if let Some(source) = self.arg_str(args, "source") {
owned.extend(["--source".to_string(), source.to_string()]);
}
if let Some(before_id) = self.arg_i64(args, "before_id").filter(|id| *id > 0) {
owned.extend(["--before-id".to_string(), before_id.to_string()]);
}
let arguments: Vec<_> = owned.iter().map(String::as_str).collect();
self.monitor_cli(&arguments)
}
fn monitor_cli(&self, args: &[&str]) -> Result<String, String> {
let (stdout, stderr, code) = run_command("akurai-monitorctl", args)?;
if code != 0 {
return Err(format!("monitor CLI failed (exit {code}): {stderr}"));
}
Ok(stdout)
}
fn monitor_get(&self, path: &str) -> Result<String, String> {
let url = format!("{}{path}", self.monitor_url);
// Use curl since we're in a std-only environment (no reqwest)
let (stdout, stderr, code) = run_command("curl", &["-sS", "--max-time", "10", &url])?;
if code != 0 {
return Err(format!(
"monitor request failed (exit {code}): {stderr}\nurl: {url}"
));
}
Ok(stdout)
}
// ── Safety wrapper helpers ──────────────────────────────────────────
/// Run bunfork-backup as a pre-mutation safety net. Aborts the mutation
/// if the backup step fails.
fn run_pre_mutation_backup(&self) -> Result<(), String> {
let (stdout, stderr, code) = run_command(&self.akurai_ec2_bin, &["bunfork-backup", "run"])?;
if code != 0 {
return Err(format!(
"Pre-mutation backup FAILED (exit {code}). Mutation aborted.\nBackup stdout: {stdout}\nBackup stderr: {stderr}"
));
}
Ok(())
}
/// Run the target app's validate.sh after a mutation. If validation fails,
/// attempt rollback and return a structured error.
fn run_post_mutation_validate(&self, app_name: &str) -> Result<(), String> {
// Find the validate path from AKURAI_APPS.toml if available.
// We read the TOML directly to find the validate script path.
let validate_path = self.resolve_app_validate(app_name);
if let Some(vpath) = validate_path {
let checkout = self.resolve_app_checkout(app_name)?;
let full_path = format!("{checkout}/{vpath}");
let (stdout, stderr, code) = run_command("bash", &["-c", &full_path])?;
if code != 0 {
// Validation failed — attempt rollback
let (rb_out, rb_err, rb_code) =
run_command(&self.akurai_ec2_bin, &["rollback", app_name])?;
return Err(format!(
"Post-mutation validation FAILED for {app_name} (validate exit {code}). \
Auto-rollback triggered.\n\
Validate stdout: {stdout}\n\
Validate stderr: {stderr}\n\
Rollback result (exit {rb_code}): {rb_out}\n\
Rollback stderr: {rb_err}"
));
}
} else {
// No validate script configured — run a basic HTTP health check
// Parse the health URL from AKURAI_APPS.toml if available
let health_url = self.resolve_app_health_url(app_name);
if let Some(url) = health_url {
let (stdout, stderr, code) =
run_command("curl", &["-fsS", "--max-time", "10", &url])?;
if code != 0 {
let (rb_out, rb_err, rb_code) =
run_command(&self.akurai_ec2_bin, &["rollback", app_name])?;
return Err(format!(
"Post-mutation health check FAILED for {app_name} (curl exit {code} on {url}). \
Auto-rollback triggered.\n\
Curl stdout: {stdout}\n\
Curl stderr: {stderr}\n\
Rollback result (exit {rb_code}): {rb_out}\n\
Rollback stderr: {rb_err}"
));
}
}
}
Ok(())
}
// ── AKURAI_APPS.toml helpers ────────────────────────────────────────
/// Parse AKURAI_APPS.toml to find an app's checkout directory.
fn resolve_app_checkout(&self, app_name: &str) -> Result<String, String> {
// Use a simple Python one-liner via akurai-ec2's existing TOML parsing
// to extract the checkout path for this app.
let script = format!(
r#"import tomllib, sys
with open('{apps_toml}', 'rb') as f:
cfg = tomllib.load(f)
for a in cfg.get('app', []):
if a.get('name') == '{app_name}':
print(a.get('checkout', ''))
sys.exit(0)
print('')
sys.exit(1)
"#,
apps_toml = self.apps_toml,
app_name = app_name,
);
let (stdout, stderr, code) = run_command("python3", &["-c", &script])?;
if code != 0 || stdout.trim().is_empty() {
return Err(format!(
"App '{app_name}' not found in AKURAI_APPS.toml ({})\nstderr: {stderr}",
self.apps_toml
));
}
Ok(stdout.trim().to_string())
}
/// Parse AKURAI_APPS.toml to find an app's validate script path.
fn resolve_app_validate(&self, app_name: &str) -> Option<String> {
let script = format!(
r#"import tomllib
with open('{apps_toml}', 'rb') as f:
cfg = tomllib.load(f)
for a in cfg.get('app', []):
if a.get('name') == '{app_name}':
print(a.get('validate', ''))
break
"#,
apps_toml = self.apps_toml,
app_name = app_name,
);
if let Ok((stdout, _, code)) = run_command("python3", &["-c", &script]) {
if code == 0 && !stdout.trim().is_empty() {
return Some(stdout.trim().to_string());
}
}
None
}
/// Parse AKURAI_APPS.toml to find an app's health URL.
fn resolve_app_health_url(&self, app_name: &str) -> Option<String> {
let script = format!(
r#"import tomllib
with open('{apps_toml}', 'rb') as f:
cfg = tomllib.load(f)
for a in cfg.get('app', []):
if a.get('name') == '{app_name}':
print(a.get('health', ''))
break
"#,
apps_toml = self.apps_toml,
app_name = app_name,
);
if let Ok((stdout, _, code)) = run_command("python3", &["-c", &script]) {
if code == 0 && !stdout.trim().is_empty() {
let url = stdout.trim().to_string();
if !url.is_empty() && url != "-" {
return Some(url);
}
}
}
None
}
}
// ── JSON-RPC helpers ────────────────────────────────────────────────────
fn null_val() -> Value {
Value::Null
}
fn jsonrpc_result(id: Value, result: &Value) -> String {
let response = Value::Object(vec![
("jsonrpc".into(), Value::Str("2.0".into())),
("id".into(), id),
("result".into(), result.clone()),
]);
response.to_json()
}
fn jsonrpc_error(id: Value, code: i64, message: &str) -> String {
let error = Value::Object(vec![
("code".into(), Value::Int(code)),
("message".into(), Value::Str(message.into())),
]);
let response = Value::Object(vec![
("jsonrpc".into(), Value::Str("2.0".into())),
("id".into(), id),
("error".into(), error),
]);
response.to_json()
}
mod tests {
#[allow(unused_imports)]
use super::*;
// ── JSON-RPC protocol tests ──────────────────────────────────────
#[test]
fn test_build_deploy_args_preserves_publish_and_bump_modes() {
assert_eq!(
ToolRegistry::build_deploy_args("/srv/app", None, false),
vec!["release", "-C", "/srv/app", "--yes", "publish"]
);
assert_eq!(
ToolRegistry::build_deploy_args("/srv/app", None, true),
vec!["release", "-C", "/srv/app", "--yes", "--dry-run", "publish"]
);
assert_eq!(
ToolRegistry::build_deploy_args("/srv/app", Some("patch"), true),
vec![
"release",
"-C",
"/srv/app",
"--yes",
"--mode",
"release",
"--bump",
"patch",
"--dry-run"
]
);
}
#[test]
fn test_tools_list_returns_all_tools() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let tools = reg.list_tools();
assert_eq!(tools.len(), 14, "Should have exactly 14 tools");
let names: Vec<&str> = tools
.iter()
.filter_map(|t| t.get("name").and_then(|n| n.as_str()))
.collect();
let backup = tools
.iter()
.find(|t| t.get("name").and_then(|n| n.as_str()) == Some("ec2_backup_run"))
.unwrap();
let description = backup.get("description").and_then(|d| d.as_str()).unwrap();
assert!(description.contains("tasks"));
assert!(description.contains("48"));
assert!(names.contains(&"ec2_fleet_status"));
assert!(names.contains(&"ec2_deploy"));
assert!(names.contains(&"ec2_deploy_history"));
assert!(names.contains(&"ec2_dns_check"));
assert!(names.contains(&"ec2_tls_expiry"));
assert!(names.contains(&"ec2_systemd_status"));
assert!(names.contains(&"ec2_systemd_restart"));
assert!(names.contains(&"ec2_backup_run"));
assert!(names.contains(&"monitor_hosts"));
assert!(names.contains(&"monitor_applications"));
assert!(names.contains(&"monitor_alerts"));
assert!(names.contains(&"monitor_events"));
assert!(names.contains(&"monitor_ingestion_health"));
assert!(names.contains(&"monitor_logs"));
}
#[test]
fn test_tools_list_jsonrpc() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let request = r#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#;
let response = reg.handle_request(request);
assert!(response.contains("\"jsonrpc\":\"2.0\""));
assert!(response.contains("\"id\":1"));
assert!(response.contains("\"tools\""));
assert!(response.contains("ec2_fleet_status"));
}
#[test]
fn test_tools_call_unknown_tool() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let request = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"nonexistent","arguments":{}}}"#;
let response = reg.handle_request(request);
assert!(response.contains("\"isError\":true"));
assert!(response.contains("Unknown tool"));
}
#[test]
fn test_tools_list_schema_valid() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
for tool in reg.list_tools() {
let name = tool.get("name").and_then(|n| n.as_str()).unwrap();
let schema = tool
.get("inputSchema")
.unwrap_or_else(|| panic!("{name} missing inputSchema"));
let stype = schema.get("type").and_then(|s| s.as_str()).unwrap();
assert_eq!(stype, "object", "{name}: inputSchema.type must be 'object'");
assert!(
schema.get("properties").is_some(),
"{name}: missing properties"
);
// Every required field must appear in properties
if let Some(Value::Array(reqs)) = schema.get("required") {
let props = schema.get("properties").unwrap();
for r in reqs {
let rname = r.as_str().unwrap();
assert!(
props.get(rname).is_some(),
"{name}: required field '{rname}' not in properties"
);
}
}
}
}
#[test]
fn test_tasks_backup_retention_bounds() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let args = Value::Object(vec![
("backup_type".into(), Value::Str("tasks".into())),
("retention".into(), Value::Int(721)),
]);
let error = reg.ec2_backup_run(&args).unwrap_err();
assert!(error.contains("2 through 720"));
}
// ── Mutating tool safety wrapper test ────────────────────────────
#[test]
fn test_mutating_tool_flag() {
// Verify that mutating tools have specific descriptions that mention
// they mutate state.
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let tools = reg.list_tools();
let mutating = ["ec2_deploy", "ec2_systemd_restart", "ec2_backup_run"];
for name in &mutating {
let tool = tools
.iter()
.find(|t| t.get("name").and_then(|n| n.as_str()) == Some(name));
assert!(tool.is_some(), "Missing mutating tool: {name}");
let _desc = tool
.unwrap()
.get("description")
.and_then(|d| d.as_str())
.unwrap_or("");
// Mutating tools should have inputSchema with at least one required param
let schema = tool.unwrap().get("inputSchema").unwrap();
assert!(
schema.get("properties").is_some(),
"{name}: missing properties"
);
}
}
#[test]
fn test_parse_error_returns_jsonrpc_error() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let response = reg.handle_request("not json");
assert!(response.contains("\"error\""));
assert!(response.contains("-32700"));
}
#[test]
fn test_missing_method_returns_error() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let response = reg.handle_request(r#"{"jsonrpc":"2.0","id":1}"#);
assert!(response.contains("\"error\""));
assert!(response.contains("-32600"));
}
#[test]
fn test_unknown_method_returns_error() {
let reg = ToolRegistry::new(
"http://localhost:8800".into(),
"akurai-ec2".into(),
"/tmp/nonexistent.toml".into(),
);
let response = reg.handle_request(r#"{"jsonrpc":"2.0","id":1,"method":"resources/list"}"#);
assert!(response.contains("\"error\""));
assert!(response.contains("-32601"));
}
}