From 16ee7eb953508801a4603edd056ab32f28986601 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Wed, 29 Jul 2026 11:07:19 +0200 Subject: [PATCH] prototype: add content-addressed catalog admission agent-session-id: cdd15622-c01a-4731-bbfe-ab8cce0cad54 agent-tool: Codex CLI agent-tool-version: 0.145.0 agent-model: unknown agent-runtime-profile: /nix/store/ph8rlhdj25mg71v81jsfzy6dq4xpcs9m-coding-agent-runtime-profile/share/coding-agents/profile.json agent-skills-manifest: /nix/store/lsykz8x5481xrpbgk280xh3pypk1c5jy-agent-skills-corpus/share/agent-skills/manifest.json tooling-profile: dotfiles@3649b53 --- ...sed-catalog-root-selects-admitted-seats.md | 109 ++ .../2026-07-29-content-addressed-catalog.md | 98 ++ src/agents.rs | 8 +- src/catalog_store.rs | 1439 +++++++++++++++++ src/eval_run.rs | 1 + src/lib.rs | 1 + src/main.rs | 291 +++- src/message.rs | 6 +- src/run.rs | 5 +- src/spec.rs | 9 +- tests/catalog_cli.rs | 104 ++ tests/reconcile.rs | 1 + 12 files changed, 2043 insertions(+), 29 deletions(-) create mode 100644 docs/vrs/.decisions/0002-content-addressed-catalog-root-selects-admitted-seats.md create mode 100644 docs/vrs/.experiments/2026-07-29-content-addressed-catalog.md create mode 100644 src/catalog_store.rs create mode 100644 tests/catalog_cli.rs diff --git a/docs/vrs/.decisions/0002-content-addressed-catalog-root-selects-admitted-seats.md b/docs/vrs/.decisions/0002-content-addressed-catalog-root-selects-admitted-seats.md new file mode 100644 index 00000000..869b30cd --- /dev/null +++ b/docs/vrs/.decisions/0002-content-addressed-catalog-root-selects-admitted-seats.md @@ -0,0 +1,109 @@ +# A content-addressed catalog root selects admitted Agent Spec seats + +Status: proposed + +## Context + +The recursive `agent.kdl` catalog makes declaration source, discovery path, and +mutable agent state share one directory. That is convenient for authored files, +but it cannot atomically select a fleet assembled from independently managed +Agent Specs. It also makes a dynamic agent manager choose between rewriting +Nix-managed declarations and projecting mutable symlinks into a recursively +discovered tree. + +The use case is a mixed catalog: static tooling may build and pin exact Agent +Spec bytes, while an agent manager may select or roll back seats at runtime. +Messages, context, status, and future state still need stable mutable paths when +the selected declaration object changes. + +This proposal does not change canonical authored KDL, the trusted-private-fleet +assumption, or the current recursive catalog. It incubates a second, +experimental resolution path with executable evidence. + +## Options + +| Option | Tradeoffs | +| --- | --- | +| Rewrite or symlink `agent.kdl` projections | Reuses discovery, but exposes partial multi-seat updates and conflates immutable source with mutable state. | +| One mutable head per seat | Gives seat-local CAS, but readers cannot name or validate one atomic fleet snapshot. | +| Immutable seat admissions selected by one complete catalog root | Adds object types and a globally contended root CAS, but gives one atomic visibility boundary and preserves stable resource paths. | + +## Proposed Decision + +Store exact Agent Spec bytes and immutable commits below the hidden +`.st2/catalog-v1` namespace. A `SeatAdmission` joins one exact Agent Spec ref +commit to one exact resource-binding commit. A parent-linked +`CatalogRootCommit` maps every bus id to its admission. After validating the +complete prospective graph, publish one mutable root head atomically. + +```text +exact KDL object <- ref commit ----\ + SeatAdmission <- CatalogRootCommit <- root +stable agent_dir <- binding commit / +``` + +`AgentSpec.path` is the immutable declaration source. `AgentSpec.agent_dir` is +the stable mutable state root used by messages, context, status, and runtime +state. Resolution does not create an `agent.kdl` projection. + +Static and dynamic managers use the same protocol: + +1. `prepare` imports exact bytes without changing selection. +2. `stage` publishes a ref commit and resource binding without changing the + selected root. +3. `admit` atomically selects one or many staged seats in a complete root. + +Manager fencing prevents a different manager from advancing an owned ref or +admitting ref/binding commits it does not own. The root's `manager` records the +transaction actor; it does not grant whole-root custody. A manager may CAS from +another manager's current root while preserving untouched foreign admissions +byte-identically. Operation ids make an acknowledged ref/root update replayable +after response loss. Rollback is a new parent-linked commit, not a head rewind. +Manager names are logical coordination labels under the trusted same-user +assumption; they are not authentication or authorization. + +## Validation Boundary + +Before moving the root head, st2 resolves every admission in the prospective +root and verifies: + +- every digest in the selected reachable graph and every referenced object; +- bus-id, host, identity, manager, and schema joins; +- exactly one explicit-host, explicit-identity declaration per object; +- active declarations lower to runnable Agent Specs; and +- every resource state path is a normal catalog-relative path; +- no state root is under reserved `.st2`, crosses an existing symlink + component, or is shared by two selected seats. + +Readers observe the old or new complete root across the atomic head rename. +Test-scoped failpoints prove process-level visibility around that boundary and +operation replay after response loss. They do not prove power-loss durability +for every filesystem, mount, kernel, or storage device. + +## Consequences and Limits + +- Nix can manage immutable source objects while an agent manager owns only + admission, without requiring a private projection directory. +- Mutable messages, context, and status remain ordinary files at stable + `agent_dir` paths; they are not content addressed. +- State roots are seat-exclusive, catalog-relative, outside `.st2`, and may not + traverse an existing symlink component. +- Root publication serializes writers and copies the complete admission map. + Scaling, sharding, and compaction require evidence before changing this. +- Source-relative `render copy` inputs need a future immutable resource-bundle + contract. Inline render content works now; silently reading mutable files + adjacent to an object would weaken reproducibility. +- `prepare` content-addresses exact KDL bytes, not a self-contained closure. + Workspaces, templates, hooks, and other referenced inputs are not captured. +- "Immutable" means content-addressed protocol publication: st2 refuses + replacement and verifies selected bytes under trusted same-user store + custody. Verification and later use are not sealed into one file descriptor; + verified-FD use or stronger filesystem sealing remains future hardening. +- Validation covers the selected reachable graph. Parent links record lineage + and replay identity, but ancestor history is not recursively audited. +- Discovery integration, GC, replication, replacement semantics, daemon + sockets, typed resource contracts, and an authorization framework are + explicitly outside this proposal. + +Acceptance requires the experiment record to remain green and a separate human +decision. Until then, the JSON CLI and on-disk schema are experimental. diff --git a/docs/vrs/.experiments/2026-07-29-content-addressed-catalog.md b/docs/vrs/.experiments/2026-07-29-content-addressed-catalog.md new file mode 100644 index 00000000..2ceb2fdc --- /dev/null +++ b/docs/vrs/.experiments/2026-07-29-content-addressed-catalog.md @@ -0,0 +1,98 @@ +# Experiment: content-addressed catalog admission + +Status: implemented prototype + +Date: 2026-07-29 + +Decision under test: +[`0002-content-addressed-catalog-root-selects-admitted-seats.md`](../.decisions/0002-content-addressed-catalog-root-selects-admitted-seats.md) + +## Question + +Can current st2 resolve and materialize exact content-addressed Agent Spec bytes while all +mutable agent resources remain at a stable `agent_dir`, and can it publish +multiple staged seats with one atomic root visibility change? + +## Prototype Surface + +Library: `src/catalog_store.rs` + +Experimental JSON CLI: + +```text +st2 --catalog ROOT catalog prepare SPEC +st2 --catalog ROOT catalog stage SPEC --manager M --state-relative PATH \ + --operation-id OP [--expected-ref COMMIT] [--binding-parent COMMIT] +st2 --catalog ROOT catalog admit REQUEST.json +st2 --catalog ROOT catalog publish SPEC --manager M --state-relative PATH \ + --operation-id OP [--expected-ref COMMIT] [--expected-root COMMIT] +st2 --catalog ROOT catalog head +st2 --catalog ROOT catalog inspect +``` + +`publish` is only a one-seat convenience composition of `stage` and `admit`. +`prepare` imports exact bytes and changes no ref or root. `stage` changes no +catalog root. An admit request is: + +```json +{ + "expectedRoot": null, + "manager": "eval", + "operationId": "run-42:root", + "selections": [ + { + "refCommit": "sha256-...", + "resourceBindingCommit": "sha256-..." + } + ] +} +``` + +## Executable Claims + +Focused tests cover: + +1. exact-byte object preservation, content-addressed source resolution, stable + message/context/status paths, inline materialization, and absence of a + projection; +2. atomic two-seat admission plus rejection of a cross-seat join without + changing the selected root; +3. a test-scoped failure after root-commit publication leaving the old root visible, + failure after head publication leaving the new root visible, and + operation-id replay bound to its original expected parent; +4. competing ref publishers producing one CAS winner plus manager fencing. +5. a dynamic manager adding its owned seat to a Nix-authored root while + preserving the untouched Nix admission byte-identically, and rejection when + it tries to admit the Nix-owned seat itself. +6. strict digest grammar before digest-derived paths, including traversal + negatives; +7. rejection of reserved, shared, or existing-symlink-crossing state roots; and +8. inspect resolving one captured root even when the selected head changes + between root capture and graph resolution. + +These are process-level atomic-visibility tests, not power-loss durability +proofs. The selected reachable graph is verified; commit ancestry is not +recursively audited. + +## Result + +Green in the isolated prototype worktree: 8 focused catalog-store tests, 1 CLI +transaction integration test, and all 157 library tests pass. This evidence +does not promote the proposed decision; acceptance remains a separate human +decision. + +## Known Gaps + +- The recursive discovery/reconcile path does not consume the experimental root. +- Full validation here means complete graph, digest, join, path, parse, and + runnable validation. The legacy validator's path-layout warnings and + host-local external filesystem checks are not yet adapted to immutable + object provenance. +- External `render copy` inputs are not bundled with the content-addressed + declaration. `prepare` stores exact KDL bytes, not a self-contained closure. +- Content-addressed publication and digest verification assume trusted + same-user store custody. Verify/use sealing through one file descriptor or + stronger filesystem mechanisms is future hardening. +- Manager strings are logical fencing labels, not authentication. +- No GC, replication, replacement API, daemon socket, typed resource contract, + public failpoint API, or authorization framework is included. diff --git a/src/agents.rs b/src/agents.rs index 5f45441b..1ac37e10 100644 --- a/src/agents.rs +++ b/src/agents.rs @@ -36,16 +36,16 @@ pub fn roster(catalog_root: &Path, this_host: &str) -> Vec { let mut rows: Vec = found .specs .iter() - .filter_map(|s| { - let agent_dir = s.path.parent()?; - Some(AgentRow { + .map(|s| { + let agent_dir = &s.agent_dir; + AgentRow { identity: s.bus_id(this_host), status: status::read_state(&status::status_path(agent_dir)), name: read_name(agent_dir), retired: s.retired, last_activity_ms: newest_mtime_ms(agent_dir), inbox: inbox_count(agent_dir), - }) + } }) .collect(); rows.sort_by(|a, b| a.identity.cmp(&b.identity)); diff --git a/src/catalog_store.rs b/src/catalog_store.rs new file mode 100644 index 00000000..e27d43a7 --- /dev/null +++ b/src/catalog_store.rs @@ -0,0 +1,1439 @@ +//! Experimental content-addressed Agent Spec catalog. +//! +//! Immutable declarations and commits live below `.st2/catalog-v1`. A single +//! mutable root head selects one complete catalog snapshot. No `agent.kdl` +//! projection is created: `AgentSpec::path` points at immutable source bytes, +//! while `AgentSpec::agent_dir` points at stable mutable state. + +use std::collections::BTreeMap; +use std::fs::{self, File, OpenOptions}; +use std::io::Write; +use std::os::fd::AsRawFd; +use std::path::{Component, Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; + +use anyhow::{Context, Result, bail}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; + +const SCHEMA: u32 = 1; +static TEMP_SERIAL: AtomicU64 = AtomicU64::new(0); + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PreparedSpec { + pub object: String, + pub host: String, + pub identity: String, + bytes: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RefCommit { + pub schema: u32, + pub parent: Option, + pub host: String, + pub identity: String, + pub manager: String, + pub target_object: Option, + pub operation_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RefHead { + pub commit: String, + pub value: RefCommit, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ResourceBindingCommit { + pub schema: u32, + pub parent: Option, + pub host: String, + pub identity: String, + pub manager: String, + pub state_relative: PathBuf, + pub operation_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ResourceBinding { + pub commit: String, + pub value: ResourceBindingCommit, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SeatAdmission { + pub schema: u32, + pub host: String, + pub identity: String, + pub ref_commit: String, + pub resource_binding_commit: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AdmissionObject { + pub digest: String, + pub value: SeatAdmission, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct CatalogRootCommit { + pub schema: u32, + pub parent: Option, + pub manager: String, + /// Bus id (`host.identity`) to immutable SeatAdmission digest. + pub admissions: BTreeMap, + /// Exact seat updates requested by this operation, for unambiguous replay. + pub updates: BTreeMap, + pub operation_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CatalogRoot { + pub commit: String, + pub value: CatalogRootCommit, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AdmissionSelection { + pub ref_commit: String, + pub resource_binding_commit: String, +} + +#[derive(Debug, Clone)] +pub struct ResolvedSeat { + pub root_commit: String, + pub admission: String, + pub ref_commit: String, + pub resource_binding_commit: String, + pub spec_object: String, + pub spec: crate::spec::AgentSpec, +} + +pub struct CatalogStore { + catalog_root: PathBuf, + store_root: PathBuf, +} + +impl CatalogStore { + pub fn new(catalog_root: impl Into) -> Self { + let catalog_root = catalog_root.into(); + let store_root = catalog_root.join(".st2").join("catalog-v1"); + Self { + catalog_root, + store_root, + } + } + + /// Parse and identify one exact KDL declaration without capturing referenced inputs. + pub fn prepare(&self, bytes: &[u8]) -> Result { + let text = std::str::from_utf8(bytes).context("Agent Spec is not UTF-8")?; + let mut raws = crate::kdl_format::parse_kdl(text)?; + if raws.len() != 1 { + bail!( + "a catalog object must contain exactly one agent declaration (found {})", + raws.len() + ); + } + let raw = raws.pop().expect("length checked"); + let identity = raw + .identity + .clone() + .filter(|value| !value.is_empty()) + .context("content-addressed declaration must carry identity in content")?; + let host = raw + .host + .clone() + .filter(|value| !value.is_empty()) + .context("content-addressed declaration must carry host in content")?; + if !raw.looks_like_spec() { + bail!("object is not an Agent Spec"); + } + let spec = raw.into_agent_spec( + identity.clone(), + Some(host.clone()), + PathBuf::from(""), + ); + if !spec.is_runnable() && !spec.retired { + bail!("active Agent Spec is not runnable"); + } + Ok(PreparedSpec { + object: digest("st2-agent-spec-object-v1", bytes), + host, + identity, + bytes: bytes.to_vec(), + }) + } + + /// Import exact bytes. An existing object is verified and never overwritten. + pub fn import(&self, prepared: &PreparedSpec) -> Result { + let path = self.object_path(&prepared.object)?; + publish_immutable(&path, &prepared.bytes)?; + if fs::read(&path)? != prepared.bytes { + bail!("immutable object collision at {}", path.display()); + } + Ok(path) + } + + /// Per-seat ref CAS. `expected = None` is create-only. + pub fn compare_and_set_ref( + &self, + host: &str, + identity: &str, + expected: Option<&str>, + manager: &str, + target_object: Option<&str>, + operation_id: &str, + ) -> Result { + validate_component("host", host)?; + validate_component("identity", identity)?; + validate_component("manager", manager)?; + validate_operation(operation_id)?; + validate_optional_digest(expected)?; + if let Some(object) = target_object + && !self.object_path(object)?.is_file() + { + bail!("target object '{object}' has not been imported"); + } + let _guard = FileLock::acquire(&self.ref_lock_path(host, identity))?; + let current = self.read_ref_head(host, identity)?; + if let Some(head) = ¤t + && head.value.operation_id == operation_id + && head.value.manager == manager + { + if head.value.parent.as_deref() == expected + && head.value.target_object.as_deref() == target_object + { + return Ok(head.clone()); + } + bail!("operation id replay does not match the original ref request"); + } + if current.as_ref().map(|head| head.commit.as_str()) != expected { + bail!( + "ref CAS conflict: expected {:?}, found {:?}", + expected, + current.as_ref().map(|head| head.commit.as_str()) + ); + } + if let Some(head) = ¤t + && head.value.manager != manager + { + bail!( + "manager conflict: ref is owned by '{}', caller is '{}'", + head.value.manager, + manager + ); + } + let value = RefCommit { + schema: SCHEMA, + parent: current.as_ref().map(|head| head.commit.clone()), + host: host.to_string(), + identity: identity.to_string(), + manager: manager.to_string(), + target_object: target_object.map(str::to_string), + operation_id: operation_id.to_string(), + }; + let bytes = encoded(&value)?; + let commit = digest("st2-agent-spec-ref-commit-v1", &bytes); + publish_immutable(&self.ref_commit_path(&commit)?, &bytes)?; + write_atomic( + &self.ref_head_path(host, identity), + format!("{commit}\n").as_bytes(), + )?; + Ok(RefHead { commit, value }) + } + + pub fn read_ref_head(&self, host: &str, identity: &str) -> Result> { + let Some(commit) = read_head_value(&self.ref_head_path(host, identity))? else { + return Ok(None); + }; + Ok(Some(self.read_ref_commit(&commit)?)) + } + + pub fn commit_resource_binding( + &self, + parent: Option<&str>, + host: &str, + identity: &str, + manager: &str, + state_relative: &Path, + operation_id: &str, + ) -> Result { + validate_component("host", host)?; + validate_component("identity", identity)?; + validate_component("manager", manager)?; + validate_relative_state_path(state_relative)?; + validate_state_path_custody(&self.catalog_root, state_relative)?; + validate_operation(operation_id)?; + validate_optional_digest(parent)?; + if let Some(parent) = parent { + let previous = self.read_resource_binding(parent)?; + if previous.value.host != host || previous.value.identity != identity { + bail!("resource binding parent belongs to a different seat"); + } + if previous.value.manager != manager { + bail!("resource binding manager conflict"); + } + } + let value = ResourceBindingCommit { + schema: SCHEMA, + parent: parent.map(str::to_string), + host: host.to_string(), + identity: identity.to_string(), + manager: manager.to_string(), + state_relative: state_relative.to_path_buf(), + operation_id: operation_id.to_string(), + }; + let bytes = encoded(&value)?; + let commit = digest("st2-resource-binding-commit-v1", &bytes); + publish_immutable(&self.binding_path(&commit)?, &bytes)?; + Ok(ResourceBinding { commit, value }) + } + + /// Atomically update one seat in the selected complete catalog. + pub fn admit( + &self, + expected_root: Option<&str>, + manager: &str, + selection: AdmissionSelection, + operation_id: &str, + ) -> Result { + self.admit_many(expected_root, manager, &[selection], operation_id) + } + + /// Atomically update multiple seats in one complete catalog root. + pub fn admit_many( + &self, + expected_root: Option<&str>, + manager: &str, + selections: &[AdmissionSelection], + operation_id: &str, + ) -> Result { + self.admit_many_with_hook(expected_root, manager, selections, operation_id, |_| Ok(())) + } + + fn admit_many_with_hook( + &self, + expected_root: Option<&str>, + manager: &str, + selections: &[AdmissionSelection], + operation_id: &str, + mut hook: F, + ) -> Result + where + F: FnMut(RootPublishPoint) -> Result<()>, + { + validate_component("manager", manager)?; + validate_operation(operation_id)?; + validate_optional_digest(expected_root)?; + if selections.is_empty() { + bail!("at least one seat selection is required"); + } + let mut updates = BTreeMap::new(); + for selection in selections { + let admission = self.prepare_admission(selection, manager)?; + let bus_id = format!("{}.{}", admission.value.host, admission.value.identity); + if updates + .insert(bus_id.clone(), admission.digest.clone()) + .is_some() + { + bail!("duplicate seat '{bus_id}' in publication"); + } + let bytes = encoded(&admission.value)?; + publish_immutable(&self.admission_path(&admission.digest)?, &bytes)?; + } + hook(RootPublishPoint::AdmissionsPersisted)?; + + let _guard = FileLock::acquire(&self.root_lock_path())?; + let current = self.read_root()?; + if let Some(root) = ¤t + && root.value.operation_id == operation_id + && root.value.manager == manager + { + if root.value.parent.as_deref() == expected_root && root.value.updates == updates { + return Ok(root.clone()); + } + bail!("operation id replay does not match the original root request"); + } + if current.as_ref().map(|root| root.commit.as_str()) != expected_root { + bail!( + "catalog root CAS conflict: expected {:?}, found {:?}", + expected_root, + current.as_ref().map(|root| root.commit.as_str()) + ); + } + let mut admissions = current + .as_ref() + .map(|root| root.value.admissions.clone()) + .unwrap_or_default(); + admissions.extend(updates.clone()); + let value = CatalogRootCommit { + schema: SCHEMA, + parent: current.as_ref().map(|root| root.commit.clone()), + manager: manager.to_string(), + admissions, + updates, + operation_id: operation_id.to_string(), + }; + + // The complete prospective graph is validated before it becomes visible. + self.resolve_root_value("", &value)?; + let bytes = encoded(&value)?; + let commit = digest("st2-catalog-root-commit-v1", &bytes); + publish_immutable(&self.root_commit_path(&commit)?, &bytes)?; + hook(RootPublishPoint::RootPersisted)?; + write_atomic(&self.root_head_path(), format!("{commit}\n").as_bytes())?; + hook(RootPublishPoint::HeadPublished)?; + Ok(CatalogRoot { commit, value }) + } + + pub fn read_root(&self) -> Result> { + let Some(commit) = read_head_value(&self.root_head_path())? else { + return Ok(None); + }; + let bytes = fs::read(self.root_commit_path(&commit)?) + .with_context(|| format!("catalog root references missing commit '{commit}'"))?; + let value: CatalogRootCommit = serde_json::from_slice(&bytes)?; + verify_digest("st2-catalog-root-commit-v1", &commit, &value)?; + validate_optional_digest(value.parent.as_deref())?; + for value in value.admissions.values().chain(value.updates.values()) { + validate_digest(value)?; + } + Ok(Some(CatalogRoot { commit, value })) + } + + pub fn resolve_root(&self) -> Result> { + Ok(self + .inspect_snapshot()? + .map(|(_, seats)| seats) + .unwrap_or_default()) + } + + /// Capture one selected root and resolve exactly that immutable snapshot. + pub fn inspect_snapshot(&self) -> Result)>> { + self.inspect_snapshot_with_hook(|| Ok(())) + } + + fn inspect_snapshot_with_hook( + &self, + hook: F, + ) -> Result)>> + where + F: FnOnce() -> Result<()>, + { + let Some(root) = self.read_root()? else { + return Ok(None); + }; + hook()?; + let seats = self.resolve_root_value(&root.commit, &root.value)?; + Ok(Some((root, seats))) + } + + fn resolve_root_value( + &self, + root_commit: &str, + root: &CatalogRootCommit, + ) -> Result> { + if root.schema != SCHEMA { + bail!("unsupported catalog root schema {}", root.schema); + } + let mut resolved = Vec::with_capacity(root.admissions.len()); + let mut state_paths = BTreeMap::::new(); + for (bus_id, admission_digest) in &root.admissions { + let admission = self.read_admission(admission_digest)?; + let expected_bus_id = format!("{}.{}", admission.value.host, admission.value.identity); + if *bus_id != expected_bus_id { + bail!( + "catalog root key '{bus_id}' does not match admission seat '{expected_bus_id}'" + ); + } + let spec_ref = self.read_ref_commit(&admission.value.ref_commit)?; + let binding = self.read_resource_binding(&admission.value.resource_binding_commit)?; + validate_join(&admission.value, &spec_ref.value, &binding.value)?; + validate_state_path_custody(&self.catalog_root, &binding.value.state_relative)?; + if let Some(other) = + state_paths.insert(binding.value.state_relative.clone(), bus_id.clone()) + { + bail!( + "seats '{other}' and '{bus_id}' share resource state path '{}'", + binding.value.state_relative.display() + ); + } + let object = spec_ref + .value + .target_object + .as_deref() + .context("admission references a tombstoned ref commit")?; + let object_path = self.object_path(object)?; + let bytes = fs::read(&object_path) + .with_context(|| format!("admission references missing object '{object}'"))?; + if digest("st2-agent-spec-object-v1", &bytes) != object { + bail!("Agent Spec object digest mismatch for '{object}'"); + } + let text = std::str::from_utf8(&bytes).context("Agent Spec object is not UTF-8")?; + let mut raws = crate::kdl_format::parse_kdl(text)?; + if raws.len() != 1 { + bail!("admitted Agent Spec must contain exactly one declaration"); + } + let raw = raws.pop().expect("length checked"); + if raw.identity.as_deref() != Some(&admission.value.identity) + || raw.host.as_deref() != Some(&admission.value.host) + { + bail!("Agent Spec content does not match admission seat '{bus_id}'"); + } + let mut spec = raw.into_agent_spec( + admission.value.identity.clone(), + Some(admission.value.host.clone()), + object_path, + ); + if !spec.is_runnable() && !spec.retired { + bail!("active admitted Agent Spec '{bus_id}' is not runnable"); + } + spec.agent_dir = self.catalog_root.join(&binding.value.state_relative); + resolved.push(ResolvedSeat { + root_commit: root_commit.to_string(), + admission: admission.digest, + ref_commit: spec_ref.commit, + resource_binding_commit: binding.commit, + spec_object: object.to_string(), + spec, + }); + } + Ok(resolved) + } + + fn prepare_admission( + &self, + selection: &AdmissionSelection, + manager: &str, + ) -> Result { + let spec_ref = self.read_ref_commit(&selection.ref_commit)?; + let binding = self.read_resource_binding(&selection.resource_binding_commit)?; + if spec_ref.value.manager != manager || binding.value.manager != manager { + bail!( + "manager '{manager}' cannot admit seat commits owned by '{}'", + spec_ref.value.manager + ); + } + let value = SeatAdmission { + schema: SCHEMA, + host: spec_ref.value.host.clone(), + identity: spec_ref.value.identity.clone(), + ref_commit: selection.ref_commit.clone(), + resource_binding_commit: selection.resource_binding_commit.clone(), + }; + validate_join(&value, &spec_ref.value, &binding.value)?; + if spec_ref.value.target_object.is_none() { + bail!("a tombstoned ref commit cannot be admitted"); + } + let bytes = encoded(&value)?; + Ok(AdmissionObject { + digest: digest("st2-seat-admission-v1", &bytes), + value, + }) + } + + fn read_ref_commit(&self, commit: &str) -> Result { + let bytes = fs::read(self.ref_commit_path(commit)?) + .with_context(|| format!("missing ref commit '{commit}'"))?; + let value: RefCommit = serde_json::from_slice(&bytes)?; + verify_digest("st2-agent-spec-ref-commit-v1", commit, &value)?; + validate_optional_digest(value.parent.as_deref())?; + validate_optional_digest(value.target_object.as_deref())?; + Ok(RefHead { + commit: commit.to_string(), + value, + }) + } + + fn read_resource_binding(&self, commit: &str) -> Result { + let bytes = fs::read(self.binding_path(commit)?) + .with_context(|| format!("missing resource binding commit '{commit}'"))?; + let value: ResourceBindingCommit = serde_json::from_slice(&bytes)?; + verify_digest("st2-resource-binding-commit-v1", commit, &value)?; + validate_optional_digest(value.parent.as_deref())?; + validate_relative_state_path(&value.state_relative)?; + Ok(ResourceBinding { + commit: commit.to_string(), + value, + }) + } + + fn read_admission(&self, digest_value: &str) -> Result { + let bytes = fs::read(self.admission_path(digest_value)?) + .with_context(|| format!("missing SeatAdmission '{digest_value}'"))?; + let value: SeatAdmission = serde_json::from_slice(&bytes)?; + verify_digest("st2-seat-admission-v1", digest_value, &value)?; + validate_digest(&value.ref_commit)?; + validate_digest(&value.resource_binding_commit)?; + Ok(AdmissionObject { + digest: digest_value.to_string(), + value, + }) + } + + fn object_path(&self, value: &str) -> Result { + validate_digest(value)?; + Ok(self + .store_root + .join("objects") + .join(format!("{value}.agent.kdl"))) + } + + fn ref_commit_path(&self, value: &str) -> Result { + validate_digest(value)?; + Ok(self + .store_root + .join("ref-commits") + .join(format!("{value}.json"))) + } + + fn ref_head_path(&self, host: &str, identity: &str) -> PathBuf { + self.store_root + .join("refs") + .join(host) + .join(identity) + .join("head") + } + + fn ref_lock_path(&self, host: &str, identity: &str) -> PathBuf { + self.store_root + .join("locks") + .join("refs") + .join(host) + .join(format!("{identity}.lock")) + } + + fn binding_path(&self, value: &str) -> Result { + validate_digest(value)?; + Ok(self + .store_root + .join("resource-bindings") + .join(format!("{value}.json"))) + } + + fn admission_path(&self, value: &str) -> Result { + validate_digest(value)?; + Ok(self + .store_root + .join("admissions") + .join(format!("{value}.json"))) + } + + fn root_commit_path(&self, value: &str) -> Result { + validate_digest(value)?; + Ok(self + .store_root + .join("root-commits") + .join(format!("{value}.json"))) + } + + fn root_head_path(&self) -> PathBuf { + self.store_root.join("root") + } + + fn root_lock_path(&self) -> PathBuf { + self.store_root.join("locks").join("root.lock") + } +} + +fn validate_join( + admission: &SeatAdmission, + spec_ref: &RefCommit, + binding: &ResourceBindingCommit, +) -> Result<()> { + if admission.schema != SCHEMA || spec_ref.schema != SCHEMA || binding.schema != SCHEMA { + bail!("unsupported catalog object schema"); + } + if spec_ref.host != admission.host + || spec_ref.identity != admission.identity + || binding.host != admission.host + || binding.identity != admission.identity + { + bail!("SeatAdmission contains a cross-seat join"); + } + if spec_ref.manager != binding.manager { + bail!("SeatAdmission joins commits owned by different managers"); + } + Ok(()) +} + +fn validate_component(label: &str, value: &str) -> Result<()> { + if value.is_empty() + || value == "." + || value == ".." + || value.contains('/') + || value.contains('\\') + { + bail!("invalid {label} '{value}'"); + } + Ok(()) +} + +fn validate_operation(value: &str) -> Result<()> { + if value.is_empty() { + bail!("operation id must not be empty"); + } + Ok(()) +} + +fn validate_digest(value: &str) -> Result<()> { + let Some(hex) = value.strip_prefix("sha256-") else { + bail!("invalid digest '{value}': expected sha256-<64 lowercase hex>"); + }; + if hex.len() != 64 + || !hex + .bytes() + .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte)) + { + bail!("invalid digest '{value}': expected sha256-<64 lowercase hex>"); + } + Ok(()) +} + +fn validate_optional_digest(value: Option<&str>) -> Result<()> { + if let Some(value) = value { + validate_digest(value)?; + } + Ok(()) +} + +fn validate_relative_state_path(path: &Path) -> Result<()> { + if path.as_os_str().is_empty() || path.is_absolute() { + bail!("resource state path must be a non-empty relative path"); + } + if path + .components() + .any(|component| !matches!(component, Component::Normal(_))) + { + bail!( + "resource state path '{}' must contain only normal relative components", + path.display() + ); + } + if path + .components() + .next() + .is_some_and(|component| component.as_os_str() == ".st2") + { + bail!("resource state path must not be under reserved .st2"); + } + Ok(()) +} + +fn validate_state_path_custody(catalog_root: &Path, relative: &Path) -> Result<()> { + validate_relative_state_path(relative)?; + let mut current = catalog_root.to_path_buf(); + for component in std::iter::once(None).chain( + relative + .components() + .map(|component| Some(component.as_os_str())), + ) { + if let Some(component) = component { + current.push(component); + } + match fs::symlink_metadata(¤t) { + Ok(metadata) if metadata.file_type().is_symlink() => { + bail!( + "resource state path crosses symlink component {}", + current.display() + ); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => break, + Err(error) => { + return Err(error) + .with_context(|| format!("inspecting state path {}", current.display())); + } + } + } + Ok(()) +} + +fn digest(domain: &str, bytes: &[u8]) -> String { + let mut hasher = Sha256::new(); + hasher.update(domain.as_bytes()); + hasher.update([0]); + hasher.update(bytes); + format!("sha256-{:x}", hasher.finalize()) +} + +fn encoded(value: &T) -> Result> { + let mut bytes = serde_json::to_vec_pretty(value)?; + bytes.push(b'\n'); + Ok(bytes) +} + +fn verify_digest(domain: &str, expected: &str, value: &T) -> Result<()> { + if digest(domain, &encoded(value)?) != expected { + bail!("digest mismatch for '{expected}'"); + } + Ok(()) +} + +fn read_head_value(path: &Path) -> Result> { + match fs::read_to_string(path) { + Ok(raw) => { + let value = raw.trim(); + if value.is_empty() { + bail!("empty catalog head at {}", path.display()); + } + validate_digest(value)?; + Ok(Some(value.to_string())) + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error.into()), + } +} + +fn temp_path(target: &Path) -> PathBuf { + let serial = TEMP_SERIAL.fetch_add(1, Ordering::Relaxed); + let name = target + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("value"); + target.with_file_name(format!(".{name}.tmp-{}-{serial}", std::process::id())) +} + +fn publish_immutable(path: &Path, bytes: &[u8]) -> Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + if path.exists() { + return if fs::read(path)? == bytes { + Ok(()) + } else { + bail!("refusing to replace immutable artifact {}", path.display()) + }; + } + let tmp = temp_path(path); + let result = (|| -> Result<()> { + let mut file = OpenOptions::new().write(true).create_new(true).open(&tmp)?; + file.write_all(bytes)?; + file.sync_all()?; + match fs::hard_link(&tmp, path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + if fs::read(path)? != bytes { + bail!("immutable publication collision at {}", path.display()); + } + } + Err(error) => return Err(error.into()), + } + sync_parent(path)?; + Ok(()) + })(); + let _ = fs::remove_file(&tmp); + result +} + +fn write_atomic(path: &Path, bytes: &[u8]) -> Result<()> { + let parent = path.parent().context("atomic target has no parent")?; + fs::create_dir_all(parent)?; + let tmp = temp_path(path); + let result = (|| -> Result<()> { + let mut file = OpenOptions::new().write(true).create_new(true).open(&tmp)?; + file.write_all(bytes)?; + file.sync_all()?; + fs::rename(&tmp, path)?; + sync_parent(path)?; + Ok(()) + })(); + if result.is_err() { + let _ = fs::remove_file(&tmp); + } + result +} + +fn sync_parent(path: &Path) -> Result<()> { + if let Some(parent) = path.parent() { + File::open(parent)?.sync_all()?; + } + Ok(()) +} + +struct FileLock(File); + +impl FileLock { + fn acquire(path: &Path) -> Result { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent)?; + } + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(path)?; + let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) }; + if result != 0 { + return Err(std::io::Error::last_os_error().into()); + } + Ok(Self(file)) + } +} + +impl Drop for FileLock { + fn drop(&mut self) { + unsafe { + libc::flock(self.0.as_raw_fd(), libc::LOCK_UN); + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RootPublishPoint { + AdmissionsPersisted, + RootPersisted, + HeadPublished, +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Barrier}; + + use super::*; + + fn spec(identity: &str, command: &str) -> Vec { + format!("agent \"{identity}\" {{\n host \"h\"\n command \"{command}\"\n}}\n").into_bytes() + } + + fn publish_seat( + store: &CatalogStore, + identity: &str, + command: &str, + operation: &str, + ) -> (RefHead, ResourceBinding) { + publish_seat_as(store, identity, command, "eval", operation) + } + + fn publish_seat_as( + store: &CatalogStore, + identity: &str, + command: &str, + manager: &str, + operation: &str, + ) -> (RefHead, ResourceBinding) { + let prepared = store.prepare(&spec(identity, command)).unwrap(); + store.import(&prepared).unwrap(); + let spec_ref = store + .compare_and_set_ref( + "h", + identity, + None, + manager, + Some(&prepared.object), + &format!("{operation}:ref"), + ) + .unwrap(); + let binding = store + .commit_resource_binding( + None, + "h", + identity, + manager, + Path::new(&format!("agents/h/{identity}")), + &format!("{operation}:binding"), + ) + .unwrap(); + (spec_ref, binding) + } + + #[test] + fn exact_bytes_resolve_with_stable_mutable_agent_state_without_projection() { + let tmp = tempfile::tempdir().unwrap(); + let workspace = tmp.path().join("workspace"); + fs::create_dir_all(&workspace).unwrap(); + let bytes = format!( + "agent \"worker\" {{\n host \"h\"\n workspace \"{}\"\n command \"sleep 10\"\n render {{ file \".st2/proof\" \"ok\" }}\n}}\n", + workspace.display() + ) + .into_bytes(); + let store = CatalogStore::new(tmp.path()); + let prepared = store.prepare(&bytes).unwrap(); + let object_path = store.import(&prepared).unwrap(); + let spec_ref = store + .compare_and_set_ref( + "h", + "worker", + None, + "dynamic", + Some(&prepared.object), + "op:ref", + ) + .unwrap(); + let binding = store + .commit_resource_binding( + None, + "h", + "worker", + "dynamic", + Path::new("agents/h/worker"), + "op:binding", + ) + .unwrap(); + store + .admit( + None, + "dynamic", + AdmissionSelection { + ref_commit: spec_ref.commit, + resource_binding_commit: binding.commit, + }, + "op:root", + ) + .unwrap(); + + let seats = store.resolve_root().unwrap(); + assert_eq!(seats.len(), 1); + let seat = &seats[0]; + assert_eq!(seat.spec.path, object_path); + assert_eq!(seat.spec.agent_dir, tmp.path().join("agents/h/worker")); + assert!(!seat.spec.agent_dir.join("agent.kdl").exists()); + + let inbox = crate::message::inbox_dir(&seat.spec.agent_dir); + crate::message::send_to_inbox(&inbox, "h.sender", None, None, &[], "hello").unwrap(); + assert_eq!( + crate::message::list_inbox(&inbox).unwrap()[0].body, + "hello\n" + ); + let context = crate::context::context_dir(&seat.spec.agent_dir); + crate::context::write_now(&context, "working").unwrap(); + assert_eq!( + crate::context::read(&context, crate::context::View::Now), + "working" + ); + let status = crate::status::status_path(&seat.spec.agent_dir); + crate::status::set_state(&status, crate::status::State::Available).unwrap(); + assert_eq!( + crate::status::read_state(&status), + crate::status::State::Available + ); + let before = fs::read(&object_path).unwrap(); + crate::materialize::materialize_agent(tmp.path(), &seat.spec, "h").unwrap(); + assert_eq!( + fs::read_to_string(workspace.join(".st2/proof")).unwrap(), + "ok" + ); + assert_eq!(fs::read(&object_path).unwrap(), before); + } + + #[test] + fn multi_seat_root_is_atomic_and_validates_the_complete_graph() { + let tmp = tempfile::tempdir().unwrap(); + let store = CatalogStore::new(tmp.path()); + let (a_ref, a_binding) = publish_seat(&store, "a", "sleep 1", "a"); + let (b_ref, b_binding) = publish_seat(&store, "b", "sleep 2", "b"); + let root = store + .admit_many( + None, + "eval", + &[ + AdmissionSelection { + ref_commit: a_ref.commit, + resource_binding_commit: a_binding.commit, + }, + AdmissionSelection { + ref_commit: b_ref.commit, + resource_binding_commit: b_binding.commit, + }, + ], + "root-ab", + ) + .unwrap(); + assert_eq!(root.value.admissions.len(), 2); + let seats = store.resolve_root().unwrap(); + assert_eq!( + seats + .iter() + .map(|seat| seat.spec.identity.as_str()) + .collect::>(), + ["a", "b"] + ); + + // A cross-seat ref/binding pair fails before root visibility changes. + assert!( + store + .admit( + Some(&root.commit), + "eval", + AdmissionSelection { + ref_commit: seats[0].ref_commit.clone(), + resource_binding_commit: seats[1].resource_binding_commit.clone(), + }, + "invalid", + ) + .is_err() + ); + assert_eq!(store.read_root().unwrap().unwrap(), root); + } + + #[test] + fn mixed_managers_update_only_owned_seats_and_preserve_foreign_admissions() { + let tmp = tempfile::tempdir().unwrap(); + let store = CatalogStore::new(tmp.path()); + let (nix_ref, nix_binding) = publish_seat_as(&store, "static", "sleep 1", "nix", "nix"); + let nix_selection = AdmissionSelection { + ref_commit: nix_ref.commit, + resource_binding_commit: nix_binding.commit, + }; + let nix_root = store + .admit(None, "nix", nix_selection.clone(), "root-nix") + .unwrap(); + let static_admission = nix_root.value.admissions["h.static"].clone(); + + let (dynamic_ref, dynamic_binding) = + publish_seat_as(&store, "dynamic", "sleep 2", "dynamic", "dynamic"); + let mixed_root = store + .admit( + Some(&nix_root.commit), + "dynamic", + AdmissionSelection { + ref_commit: dynamic_ref.commit, + resource_binding_commit: dynamic_binding.commit, + }, + "root-dynamic", + ) + .unwrap(); + assert_eq!(mixed_root.value.manager, "dynamic"); + assert_eq!( + mixed_root.value.admissions["h.static"], static_admission, + "untouched Nix admission must be preserved byte-identically" + ); + assert!(mixed_root.value.admissions.contains_key("h.dynamic")); + assert!( + store + .admit( + Some(&mixed_root.commit), + "dynamic", + nix_selection, + "foreign-seat", + ) + .is_err() + ); + assert_eq!(store.read_root().unwrap().unwrap(), mixed_root); + } + + #[test] + fn root_head_is_the_atomic_visibility_boundary_and_response_loss_replays() { + let tmp = tempfile::tempdir().unwrap(); + let store = CatalogStore::new(tmp.path()); + let (a_ref, binding) = publish_seat(&store, "worker", "sleep 1", "a"); + let root_a = store + .admit( + None, + "eval", + AdmissionSelection { + ref_commit: a_ref.commit.clone(), + resource_binding_commit: binding.commit.clone(), + }, + "root-a", + ) + .unwrap(); + let b = store.prepare(&spec("worker", "sleep 2")).unwrap(); + store.import(&b).unwrap(); + let b_ref = store + .compare_and_set_ref( + "h", + "worker", + Some(&a_ref.commit), + "eval", + Some(&b.object), + "b:ref", + ) + .unwrap(); + let selection = AdmissionSelection { + ref_commit: b_ref.commit, + resource_binding_commit: binding.commit, + }; + + let before_head = store.admit_many_with_hook( + Some(&root_a.commit), + "eval", + std::slice::from_ref(&selection), + "root-b-before", + |point| { + if point == RootPublishPoint::RootPersisted { + bail!("simulated crash"); + } + Ok(()) + }, + ); + assert!(before_head.is_err()); + assert_eq!(store.read_root().unwrap().unwrap(), root_a); + + let after_head = store.admit_many_with_hook( + Some(&root_a.commit), + "eval", + std::slice::from_ref(&selection), + "root-b", + |point| { + if point == RootPublishPoint::HeadPublished { + bail!("simulated response loss"); + } + Ok(()) + }, + ); + assert!(after_head.is_err()); + let visible = store.read_root().unwrap().unwrap(); + let replayed = store + .admit_many( + Some(&root_a.commit), + "eval", + &[selection.clone()], + "root-b", + ) + .unwrap(); + assert_eq!(replayed, visible); + assert!( + store + .admit_many(Some(&visible.commit), "eval", &[selection], "root-b",) + .is_err(), + "same operation id with a different expected parent is not a replay" + ); + } + + #[test] + fn ref_cas_has_one_winner_and_manager_fencing() { + let tmp = tempfile::tempdir().unwrap(); + let store = CatalogStore::new(tmp.path()); + let a = store.prepare(&spec("worker", "sleep 1")).unwrap(); + let b = store.prepare(&spec("worker", "sleep 2")).unwrap(); + store.import(&a).unwrap(); + store.import(&b).unwrap(); + let first = store + .compare_and_set_ref("h", "worker", None, "nix", Some(&a.object), "first") + .unwrap(); + let barrier = Arc::new(Barrier::new(3)); + let attempts = [a.object, b.object] + .into_iter() + .enumerate() + .map(|(index, object)| { + let barrier = Arc::clone(&barrier); + let root = tmp.path().to_path_buf(); + let expected = first.commit.clone(); + std::thread::spawn(move || { + barrier.wait(); + CatalogStore::new(root).compare_and_set_ref( + "h", + "worker", + Some(&expected), + "nix", + Some(&object), + &format!("race-{index}"), + ) + }) + }) + .collect::>(); + barrier.wait(); + let results = attempts + .into_iter() + .map(|thread| thread.join().unwrap()) + .collect::>(); + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + assert_eq!(results.iter().filter(|result| result.is_err()).count(), 1); + let current = store.read_ref_head("h", "worker").unwrap().unwrap(); + assert!( + store + .compare_and_set_ref( + "h", + "worker", + Some(¤t.commit), + "nix", + current.value.target_object.as_deref(), + ¤t.value.operation_id, + ) + .is_err(), + "same operation id with a different expected parent is not a replay" + ); + assert!( + store + .compare_and_set_ref("h", "worker", Some(¤t.commit), "other", None, "steal",) + .is_err() + ); + } + + #[test] + fn digest_grammar_is_checked_before_any_digest_derived_path() { + let tmp = tempfile::tempdir().unwrap(); + let store = CatalogStore::new(tmp.path()); + for invalid in [ + "../escape", + "sha256-../../escape", + "sha256-ABCDEF", + "sha256-abc", + "md5-0000000000000000000000000000000000000000000000000000000000000000", + ] { + assert!(store.object_path(invalid).is_err()); + assert!(store.ref_commit_path(invalid).is_err()); + assert!(store.binding_path(invalid).is_err()); + assert!(store.admission_path(invalid).is_err()); + assert!(store.root_commit_path(invalid).is_err()); + } + assert!( + store + .compare_and_set_ref( + "h", + "worker", + None, + "eval", + Some("../../escape"), + "bad-target", + ) + .is_err() + ); + assert!( + store + .commit_resource_binding( + Some("../../escape"), + "h", + "worker", + "eval", + Path::new("agents/h/worker"), + "bad-parent", + ) + .is_err() + ); + write_atomic(&store.root_head_path(), b"../../escape\n").unwrap(); + assert!(store.read_root().is_err()); + assert!(!tmp.path().join("escape").exists()); + } + + #[test] + fn state_roots_are_unique_non_reserved_and_do_not_cross_symlinks() { + use std::os::unix::fs::symlink; + + let tmp = tempfile::tempdir().unwrap(); + let store = CatalogStore::new(tmp.path()); + assert!( + store + .commit_resource_binding( + None, + "h", + "reserved", + "eval", + Path::new(".st2/agent-state"), + "reserved", + ) + .is_err() + ); + fs::create_dir_all(tmp.path().join("real-agents")).unwrap(); + symlink( + tmp.path().join("real-agents"), + tmp.path().join("linked-agents"), + ) + .unwrap(); + assert!( + store + .commit_resource_binding( + None, + "h", + "linked", + "eval", + Path::new("linked-agents/h/linked"), + "linked", + ) + .is_err() + ); + + let (a_ref, _) = publish_seat(&store, "a", "sleep 1", "a"); + let (b_ref, _) = publish_seat(&store, "b", "sleep 2", "b"); + let a_binding = store + .commit_resource_binding( + None, + "h", + "a", + "eval", + Path::new("shared-state"), + "a-shared", + ) + .unwrap(); + let b_binding = store + .commit_resource_binding( + None, + "h", + "b", + "eval", + Path::new("shared-state"), + "b-shared", + ) + .unwrap(); + assert!( + store + .admit_many( + None, + "eval", + &[ + AdmissionSelection { + ref_commit: a_ref.commit, + resource_binding_commit: a_binding.commit, + }, + AdmissionSelection { + ref_commit: b_ref.commit, + resource_binding_commit: b_binding.commit, + }, + ], + "duplicate-state", + ) + .is_err() + ); + assert!(store.read_root().unwrap().is_none()); + } + + #[test] + fn inspect_resolves_the_single_root_snapshot_it_captured() { + let tmp = tempfile::tempdir().unwrap(); + let store = CatalogStore::new(tmp.path()); + let (a_ref, binding) = publish_seat(&store, "worker", "sleep 1", "a"); + let root_a = store + .admit( + None, + "eval", + AdmissionSelection { + ref_commit: a_ref.commit.clone(), + resource_binding_commit: binding.commit.clone(), + }, + "root-a", + ) + .unwrap(); + let b = store.prepare(&spec("worker", "sleep 2")).unwrap(); + store.import(&b).unwrap(); + let b_ref = store + .compare_and_set_ref( + "h", + "worker", + Some(&a_ref.commit), + "eval", + Some(&b.object), + "b-ref", + ) + .unwrap(); + let selection = AdmissionSelection { + ref_commit: b_ref.commit, + resource_binding_commit: binding.commit, + }; + + let (captured, seats) = store + .inspect_snapshot_with_hook(|| { + store.admit(Some(&root_a.commit), "eval", selection, "root-b")?; + Ok(()) + }) + .unwrap() + .unwrap(); + assert_eq!(captured, root_a); + assert_eq!(seats[0].root_commit, root_a.commit); + assert!( + seats[0] + .spec + .tasks + .iter() + .any(|task| task.command.as_deref() == Some("sleep 1")) + ); + assert_ne!(store.read_root().unwrap().unwrap().commit, captured.commit); + } +} diff --git a/src/eval_run.rs b/src/eval_run.rs index 3395c80c..9e7732fe 100644 --- a/src/eval_run.rs +++ b/src/eval_run.rs @@ -96,6 +96,7 @@ pub fn spec_to_agent_specs(agents: &[SpecAgent], host: &str, root: &Path) -> Vec restart: None, tasks, path: path.clone(), + agent_dir: root.to_path_buf(), } }) .collect() diff --git a/src/lib.rs b/src/lib.rs index 9def522b..5e52be3e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ //! each declaration's command, environment, hooks, and workspace materialization block. pub mod agents; +pub mod catalog_store; pub mod compile_agent; pub mod context; pub mod ding; diff --git a/src/main.rs b/src/main.rs index e1ce6168..69206c07 100644 --- a/src/main.rs +++ b/src/main.rs @@ -65,6 +65,9 @@ enum Command { /// The stable wire format is a `-.md` Markdown file. #[command(subcommand)] Message(MessageCmd), + /// EXPERIMENTAL: content-addressed Agent Spec catalog operations. Every result is JSON. + #[command(subcommand)] + Catalog(CatalogCmd), /// An agent's working-state context for lossless restart: read/write/append. #[command(subcommand)] Context(ContextCmd), @@ -331,6 +334,56 @@ enum HooksCmd { Verify, } +#[derive(Subcommand)] +enum CatalogCmd { + /// Parse and persist exact Agent Spec bytes without changing any ref or root. + Prepare { + spec: PathBuf, + }, + /// Stage one immutable ref commit and resource-binding commit without changing the root. + Stage { + spec: PathBuf, + #[arg(long)] + manager: String, + #[arg(long)] + state_relative: PathBuf, + #[arg(long)] + operation_id: String, + #[arg(long)] + expected_ref: Option, + #[arg(long)] + binding_parent: Option, + }, + /// Print the selected immutable catalog root, or JSON null when absent. + Head, + /// Resolve and verify the complete selected catalog graph. + Inspect, + /// Import one exact spec, advance its ref and resource binding, then admit it atomically. + Publish { + spec: PathBuf, + #[arg(long)] + manager: String, + /// Stable mutable state directory relative to the catalog root. + #[arg(long)] + state_relative: PathBuf, + #[arg(long)] + operation_id: String, + #[arg(long)] + expected_ref: Option, + #[arg(long)] + expected_root: Option, + #[arg(long)] + binding_parent: Option, + }, + /// Atomically admit multiple already-published ref/binding pairs from a JSON request. + /// + /// Request shape: {expectedRoot?, manager, operationId, + /// selections:[{refCommit, resourceBindingCommit}]}. Use `-` for stdin. + Admit { + request: PathBuf, + }, +} + #[derive(Subcommand)] enum ResourceCmd { /// Link a resource (a URL you produced or reference) into your resource list. @@ -534,6 +587,10 @@ fn main() -> Result<()> { up(&root, host, once, materialize_only, interval, agent) } Command::Message(cmd) => message_cmd(cmd), + Command::Catalog(cmd) => { + let root = catalog_arg(None)?; + catalog_cmd(&root, cmd) + } Command::Context(cmd) => context_cmd(cmd), Command::Resource(cmd) => resource_cmd(cmd), Command::Service(cmd) => service_cmd(cmd), @@ -676,6 +733,206 @@ fn hooks_cmd(command: HooksCmd) -> Result<()> { Ok(()) } +#[derive(serde::Deserialize)] +#[serde(rename_all = "camelCase")] +struct CatalogAdmitRequest { + expected_root: Option, + manager: String, + operation_id: String, + selections: Vec, +} + +struct StagedCatalogSeat { + host: String, + identity: String, + object: String, + object_path: PathBuf, + ref_commit: String, + resource_binding_commit: String, +} + +#[allow(clippy::too_many_arguments)] +fn catalog_stage( + store: &st2::catalog_store::CatalogStore, + spec: &Path, + manager: &str, + state_relative: &Path, + operation_id: &str, + expected_ref: Option<&str>, + binding_parent: Option<&str>, +) -> Result { + let bytes = std::fs::read(spec) + .with_context(|| format!("reading Agent Spec {}", spec.display()))?; + let prepared = store.prepare(&bytes)?; + let object_path = store.import(&prepared)?; + let spec_ref = store.compare_and_set_ref( + &prepared.host, + &prepared.identity, + expected_ref, + manager, + Some(&prepared.object), + &format!("{operation_id}:ref"), + )?; + let binding = store.commit_resource_binding( + binding_parent, + &prepared.host, + &prepared.identity, + manager, + state_relative, + &format!("{operation_id}:binding"), + )?; + Ok(StagedCatalogSeat { + host: prepared.host, + identity: prepared.identity, + object: prepared.object, + object_path, + ref_commit: spec_ref.commit, + resource_binding_commit: binding.commit, + }) +} + +fn catalog_cmd(root: &Path, command: CatalogCmd) -> Result<()> { + use st2::catalog_store::{AdmissionSelection, CatalogStore}; + + let store = CatalogStore::new(root); + let value = match command { + CatalogCmd::Prepare { spec } => { + let bytes = std::fs::read(&spec) + .with_context(|| format!("reading Agent Spec {}", spec.display()))?; + let prepared = store.prepare(&bytes)?; + let object_path = store.import(&prepared)?; + serde_json::json!({ + "experimental": true, + "object": prepared.object, + "objectPath": object_path, + "host": prepared.host, + "identity": prepared.identity, + "bytes": bytes.len(), + }) + } + CatalogCmd::Stage { + spec, + manager, + state_relative, + operation_id, + expected_ref, + binding_parent, + } => { + let staged = catalog_stage( + &store, + &spec, + &manager, + &state_relative, + &operation_id, + expected_ref.as_deref(), + binding_parent.as_deref(), + )?; + serde_json::json!({ + "experimental": true, + "host": staged.host, + "identity": staged.identity, + "object": staged.object, + "objectPath": staged.object_path, + "refCommit": staged.ref_commit, + "resourceBindingCommit": staged.resource_binding_commit, + "rootChanged": false, + }) + } + CatalogCmd::Head => match store.read_root()? { + Some(root) => serde_json::json!({ + "experimental": true, + "commit": root.commit, + "value": root.value, + }), + None => serde_json::Value::Null, + }, + CatalogCmd::Inspect => { + let snapshot = store.inspect_snapshot()?; + let (root, seats) = match snapshot { + Some(snapshot) => (Some(snapshot.0), snapshot.1), + None => (None, Vec::new()), + }; + serde_json::json!({ + "experimental": true, + "rootCommit": root.as_ref().map(|root| &root.commit), + "manager": root.as_ref().map(|root| &root.value.manager), + "seats": seats.into_iter().map(|seat| serde_json::json!({ + "busId": seat.spec.bus_id(""), + "admission": seat.admission, + "refCommit": seat.ref_commit, + "resourceBindingCommit": seat.resource_binding_commit, + "specObject": seat.spec_object, + "sourcePath": seat.spec.path, + "agentDir": seat.spec.agent_dir, + })).collect::>(), + }) + } + CatalogCmd::Publish { + spec, + manager, + state_relative, + operation_id, + expected_ref, + expected_root, + binding_parent, + } => { + let staged = catalog_stage( + &store, + &spec, + &manager, + &state_relative, + &operation_id, + expected_ref.as_deref(), + binding_parent.as_deref(), + )?; + let root_commit = store.admit( + expected_root.as_deref(), + &manager, + AdmissionSelection { + ref_commit: staged.ref_commit.clone(), + resource_binding_commit: staged.resource_binding_commit.clone(), + }, + &format!("{operation_id}:root"), + )?; + serde_json::json!({ + "experimental": true, + "host": staged.host, + "identity": staged.identity, + "object": staged.object, + "objectPath": staged.object_path, + "refCommit": staged.ref_commit, + "resourceBindingCommit": staged.resource_binding_commit, + "rootCommit": root_commit.commit, + }) + } + CatalogCmd::Admit { request } => { + let bytes = if request == Path::new("-") { + use std::io::Read as _; + let mut bytes = Vec::new(); + std::io::stdin().read_to_end(&mut bytes)?; + bytes + } else { + std::fs::read(&request) + .with_context(|| format!("reading admission request {}", request.display()))? + }; + let request: CatalogAdmitRequest = serde_json::from_slice(&bytes)?; + let root_commit = store.admit_many( + request.expected_root.as_deref(), + &request.manager, + &request.selections, + &request.operation_id, + )?; + serde_json::json!({ + "experimental": true, + "rootCommit": root_commit.commit, + "seats": root_commit.value.admissions.len(), + }) + } + }; + println!("{}", serde_json::to_string_pretty(&value)?); + Ok(()) +} + fn down_cmd(root: &Path, host: Option) -> Result<()> { // A single-file team spec: tear down the DECLARED team's sessions (symmetric with `st2 up`/`st2 ls` // over a spec — the "stop the fleet cleanly" verb). A catalog dir falls through to catalog teardown. @@ -1039,24 +1296,22 @@ fn doctor_cmd(root: &Path, host: Option, require_supervisor: bool) -> Re "session dead/missing", ); } - if let Some(dir) = spec.path.parent() { - let path = st2::status::status_path(dir); - if !path.is_file() { - report_check( - &mut problems, - false, - &format!("{bus_id} presence missing"), - "no status file — is its ding refreshing?", - ); - } else { - let state = st2::status::read_state(&path); - report_check( - &mut problems, - state != st2::status::State::Unknown, - &format!("{bus_id} presence fresh (is `{}`)", state.as_str()), - "rotted to `unknown` — is its ding refreshing?", - ); - } + let path = st2::status::status_path(&spec.agent_dir); + if !path.is_file() { + report_check( + &mut problems, + false, + &format!("{bus_id} presence missing"), + "no status file — is its ding refreshing?", + ); + } else { + let state = st2::status::read_state(&path); + report_check( + &mut problems, + state != st2::status::State::Unknown, + &format!("{bus_id} presence fresh (is `{}`)", state.as_str()), + "rotted to `unknown` — is its ding refreshing?", + ); } } diff --git a/src/message.rs b/src/message.rs index 892a6109..51c8a4ac 100644 --- a/src/message.rs +++ b/src/message.rs @@ -343,7 +343,7 @@ pub fn resolve_agent_dir(catalog_root: &Path, recipient: &str, this_host: &str) .specs .into_iter() .find(|s| s.bus_id(this_host) == recipient || s.identity == recipient) - .and_then(|s| s.path.parent().map(Path::to_path_buf)) + .map(|s| s.agent_dir) } /// The default subject for a reply to a message whose subject was `original`: the original prefixed @@ -379,9 +379,7 @@ pub fn collect_thread(catalog_root: &Path, filename: &str) -> Vec { let found = crate::discover(catalog_root); let mut all: HashMap = HashMap::new(); for spec in &found.specs { - let Some(dir) = spec.path.parent() else { - continue; - }; + let dir = &spec.agent_dir; for d in [inbox_dir(dir), archive_dir(dir)] { for m in list_dir(&d).unwrap_or_default() { all.entry(m.filename.clone()).or_insert(m); diff --git a/src/run.rs b/src/run.rs index 6785a53a..0f6225d1 100644 --- a/src/run.rs +++ b/src/run.rs @@ -595,7 +595,7 @@ pub fn execute( } for launch in &plan.launch { - let spec_dir = launch.spec.path.parent().unwrap_or_else(|| Path::new(".")); + let spec_dir = &launch.spec.agent_dir; let policy = launch.spec.restart_policy(); for target in &launch.tasks { let now = Instant::now(); @@ -842,7 +842,7 @@ fn gate_codex_launches<'a, V, F>( }) else { continue; }; - let spec_dir = launch.spec.path.parent().unwrap_or_else(|| Path::new(".")); + let spec_dir = &launch.spec.agent_dir; let workspace = resolve_task_cwd(agent, spec_dir, catalog_root); if !workspaces.contains(&workspace) { workspaces.push(workspace); @@ -1335,6 +1335,7 @@ mod tests { restart: None, tasks: vec![], path: std::path::PathBuf::from("/x"), + agent_dir: std::path::PathBuf::from("/"), } } diff --git a/src/spec.rs b/src/spec.rs index a34ce341..2d5821f4 100644 --- a/src/spec.rs +++ b/src/spec.rs @@ -41,8 +41,10 @@ pub struct AgentSpec { pub restart: Option, /// The runnable tasks (`pty` + `exec`), sorted by name for determinism. pub tasks: Vec, - /// Where this spec was loaded from — the anchor for its resources and for edits. + /// Immutable declaration source used for render parsing and source-relative inputs. pub path: PathBuf, + /// Stable mutable state root used for resources, status, and the default task cwd. + pub agent_dir: PathBuf, } /// The kind of job. Only `service` (long-running) remains — `type = batch` is retired; the native @@ -317,6 +319,10 @@ impl RawSpec { // `service` is the only job type; a stray `type` string is caught by validate (unknown-type). let job_type = JobType::Service; + let agent_dir = path + .parent() + .unwrap_or_else(|| std::path::Path::new(".")) + .to_path_buf(); AgentSpec { identity, host, @@ -329,6 +335,7 @@ impl RawSpec { restart: self.restart.map(RawRestart::lower), tasks, path, + agent_dir, } } } diff --git a/tests/catalog_cli.rs b/tests/catalog_cli.rs new file mode 100644 index 00000000..bb670061 --- /dev/null +++ b/tests/catalog_cli.rs @@ -0,0 +1,104 @@ +//! End-to-end coverage for the experimental JSON catalog transaction surface. + +use std::fs; +use std::path::Path; +use std::process::{Command, Output}; + +fn run(root: &Path, args: &[&str]) -> Output { + Command::new(env!("CARGO_BIN_EXE_st2")) + .arg("--catalog") + .arg(root) + .args(["catalog"]) + .args(args) + .output() + .unwrap() +} + +fn json(output: Output) -> serde_json::Value { + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice(&output.stdout).unwrap() +} + +fn write_spec(path: &Path, identity: &str) { + fs::write( + path, + format!("agent \"{identity}\" {{\n host \"h\"\n command \"sleep 10\"\n}}\n"), + ) + .unwrap(); +} + +#[test] +fn prepare_stage_then_atomic_multi_seat_admit_has_explicit_visibility_boundaries() { + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path().join("catalog"); + fs::create_dir_all(&root).unwrap(); + let a = tmp.path().join("a.kdl"); + let b = tmp.path().join("b.kdl"); + write_spec(&a, "a"); + write_spec(&b, "b"); + + let prepared = json(run(&root, &["prepare", a.to_str().unwrap()])); + assert!(Path::new(prepared["objectPath"].as_str().unwrap()).is_file()); + assert_eq!(json(run(&root, &["head"])), serde_json::Value::Null); + + let staged_a = json(run( + &root, + &[ + "stage", + a.to_str().unwrap(), + "--manager", + "eval", + "--state-relative", + "agents/h/a", + "--operation-id", + "a", + ], + )); + let staged_b = json(run( + &root, + &[ + "stage", + b.to_str().unwrap(), + "--manager", + "eval", + "--state-relative", + "agents/h/b", + "--operation-id", + "b", + ], + )); + assert_eq!(staged_a["rootChanged"], false); + assert_eq!(staged_b["rootChanged"], false); + assert_eq!(json(run(&root, &["head"])), serde_json::Value::Null); + + let request = tmp.path().join("admit.json"); + fs::write( + &request, + serde_json::to_vec_pretty(&serde_json::json!({ + "expectedRoot": null, + "manager": "eval", + "operationId": "root-ab", + "selections": [ + { + "refCommit": staged_a["refCommit"], + "resourceBindingCommit": staged_a["resourceBindingCommit"], + }, + { + "refCommit": staged_b["refCommit"], + "resourceBindingCommit": staged_b["resourceBindingCommit"], + } + ] + })) + .unwrap(), + ) + .unwrap(); + let admitted = json(run(&root, &["admit", request.to_str().unwrap()])); + assert_eq!(admitted["seats"], 2); + let inspected = json(run(&root, &["inspect"])); + assert_eq!(inspected["rootCommit"], admitted["rootCommit"]); + assert_eq!(inspected["seats"].as_array().unwrap().len(), 2); +} diff --git a/tests/reconcile.rs b/tests/reconcile.rs index 5eca7d12..68bac51b 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -33,6 +33,7 @@ fn spec(identity: &str, host: Option<&str>, job_type: JobType, retired: bool, ta restart: None, tasks, path: PathBuf::from(format!("/cat/agents/{}/{identity}/agent.kdl", host.unwrap_or("this"))), + agent_dir: PathBuf::from(format!("/cat/agents/{}/{identity}", host.unwrap_or("this"))), } }