diff --git a/Cargo.lock b/Cargo.lock index eccd826..4da692f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,7 @@ dependencies = [ "rusqlite", "serde", "serde_json", + "serde_yaml", "tokio", "tokio-stream", "tonic", @@ -1631,6 +1632,19 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_yaml" +version = "0.9.34+deprecated" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47" +dependencies = [ + "indexmap 2.14.0", + "itoa", + "ryu", + "serde", + "unsafe-libyaml", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -2048,6 +2062,12 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "unsafe-libyaml" +version = "0.2.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" + [[package]] name = "untrusted" version = "0.9.0" diff --git a/core/agentsd/Cargo.toml b/core/agentsd/Cargo.toml index 400b00f..5c28e51 100644 --- a/core/agentsd/Cargo.toml +++ b/core/agentsd/Cargo.toml @@ -25,3 +25,4 @@ reqwest = { workspace = true } libc = { workspace = true } cron = "0.12" chrono = "0.4" +serde_yaml = "0.9" diff --git a/core/agentsd/src/admin.rs b/core/agentsd/src/admin.rs new file mode 100644 index 0000000..dbee7be --- /dev/null +++ b/core/agentsd/src/admin.rs @@ -0,0 +1,53 @@ +use crate::manifest::PluginManifest; +use crate::plugin::PluginManager; +use crate::supervisor::Supervisor; +use std::path::PathBuf; +use std::sync::Arc; + +/// Admin service: implements the gRPC Admin contract (proto/agentsd.proto) +/// wiring manifest + resolver + supervisor + PluginManager together. +/// +/// CONTRACT: +/// - Installed-plugin registry persisted at `data/installed.json`: +/// [{ id, dir, enabled }] — `dir` is the directory containing plugin.yaml. +/// - install(path): load+validate manifest, reject duplicate id, add to registry. +/// - uninstall(id): must be stopped first; removes from registry (files untouched). +/// - enable/disable(id): flips `enabled`; disable does NOT stop a running plugin +/// (matches systemctl semantics); start refuses when disabled. +/// - start(id): resolve transitive deps (resolver), start each in topo order, +/// skipping already-running ones; waits briefly for gRPC registration. +/// - stop(id): warn (in message) if running dependents exist, then stop. +/// - restart(id) = stop + start. +/// - status/list: merge installed registry + PluginManager records + Supervisor state. +/// - logs(id, n): Supervisor::tail_log. +/// - autostart(): called from server startup — start all enabled+autoload plugins +/// in global topo order; failures are logged, not fatal. +pub struct AdminState { + pub plugins: Arc, + pub supervisor: Arc, + pub installed_path: PathBuf, // data/installed.json +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct InstalledPlugin { + pub id: String, + pub dir: PathBuf, + pub enabled: bool, +} + +impl AdminState { + pub fn new(plugins: Arc, supervisor: Arc) -> Self { + todo!("subagent C: implement") + } + + pub fn load_manifest(&self, id: &str) -> Result { + todo!("subagent C: implement") + } + + pub async fn autostart(self: &Arc) { + todo!("subagent C: implement") + } +} + +// subagent C: implement `pub struct AdminSvc` + `#[tonic::async_trait] impl AdminService for AdminSvc` +// and add `.add_service(AdminServer::new(...))` wiring in server.rs. diff --git a/core/agentsd/src/cli.rs b/core/agentsd/src/cli.rs new file mode 100644 index 0000000..574d71d --- /dev/null +++ b/core/agentsd/src/cli.rs @@ -0,0 +1,29 @@ +/// CLI subcommands (= systemctl). Invoked when argv[1] is a known command; +/// otherwise main.rs falls through to daemon mode. +/// +/// CONTRACT: +/// - Talks to a RUNNING agentsd daemon via gRPC Admin service at +/// AGENTSD_ADDR (default http://[::1]:50051). If the daemon is not +/// reachable, print a clear error and exit code 2. +/// - Commands: +/// agentsd start agentsd stop agentsd restart +/// agentsd status [] agentsd list agentsd logs [-n N] +/// agentsd install agentsd uninstall +/// agentsd enable agentsd disable +/// - `status` without id behaves like `list`. +/// - Exit codes: 0 ok, 1 operation failed (server said no), 2 cannot connect/usage. +/// - Output: human-readable; `list` is an aligned table: +/// ID TYPE STATUS ENABLED PID RESTARTS VERSION +pub const COMMANDS: &[&str] = &[ + "start", "stop", "restart", "status", "list", "logs", "install", "uninstall", "enable", + "disable", +]; + +pub fn is_cli_invocation(args: &[String]) -> bool { + args.first().is_some_and(|c| COMMANDS.contains(&c.as_str())) +} + +/// Run the CLI. Returns the process exit code. +pub async fn run(args: Vec) -> i32 { + todo!("subagent D: implement") +} diff --git a/core/agentsd/src/main.rs b/core/agentsd/src/main.rs index 10565a3..129d973 100644 --- a/core/agentsd/src/main.rs +++ b/core/agentsd/src/main.rs @@ -1,8 +1,13 @@ +pub mod admin; pub mod callback; +pub mod cli; +pub mod manifest; pub mod paths; pub mod permission; mod plugin; +pub mod resolver; mod server; +pub mod supervisor; mod syscall; use anyhow::Result; @@ -10,6 +15,11 @@ use tracing_subscriber::EnvFilter; #[tokio::main] async fn main() -> Result<()> { + let args: Vec = std::env::args().skip(1).collect(); + if cli::is_cli_invocation(&args) { + std::process::exit(cli::run(args).await); + } + tracing_subscriber::fmt() .with_env_filter(EnvFilter::from_default_env().add_directive("agentsd=info".parse()?)) .init(); diff --git a/core/agentsd/src/manifest.rs b/core/agentsd/src/manifest.rs new file mode 100644 index 0000000..00ecebb --- /dev/null +++ b/core/agentsd/src/manifest.rs @@ -0,0 +1,75 @@ +use serde::{Deserialize, Serialize}; +use std::path::{Path, PathBuf}; + +/// Parsed plugin.yaml manifest. +/// +/// CONTRACT (do not change field names/types without updating admin.rs, cli.rs, supervisor.rs): +/// - `exports` accepts either a YAML list of {name, description} or the literal string "auto" +/// (auto means exports are discovered at runtime via PluginBridge.Register). +/// - `enabled`/`autoload` default to true when absent. +/// - `entry` is the shell command line used by the supervisor to launch the plugin. +/// - `port` is informational (plugins bind via PLUGIN_LISTEN_ADDR env or their own default). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PluginManifest { + pub id: String, + pub name: String, + pub version: String, + #[serde(rename = "type")] + pub plugin_type: String, // "system" | "basic" | "standard" + #[serde(default)] + pub syscalls: Vec, + #[serde(default)] + pub depends: Vec, + #[serde(default)] + pub exports: ManifestExports, + pub entry: String, + #[serde(default)] + pub port: Option, + #[serde(default = "default_true")] + pub enabled: bool, + #[serde(default = "default_true")] + pub autoload: bool, + /// Directory containing plugin.yaml. Not part of the YAML itself; set by the loader. + #[serde(skip)] + pub dir: PathBuf, +} + +fn default_true() -> bool { + true +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)] +#[serde(untagged)] +pub enum ManifestExports { + #[default] + #[serde(skip)] + None, + Auto(String), // must be the literal "auto" + List(Vec), +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ManifestExport { + pub name: String, + #[serde(default)] + pub description: String, +} + +impl PluginManifest { + /// Load and validate `/plugin.yaml`. + pub fn load(dir: &Path) -> Result { + todo!("subagent A: implement") + } + + /// Validation: non-empty id/name/version/entry, type in {system,basic,standard}, + /// exports Auto variant only allows the literal "auto", id charset [a-zA-Z0-9_-]+. + pub fn validate(&self) -> Result<(), String> { + todo!("subagent A: implement") + } +} + +/// Scan a directory of plugin dirs (e.g. `plugins/`), loading every subdir with a plugin.yaml. +/// Invalid manifests are returned as (dir, error) so callers can report instead of dying. +pub fn scan_plugins_dir(root: &Path) -> (Vec, Vec<(PathBuf, String)>) { + todo!("subagent A: implement") +} diff --git a/core/agentsd/src/resolver.rs b/core/agentsd/src/resolver.rs new file mode 100644 index 0000000..0ab8d83 --- /dev/null +++ b/core/agentsd/src/resolver.rs @@ -0,0 +1,40 @@ +use std::collections::HashMap; + +/// Dependency resolution for plugin start ordering. +/// +/// CONTRACT: +/// - Input: map of plugin id -> list of dependency ids (the `depends` field). +/// - `resolve_order(ids, deps)` returns the full topological order for starting +/// ALL given plugins: dependencies come BEFORE dependents. Deterministic: +/// ties broken by lexicographic id order. +/// - `start_order_for(target, deps)` returns the chain needed to start `target`: +/// its transitive deps (topo-sorted) followed by `target` itself. Plugins not +/// reachable from `target` are excluded. +/// - Errors are returned as human-readable strings: +/// - cycle: "dependency cycle: a -> b -> a" +/// - missing: "plugin 'x' depends on 'y' which is not installed" +#[derive(Debug, Clone)] +pub struct DependencyGraph { + deps: HashMap>, +} + +impl DependencyGraph { + pub fn new(deps: HashMap>) -> Self { + Self { deps } + } + + /// Topological order over all known plugins (deps before dependents). + pub fn resolve_order(&self) -> Result, String> { + todo!("subagent A: implement (Kahn's algorithm, lexicographic tie-break)") + } + + /// Transitive-dependency start chain for one target (deps first, target last). + pub fn start_order_for(&self, target: &str) -> Result, String> { + todo!("subagent A: implement") + } + + /// Plugins that (transitively) depend on `target` — used to warn on stop. + pub fn dependents_of(&self, target: &str) -> Vec { + todo!("subagent A: implement") + } +} diff --git a/core/agentsd/src/supervisor.rs b/core/agentsd/src/supervisor.rs new file mode 100644 index 0000000..12bcbee --- /dev/null +++ b/core/agentsd/src/supervisor.rs @@ -0,0 +1,78 @@ +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; + +/// Process supervisor: spawns plugin processes from their manifest `entry`, +/// captures stdout/stderr to `data/logs/.log`, watches for exit, and +/// restarts according to policy. +/// +/// CONTRACT (admin.rs depends on these signatures): +/// - All methods are cheap to call from async context (no blocking locks held +/// across awaits; spawn/wait runs on tokio tasks). +/// - Supervisor does NOT know about gRPC registration; it manages OS processes +/// only. admin.rs correlates "process running" with "plugin registered". +/// - On Windows, stop() must kill the whole process tree (entry commands are +/// `cargo run ...` / `python ...` which spawn children). +#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize, serde::Deserialize)] +pub enum RestartPolicy { + Always, + OnFailure, + Never, +} + +#[derive(Debug, Clone)] +pub struct SupervisedState { + pub pid: u32, + pub started_at_ms: u64, + pub restart_count: u32, + /// true while the watchdog considers this plugin managed (stop() clears it) + pub active: bool, +} + +pub struct Supervisor { + inner: Arc>>, + logs_dir: PathBuf, +} + +impl Supervisor { + pub fn new(logs_dir: PathBuf) -> Self { + todo!("subagent B: implement") + } + + /// Spawn `entry` (shell command line) with cwd = plugin dir. + /// stdout+stderr appended to `/.log` with timestamps. + /// A watchdog task waits for exit and applies `policy`: + /// - exponential backoff starting at 1s, doubling, capped at 30s + /// - after 5 consecutive failures within a window, give up and call `on_give_up` + /// - clean stop via stop() must NOT trigger restart + pub fn start( + &self, + id: &str, + entry: &str, + cwd: &PathBuf, + env: Vec<(String, String)>, + policy: RestartPolicy, + on_exit: Box) + Send + Sync>, + ) -> Result { + todo!("subagent B: implement") + } + + /// Kill the process tree. Marks state inactive first so the watchdog + /// doesn't restart it. No-op error if not running. + pub fn stop(&self, id: &str) -> Result<(), String> { + todo!("subagent B: implement") + } + + pub fn state(&self, id: &str) -> Option { + todo!("subagent B: implement") + } + + pub fn is_running(&self, id: &str) -> bool { + todo!("subagent B: implement") + } + + /// Read last `n` lines of `/.log`. + pub fn tail_log(&self, id: &str, n: usize) -> Result, String> { + todo!("subagent B: implement") + } +} diff --git a/proto/agentsd.proto b/proto/agentsd.proto index 0835818..ff4c9ed 100644 --- a/proto/agentsd.proto +++ b/proto/agentsd.proto @@ -214,5 +214,53 @@ message CallResponse { bytes result = 1; bool ok = 2; string error = 3; } message HeartbeatRequest { string plugin_id = 1; } message HeartbeatResponse { bool ok = 1; } +// ========== Admin (plugin lifecycle management, = systemctl) ========== +service Admin { + rpc Start(AdminIdRequest) returns (AdminResponse); + rpc Stop(AdminIdRequest) returns (AdminResponse); + rpc Restart(AdminIdRequest) returns (AdminResponse); + rpc Status(AdminIdRequest) returns (AdminStatusResponse); + rpc ListPlugins(Empty) returns (AdminListResponse); + rpc Logs(AdminLogsRequest) returns (AdminLogsResponse); + rpc Install(AdminInstallRequest) returns (AdminResponse); + rpc Uninstall(AdminIdRequest) returns (AdminResponse); + rpc Enable(AdminIdRequest) returns (AdminResponse); + rpc Disable(AdminIdRequest) returns (AdminResponse); +} + +message AdminIdRequest { string id = 1; } +message AdminResponse { bool ok = 1; string error = 2; string message = 3; } +message AdminStatusResponse { + bool ok = 1; + string error = 2; + AdminPluginState plugin = 3; +} +message AdminListResponse { repeated AdminPluginState plugins = 1; } +message AdminPluginState { + string id = 1; + string name = 2; + string version = 3; + string plugin_type = 4; + string status = 5; // "registered" | "running" | "stopped" | "failed: " + bool enabled = 6; + bool installed = 7; + uint32 pid = 8; // 0 if not running under supervisor + uint64 started_at_ms = 9; // 0 if not running + uint32 restart_count = 10; + repeated string depends = 11; +} +message AdminLogsRequest { + string id = 1; + uint32 tail_lines = 2; // 0 = default (100) +} +message AdminLogsResponse { + bool ok = 1; + string error = 2; + repeated string lines = 3; +} +message AdminInstallRequest { + string path = 1; // directory containing plugin.yaml +} + // ========== Common ========== message Empty {}