diff --git a/README.md b/README.md index 812033f9..bc21734e 100644 --- a/README.md +++ b/README.md @@ -360,6 +360,15 @@ record. It also does not serialize catalog or runtime writers, reconcile tasks, control-plane cutover. Consumers that require a zero-write boundary under concurrent root deletion or a transactional declaration boundary need a separate protocol. +For cleanup planners that need the same evidence aggregated by explicit Agent Spec workspace, use: + +```sh +st2 workspace-activity --catalog "$CATALOG" --host --ttl 60 --json +``` + +The short-lived `st2.workspace-activity.v1` envelope is read-only and fail-closed. It is not deletion +authority; see [the workspace activity contract](docs/workspace-activity.md). + ### Staged control-plane replacement gate `st2 up` is a replaceable control plane, not the lifetime owner of an agent. Stopping it normally diff --git a/docs/workspace-activity.md b/docs/workspace-activity.md new file mode 100644 index 00000000..47473d4a --- /dev/null +++ b/docs/workspace-activity.md @@ -0,0 +1,34 @@ +# Workspace activity snapshot + +`st2 workspace-activity --json` emits short-lived, read-only evidence about activity in explicit +Agent Spec `workspace` paths on one host. Suspended and retired declarations remain in the snapshot +so a retained live generation cannot disappear from cleanup evidence. The command reuses st2's PTY +and exec generation observers; it does not scan arbitrary processes, reconcile tasks, authorize +cleanup, or delete anything. + +The `st2.workspace-activity.v1` envelope contains `schemaVersion`, `producer`, an `epoch` bound to +the canonical catalog, host, and catalog generation, `capturedAt`, `expiresAt`, `complete`, `errors`, +and lexically sorted `claims`. Each claim contains a canonical workspace path, sorted owning agent +IDs, sorted positively running runtime IDs, and the derived `active` boolean. + +Consumers must fail closed unless `complete` is true, the snapshot has not expired, and the epoch is +the one they admitted. An inactive claim means only that st2 observed no running generation for its +declared tasks in this snapshot. Cleanup still needs its own filesystem/process liveness checks and +must revalidate immediately before mutation. + +The TTL must be between 1 and 300 seconds. Out-of-range values emit an incomplete envelope whose +`expiresAt` equals `capturedAt`, then exit non-zero. + +This v1 precursor identifies active runtime IDs but is not a generation-bound lease: generation +PID/creation evidence remains available from `st2 tasks --json`. A deletion transaction must obtain +and revalidate that stronger evidence rather than treating this snapshot as a lock. + +Example: + +```console +st2 --catalog "$CATALOG" workspace-activity --host dev3 --ttl 60 --json +``` + +Catalog discovery errors, an unavailable PTY/exec backend, declaration drift, catalog-generation +drift, and an unresolvable declared workspace all make the envelope incomplete and the command exits +non-zero after printing the JSON evidence. diff --git a/src/catalog_lock.rs b/src/catalog_lock.rs index 5eea2370..cf751642 100644 --- a/src/catalog_lock.rs +++ b/src/catalog_lock.rs @@ -20,6 +20,14 @@ pub const GENERATION_INTENT_FILE: &str = "catalog-generation-incomplete"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct CatalogReadFence(Option); +impl CatalogReadFence { + /// The declaration generation observed by this read fence. `None` means the catalog predates + /// generation receipts; read-only consumers must treat that as an unversioned epoch. + pub fn generation(self) -> Option { + self.0 + } +} + pub fn read_fence(catalog: &Path) -> Result { let first = read_generation(catalog)?; ensure_authoring_complete(catalog)?; diff --git a/src/exec_backend.rs b/src/exec_backend.rs index 13f5b47a..d542f1e5 100644 --- a/src/exec_backend.rs +++ b/src/exec_backend.rs @@ -898,7 +898,7 @@ fn process_created_at(start_time_ticks: u64) -> anyhow::Result { Ok(UNIX_EPOCH + Duration::from_micros(start_time_ticks)) } -pub(crate) fn rfc3339_utc(time: SystemTime) -> anyhow::Result { +pub fn rfc3339_utc(time: SystemTime) -> anyhow::Result { let duration = time.duration_since(UNIX_EPOCH)?; let seconds = duration.as_secs() as libc::time_t; let millis = duration.subsec_millis(); diff --git a/src/lib.rs b/src/lib.rs index 4eab71f6..7b1d10e2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -33,6 +33,7 @@ pub mod status; pub mod task_inventory; pub mod validate; pub mod version; +pub mod workspace_activity; mod watch; // The declaration model and the catalog walk live in the `agent-spec` crate, so st2 and any other diff --git a/src/main.rs b/src/main.rs index de118588..2e31f748 100644 --- a/src/main.rs +++ b/src/main.rs @@ -261,6 +261,20 @@ enum Command { #[arg(long)] json: bool, }, + /// Emit a short-lived, fail-closed snapshot of runtime activity for Agent Spec workspaces. + /// This is read-only evidence for external cleanup planners, never deletion authority. + WorkspaceActivity { + /// Host whose declared workspaces and task generations to inspect. Defaults to this host. + #[arg(long)] + host: Option, + /// Snapshot lifetime in seconds (1..=300). Consumers must reject a snapshot after + /// `expiresAt`. + #[arg(long, default_value_t = 60)] + ttl: u64, + /// Emit the versioned machine-readable envelope. Required in v1. + #[arg(long)] + json: bool, + }, /// Clear one task's park after fixing what crash-looped it. A task parked by its `restart{}` /// policy (mode=fail) stays parked for the rest of the supervisor run, and this is its per-task /// exit: the running supervisor relaunches exactly this task on its next pass, leaving every @@ -996,6 +1010,13 @@ fn main() -> Result<()> { let catalog = catalog_arg(None)?; tasks_cmd(&catalog, host) } + Command::WorkspaceActivity { host, ttl, json } => { + if !json { + anyhow::bail!("`st2 workspace-activity` v1 requires --json"); + } + let catalog = catalog_arg(None)?; + workspace_activity_cmd(&catalog, host, ttl) + } Command::Unpark { task, host } => { let catalog = catalog_arg(None)?; unpark_cmd(&catalog, &task, host) @@ -1574,6 +1595,17 @@ fn tasks_cmd(root: &Path, host: Option) -> Result<()> { } } +fn workspace_activity_cmd(root: &Path, host: Option, ttl: u64) -> Result<()> { + let host = host.unwrap_or_else(detect_host); + let snapshot = st2::workspace_activity::snapshot(root, &host, Duration::from_secs(ttl)); + println!("{}", snapshot.to_json()); + if snapshot.complete() { + Ok(()) + } else { + anyhow::bail!("workspace activity snapshot incomplete") + } +} + /// Ask this host's supervisor to release one parked task. /// /// The request is a file the supervisor drains at the top of its next pass, not a direct mutation: diff --git a/src/workspace_activity.rs b/src/workspace_activity.rs new file mode 100644 index 00000000..b08950a4 --- /dev/null +++ b/src/workspace_activity.rs @@ -0,0 +1,319 @@ +//! Short-lived, read-only workspace activity snapshots. +//! +//! The snapshot joins explicit Agent Spec `workspace` declarations to the same PTY/exec generation +//! observers used by `st2 tasks --json`. It is evidence for a cleanup planner, not cleanup authority: +//! st2 never deletes a workspace and a consumer must reject incomplete or expired snapshots. + +use std::collections::{BTreeMap, BTreeSet}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, SystemTime}; + +use serde::Serialize; + +use crate::task_inventory::{DesiredRuntime, ObservedState, RuntimeObservation, RuntimeObserver}; +use crate::{SystemRunner, discover, exec_state_dir}; + +pub const SCHEMA_VERSION: &str = "st2.workspace-activity.v1"; +pub const PRODUCER: &str = "st2"; +pub const MIN_TTL: Duration = Duration::from_secs(1); +pub const MAX_TTL: Duration = Duration::from_secs(5 * 60); + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct WorkspaceActivitySnapshot { + schema_version: &'static str, + producer: &'static str, + epoch: Epoch, + captured_at: String, + expires_at: String, + complete: bool, + errors: Vec, + claims: Vec, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct Epoch { + catalog: PathBuf, + host: String, + catalog_generation: Option, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct WorkspaceClaim { + workspace: PathBuf, + agents: Vec, + active_runtime_ids: Vec, + active: bool, +} + +impl WorkspaceActivitySnapshot { + pub fn complete(&self) -> bool { + self.complete + } + + pub fn to_json(&self) -> String { + serde_json::to_string(self).expect("workspace activity snapshot is serializable") + } + + fn incomplete( + catalog: PathBuf, + host: String, + now: SystemTime, + ttl: Duration, + error: String, + ) -> Self { + let captured_at = timestamp(now); + let expires_at = timestamp(now.checked_add(ttl).unwrap_or(now)); + Self { + schema_version: SCHEMA_VERSION, + producer: PRODUCER, + epoch: Epoch { + catalog, + host, + catalog_generation: None, + }, + captured_at, + expires_at, + complete: false, + errors: vec![error], + claims: Vec::new(), + } + } +} + +#[derive(Debug)] +struct DeclaredWorkspace { + path: PathBuf, + agent: String, + runtimes: Vec, +} + +/// Capture one bounded workspace-activity observation. All failures are represented in the JSON +/// envelope and make the command fail after printing it, so absence can never be inferred from an +/// unavailable catalog or runtime backend. +pub fn snapshot(root: &Path, host: &str, ttl: Duration) -> WorkspaceActivitySnapshot { + snapshot_at(root, host, ttl, SystemTime::now()) +} + +fn snapshot_at( + root: &Path, + host: &str, + ttl: Duration, + now: SystemTime, +) -> WorkspaceActivitySnapshot { + if !(MIN_TTL..=MAX_TTL).contains(&ttl) { + return WorkspaceActivitySnapshot::incomplete( + root.to_path_buf(), + host.to_owned(), + now, + Duration::ZERO, + format!( + "snapshot TTL must be between {} and {} seconds", + MIN_TTL.as_secs(), + MAX_TTL.as_secs() + ), + ); + } + let catalog = match root.canonicalize() { + Ok(path) => path, + Err(error) => { + return WorkspaceActivitySnapshot::incomplete( + root.to_path_buf(), + host.to_owned(), + now, + ttl, + format!("canonicalize catalog {}: {error}", root.display()), + ); + } + }; + let before = match crate::catalog_lock::read_fence(&catalog) { + Ok(fence) => fence, + Err(error) => { + return WorkspaceActivitySnapshot::incomplete( + catalog, + host.to_owned(), + now, + ttl, + error.to_string(), + ); + } + }; + let found = discover(&catalog); + let mut errors = found + .errors + .iter() + .map(|error| format!("catalog file {}: {}", error.path.display(), error.message)) + .collect::>(); + let mut declared = Vec::new(); + for spec in &found.specs { + if spec.resolved_host(host) != host { + continue; + } + let Some(raw_workspace) = spec.workspace.as_deref() else { + continue; + }; + let expanded = crate::expand::expand_catalog(raw_workspace, &catalog); + let spec_dir = spec.path.parent().unwrap_or(&catalog); + let path = match spec_dir.join(&expanded).canonicalize() { + Ok(path) => path, + Err(error) => { + errors.push(format!( + "canonicalize workspace {expanded:?} for {}: {error}", + spec.bus_id(host) + )); + continue; + } + }; + let bus_id = spec.bus_id(host); + let runtimes = spec + .tasks + .iter() + .filter(|task| { + !spec.desired_state.is_running() || task.command.is_some() || task.argv.is_some() + }) + .map(|task| DesiredRuntime { + runtime_id: task + .id + .clone() + .unwrap_or_else(|| format!("{bus_id}.{}", task.name)), + kind: task.kind, + }) + .collect(); + declared.push(DeclaredWorkspace { + path, + agent: bus_id, + runtimes, + }); + } + let desired = declared + .iter() + .flat_map(|workspace| workspace.runtimes.iter().cloned()) + .collect::>(); + let mut runtime_owners = BTreeMap::::new(); + for runtime in &desired { + *runtime_owners + .entry(runtime.runtime_id.clone()) + .or_default() += 1; + } + for (runtime_id, owners) in runtime_owners { + if owners > 1 { + errors.push(format!( + "duplicate runtime id {runtime_id:?} is declared {owners} times" + )); + } + } + let runner = SystemRunner::new(catalog.clone(), exec_state_dir(host)); + let observed = runner.observe(&desired); + errors.extend(observed.errors.iter().cloned()); + if !observed.complete && observed.errors.is_empty() { + errors.push("runtime observer reported an incomplete batch".into()); + } + let desired_ids = desired + .iter() + .map(|runtime| runtime.runtime_id.as_str()) + .collect::>(); + let mut observations = BTreeMap::::new(); + for observation in observed.observations { + if !desired_ids.contains(observation.runtime_id.as_str()) { + errors.push(format!( + "runtime observer returned undeclared id {:?}", + observation.runtime_id + )); + continue; + } + let runtime_id = observation.runtime_id.clone(); + if observations + .insert(runtime_id.clone(), observation) + .is_some() + { + errors.push(format!( + "runtime observer returned duplicate id {runtime_id:?}" + )); + } + } + let mut claims = BTreeMap::, BTreeSet)>::new(); + for workspace in declared { + let claim = claims.entry(workspace.path).or_default(); + claim.0.insert(workspace.agent); + for runtime in workspace.runtimes { + match observations.get(&runtime.runtime_id).map(|row| &row.state) { + Some(ObservedState::Running(_)) => { + claim.1.insert(runtime.runtime_id); + } + Some(ObservedState::Indeterminate(error)) => errors.push(error.clone()), + None if !observed.complete => errors.push(format!( + "runtime observation incomplete for {:?}", + runtime.runtime_id + )), + _ => {} + } + } + } + let after_found = discover(&catalog); + if !crate::task_inventory::same_discovery(&found, &after_found) { + errors.push("catalog declarations changed during workspace activity observation".into()); + } + match crate::catalog_lock::read_fence(&catalog) { + Ok(after) if after == before => {} + Ok(_) => { + errors.push("catalog generation changed during workspace activity observation".into()) + } + Err(error) => errors.push(error.to_string()), + } + errors.sort(); + errors.dedup(); + let claims = claims + .into_iter() + .map(|(workspace, (agents, active_runtime_ids))| { + let active_runtime_ids = active_runtime_ids.into_iter().collect::>(); + WorkspaceClaim { + workspace, + agents: agents.into_iter().collect(), + active: !active_runtime_ids.is_empty(), + active_runtime_ids, + } + }) + .collect(); + let captured_at = timestamp(now); + let expires_at = timestamp(now.checked_add(ttl).unwrap_or(now)); + WorkspaceActivitySnapshot { + schema_version: SCHEMA_VERSION, + producer: PRODUCER, + epoch: Epoch { + catalog, + host: host.to_owned(), + catalog_generation: before.generation(), + }, + captured_at, + expires_at, + complete: errors.is_empty() && observed.complete, + errors, + claims, + } +} + +fn timestamp(time: SystemTime) -> String { + crate::exec_backend::rfc3339_utc(time).unwrap_or_else(|_| "1970-01-01T00:00:00.000Z".into()) +} + +#[cfg(test)] +mod tests { + use std::time::{Duration, UNIX_EPOCH}; + + use super::snapshot_at; + + #[test] + fn expiry_is_derived_from_the_capture_time_and_ttl() { + let snapshot = snapshot_at( + std::path::Path::new("/definitely/missing/st2-catalog"), + "host", + Duration::from_secs(30), + UNIX_EPOCH + Duration::from_secs(86_400), + ); + + assert_eq!(snapshot.captured_at, "1970-01-02T00:00:00.000Z"); + assert_eq!(snapshot.expires_at, "1970-01-02T00:00:30.000Z"); + } +} diff --git a/tests/workspace_activity_cli.rs b/tests/workspace_activity_cli.rs new file mode 100644 index 00000000..689bcf6c --- /dev/null +++ b/tests/workspace_activity_cli.rs @@ -0,0 +1,218 @@ +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output}; + +fn fixture(pty_json: &str) -> (tempfile::TempDir, PathBuf, PathBuf, PathBuf) { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let workspace = tmp.path().join("workspace"); + let pty_root = tmp.path().join("pty-root"); + let bin = tmp.path().join("bin"); + fs::create_dir_all(catalog.join("agents/h/worker")).unwrap(); + fs::create_dir(&workspace).unwrap(); + fs::create_dir(&pty_root).unwrap(); + fs::create_dir(&bin).unwrap(); + fs::write( + catalog.join("catalog.kdl"), + format!( + "catalog {{ pty-root {:?} }}\n", + pty_root.display().to_string() + ), + ) + .unwrap(); + fs::write( + catalog.join("agents/h/worker/agent.kdl"), + format!( + "agent \"worker\" {{\n host \"h\"\n workspace {:?}\n pty \"agent\" {{ id \"h.worker\"; argv \"agent-bin\" }}\n}}\n", + workspace.display().to_string() + ), + ) + .unwrap(); + write_executable( + &bin.join("pty"), + &format!( + "#!/bin/sh\nprintf '%s\\n' '{}'\n", + pty_json.replace('\'', "'\"'\"'") + ), + ); + (tmp, catalog, workspace, bin) +} + +fn write_executable(path: &Path, body: &str) { + fs::write(path, body).unwrap(); + let mut permissions = fs::metadata(path).unwrap().permissions(); + permissions.set_mode(0o755); + fs::set_permissions(path, permissions).unwrap(); +} + +fn snapshot_with_ttl(catalog: &Path, bin: &Path, state: &Path, ttl: &str) -> Output { + Command::new(env!("CARGO_BIN_EXE_st2")) + .args([ + "workspace-activity", + "--host", + "h", + "--ttl", + ttl, + "--json", + "--catalog", + ]) + .arg(catalog) + .env("PATH", bin) + .env("XDG_STATE_HOME", state) + .env_remove("CATALOG") + .env_remove("ST_ROOT") + .env_remove("PTY_ROOT") + .output() + .unwrap() +} + +fn snapshot(catalog: &Path, bin: &Path, state: &Path) -> Output { + snapshot_with_ttl(catalog, bin, state, "30") +} + +#[test] +fn reports_sorted_active_workspace_claim_without_mutating_state() { + let (tmp, catalog, workspace, bin) = fixture( + r#"[{"name":"h.worker","status":"running","pid":77,"createdAt":"2026-08-13T10:00:00.000Z"}]"#, + ); + let output = snapshot(&catalog, &bin, &tmp.path().join("state")); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["schemaVersion"], "st2.workspace-activity.v1"); + assert_eq!(value["producer"], "st2"); + assert_eq!(value["complete"], true); + assert_eq!( + value["claims"][0]["workspace"], + workspace.display().to_string() + ); + assert_eq!( + value["claims"][0]["agents"], + serde_json::json!(["h.worker"]) + ); + assert_eq!( + value["claims"][0]["activeRuntimeIds"], + serde_json::json!(["h.worker"]) + ); + assert_eq!(value["claims"][0]["active"], true); + assert!(value["capturedAt"].as_str().unwrap().ends_with('Z')); + assert!(value["expiresAt"].as_str().unwrap().ends_with('Z')); + assert!(!tmp.path().join("state").exists()); +} + +#[test] +fn runtime_observer_failure_prints_incomplete_envelope_and_fails() { + let (tmp, catalog, _workspace, bin) = fixture("not-json"); + let output = snapshot(&catalog, &bin, &tmp.path().join("state")); + assert!(!output.status.success()); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["complete"], false); + assert!(!value["errors"].as_array().unwrap().is_empty()); +} + +#[test] +fn accepts_ttl_boundaries() { + let (tmp, catalog, _workspace, bin) = fixture("[]"); + for ttl in ["1", "300"] { + let output = snapshot_with_ttl(&catalog, &bin, &tmp.path().join("state"), ttl); + assert!( + output.status.success(), + "TTL {ttl} was unexpectedly rejected" + ); + } +} + +#[test] +fn rejects_zero_and_overlong_ttls_with_an_incomplete_envelope() { + let (tmp, catalog, _workspace, bin) = fixture("[]"); + for ttl in ["0", "301"] { + let output = snapshot_with_ttl(&catalog, &bin, &tmp.path().join("state"), ttl); + assert!(!output.status.success(), "TTL {ttl} unexpectedly succeeded"); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["complete"], false); + assert_eq!(value["expiresAt"], value["capturedAt"]); + assert!( + value["errors"][0] + .as_str() + .unwrap() + .contains("TTL must be between") + ); + } +} + +#[test] +fn catalog_discovery_failure_prints_incomplete_envelope_and_fails() { + let (tmp, catalog, _workspace, bin) = fixture("[]"); + fs::write( + catalog.join("agents/h/worker/agent.kdl"), + "not valid kdl {{", + ) + .unwrap(); + + let output = snapshot(&catalog, &bin, &tmp.path().join("state")); + assert!(!output.status.success()); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["complete"], false); + assert!( + value["errors"][0] + .as_str() + .unwrap() + .contains("catalog file") + ); +} + +#[test] +fn duplicate_runtime_ids_print_incomplete_envelope_and_fail() { + let (tmp, catalog, workspace, bin) = fixture("[]"); + let duplicate_dir = catalog.join("agents/h/duplicate"); + fs::create_dir_all(&duplicate_dir).unwrap(); + fs::write( + duplicate_dir.join("agent.kdl"), + format!( + "agent \"duplicate\" {{\n host \"h\"\n workspace {:?}\n pty \"agent\" {{ id \"h.worker\"; argv \"agent-bin\" }}\n}}\n", + workspace.display().to_string() + ), + ) + .unwrap(); + + let output = snapshot(&catalog, &bin, &tmp.path().join("state")); + assert!(!output.status.success()); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["complete"], false); + assert!( + value["errors"] + .as_array() + .unwrap() + .iter() + .any(|error| error.as_str().unwrap().contains("duplicate runtime id")) + ); +} + +#[test] +fn relative_workspace_is_resolved_from_the_declaring_spec_directory() { + let (tmp, catalog, workspace, bin) = fixture("[]"); + let declaration = catalog.join("agents/h/worker/agent.kdl"); + let relative_workspace = declaration.parent().unwrap().join("relative-workspace"); + fs::create_dir(&relative_workspace).unwrap(); + let contents = fs::read_to_string(&declaration).unwrap().replace( + &format!("workspace {:?}", workspace.display().to_string()), + "workspace \"relative-workspace\"", + ); + fs::write(declaration, contents).unwrap(); + + let output = snapshot(&catalog, &bin, &tmp.path().join("state")); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!( + value["claims"][0]["workspace"], + relative_workspace.display().to_string() + ); +}