diff --git a/src/agent_author.rs b/src/agent_author.rs index 566d757b..acbc95d9 100644 --- a/src/agent_author.rs +++ b/src/agent_author.rs @@ -1720,7 +1720,7 @@ fn verify_desired_state_candidate( )) } -fn exact_agent_node<'a>( +pub(crate) fn exact_agent_node<'a>( document: &'a KdlDocument, expected_identity: &str, expected_host: &str, @@ -1764,7 +1764,7 @@ fn exact_agent_node<'a>( } } -fn agent_identity_parts(node: &KdlNode) -> (Option, Option) { +pub(crate) fn agent_identity_parts(node: &KdlNode) -> (Option, Option) { let mut identity = node .get(0) .and_then(|value| value.as_string()) @@ -1793,7 +1793,7 @@ fn agent_identity_parts(node: &KdlNode) -> (Option, Option) { (host, identity) } -fn is_nix_managed(node: &KdlNode) -> bool { +pub(crate) fn is_nix_managed(node: &KdlNode) -> bool { node.children().is_some_and(|children| { children .nodes() @@ -1855,7 +1855,7 @@ fn parse_field_value(node: &KdlNode, field: PresentationField) -> Result<&str, A }) } -fn quoted(value: &str) -> Result { +pub(crate) fn quoted(value: &str) -> Result { serde_json::to_string(value).map_err(|error| { AuthorError::new( "unsafe-source-edit", @@ -1900,7 +1900,7 @@ fn insert_field( ) } -fn insert_node(text: &str, target: &KdlNode, authored: &str) -> Result { +pub(crate) fn insert_node(text: &str, target: &KdlNode, authored: &str) -> Result { let span = target.span(); let start = span.offset(); let end = start + span.len(); @@ -2038,7 +2038,7 @@ fn remove_field(text: &str, node: &KdlNode) -> Result { Ok(replacement) } -fn line_indent(text: &str, offset: usize) -> Option { +pub(crate) fn line_indent(text: &str, offset: usize) -> Option { let prefix = text.get(..offset)?; let start = prefix.rfind('\n').map_or(0, |newline| newline + 1); let indent = prefix.get(start..)?; diff --git a/src/catalog_archive.rs b/src/catalog_archive.rs index 1cac25f3..1f795f1f 100644 --- a/src/catalog_archive.rs +++ b/src/catalog_archive.rs @@ -84,10 +84,18 @@ pub struct UnarchiveRequest { } /// The durable trace an archived identity leaves in the live catalog. +/// +/// The schema stays `st2.catalog-archive-tombstone.v1` across the addition of `agentId`: an added +/// optional field changes no existing field's meaning, and this struct does not +/// `deny_unknown_fields`, so a reader of either vintage parses a record of either vintage. Compare +/// `docs/vrs/05-harness-state/spec.md:15-30`, where a version bump IS required — there an existing +/// field's meaning changes, which no tolerant reader can absorb. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct Tombstone { pub schema: String, + /// The legacy `.` bus identity this subject was archived under. Still the + /// tombstone's positional key; see `agent_id` for the immutable subject ID. pub id: String, pub host: String, pub identity: String, @@ -96,6 +104,11 @@ pub struct Tombstone { pub reason: Option, /// Catalog-relative location of the moved directory, so a moved catalog stays readable. pub archive_root: String, + /// The archived subject's immutable agent ID (R24), frozen from the archived declaration's + /// explicit `id`. `None` on a tombstone written before ID migration reached this subject — + /// including every tombstone written by a pre-DELTA-003 st2. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub agent_id: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -108,6 +121,8 @@ pub struct ArchivedEntry { pub to: String, pub archived_at: u64, pub reason: Option, + /// The archived subject's immutable agent ID (R24), mirroring its tombstone's `agentId`. + pub agent_id: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -162,6 +177,8 @@ struct Candidate { identity: String, reason: Option, from: PathBuf, + /// The declaration's explicit `id` (R24), threaded to the tombstone and the receipt row. + agent_id: Option, } /// Archive every selected identity under one exclusive authoring lock and one generation commit. @@ -319,10 +336,37 @@ pub fn unarchive(request: UnarchiveRequest) -> Result { ); let tombstone_path = host_root.join(format!("{}{TOMBSTONE_SUFFIX}", request.identity)); - let archived_at = read_tombstone(&tombstone_path) - .ok() - .flatten() - .map(|tombstone| tombstone.archived_at); + // An absent or unreadable tombstone stays tolerated: `unarchive` is the exact reverse move + // even for an archived directory the control plane can no longer explain. + let tombstone = read_tombstone(&tombstone_path).ok().flatten(); + let archived_at = tombstone.as_ref().map(|tombstone| tombstone.archived_at); + if let Some(expected) = tombstone.as_ref().and_then(|it| it.agent_id.as_deref()) { + // A migrated tombstone froze this subject's immutable agent ID (R24). The declaration it + // covers must still carry the same one, or restoring it would re-enter the live plane + // under an identity the catalog never recorded for this subject. + let declaration = from.join("agent.kdl"); + let shown = relative(&catalog, &declaration) + .unwrap_or_else(|| declaration.display().to_string()); + let declared = agent_spec::discovery::parse_declared(&declaration).with_context(|| { + format!("read {shown} to check its agent ID against its tombstone's agentId") + })?; + let declared_id = match declared.as_slice() { + [single] => single.id.as_deref(), + many => anyhow::bail!( + "{shown} declares {} agents; a tombstoned identity must be one canonical declaration", + many.len() + ), + }; + anyhow::ensure!( + declared_id == Some(expected), + "tombstone records agent ID '{expected}' but {shown} declares {}; repair one of them before unarchiving", + declared_id.map_or_else(|| "no explicit id".to_owned(), |id| format!("'{id}'")) + ); + } + // DELTA-003: refusing an UNMIGRATED archive (a tombstone with no `agentId`) after ID migration + // has activated, and validating the restored ID's uniqueness against the prospective + // live-and-archived set, both land with the writer PR (D3). Until then a legacy tombstone + // restores exactly as it always did. let parent = to.parent().context("live identity path has no parent")?; fs::create_dir_all(parent) @@ -613,6 +657,10 @@ fn plan( identity, reason: spec.desired_state.reason().map(str::to_owned), from, + // A migrated declaration carries its immutable ID explicitly; a legacy one has none + // yet, and freezing `.` here is the migration verb's job, not this + // one's — archival must not mint identity. + agent_id: spec.id.clone(), }); } refused.sort_by(|left, right| left.id.cmp(&right.id)); @@ -662,6 +710,7 @@ fn move_out(catalog: &Path, root: &Path, candidate: &Candidate, archived_at: u64 archived_at, reason: candidate.reason.clone(), archive_root: relative(catalog, &to).context("archive destination escaped the catalog")?, + agent_id: candidate.agent_id.clone(), }; let mut body = serde_json::to_vec_pretty(&tombstone)?; body.push(b'\n'); @@ -691,6 +740,7 @@ fn entry(catalog: &Path, candidate: &Candidate, archived_at: u64) -> ArchivedEnt to: relative(catalog, &to).unwrap_or_else(|| to.display().to_string()), archived_at, reason: candidate.reason.clone(), + agent_id: candidate.agent_id.clone(), } } @@ -874,6 +924,9 @@ fn read_tombstone(path: &Path) -> Result> { "unknown archive tombstone schema '{}'", tombstone.schema ); + // `id` is the legacy bus-identity key, so it must still agree with the pair it is composed + // of: a record where they disagree is corrupt, whatever its vintage. `agent_id` is the + // immutable one and is deliberately unconstrained by this pair. anyhow::ensure!( tombstone.id == format!("{}.{}", tombstone.host, tombstone.identity), "archive tombstone id does not match its host and identity" @@ -1024,4 +1077,31 @@ mod tests { "only the local supervisor may reconcile its own host's rows" ); } + + /// The claim that keeps `TOMBSTONE_SCHEMA` at v1: a record of either vintage parses, and a key + /// this build has never heard of is tolerated rather than rejected. + #[test] + fn a_tombstone_parses_without_an_agent_id_and_with_an_unknown_key() { + let legacy: Tombstone = serde_json::from_str( + r#"{"schema":"st2.catalog-archive-tombstone.v1","id":"h.gone","host":"h", + "identity":"gone","archivedAt":7,"reason":null,"archiveRoot":".st2/archive/h/gone"}"#, + ) + .unwrap(); + assert_eq!(legacy.agent_id, None); + + let forward: Tombstone = serde_json::from_str( + r#"{"schema":"st2.catalog-archive-tombstone.v1","id":"h.gone","host":"h", + "identity":"gone","archivedAt":7,"reason":null,"archiveRoot":".st2/archive/h/gone", + "agentId":"0199b8f4-8d3a-7c21-9a44-6f85b7320ea1","reassignedFrom":"h.gone"}"#, + ) + .unwrap(); + assert_eq!( + forward.agent_id.as_deref(), + Some("0199b8f4-8d3a-7c21-9a44-6f85b7320ea1") + ); + + // And the round trip of a legacy record re-emits exactly the legacy shape. + let bytes = serde_json::to_string(&legacy).unwrap(); + assert!(!bytes.contains("agentId"), "{bytes}"); + } } diff --git a/src/catalog_graph.rs b/src/catalog_graph.rs index 8a9109bd..ed351a99 100644 --- a/src/catalog_graph.rs +++ b/src/catalog_graph.rs @@ -37,6 +37,10 @@ pub struct GraphArchived { pub reason: Option, /// Catalog-relative location of the moved identity directory. pub archive_root: String, + /// The archived subject's immutable agent ID (R24), projected verbatim from the tombstone's + /// `agentId`. `null` for a tombstone written before ID migration reached this subject. Newest + /// field in the row and additive, so `CATALOG_GRAPH_SCHEMA` does not move. + pub agent_id: Option, } #[derive(Debug, Serialize)] @@ -220,6 +224,7 @@ pub fn snapshot(root: &Path, this_host: &str) -> Result { archived_at: tombstone.archived_at, reason: tombstone.reason, archive_root: tombstone.archive_root, + agent_id: tombstone.agent_id, }) .collect(); diff --git a/src/catalog_migrate_ids.rs b/src/catalog_migrate_ids.rs new file mode 100644 index 00000000..fc365be3 --- /dev/null +++ b/src/catalog_migrate_ids.rs @@ -0,0 +1,1317 @@ +//! `st2 catalog migrate-ids` — freeze every legacy subject's immutable agent ID in one transaction. +//! +//! Decision 0015 separates the immutable catalog-global agent ID from the mutable address that the +//! positional `identity` currently serves as. Migration is the step that makes an existing catalog +//! expressible in that model without re-keying any durable state: a live subject freezes its +//! existing `.` bus identity as its explicit `id`, so every runtime identifier, +//! task ID, socket path, and declaration-anchored state path keeps its exact bytes. A structurally +//! archived subject freezes the same bytes when they remain unique across the combined +//! live-and-archived subject set; an archived collision receives a generated UUIDv7 in both its +//! declaration and its tombstone, and the reassignment is recorded durably so a reader of a +//! version-1 durable record never retypes colliding bytes into the wrong subject. +//! +//! Supervisor references resolve against the combined *pre-migration* index and are rewritten to +//! the parent's migrated ID inside the same transaction. A missing or ambiguous reference refuses +//! before any write: the operator unarchives and repairs that declaration through the ordinary +//! pre-activation authoring path, then retries. +//! +//! The transaction is the same shape `catalog_archive` uses — one exclusive authoring lock, one +//! strict discovery, a pure plan, then every write inside one generation commit. A durable marker +//! makes an interrupted run resumable rather than indeterminate. + +use std::collections::{BTreeMap, BTreeSet}; +use std::fs; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use kdl::{KdlDocument, KdlNode}; +use serde::{Deserialize, Serialize}; + +use crate::agent_author::{ + agent_identity_parts, exact_agent_node, insert_node, is_nix_managed, quoted, +}; +use crate::catalog_archive::{self, TOMBSTONE_SCHEMA, Tombstone}; +use crate::catalog_lock::{CONTROL_DIR, CatalogLock}; +use crate::catalog_transaction; + +pub const MIGRATE_SCHEMA: &str = "st2.catalog-migrate-ids.v1"; +/// The durable record of every legacy bus identity migration reassigned. +pub const MIGRATION_RECORD_SCHEMA: &str = "st2.agent-id-migration.v1"; +pub const MARKER_SCHEMA: &str = "st2.catalog-migrate-ids-incomplete.v1"; + +const MARKER_FILE: &str = "migrate-ids-incomplete"; +const MIGRATION_RECORD_FILE: &str = "agent-id-migration.json"; +const TOMBSTONE_SUFFIX: &str = ".tombstone.json"; +const DECLARATION_STEMS: [&str; 1] = ["agent"]; +const KDL_EXTENSION: &str = "kdl"; +const FOREIGN_EXTENSIONS: [&str; 2] = ["toml", "json"]; + +#[derive(Debug, Clone)] +pub struct MigrateRequest { + pub catalog: PathBuf, + /// Host used to freeze a `.` bus identity for a declaration that omits `host`, + /// and to resolve a bare-identity supervisor reference. + pub host: String, + pub dry_run: bool, + pub resume: bool, +} + +/// Which plane a subject's declaration lives in. Both are migrated in the same transaction. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Plane { + Live, + Archived, +} + +/// Where a subject's frozen ID came from. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum IdSource { + /// The subject's existing `.` bus identity, frozen verbatim. + FrozenBusIdentity, + /// A generated UUIDv7, because the bus identity was already claimed in the combined set. + GeneratedUuidV7, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PlannedId { + pub agent_id: String, + pub host: String, + pub identity: String, + pub plane: Plane, + pub source: IdSource, + /// Catalog-relative declaration path, so a moved catalog stays readable. + pub declaration: String, +} + +/// One legacy bus identity that migration could not give to both claimants. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Reassignment { + pub legacy_bus_identity: String, + /// The subject that kept the colliding bytes as its own immutable ID. + pub kept_by_agent_id: String, + pub kept_by_plane: Plane, + /// The immutable ID the other claimant received instead. + pub reassigned_agent_id: String, + pub reassigned_host: String, + pub reassigned_identity: String, + pub reassigned_plane: Plane, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SupervisorRewrite { + pub declaration: String, + pub host: String, + pub identity: String, + /// The legacy reference as written. + pub from: String, + /// The parent's migrated immutable ID. + pub to_agent_id: String, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum MigrateStatus { + Migrated, + Unchanged, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct MigrateResult { + pub schema: &'static str, + pub host: String, + pub dry_run: bool, + pub resumed: bool, + pub status: MigrateStatus, + /// The catalog generation after the transaction, or the observed one for a dry run. + pub generation: Option, + pub assigned: Vec, + pub reassigned: Vec, + pub supervisor_rewrites: Vec, + /// Subjects that already carry an explicit `id`; their bytes are left untouched. + pub already_migrated: Vec, + /// Migrated declarations that carry `meta { managed-by "nix" }`. Their upstream generator must + /// emit `id` before the next activation re-projects the file without one. + pub nix_owned: Vec, +} + +/// The durable reassignment record consulted by readers of version-1 durable records. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MigrationRecord { + pub schema: String, + pub migrated_at_ms: u64, + pub reassigned: Vec, +} + +/// Read the durable reassignment record, if this catalog has one. +/// +/// An absent record means no legacy bus identity was reassigned, which is the ordinary case: it is +/// written only when an archived subject's bytes were already claimed. A record whose schema this +/// version does not own is an error rather than an empty answer — silently reading it as "nothing +/// was reassigned" would let a reader retype colliding legacy bytes into the wrong subject, which +/// is exactly what the record exists to prevent. +pub fn read_migration_record(catalog: &Path) -> Result> { + let path = migration_record_path(catalog); + let metadata = match fs::symlink_metadata(&path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => { + return Err(error) + .with_context(|| format!("inspect agent-id migration record {}", path.display())); + } + }; + anyhow::ensure!( + metadata.is_file() && !metadata.file_type().is_symlink(), + "agent-id migration record is not a real regular file" + ); + let bytes = fs::read(&path) + .with_context(|| format!("read agent-id migration record {}", path.display()))?; + let record: MigrationRecord = + serde_json::from_slice(&bytes).context("parse agent-id migration record")?; + anyhow::ensure!( + record.schema == MIGRATION_RECORD_SCHEMA, + "unknown agent-id migration record schema '{}'", + record.schema + ); + Ok(Some(record)) +} + +/// The immutable ID a version-1 durable record's legacy endpoint denotes for `state_owner`. +/// +/// Migration froze most legacy bus identities as the ID of the subject that already held them, so +/// those bytes need no translation. A reassigned identity denotes two subjects, and the record's +/// own state owner is the only endpoint whose subject is provable: the sender for a sender-owned +/// row, the recipient for an inbox row. Any other colliding endpoint is unattributed +/// (`MESSAGE-R04`), and this returns `None` for it so the caller renders the legacy bytes as a +/// historical address rather than addressing the live replacement. +pub fn attribute_legacy_endpoint( + record: Option<&MigrationRecord>, + endpoint: &str, + state_owner: Option<&str>, +) -> Option { + let reassigned = record.and_then(|record| { + record + .reassigned + .iter() + .find(|entry| entry.legacy_bus_identity == endpoint) + }); + let Some(reassigned) = reassigned else { + // Not reassigned: the bytes are the keeping subject's frozen ID. + return Some(endpoint.to_owned()); + }; + match state_owner { + Some(owner) if owner == endpoint => Some(reassigned.kept_by_agent_id.clone()), + _ => None, + } +} + +pub fn migration_record_path(catalog: &Path) -> PathBuf { + catalog.join(CONTROL_DIR).join(MIGRATION_RECORD_FILE) +} + +pub fn marker_path(catalog: &Path) -> PathBuf { + catalog.join(CONTROL_DIR).join(MARKER_FILE) +} + +/// One subject in the combined pre-migration index. +#[derive(Debug, Clone)] +struct Subject { + plane: Plane, + host: String, + identity: String, + declaration: PathBuf, + declared_id: Option, + supervisor: Option, + nix_owned: bool, + /// The tombstone beside an archived subject's directory. + tombstone: Option, +} + +impl Subject { + fn bus_identity(&self) -> String { + format!("{}.{}", self.host, self.identity) + } +} + +/// One text edit to apply to one agent node. +#[derive(Debug, Clone, PartialEq, Eq)] +enum Edit { + /// Insert `id ""` into an agent that has none. + InsertId(String), + /// Rewrite `supervisor ""` to the parent's migrated ID. + RewriteSupervisor { from: String, to: String }, +} + +/// Every edit one declaration file needs, keyed by the agent node it applies to. +#[derive(Debug, Clone, Default)] +struct FileEdits { + /// `(host, identity, edit)` — the node selector plus what to do to it. + edits: Vec<(String, String, Edit)>, +} + +#[derive(Debug, Clone, Default)] +struct Plan { + assigned: Vec, + already_migrated: Vec, + reassigned: Vec, + supervisor_rewrites: Vec, + nix_owned: Vec, + files: BTreeMap, + /// `(tombstone path, agent id)` for every archived subject whose tombstone must record its ID. + tombstones: Vec<(PathBuf, String)>, +} + +impl Plan { + fn is_empty(&self) -> bool { + self.files.is_empty() && self.tombstones.is_empty() + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct Marker { + schema: String, + catalog: String, + host: String, + started_at_ms: u64, + /// Every ID this transaction intends to freeze, so a resume cannot silently plan another one. + assigned: Vec, + reassigned: Vec, + supervisor_rewrites: Vec, +} + +/// Migrate every live and structurally archived legacy subject in one transaction. +pub fn migrate_ids(request: MigrateRequest) -> Result { + anyhow::ensure!( + !(request.dry_run && request.resume), + "--dry-run and --resume are mutually exclusive" + ); + anyhow::ensure!( + !request.host.is_empty() && !request.host.contains('/') && !request.host.contains('.'), + "host '{}' must be one non-empty path component without `.`", + request.host + ); + let catalog = request + .catalog + .canonicalize() + .with_context(|| format!("canonicalize catalog {}", request.catalog.display()))?; + let lock = CatalogLock::exclusive(&catalog)?; + + let existing_marker = read_marker(&catalog)?; + if let Some(marker) = existing_marker.as_ref() { + anyhow::ensure!( + request.resume || request.dry_run, + "a previous `st2 catalog migrate-ids` did not complete; rerun with --resume (marker: {})", + marker_path(&catalog).display() + ); + anyhow::ensure!( + marker.catalog == catalog.display().to_string(), + "migration marker names catalog '{}', not {}", + marker.catalog, + catalog.display() + ); + } else { + anyhow::ensure!( + !request.resume, + "nothing to resume: no migration marker at {}", + marker_path(&catalog).display() + ); + } + + let subjects = collect_subjects(&catalog, &request.host)?; + let plan = plan(&catalog, &request.host, &subjects)?; + + if let Some(marker) = existing_marker.as_ref() { + verify_resumable(marker, &plan)?; + } + + if request.dry_run || plan.is_empty() { + return Ok(MigrateResult { + schema: MIGRATE_SCHEMA, + host: request.host, + dry_run: request.dry_run, + resumed: request.resume, + status: if plan.is_empty() { + MigrateStatus::Unchanged + } else { + MigrateStatus::Migrated + }, + generation: lock.generation()?, + assigned: plan.assigned, + reassigned: plan.reassigned, + supervisor_rewrites: plan.supervisor_rewrites, + already_migrated: plan.already_migrated, + nix_owned: plan.nix_owned, + }); + } + + // Prove the catalog admits BEFORE rewriting a byte. Migration re-admits the whole live plane + // after its writes, and a plane that already fails admission would fail that re-admission for + // a reason migration did not cause — leaving every declaration rewritten, the generation + // unmoved, and a marker whose resume can only fail the same way. Refusing up front keeps the + // pre-existing fault the operator's to repair, exactly as `catalog apply` does. + catalog_transaction::validate_full_catalog(&catalog).context( + "refusing to migrate: the catalog does not currently admit, so a rewritten plane could not be re-admitted either; repair the declarations named above and retry", + )?; + + if existing_marker.is_none() { + write_marker( + &lock, + &catalog, + &Marker { + schema: MARKER_SCHEMA.to_owned(), + catalog: catalog.display().to_string(), + host: request.host.clone(), + started_at_ms: crate::message::now_ms(), + assigned: plan.assigned.clone(), + reassigned: plan.reassigned.clone(), + supervisor_rewrites: plan.supervisor_rewrites.clone(), + }, + )?; + } + + let generation = lock.begin_generation_commit()?; + // Declarations before tombstones: a crash between them leaves a migrated declaration whose + // tombstone has not caught up, which `--resume` finishes. The opposite order would leave a + // tombstone advertising an ID its declaration does not carry. + for (path, edits) in &plan.files { + apply_file_edits(&lock, &catalog, path, edits)?; + test_checkpoint("migrate-ids-declaration-written"); + } + for (path, agent_id) in &plan.tombstones { + apply_tombstone_id(&lock, &catalog, path, agent_id)?; + } + if !plan.reassigned.is_empty() { + write_migration_record( + &lock, + &catalog, + &MigrationRecord { + schema: MIGRATION_RECORD_SCHEMA.to_owned(), + migrated_at_ms: crate::message::now_ms(), + reassigned: plan.reassigned.clone(), + }, + )?; + } + // Re-admit the whole live plane before the generation moves: a migration that produced an + // inadmissible catalog must fail with its marker intact rather than publish a broken plane. + catalog_transaction::validate_full_catalog(&catalog) + .context("migrated catalog fails full validation")?; + reparse_archived(&plan)?; + generation.commit()?; + clear_marker(&catalog)?; + + Ok(MigrateResult { + schema: MIGRATE_SCHEMA, + host: request.host, + dry_run: false, + resumed: request.resume, + status: MigrateStatus::Migrated, + generation: lock.generation()?, + assigned: plan.assigned, + reassigned: plan.reassigned, + supervisor_rewrites: plan.supervisor_rewrites, + already_migrated: plan.already_migrated, + nix_owned: plan.nix_owned, + }) +} + +/// Build the combined pre-migration live-and-archived subject index. +/// +/// Both halves must be complete. A declaration that failed to parse could be the one naming a +/// subject as its `supervisor`, and an unexplained archive entry could be a subject whose bytes a +/// live freeze would then steal — so partial input refuses rather than migrating a guess. +fn collect_subjects(catalog: &Path, host: &str) -> Result> { + let found = crate::discover_strict(catalog); + anyhow::ensure!( + found.errors.is_empty(), + "refusing to migrate: catalog discovery is incomplete, so a subject or supervisor reference could be hidden:\n{}", + found + .errors + .iter() + .map(|error| format!(" {}: {}", error.path.display(), error.message)) + .collect::>() + .join("\n") + ); + + let mut subjects = Vec::new(); + for spec in &found.specs { + let declaration = spec.path.clone(); + subjects.push(Subject { + plane: Plane::Live, + host: spec.resolved_host(host).to_owned(), + identity: spec.identity.clone(), + declaration: declaration.clone(), + declared_id: spec.id.clone(), + supervisor: spec.supervisor.clone(), + nix_owned: declaration_is_nix_owned(&declaration, &spec.identity)?, + tombstone: None, + }); + } + + let observation = catalog_archive::observe(catalog)?; + anyhow::ensure!( + observation.issues.is_empty(), + "refusing to migrate: the structural archive has unexplained state, so a frozen ID could collide with a subject this run cannot see:\n{}", + observation + .issues + .iter() + .map(|issue| format!(" {}: {}", issue.path, issue.message)) + .collect::>() + .join("\n") + ); + let archive_root = catalog_archive::archive_root(catalog); + for tombstone in &observation.archived { + let directory = archive_root.join(&tombstone.host).join(&tombstone.identity); + let declaration = archived_declaration(&directory).with_context(|| { + format!( + "locate the archived declaration of {}.{}", + tombstone.host, tombstone.identity + ) + })?; + let (declared_id, supervisor, nix_owned) = archived_declaration_facts( + &declaration, + &tombstone.host, + &tombstone.identity, + )?; + subjects.push(Subject { + plane: Plane::Archived, + host: tombstone.host.clone(), + identity: tombstone.identity.clone(), + declaration, + declared_id, + supervisor, + nix_owned, + tombstone: Some( + archive_root + .join(&tombstone.host) + .join(format!("{}{TOMBSTONE_SUFFIX}", tombstone.identity)), + ), + }); + } + Ok(subjects) +} + +/// The canonical declaration inside an archived identity directory. +fn archived_declaration(directory: &Path) -> Result { + for stem in DECLARATION_STEMS { + let kdl = directory.join(format!("{stem}.{KDL_EXTENSION}")); + if kdl.is_file() { + return Ok(kdl); + } + for extension in FOREIGN_EXTENSIONS { + let foreign = directory.join(format!("{stem}.{extension}")); + if foreign.is_file() { + return Ok(foreign); + } + } + } + anyhow::bail!( + "archived identity {} holds no `agent.{{kdl,toml,json}}` declaration", + directory.display() + ) +} + +/// `(declared id, supervisor, nix-owned)` of one archived declaration. +/// +/// Archived declarations are structurally undiscoverable — `.st2` is excluded at every depth — so +/// they are read directly rather than through catalog discovery. +fn archived_declaration_facts( + declaration: &Path, + host: &str, + identity: &str, +) -> Result<(Option, Option, bool)> { + if declaration.extension().and_then(|ext| ext.to_str()) != Some(KDL_EXTENSION) { + // A foreign-format archived declaration is readable but not editable here; `plan` turns + // that into a refusal only if the subject actually needs an edit. + return Ok((None, None, false)); + } + let text = fs::read_to_string(declaration) + .with_context(|| format!("read archived declaration {}", declaration.display()))?; + let document = KdlDocument::parse(&text) + .map_err(|error| anyhow::anyhow!("parse {}: {error}", declaration.display()))?; + let node = exact_agent_node(&document, identity, host, identity) + .map_err(|error| anyhow::anyhow!("{}: {error}", declaration.display()))?; + Ok(( + child_string(node, "id"), + child_string(node, "supervisor"), + is_nix_managed(node), + )) +} + +fn child_string(node: &KdlNode, name: &str) -> Option { + node.children() + .into_iter() + .flat_map(|children| children.nodes()) + .find(|child| child.name().value() == name) + .and_then(|child| child.get(0)) + .and_then(|value| value.as_string()) + .map(str::to_owned) +} + +fn declaration_is_nix_owned(declaration: &Path, identity: &str) -> Result { + if declaration.extension().and_then(|ext| ext.to_str()) != Some(KDL_EXTENSION) { + return Ok(false); + } + let text = fs::read_to_string(declaration) + .with_context(|| format!("read declaration {}", declaration.display()))?; + let document = KdlDocument::parse(&text) + .map_err(|error| anyhow::anyhow!("parse {}: {error}", declaration.display()))?; + Ok(document + .nodes() + .iter() + .filter(|node| node.name().value() == "agent") + .filter(|node| { + let (_, declared) = agent_identity_parts(node); + declared.as_deref() == Some(identity) || declared.is_none() + }) + .any(is_nix_managed)) +} + +/// Decide every ID and every rewrite without touching a byte. +fn plan(catalog: &Path, host: &str, subjects: &[Subject]) -> Result { + // Live first: a live subject always keeps its own bus identity, so its claim is decided before + // any archived subject can compete for the same bytes. + let mut ordered: Vec<&Subject> = subjects.iter().collect(); + ordered.sort_by(|left, right| { + left.plane + .cmp(&right.plane) + .then_with(|| left.host.cmp(&right.host)) + .then_with(|| left.identity.cmp(&right.identity)) + }); + + let mut claimed: BTreeMap = BTreeMap::new(); + // Explicit IDs already in the catalog are claimed before anything is frozen: a legacy freeze + // may never take an ID a migrated subject already owns. + for subject in &ordered { + if let Some(id) = subject.declared_id.as_deref() { + if let Some((_, plane)) = claimed.insert(id.to_owned(), (id.to_owned(), subject.plane)) { + anyhow::bail!( + "refusing to migrate: agent id '{id}' is declared by more than one subject (second: {} {}.{}, first plane {plane:?})", + plane_label(subject.plane), + subject.host, + subject.identity + ); + } + } + } + + let mut plan = Plan::default(); + let mut assigned_ids: BTreeMap = BTreeMap::new(); + for subject in &ordered { + let bus_identity = subject.bus_identity(); + let planned = PlannedId { + agent_id: String::new(), + host: subject.host.clone(), + identity: subject.identity.clone(), + plane: subject.plane, + source: IdSource::FrozenBusIdentity, + declaration: relative(catalog, &subject.declaration), + }; + if let Some(id) = subject.declared_id.as_deref() { + plan.already_migrated.push(PlannedId { + agent_id: id.to_owned(), + ..planned + }); + assigned_ids.insert(bus_identity, id.to_owned()); + continue; + } + let (agent_id, source) = match claimed.get(&bus_identity) { + None => (bus_identity.clone(), IdSource::FrozenBusIdentity), + Some((kept_by, kept_plane)) => { + anyhow::ensure!( + subject.plane == Plane::Archived, + "refusing to migrate: live subject {bus_identity} cannot freeze its bus identity because '{kept_by}' is already claimed; resolve the duplicate declaration first" + ); + let generated = crate::uuid_v7::uuid_v7()?; + plan.reassigned.push(Reassignment { + legacy_bus_identity: bus_identity.clone(), + kept_by_agent_id: kept_by.clone(), + kept_by_plane: *kept_plane, + reassigned_agent_id: generated.clone(), + reassigned_host: subject.host.clone(), + reassigned_identity: subject.identity.clone(), + reassigned_plane: subject.plane, + }); + (generated, IdSource::GeneratedUuidV7) + } + }; + agent_spec::validate_agent_id(&agent_id).with_context(|| { + format!( + "refusing to migrate {}: the frozen id is not a usable agent id", + subject.declaration.display() + ) + })?; + anyhow::ensure!( + claimed + .insert(agent_id.clone(), (agent_id.clone(), subject.plane)) + .is_none(), + "refusing to migrate: generated agent id '{agent_id}' collides with an existing one" + ); + assigned_ids.insert(bus_identity, agent_id.clone()); + + refuse_foreign_format(&subject.declaration)?; + plan.files + .entry(subject.declaration.clone()) + .or_default() + .edits + .push(( + subject.host.clone(), + subject.identity.clone(), + Edit::InsertId(agent_id.clone()), + )); + if subject.nix_owned { + plan.nix_owned.push(relative(catalog, &subject.declaration)); + } + if let Some(tombstone) = subject.tombstone.as_ref() { + plan.tombstones + .push((tombstone.clone(), agent_id.clone())); + } + plan.assigned.push(PlannedId { + agent_id, + source, + ..planned + }); + } + + // Supervisor references resolve against the combined pre-migration index — bus identities as + // written — and are rewritten to the parent's migrated ID. + let index: BTreeMap = subjects + .iter() + .map(|subject| (subject.bus_identity(), subject)) + .collect(); + for subject in &ordered { + let Some(reference) = subject.supervisor.as_deref() else { + continue; + }; + let parent = resolve_supervisor(&index, reference, &subject.host, host).with_context( + || { + format!( + "legacy-supervisor-unresolved: {} declares supervisor '{reference}'", + relative(catalog, &subject.declaration) + ) + }, + )?; + let parent_id = assigned_ids + .get(&parent.bus_identity()) + .cloned() + .with_context(|| { + format!( + "legacy-supervisor-unresolved: {} declares supervisor '{reference}', whose subject has no migrated id", + relative(catalog, &subject.declaration) + ) + })?; + if reference == parent_id { + continue; + } + refuse_foreign_format(&subject.declaration)?; + plan.files + .entry(subject.declaration.clone()) + .or_default() + .edits + .push(( + subject.host.clone(), + subject.identity.clone(), + Edit::RewriteSupervisor { + from: reference.to_owned(), + to: parent_id.clone(), + }, + )); + plan.supervisor_rewrites.push(SupervisorRewrite { + declaration: relative(catalog, &subject.declaration), + host: subject.host.clone(), + identity: subject.identity.clone(), + from: reference.to_owned(), + to_agent_id: parent_id, + }); + } + + plan.nix_owned.sort(); + plan.nix_owned.dedup(); + Ok(plan) +} + +/// A reference matches a full `.` bus identity, or a bare identity on the +/// referring declaration's own resolved host. Absence and ambiguity both refuse. +fn resolve_supervisor<'a>( + index: &BTreeMap, + reference: &str, + referring_host: &str, + default_host: &str, +) -> Result<&'a Subject> { + let mut candidates: Vec<&&Subject> = Vec::new(); + if let Some(subject) = index.get(reference) { + candidates.push(subject); + } + for host in [referring_host, default_host] { + if let Some(subject) = index.get(&format!("{host}.{reference}")) { + candidates.push(subject); + } + } + candidates.dedup_by(|left, right| { + left.bus_identity() == right.bus_identity() && left.plane == right.plane + }); + match candidates.as_slice() { + [subject] => Ok(subject), + [] => anyhow::bail!( + "no live or archived subject matches it; unarchive and repair the declaration, then retry" + ), + _ => anyhow::bail!( + "it matches {} subjects; unarchive and repair the declaration, then retry", + candidates.len() + ), + } +} + +fn refuse_foreign_format(declaration: &Path) -> Result<()> { + let extension = declaration.extension().and_then(|ext| ext.to_str()); + anyhow::ensure!( + extension == Some(KDL_EXTENSION), + "unsupported-declaration-format: {} is not canonical KDL, so migration cannot author its `id`; convert it to KDL first", + declaration.display() + ); + Ok(()) +} + +fn plane_label(plane: Plane) -> &'static str { + match plane { + Plane::Live => "live", + Plane::Archived => "archived", + } +} + +/// A resume may only finish work the original transaction planned. +fn verify_resumable(marker: &Marker, plan: &Plan) -> Result<()> { + let recorded: BTreeSet<(String, String, String)> = marker + .assigned + .iter() + .map(|planned| { + ( + planned.host.clone(), + planned.identity.clone(), + planned.agent_id.clone(), + ) + }) + .collect(); + for planned in &plan.assigned { + anyhow::ensure!( + recorded.contains(&( + planned.host.clone(), + planned.identity.clone(), + planned.agent_id.clone() + )), + "refusing to resume: {}.{} would now receive id '{}', which the interrupted transaction did not plan", + planned.host, + planned.identity, + planned.agent_id + ); + } + // Every already-migrated subject the marker planned must carry exactly the planned ID: an + // outside writer that gave it a different one makes the resume indeterminate. + let applied: BTreeMap<(String, String), String> = plan + .already_migrated + .iter() + .map(|planned| { + ( + (planned.host.clone(), planned.identity.clone()), + planned.agent_id.clone(), + ) + }) + .collect(); + for planned in &marker.assigned { + if let Some(observed) = applied.get(&(planned.host.clone(), planned.identity.clone())) { + anyhow::ensure!( + observed == &planned.agent_id, + "refusing to resume: {}.{} carries id '{observed}', not the planned '{}'", + planned.host, + planned.identity, + planned.agent_id + ); + } + } + Ok(()) +} + +/// Apply every planned edit to one declaration file and publish it atomically. +/// +/// Edits are applied one at a time against a freshly parsed document, because each insertion moves +/// every later source span. The file is re-parsed and re-checked before it is published. +fn apply_file_edits( + lock: &CatalogLock, + catalog: &Path, + path: &Path, + edits: &FileEdits, +) -> Result<()> { + let original = fs::read_to_string(path) + .with_context(|| format!("read declaration {}", path.display()))?; + let mode = fs::symlink_metadata(path) + .with_context(|| format!("stat declaration {}", path.display()))?; + anyhow::ensure!( + mode.is_file() && !mode.file_type().is_symlink(), + "declaration is not a real regular file: {}", + path.display() + ); + + let mut text = original.clone(); + for (host, identity, edit) in &edits.edits { + let document = KdlDocument::parse(&text) + .map_err(|error| anyhow::anyhow!("parse {}: {error}", path.display()))?; + let node = exact_agent_node(&document, identity, host, identity) + .map_err(|error| anyhow::anyhow!("{}: {error}", path.display()))?; + text = match edit { + Edit::InsertId(agent_id) => { + let authored = format!( + "id {}", + quoted(agent_id).map_err(|error| anyhow::anyhow!("{error}"))? + ); + insert_node(&text, node, &authored).map_err(|error| anyhow::anyhow!("{error}"))? + } + Edit::RewriteSupervisor { from, to } => { + rewrite_supervisor(&text, node, from, to).with_context(|| { + format!("rewrite supervisor in {}", path.display()) + })? + } + }; + } + + verify_file(&text, path, edits)?; + + let directory = path + .parent() + .with_context(|| format!("declaration {} has no parent", path.display()))?; + let control = catalog_transaction::retained_dir_path(lock.control())?; + let mut temporary = tempfile::Builder::new() + .prefix("catalog-migrate-ids-") + .tempfile_in(&control) + .with_context(|| format!("stage migrated declaration {}", path.display()))?; + use std::io::Write as _; + use std::os::unix::fs::PermissionsExt as _; + temporary + .as_file_mut() + .set_permissions(fs::Permissions::from_mode(mode.permissions().mode() & 0o7777))?; + temporary.write_all(text.as_bytes())?; + temporary.as_file().sync_all()?; + // The exclusive authoring lock excludes cooperating writers, not a direct same-UID write, so + // the exact preimage is rechecked immediately before publication. + let observed = fs::read_to_string(path) + .with_context(|| format!("re-read declaration {}", path.display()))?; + anyhow::ensure!( + observed == original, + "declaration {} changed while the migration was authored", + path.display() + ); + catalog_transaction::persist_tempfile_from_control( + lock.control(), + catalog, + temporary, + path, + ) + .with_context(|| format!("publish migrated declaration {}", path.display()))?; + catalog_transaction::open_dir_beneath(catalog, directory) + .and_then(|directory| directory.sync_all()) + .with_context(|| format!("sync declaration directory {}", directory.display()))?; + Ok(()) +} + +/// Replace the `supervisor` value in place, preserving every other byte. +fn rewrite_supervisor(text: &str, node: &KdlNode, from: &str, to: &str) -> Result { + let children = node + .children() + .context("agent declares no child block, so it cannot declare a supervisor")?; + let matches = children + .nodes() + .iter() + .filter(|child| child.name().value() == "supervisor") + .collect::>(); + let [supervisor] = matches.as_slice() else { + anyhow::bail!( + "agent declares `supervisor` {} times; exactly one is required to rewrite it", + matches.len() + ); + }; + anyhow::ensure!( + supervisor.children().is_none() + && supervisor.entries().len() == 1 + && supervisor.entries()[0].name().is_none(), + "`supervisor` must contain exactly one positional string" + ); + anyhow::ensure!( + supervisor.get(0).and_then(|value| value.as_string()) == Some(from), + "`supervisor` no longer reads '{from}'" + ); + let span = supervisor.entries()[0].span(); + let range = span.offset()..span.offset() + span.len(); + anyhow::ensure!( + text.get(range.clone()).is_some(), + "supervisor value span falls outside the declaration" + ); + let mut replacement = text.to_owned(); + replacement.replace_range( + range, + "ed(to).map_err(|error| anyhow::anyhow!("{error}"))?, + ); + Ok(replacement) +} + +/// Prove the edited text parses and says exactly what the plan intended. +fn verify_file(text: &str, path: &Path, edits: &FileEdits) -> Result<()> { + let document = KdlDocument::parse(text) + .map_err(|error| anyhow::anyhow!("migration produced invalid KDL in {}: {error}", path.display()))?; + for (host, identity, edit) in &edits.edits { + let node = exact_agent_node(&document, identity, host, identity) + .map_err(|error| anyhow::anyhow!("{}: {error}", path.display()))?; + match edit { + Edit::InsertId(agent_id) => { + let observed = node + .children() + .into_iter() + .flat_map(|children| children.nodes()) + .filter(|child| child.name().value() == "id") + .collect::>(); + anyhow::ensure!( + observed.len() == 1, + "migration produced {} `id` fields in {}", + observed.len(), + path.display() + ); + anyhow::ensure!( + observed[0].get(0).and_then(|value| value.as_string()) == Some(agent_id.as_str()), + "migration did not write id '{agent_id}' in {}", + path.display() + ); + } + Edit::RewriteSupervisor { to, .. } => { + anyhow::ensure!( + child_string(node, "supervisor").as_deref() == Some(to.as_str()), + "migration did not rewrite supervisor to '{to}' in {}", + path.display() + ); + } + } + } + Ok(()) +} + +/// Record an archived subject's migrated ID in its tombstone. +fn apply_tombstone_id( + lock: &CatalogLock, + catalog: &Path, + path: &Path, + agent_id: &str, +) -> Result<()> { + let bytes = fs::read(path).with_context(|| format!("read tombstone {}", path.display()))?; + let mut tombstone: Tombstone = + serde_json::from_slice(&bytes).with_context(|| format!("parse {}", path.display()))?; + anyhow::ensure!( + tombstone.schema == TOMBSTONE_SCHEMA, + "unknown archive tombstone schema '{}'", + tombstone.schema + ); + if tombstone.agent_id.as_deref() == Some(agent_id) { + return Ok(()); + } + tombstone.agent_id = Some(agent_id.to_owned()); + let mut serialized = serde_json::to_vec_pretty(&tombstone)?; + serialized.push(b'\n'); + write_control_file(lock, catalog, path, &serialized) +} + +fn write_migration_record( + lock: &CatalogLock, + catalog: &Path, + record: &MigrationRecord, +) -> Result<()> { + let mut serialized = serde_json::to_vec_pretty(record)?; + serialized.push(b'\n'); + write_control_file(lock, catalog, &migration_record_path(catalog), &serialized) +} + +/// Atomically replace one file beneath the catalog through the retained control capability. +fn write_control_file( + lock: &CatalogLock, + catalog: &Path, + path: &Path, + bytes: &[u8], +) -> Result<()> { + let directory = path + .parent() + .with_context(|| format!("{} has no parent", path.display()))?; + let control = catalog_transaction::retained_dir_path(lock.control())?; + let mut temporary = tempfile::Builder::new() + .prefix("catalog-migrate-ids-") + .tempfile_in(&control) + .with_context(|| format!("stage {}", path.display()))?; + use std::io::Write as _; + use std::os::unix::fs::PermissionsExt as _; + temporary + .as_file_mut() + .set_permissions(fs::Permissions::from_mode(0o600))?; + temporary.write_all(bytes)?; + temporary.as_file().sync_all()?; + catalog_transaction::persist_tempfile_from_control(lock.control(), catalog, temporary, path) + .with_context(|| format!("publish {}", path.display()))?; + catalog_transaction::open_dir_beneath(catalog, directory) + .and_then(|directory| directory.sync_all()) + .with_context(|| format!("sync {}", directory.display()))?; + Ok(()) +} + +/// Prove every rewritten archived declaration still parses as exactly one agent. +/// +/// Archived declarations are outside `validate_full_catalog`'s projection by construction, so they +/// get their own re-admission pass rather than none. +fn reparse_archived(plan: &Plan) -> Result<()> { + for (tombstone, agent_id) in &plan.tombstones { + let directory = tombstone + .parent() + .and_then(|host_root| { + tombstone + .file_name() + .and_then(|name| name.to_str()) + .and_then(|name| name.strip_suffix(TOMBSTONE_SUFFIX)) + .map(|identity| host_root.join(identity)) + }) + .with_context(|| format!("derive archived directory from {}", tombstone.display()))?; + let declaration = archived_declaration(&directory)?; + let text = fs::read_to_string(&declaration) + .with_context(|| format!("re-read archived declaration {}", declaration.display()))?; + let document = KdlDocument::parse(&text).map_err(|error| { + anyhow::anyhow!( + "migrated archived declaration {} is invalid KDL: {error}", + declaration.display() + ) + })?; + let agents = document + .nodes() + .iter() + .filter(|node| node.name().value() == "agent") + .collect::>(); + anyhow::ensure!( + agents.len() == 1, + "archived declaration {} holds {} agents; exactly one is required", + declaration.display(), + agents.len() + ); + anyhow::ensure!( + child_string(agents[0], "id").as_deref() == Some(agent_id.as_str()), + "migrated archived declaration {} does not carry id '{agent_id}'", + declaration.display() + ); + } + Ok(()) +} + +fn read_marker(catalog: &Path) -> Result> { + let path = marker_path(catalog); + let metadata = match fs::symlink_metadata(&path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => { + return Err(error).with_context(|| format!("inspect {}", path.display())); + } + }; + anyhow::ensure!( + metadata.is_file() && !metadata.file_type().is_symlink(), + "migration marker is not a real regular file: {}", + path.display() + ); + let bytes = fs::read(&path).with_context(|| format!("read {}", path.display()))?; + let marker: Marker = + serde_json::from_slice(&bytes).with_context(|| format!("parse {}", path.display()))?; + anyhow::ensure!( + marker.schema == MARKER_SCHEMA, + "unknown migration marker schema '{}'", + marker.schema + ); + Ok(Some(marker)) +} + +fn write_marker(lock: &CatalogLock, catalog: &Path, marker: &Marker) -> Result<()> { + let mut serialized = serde_json::to_vec_pretty(marker)?; + serialized.push(b'\n'); + write_control_file(lock, catalog, &marker_path(catalog), &serialized)?; + test_checkpoint("migrate-ids-marker-written"); + Ok(()) +} + +fn clear_marker(catalog: &Path) -> Result<()> { + let path = marker_path(catalog); + match fs::remove_file(&path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error).with_context(|| format!("clear {}", path.display())), + } + catalog_transaction::sync_dir(&catalog.join(CONTROL_DIR)) +} + +fn relative(catalog: &Path, path: &Path) -> String { + path.strip_prefix(catalog) + .map(|relative| relative.display().to_string()) + .unwrap_or_else(|_| path.display().to_string()) +} + +#[cfg(debug_assertions)] +fn test_checkpoint(point: &str) { + if std::env::var("ST2_TEST_MIGRATE_IDS_ABORT_AT").ok().as_deref() == Some(point) { + std::process::abort(); + } +} + +#[cfg(not(debug_assertions))] +fn test_checkpoint(_point: &str) {} + +#[cfg(test)] +mod tests { + use super::*; + + fn record(entries: &[(&str, &str, &str)]) -> MigrationRecord { + MigrationRecord { + schema: MIGRATION_RECORD_SCHEMA.to_owned(), + migrated_at_ms: 1, + reassigned: entries + .iter() + .map(|(legacy, kept, reassigned)| Reassignment { + legacy_bus_identity: (*legacy).to_owned(), + kept_by_agent_id: (*kept).to_owned(), + kept_by_plane: Plane::Live, + reassigned_agent_id: (*reassigned).to_owned(), + reassigned_host: "h".to_owned(), + reassigned_identity: "gone".to_owned(), + reassigned_plane: Plane::Archived, + }) + .collect(), + } + } + + /// Migration froze most legacy bus identities as the ID of the subject that already held them, + /// so an untouched endpoint needs no translation and no record lookup. + #[test] + fn an_untouched_legacy_endpoint_is_its_own_migrated_id() { + let record = record(&[("h.gone", "h.gone", "0199b8f4-8d3a-7c21-9a44-6f85b7320ea1")]); + assert_eq!( + attribute_legacy_endpoint(Some(&record), "h.other", Some("h.other")).as_deref(), + Some("h.other") + ); + // No record at all means nothing was ever reassigned. + assert_eq!( + attribute_legacy_endpoint(None, "h.gone", Some("h.gone")).as_deref(), + Some("h.gone") + ); + } + + /// A reassigned identity denotes two subjects. Only the row's own state owner is provable, and + /// every other colliding endpoint stays unattributed rather than addressing the live + /// replacement (`MESSAGE-R04`). + #[test] + fn a_reassigned_legacy_endpoint_resolves_only_for_the_rows_state_owner() { + let record = record(&[("h.gone", "h.gone", "0199b8f4-8d3a-7c21-9a44-6f85b7320ea1")]); + assert_eq!( + attribute_legacy_endpoint(Some(&record), "h.gone", Some("h.gone")).as_deref(), + Some("h.gone"), + "the state owner's own endpoint resolves to the keeping subject" + ); + assert_eq!( + attribute_legacy_endpoint(Some(&record), "h.gone", Some("h.reader")), + None, + "the same bytes at the other endpoint are unattributed" + ); + assert_eq!( + attribute_legacy_endpoint(Some(&record), "h.gone", None), + None, + "an ownerless row cannot attribute colliding bytes either" + ); + } + + /// A reference matches a full bus identity or a bare identity on the referring host; both + /// readings existing at once is undecidable rather than first-wins. + #[test] + fn supervisor_resolution_is_fail_closed_on_absence_and_ambiguity() { + let subject = |host: &str, identity: &str| Subject { + plane: Plane::Live, + host: host.to_owned(), + identity: identity.to_owned(), + declaration: PathBuf::from(format!("agents/{host}/{identity}/agent.kdl")), + declared_id: None, + supervisor: None, + nix_owned: false, + tombstone: None, + }; + let subjects = vec![ + subject("h", "root"), + subject("h", "a.b"), + subject("a", "b"), + subject("h", "only-here"), + ]; + let index: BTreeMap = subjects + .iter() + .map(|subject| (subject.bus_identity(), subject)) + .collect(); + + // A bare identity on the referring host. + assert_eq!( + resolve_supervisor(&index, "only-here", "h", "h") + .unwrap() + .identity, + "only-here" + ); + // A fully qualified bus identity. + assert_eq!( + resolve_supervisor(&index, "h.only-here", "h", "h") + .unwrap() + .identity, + "only-here" + ); + // Absent. + assert!(resolve_supervisor(&index, "ghost", "h", "h").is_err()); + // Both readings exist: host `a`'s `b`, and host `h`'s dotted identity `a.b`. + let error = resolve_supervisor(&index, "a.b", "h", "h").unwrap_err(); + assert!( + format!("{error}").contains("matches 2 subjects"), + "{error}" + ); + } + + /// A resume may only finish work the interrupted transaction planned. + #[test] + fn a_resume_refuses_an_id_the_interrupted_transaction_did_not_plan() { + let planned = |identity: &str, agent_id: &str| PlannedId { + agent_id: agent_id.to_owned(), + host: "h".to_owned(), + identity: identity.to_owned(), + plane: Plane::Live, + source: IdSource::FrozenBusIdentity, + declaration: format!("agents/h/{identity}/agent.kdl"), + }; + let marker = Marker { + schema: MARKER_SCHEMA.to_owned(), + catalog: "/catalog".to_owned(), + host: "h".to_owned(), + started_at_ms: 1, + assigned: vec![planned("worker", "h.worker")], + reassigned: Vec::new(), + supervisor_rewrites: Vec::new(), + }; + + let mut plan = Plan { + assigned: vec![planned("worker", "h.worker")], + ..Plan::default() + }; + verify_resumable(&marker, &plan).expect("the planned assignment resumes"); + + plan.assigned = vec![planned("worker", "0199b8f4-8d3a-7c21-9a44-6f85b7320ea1")]; + let error = verify_resumable(&marker, &plan).unwrap_err(); + assert!(format!("{error}").contains("did not plan"), "{error}"); + + // An outside writer that gave a planned subject a different id makes the resume + // indeterminate rather than silently correct. + plan.assigned = Vec::new(); + plan.already_migrated = vec![planned("worker", "someone-elses-id")]; + let error = verify_resumable(&marker, &plan).unwrap_err(); + assert!(format!("{error}").contains("not the planned"), "{error}"); + } +} diff --git a/src/lib.rs b/src/lib.rs index adfd10f9..6f0fd851 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -11,6 +11,7 @@ pub mod catalog; pub mod catalog_archive; pub mod catalog_graph; pub mod catalog_lock; +pub mod catalog_migrate_ids; pub mod catalog_transaction; pub mod claude_channel; pub mod claude_mcp; @@ -54,6 +55,7 @@ pub mod status; pub mod supervisor_chain; pub mod task_inventory; pub mod telemetry; +pub mod uuid_v7; pub mod validate; pub mod version; mod watch; diff --git a/src/main.rs b/src/main.rs index bf1be025..6117531e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -618,6 +618,24 @@ enum CatalogCmd { #[arg(long)] json: bool, }, + /// Freeze every live and structurally archived legacy subject's immutable agent `id` and + /// rewrite every supervisor reference to the parent's migrated id, in one transaction. + MigrateIds { + /// Host used to freeze a `.` bus identity for a declaration that omits + /// `host`, and to resolve a bare-identity supervisor reference. Defaults to this host. + #[arg(long)] + host: Option, + /// Decide every id and rewrite and print the plan without writing anything. + #[arg(long, conflicts_with = "resume")] + dry_run: bool, + /// Finish an interrupted migration. Only work the interrupted transaction planned is + /// applied. + #[arg(long)] + resume: bool, + /// Emit the typed migration receipt as JSON. + #[arg(long)] + json: bool, + }, /// Move one archived identity back into the live catalog. The exact reverse of `archive`. Unarchive { /// Archived identity to restore. @@ -1542,6 +1560,69 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< } Ok(()) } + Command::Catalog(CatalogCmd::MigrateIds { + host, + dry_run, + resume, + json, + }) => { + let result = + st2::catalog_migrate_ids::migrate_ids(st2::catalog_migrate_ids::MigrateRequest { + catalog: catalog_arg(None)?, + host: host.unwrap_or_else(detect_host), + dry_run, + resume, + })?; + if json { + println!("{}", serde_json::to_string_pretty(&result)?); + } else { + let verb = if result.dry_run { "would-freeze" } else { "froze" }; + for planned in &result.assigned { + println!( + "{verb} {} {}.{} [{}] {}", + planned.agent_id, + planned.host, + planned.identity, + match planned.plane { + st2::catalog_migrate_ids::Plane::Live => "live", + st2::catalog_migrate_ids::Plane::Archived => "archived", + }, + planned.declaration + ); + } + for reassignment in &result.reassigned { + println!( + "reassigned {} kept by {} -> {} ({}.{})", + reassignment.legacy_bus_identity, + reassignment.kept_by_agent_id, + reassignment.reassigned_agent_id, + reassignment.reassigned_host, + reassignment.reassigned_identity + ); + } + for rewrite in &result.supervisor_rewrites { + println!( + "supervisor {} {} -> {}", + rewrite.declaration, rewrite.from, rewrite.to_agent_id + ); + } + for declaration in &result.nix_owned { + println!("nix-owned {declaration}"); + } + println!( + "{} assigned={} reassigned={} rewrites={} already={}", + match result.status { + st2::catalog_migrate_ids::MigrateStatus::Migrated => "migrated", + st2::catalog_migrate_ids::MigrateStatus::Unchanged => "unchanged", + }, + result.assigned.len(), + result.reassigned.len(), + result.supervisor_rewrites.len(), + result.already_migrated.len() + ); + } + Ok(()) + } Command::Catalog(CatalogCmd::Unarchive { identity, host, diff --git a/src/uuid_v7.rs b/src/uuid_v7.rs new file mode 100644 index 00000000..3f77bac8 --- /dev/null +++ b/src/uuid_v7.rs @@ -0,0 +1,175 @@ +//! UUIDv7 generation for immutable agent IDs (R24). +//! +//! New agent subjects — and archived legacy subjects whose `.` bus identity is +//! already claimed when `st2 catalog migrate-ids` freezes IDs — need an identifier that is unique +//! without coordination and sorts chronologically, so a catalog stays readable in time order. That +//! is exactly RFC 9562 §5.7 UUIDv7. +//! +//! This lives in-tree rather than as a `uuid` crate dependency because the whole implementation is +//! the forty lines below: a 48-bit big-endian millisecond timestamp, four version bits, and 74 +//! random bits, rendered as canonical lowercase hex. The one non-trivial part — the entropy source +//! — already had to be written here anyway, since a weak fallback would silently break the +//! uniqueness the catalog relies on. + +/// A fresh canonical UUIDv7 from the current wall clock. +/// +/// Fallible rather than panicking: every caller sits inside an `anyhow` catalog transaction that +/// must refuse cleanly before writing, and a missing entropy source is a real environment failure +/// worth reporting rather than an abort in the middle of a commit. +pub fn uuid_v7() -> anyhow::Result { + Ok(uuid_v7_from(crate::message::now_ms(), random_bytes()?)) +} + +/// The pure core: render `unix_ts_ms` plus 74 bits taken from `random` as a canonical UUIDv7. +/// +/// Layout per RFC 9562 §5.7 — bits 0..48 are the big-endian millisecond timestamp, bits 48..52 the +/// version `0b0111`, bits 52..64 `rand_a`, bits 64..66 the variant `0b10`, bits 66..128 `rand_b`. +/// Only the low 48 bits of `unix_ts_ms` are representable; anything above is dropped, which is the +/// year-10889 horizon. +pub(crate) fn uuid_v7_from(unix_ts_ms: u64, random: [u8; 10]) -> String { + let mut bytes = [0u8; 16]; + bytes[0..6].copy_from_slice(&unix_ts_ms.to_be_bytes()[2..8]); + // version nibble + the top 4 bits of the 12-bit `rand_a` + bytes[6] = 0x70 | (random[0] & 0x0f); + bytes[7] = random[1]; + // variant bits + the top 6 bits of the 62-bit `rand_b` + bytes[8] = 0x80 | (random[2] & 0x3f); + bytes[9..16].copy_from_slice(&random[3..10]); + + let mut out = String::with_capacity(36); + for (i, byte) in bytes.iter().enumerate() { + if matches!(i, 4 | 6 | 8 | 10) { + out.push('-'); + } + out.push(char::from_digit((byte >> 4) as u32, 16).expect("nibble is a hex digit")); + out.push(char::from_digit((byte & 0x0f) as u32, 16).expect("nibble is a hex digit")); + } + out +} + +/// 10 bytes from the OS CSPRNG, or a hard error. +/// +/// Linux and Android go through `getrandom(2)`, which needs no file descriptor and cannot be +/// shadowed by a tampered `/dev`. Every other supported target (macOS, the BSDs) has no +/// `libc::getrandom`, so it reads `/dev/urandom` — after confirming the opened descriptor really is +/// a character device, so a planted regular file cannot feed us chosen "randomness". A short read +/// or a failing call is an error: never fall back to a clock- or pid-derived value, because the +/// uniqueness of an immutable agent ID depends on these bits. +fn random_bytes() -> anyhow::Result<[u8; 10]> { + let mut buf = [0u8; 10]; + #[cfg(any(target_os = "linux", target_os = "android"))] + { + let read = unsafe { libc::getrandom(buf.as_mut_ptr().cast(), buf.len(), 0) }; + anyhow::ensure!( + read == buf.len() as isize, + "getrandom(2) returned {read} of {} bytes for a UUIDv7: {}", + buf.len(), + std::io::Error::last_os_error() + ); + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + use anyhow::Context as _; + use std::io::Read as _; + let mut file = + std::fs::File::open("/dev/urandom").context("opening /dev/urandom for a UUIDv7")?; + let file_type = std::os::unix::fs::FileTypeExt::is_char_device( + &file + .metadata() + .context("stat /dev/urandom for a UUIDv7")? + .file_type(), + ); + anyhow::ensure!( + file_type, + "/dev/urandom is not a character device — refusing to seed a UUIDv7 from it" + ); + file.read_exact(&mut buf) + .context("reading 10 bytes from /dev/urandom for a UUIDv7")?; + } + Ok(buf) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The RFC 9562 §5.7 example vector — pinned bytes in, one exact string out, so a change to the + /// field layout fails here instead of silently producing a plausible-looking UUID. + #[test] + fn pinned_vector_renders_exactly() { + let random = [0x0c, 0xc3, 0x98, 0xc4, 0xdc, 0x0c, 0x0c, 0x07, 0x39, 0x8f]; + assert_eq!( + uuid_v7_from(0x017f_22e2_79b0, random), + "017f22e2-79b0-7cc3-98c4-dc0c0c07398f" + ); + } + + #[test] + fn version_and_variant_bits_are_fixed() { + for ts in [0u64, 1, 1_757_000_000_000, (1 << 48) - 1] { + for fill in [0x00u8, 0xff, 0x5a] { + let id = uuid_v7_from(ts, [fill; 10]); + let version = id.as_bytes()[14] as char; + assert_eq!(version, '7', "version nibble of {id}"); + let variant = id.as_bytes()[19] as char; + assert!( + matches!(variant, '8' | '9' | 'a' | 'b'), + "variant bits of {id} must be 0b10" + ); + } + } + } + + #[test] + fn timestamp_round_trips() { + for ts in [0u64, 1, 42, 1_757_000_000_000, (1 << 48) - 1] { + let id = uuid_v7_from(ts, [0xa5; 10]); + let hex: String = id.chars().filter(|c| *c != '-').take(12).collect(); + let parsed = u64::from_str_radix(&hex, 16).expect("first 48 bits are hex"); + assert_eq!(parsed, ts, "round-trip of {id}"); + } + } + + #[test] + fn canonical_form_is_lowercase_hyphenated_36_chars() { + let id = uuid_v7_from(1_757_000_000_000, [0xde; 10]); + assert_eq!(id.len(), 36, "{id}"); + for (i, c) in id.char_indices() { + if matches!(i, 8 | 13 | 18 | 23) { + assert_eq!(c, '-', "hyphen at {i} of {id}"); + } else { + assert!( + c.is_ascii_digit() || ('a'..='f').contains(&c), + "{c} at {i} of {id} must be lowercase hex" + ); + } + } + } + + #[test] + fn consecutive_live_calls_differ() { + let first = uuid_v7().expect("entropy source"); + let second = uuid_v7().expect("entropy source"); + assert_ne!(first, second); + } + + #[test] + fn output_is_a_valid_agent_id() { + let id = uuid_v7().expect("entropy source"); + agent_spec::validate_agent_id(&id).expect("a UUIDv7 is a valid agent id"); + } + + #[test] + fn increasing_timestamps_sort_lexicographically() { + let ids: Vec = [0u64, 1, 1 << 8, 1 << 16, 1_757_000_000_000, (1 << 48) - 1] + .into_iter() + // Constant randomness would make the ordering trivially the timestamp's; vary the + // random tail in the opposite direction so only the timestamp prefix can carry it. + .enumerate() + .map(|(i, ts)| uuid_v7_from(ts, [0xff - i as u8; 10])) + .collect(); + let mut sorted = ids.clone(); + sorted.sort(); + assert_eq!(ids, sorted, "v7 ids must sort in timestamp order"); + } +} diff --git a/tests/catalog_archive.rs b/tests/catalog_archive.rs index d1c85c67..1bd34038 100755 --- a/tests/catalog_archive.rs +++ b/tests/catalog_archive.rs @@ -666,3 +666,185 @@ fn unarchive_refuses_to_overwrite_a_live_declaration() { ); assert!(root.join(".st2/archive/h/gone/agent.kdl").is_file()); } + +// ---- DELTA-003: the tombstone carries the archived subject's immutable agent ID ------------- + +const MIGRATED_ID: &str = "0199b8f4-8d3a-7c21-9a44-6f85b7320ea1"; + +/// A retired seat whose declaration already carries an explicit immutable `id` (R24). +fn migrated_retired_seat(root: &Path, identity: &str, id: &str) { + retired_seat(root, identity, RETIRED); + write( + root, + &format!("agents/h/{identity}/agent.kdl"), + &format!( + "agent \"{identity}\" {{\n id \"{id}\"\n host \"h\"\n {RETIRED}\n command \"true\"\n}}\n" + ), + ); +} + +/// The root slot a retired declaration never holds, so whole-catalog validation stays green. +fn keeper(root: &Path) { + write( + root, + "agents/h/keeper/agent.kdl", + "agent \"keeper\" { host \"h\"; command \"true\" }\n", + ); +} + +fn archive_gone(root: &Path, bin: &Path) -> Output { + st2( + root, + bin, + &[ + "catalog", + "archive", + "--identity", + "gone", + "--host", + "h", + "--json", + ], + ) +} + +#[test] +fn archive_freezes_an_explicit_agent_id_in_the_tombstone_and_the_graph_row() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + pty_shim(&bin, "[]"); + migrated_retired_seat(root, "gone", MIGRATED_ID); + + let output = archive_gone(root, &bin); + assert!( + output.status.success(), + "stderr:\n{}", + String::from_utf8_lossy(&output.stderr) + ); + let receipt = json(&output); + assert_eq!(receipt["archived"][0]["id"], "h.gone"); + assert_eq!(receipt["archived"][0]["agentId"], MIGRATED_ID); + + let tombstone: serde_json::Value = + serde_json::from_slice(&fs::read(root.join(".st2/archive/h/gone.tombstone.json")).unwrap()) + .unwrap(); + assert_eq!(tombstone["schema"], "st2.catalog-archive-tombstone.v1"); + assert_eq!( + tombstone["id"], "h.gone", + "`id` stays the legacy bus-identity key" + ); + assert_eq!(tombstone["agentId"], MIGRATED_ID); + + let graph = json(&st2( + root, + &bin, + &["catalog", "graph", "--host", "h", "--json"], + )); + assert_eq!(graph["complete"], true, "{graph:#}"); + let rows = graph["archived"].as_array().unwrap(); + assert_eq!(rows.len(), 1, "{graph:#}"); + assert_eq!(rows[0]["id"], "h.gone"); + assert_eq!(rows[0]["agentId"], MIGRATED_ID); +} + +#[test] +fn a_legacy_tombstone_omits_the_agent_id_key_and_still_round_trips() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + pty_shim(&bin, "[]"); + retired_seat(root, "gone", RETIRED); + keeper(root); + + assert!(archive_gone(root, &bin).status.success()); + + // The serialized bytes must not carry the key at all: a pre-DELTA-003 reader has to see the + // exact shape it always saw, which is what keeps the schema at v1. + let bytes = fs::read(root.join(".st2/archive/h/gone.tombstone.json")).unwrap(); + let text = String::from_utf8(bytes.clone()).unwrap(); + assert!( + !text.contains("agentId"), + "an unmigrated subject's tombstone must omit the key entirely:\n{text}" + ); + let tombstone: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert!(tombstone.get("agentId").is_none(), "{tombstone:#}"); + + let graph = json(&st2( + root, + &bin, + &["catalog", "graph", "--host", "h", "--json"], + )); + assert_eq!(graph["complete"], true, "{graph:#}"); + assert!( + graph["archived"][0]["agentId"].is_null(), + "a legacy tombstone projects a null agent ID: {graph:#}" + ); + + // Both sides absent: unarchive has nothing to reconcile and restores as it always did. + let restored = st2( + root, + &bin, + &["catalog", "unarchive", "gone", "--host", "h", "--json"], + ); + assert!( + restored.status.success(), + "stderr:\n{}", + String::from_utf8_lossy(&restored.stderr) + ); + assert!(root.join("agents/h/gone/agent.kdl").is_file()); +} + +#[test] +fn unarchive_refuses_when_the_tombstone_and_the_declaration_disagree_on_the_agent_id() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + pty_shim(&bin, "[]"); + migrated_retired_seat(root, "gone", MIGRATED_ID); + keeper(root); + assert!(archive_gone(root, &bin).status.success()); + + // Rewrite the archived declaration's immutable ID out from under its tombstone. + let archived_spec = root.join(".st2/archive/h/gone/agent.kdl"); + let original = fs::read_to_string(&archived_spec).unwrap(); + fs::write( + &archived_spec, + original.replace(MIGRATED_ID, "0199b8f4-8d3a-7c21-9a44-000000000000"), + ) + .unwrap(); + + let refused = st2( + root, + &bin, + &["catalog", "unarchive", "gone", "--host", "h", "--json"], + ); + assert!(!refused.status.success()); + let stderr = String::from_utf8_lossy(&refused.stderr); + assert!( + stderr.contains("tombstone records agent ID") && stderr.contains(MIGRATED_ID), + "stderr:\n{stderr}" + ); + assert!( + archived_spec.is_file() && !root.join("agents/h/gone").exists(), + "a refused unarchive moves nothing" + ); + + // Agreeing again admits the exact same restore. + fs::write(&archived_spec, &original).unwrap(); + let restored = st2( + root, + &bin, + &["catalog", "unarchive", "gone", "--host", "h", "--json"], + ); + assert!( + restored.status.success(), + "stderr:\n{}", + String::from_utf8_lossy(&restored.stderr) + ); + assert_eq!( + fs::read_to_string(root.join("agents/h/gone/agent.kdl")).unwrap(), + original + ); + assert!(!root.join(".st2/archive/h/gone.tombstone.json").exists()); +} diff --git a/tests/catalog_migrate_ids.rs b/tests/catalog_migrate_ids.rs new file mode 100644 index 00000000..8fbbe78d --- /dev/null +++ b/tests/catalog_migrate_ids.rs @@ -0,0 +1,661 @@ +#![cfg(unix)] +//! `st2 catalog migrate-ids` freezes every legacy subject's immutable agent ID in one transaction. +//! +//! The headline fixture mirrors the shape a real catalog presents at migration time: one counted +//! root per host, a supervisor chain, hundreds of structurally archived subjects, and one archived +//! subject whose `.` bytes a live re-projection has already reclaimed — the +//! oscillation an archiving supervisor and a re-projecting generator produce together. + +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output}; + +fn write(root: &Path, relative: &str, body: &str) { + let path = root.join(relative); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, body).unwrap(); +} + +fn fixture(temporary: &tempfile::TempDir) -> (PathBuf, PathBuf) { + let catalog = temporary.path().join("catalog"); + fs::create_dir_all(&catalog).unwrap(); + (catalog, temporary.path().join("bin")) +} + +fn st2(root: &Path, bin: &Path, args: &[&str]) -> Output { + st2_with(root, bin, args, &[]) +} + +fn st2_with(root: &Path, bin: &Path, args: &[&str], env: &[(&str, &str)]) -> Output { + let home = root.parent().unwrap().join("home"); + fs::create_dir_all(&home).unwrap(); + let mut command = Command::new(env!("CARGO_BIN_EXE_st2")); + command + .args(["--catalog", root.to_str().unwrap()]) + .args(args) + .env("PATH", bin) + .env("HOME", &home) + .env("XDG_STATE_HOME", home.join("state")) + .env("PTY_ROOT", home.join("pty")) + .env_remove("CATALOG") + .env_remove("ST_ROOT"); + for (key, value) in env { + command.env(key, value); + } + command.output().unwrap() +} + +fn json(output: &Output) -> serde_json::Value { + serde_json::from_slice(&output.stdout).unwrap_or_else(|error| { + panic!( + "stdout is not JSON ({error}):\nstdout={}\nstderr={}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ) + }) +} + +fn ok(output: &Output) { + assert!( + output.status.success(), + "stdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +fn stderr(output: &Output) -> String { + String::from_utf8_lossy(&output.stderr).into_owned() +} + +fn generation(root: &Path) -> String { + fs::read_to_string(root.join(".st2/catalog-generation")).unwrap_or_default() +} + +/// The explicit `id` a declaration file carries, if any. +fn declared_id(path: &Path) -> Option { + let text = fs::read_to_string(path).unwrap_or_else(|error| panic!("{}: {error}", path.display())); + text.lines() + .map(str::trim) + .find_map(|line| line.strip_prefix("id \"")) + .and_then(|rest| rest.strip_suffix('"')) + .map(str::to_owned) +} + +/// The `supervisor` value a declaration file carries, if any. +fn declared_supervisor(path: &Path) -> Option { + let text = fs::read_to_string(path).unwrap(); + text.lines() + .map(str::trim) + .find_map(|line| line.strip_prefix("supervisor \"")) + .and_then(|rest| rest.strip_suffix('"')) + .map(str::to_owned) +} + +fn live(root: &Path, host: &str, identity: &str, body: &str) { + write( + root, + &format!("agents/{host}/{identity}/agent.kdl"), + &format!("agent \"{identity}\" {{\n host \"{host}\"\n{body} argv \"true\"\n}}\n"), + ); +} + +/// One structurally archived subject: its moved directory plus the tombstone that explains it. +fn archived(root: &Path, host: &str, identity: &str, supervisor: Option<&str>) { + let supervisor_line = supervisor + .map(|value| format!(" supervisor \"{value}\"\n")) + .unwrap_or_default(); + write( + root, + &format!(".st2/archive/{host}/{identity}/agent.kdl"), + &format!( + "agent \"{identity}\" {{\n host \"{host}\"\n{supervisor_line} desired-state \"retired\" reason=\"Work finished\"\n argv \"true\"\n}}\n" + ), + ); + write( + root, + &format!(".st2/archive/{host}/{identity}.tombstone.json"), + &format!( + "{{\n \"schema\": \"st2.catalog-archive-tombstone.v1\",\n \"id\": \"{host}.{identity}\",\n \"host\": \"{host}\",\n \"identity\": \"{identity}\",\n \"archivedAt\": 1750000000000,\n \"reason\": \"Work finished\",\n \"archiveRoot\": \".st2/archive/{host}/{identity}\"\n}}\n" + ), + ); +} + +const ARCHIVED_COUNT: usize = 600; +const LIVE_COUNT: usize = 40; +/// The identity a re-projecting generator recreated live after the supervisor archived it, so both +/// planes claim one `.`. +const COLLIDING: &str = "reprojected"; + +/// A catalog shaped like a real one at migration time. +fn dev3_shaped(root: &Path) { + live(root, "h", "root", ""); + // A five-deep supervisor chain, so the rewrite is proved on more than a flat fan-out. + live(root, "h", "chain-1", " supervisor \"root\"\n"); + for depth in 2..=5 { + live( + root, + "h", + &format!("chain-{depth}"), + &format!(" supervisor \"chain-{}\"\n", depth - 1), + ); + } + // Identities that themselves contain dots, which is what a real catalog's semantic routes look + // like — the reference `dotfiles.worker` is still a BARE identity, not a host-qualified one. + live(root, "h", "dotfiles.worker", " supervisor \"root\"\n"); + live( + root, + "h", + "dotfiles.worker.child", + " supervisor \"dotfiles.worker\"\n", + ); + for index in 0..LIVE_COUNT { + live( + root, + "h", + &format!("live-{index:03}"), + " supervisor \"root\"\n", + ); + } + // The live half of the collision. + live(root, "h", COLLIDING, " supervisor \"root\"\n"); + + for index in 0..ARCHIVED_COUNT { + archived(root, "h", &format!("arch-{index:03}"), Some("root")); + } + // One archived subject supervised by another archived subject: supervisor resolution must see + // the archived half of the combined index, not only the live catalog. + archived(root, "h", "arch-child", Some("arch-000")); + // The archived half of the collision. + archived(root, "h", COLLIDING, Some("root")); +} + +#[test] +fn a_dev3_shaped_catalog_freezes_live_bus_identities_and_reassigns_only_an_archived_collision() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + dev3_shaped(root); + + let before = generation(root); + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + ok(&output); + let receipt = json(&output); + assert_eq!(receipt["schema"], "st2.catalog-migrate-ids.v1"); + assert_eq!(receipt["status"], "migrated"); + assert_eq!(receipt["dryRun"], false); + + let assigned = receipt["assigned"].as_array().unwrap(); + // every live subject + every archived subject + let live_total = LIVE_COUNT + 1 /* root */ + 5 /* chain */ + 2 /* dotted */ + 1 /* colliding */; + let archived_total = ARCHIVED_COUNT + 1 /* arch-child */ + 1 /* colliding */; + assert_eq!(assigned.len(), live_total + archived_total, "{receipt:#}"); + + // A live subject freezes its own bus identity, so nothing runtime-visible moves. + for index in 0..LIVE_COUNT { + let identity = format!("live-{index:03}"); + let path = root.join(format!("agents/h/{identity}/agent.kdl")); + assert_eq!( + declared_id(&path).as_deref(), + Some(format!("h.{identity}").as_str()), + "{identity} must freeze its bus identity" + ); + } + assert_eq!( + declared_id(&root.join("agents/h/root/agent.kdl")).as_deref(), + Some("h.root") + ); + assert_eq!( + declared_id(&root.join("agents/h/dotfiles.worker/agent.kdl")).as_deref(), + Some("h.dotfiles.worker") + ); + + // A non-colliding archived subject freezes the same bytes. + assert_eq!( + declared_id(&root.join(".st2/archive/h/arch-000/agent.kdl")).as_deref(), + Some("h.arch-000") + ); + + // The live claimant of the colliding bytes keeps them; the archived one receives UUIDv7. + assert_eq!( + declared_id(&root.join(format!("agents/h/{COLLIDING}/agent.kdl"))).as_deref(), + Some(format!("h.{COLLIDING}").as_str()) + ); + let reassigned_id = + declared_id(&root.join(format!(".st2/archive/h/{COLLIDING}/agent.kdl"))).unwrap(); + assert_eq!(reassigned_id.len(), 36, "{reassigned_id} must be a UUID"); + assert_eq!( + reassigned_id.as_bytes()[14], b'7', + "{reassigned_id} must be version 7" + ); + assert!( + matches!(reassigned_id.as_bytes()[19], b'8' | b'9' | b'a' | b'b'), + "{reassigned_id} must carry the RFC 9562 variant" + ); + + let reassigned = receipt["reassigned"].as_array().unwrap(); + assert_eq!(reassigned.len(), 1, "only the collision is reassigned"); + assert_eq!(reassigned[0]["legacyBusIdentity"], format!("h.{COLLIDING}")); + assert_eq!(reassigned[0]["keptByAgentId"], format!("h.{COLLIDING}")); + assert_eq!(reassigned[0]["keptByPlane"], "live"); + assert_eq!(reassigned[0]["reassignedAgentId"], reassigned_id); + assert_eq!(reassigned[0]["reassignedPlane"], "archived"); + + // The tombstone records the same immutable ID, so the archived subject stays identifiable. + let tombstone: serde_json::Value = serde_json::from_slice( + &fs::read(root.join(format!(".st2/archive/h/{COLLIDING}.tombstone.json"))).unwrap(), + ) + .unwrap(); + assert_eq!(tombstone["agentId"], reassigned_id); + assert_eq!(tombstone["id"], format!("h.{COLLIDING}")); + + // The durable reassignment record is what keeps a version-1 durable record readable. + let record: serde_json::Value = + serde_json::from_slice(&fs::read(root.join(".st2/agent-id-migration.json")).unwrap()) + .unwrap(); + assert_eq!(record["schema"], "st2.agent-id-migration.v1"); + assert_eq!(record["reassigned"].as_array().unwrap().len(), 1); + assert_eq!(record["reassigned"][0]["keptByAgentId"], format!("h.{COLLIDING}")); + assert_eq!(record["reassigned"][0]["reassignedAgentId"], reassigned_id); + + // Every supervisor reference now names the parent's migrated ID. + assert_eq!( + declared_supervisor(&root.join("agents/h/chain-5/agent.kdl")).as_deref(), + Some("h.chain-4") + ); + assert_eq!( + declared_supervisor(&root.join("agents/h/dotfiles.worker.child/agent.kdl")).as_deref(), + Some("h.dotfiles.worker") + ); + assert_eq!( + declared_supervisor(&root.join(".st2/archive/h/arch-child/agent.kdl")).as_deref(), + Some("h.arch-000"), + "an archived parent resolves through the combined index" + ); + let rewrites = receipt["supervisorRewrites"].as_array().unwrap(); + assert_eq!( + rewrites.len(), + live_total - 1 + archived_total, + "every subject except the root declares a supervisor" + ); + + assert_ne!(generation(root), before, "the transaction commits one generation"); + assert!( + !root.join(".st2/migrate-ids-incomplete").exists(), + "a completed transaction leaves no marker" + ); + + // Re-running is a no-op: nothing is written and the generation does not move. + let after_first = generation(root); + let again = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + ok(&again); + let second = json(&again); + assert_eq!(second["status"], "unchanged"); + assert_eq!(second["assigned"].as_array().unwrap().len(), 0); + assert_eq!( + second["alreadyMigrated"].as_array().unwrap().len(), + live_total + archived_total + ); + assert_eq!(generation(root), after_first, "a no-op advances no generation"); +} + +#[test] +fn a_dry_run_reports_the_whole_plan_and_writes_nothing() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + live(root, "h", "worker", " supervisor \"root\"\n"); + archived(root, "h", "gone", Some("root")); + + let before = generation(root); + let output = st2( + root, + &bin, + &[ + "catalog", + "migrate-ids", + "--host", + "h", + "--dry-run", + "--json", + ], + ); + ok(&output); + let receipt = json(&output); + assert_eq!(receipt["dryRun"], true); + assert_eq!(receipt["status"], "migrated"); + assert_eq!(receipt["assigned"].as_array().unwrap().len(), 3); + assert_eq!(receipt["supervisorRewrites"].as_array().unwrap().len(), 2); + + assert_eq!(declared_id(&root.join("agents/h/worker/agent.kdl")), None); + assert_eq!( + declared_id(&root.join(".st2/archive/h/gone/agent.kdl")), + None + ); + assert_eq!( + declared_supervisor(&root.join("agents/h/worker/agent.kdl")).as_deref(), + Some("root") + ); + assert_eq!(generation(root), before); + assert!(!root.join(".st2/migrate-ids-incomplete").exists()); + assert!(!root.join(".st2/agent-id-migration.json").exists()); +} + +#[test] +fn an_unresolved_supervisor_reference_refuses_before_any_write() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + live(root, "h", "orphan", " supervisor \"ghost\"\n"); + + let before = generation(root); + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + assert!(!output.status.success()); + let message = stderr(&output); + assert!( + message.contains("legacy-supervisor-unresolved"), + "expected the named refusal:\n{message}" + ); + assert!( + message.contains("supervisor 'ghost'"), + "the refusal must name the reference:\n{message}" + ); + assert!( + message.contains("agents/h/orphan/agent.kdl"), + "the refusal must name the declaration:\n{message}" + ); + + assert_eq!(declared_id(&root.join("agents/h/root/agent.kdl")), None); + assert_eq!(declared_id(&root.join("agents/h/orphan/agent.kdl")), None); + assert_eq!(generation(root), before); + assert!(!root.join(".st2/migrate-ids-incomplete").exists()); +} + +#[test] +fn an_ambiguous_supervisor_reference_refuses_before_any_write() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + // `a.b` is readable two ways against the combined index: host `a`'s subject `b`, and host + // `h`'s dotted identity `a.b`. Both exist, so the reference is undecidable. + live(root, "h", "root", ""); + live(root, "a", "b", ""); + live(root, "h", "a.b", " supervisor \"root\"\n"); + live(root, "h", "child", " supervisor \"a.b\"\n"); + + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + assert!(!output.status.success()); + let message = stderr(&output); + assert!( + message.contains("legacy-supervisor-unresolved") && message.contains("matches 2 subjects"), + "expected an ambiguity refusal:\n{message}" + ); + assert_eq!(declared_id(&root.join("agents/h/child/agent.kdl")), None); +} + +#[test] +fn a_non_kdl_declaration_refuses_rather_than_being_rewritten() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + write( + root, + "agents/h/legacy/agent.toml", + "identity = \"legacy\"\nhost = \"h\"\nsupervisor = \"root\"\nargv = [\"true\"]\n", + ); + + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + assert!(!output.status.success()); + let message = stderr(&output); + assert!( + message.contains("unsupported-declaration-format") + && message.contains("agents/h/legacy/agent.toml"), + "expected a format refusal naming the declaration:\n{message}" + ); + assert_eq!(declared_id(&root.join("agents/h/root/agent.kdl")), None); +} + +#[test] +fn a_nix_owned_declaration_is_migrated_and_reported_for_its_generator() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + write( + root, + "agents/h/projected/agent.kdl", + "agent \"projected\" {\n host \"h\"\n supervisor \"root\"\n argv \"true\"\n meta { managed-by \"nix\" }\n}\n", + ); + + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + ok(&output); + let receipt = json(&output); + // Migration is not interactive authoring: the Nix marker guards `st2 rename`, not the one + // transaction that has to reach the whole plane. The receipt names it so the generator can be + // taught to emit `id` before the next activation re-projects the file without one. + assert_eq!( + receipt["nixOwned"].as_array().unwrap(), + &vec![serde_json::json!("agents/h/projected/agent.kdl")] + ); + assert_eq!( + declared_id(&root.join("agents/h/projected/agent.kdl")).as_deref(), + Some("h.projected") + ); + let text = fs::read_to_string(root.join("agents/h/projected/agent.kdl")).unwrap(); + assert!( + text.contains("meta { managed-by \"nix\" }"), + "unrelated source bytes are preserved:\n{text}" + ); +} + +#[test] +fn a_catalog_that_does_not_currently_admit_refuses_before_any_write() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + // Two counted roots on one host: the catalog already fails admission, so a rewritten plane + // could not be re-admitted either. + live(root, "h", "root", ""); + live(root, "h", "second-root", ""); + + let before = generation(root); + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + assert!(!output.status.success()); + let message = stderr(&output); + assert!( + message.contains("does not currently admit") && message.contains("root-count"), + "expected the pre-migration admission refusal:\n{message}" + ); + assert_eq!(declared_id(&root.join("agents/h/root/agent.kdl")), None); + assert_eq!(generation(root), before); + assert!(!root.join(".st2/migrate-ids-incomplete").exists()); +} + +#[test] +fn an_explicit_id_is_left_alone_and_blocks_a_freeze_that_would_claim_it() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + // A migrated subject already owns the bytes `h.worker`, which is exactly what the legacy + // subject `worker` would freeze. The transaction may not hand one ID to two subjects. + live( + root, + "h", + "impostor", + " id \"h.worker\"\n supervisor \"root\"\n", + ); + live(root, "h", "worker", " supervisor \"root\"\n"); + + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + assert!(!output.status.success()); + let message = stderr(&output); + assert!( + message.contains("cannot freeze its bus identity"), + "expected a freeze refusal:\n{message}" + ); + assert_eq!( + declared_id(&root.join("agents/h/impostor/agent.kdl")).as_deref(), + Some("h.worker"), + "the already-migrated declaration is untouched" + ); + assert_eq!(declared_id(&root.join("agents/h/worker/agent.kdl")), None); +} + +#[test] +fn an_interrupted_transaction_refuses_a_plain_rerun_and_resumes_exactly_the_remaining_work() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + for index in 0..4 { + live( + root, + "h", + &format!("worker-{index}"), + " supervisor \"root\"\n", + ); + } + + let before = generation(root); + // Abort after the first declaration is published: the marker exists, one file is migrated, and + // the generation has not moved. + let aborted = st2_with( + root, + &bin, + &["catalog", "migrate-ids", "--host", "h", "--json"], + &[( + "ST2_TEST_MIGRATE_IDS_ABORT_AT", + "migrate-ids-declaration-written", + )], + ); + assert!(!aborted.status.success(), "the run must not complete"); + assert!( + root.join(".st2/migrate-ids-incomplete").is_file(), + "an interrupted transaction leaves its marker" + ); + let migrated_before_resume = ["root", "worker-0", "worker-1", "worker-2", "worker-3"] + .into_iter() + .filter(|identity| { + declared_id(&root.join(format!("agents/h/{identity}/agent.kdl"))).is_some() + }) + .count(); + assert_eq!( + migrated_before_resume, 1, + "exactly one declaration was published before the abort" + ); + assert_eq!(generation(root), before, "an aborted run commits no generation"); + + // A plain rerun refuses rather than planning a second transaction over a half-migrated plane. + let plain = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + assert!(!plain.status.success()); + assert!( + stderr(&plain).contains("--resume"), + "the refusal must name the recovery path:\n{}", + stderr(&plain) + ); + + let resumed = st2( + root, + &bin, + &[ + "catalog", + "migrate-ids", + "--host", + "h", + "--resume", + "--json", + ], + ); + ok(&resumed); + let receipt = json(&resumed); + assert_eq!(receipt["resumed"], true); + assert_eq!(receipt["status"], "migrated"); + assert_eq!( + receipt["assigned"].as_array().unwrap().len(), + 4, + "only the remaining declarations are assigned" + ); + assert_eq!( + receipt["alreadyMigrated"].as_array().unwrap().len(), + 1, + "the declaration published before the abort is already migrated" + ); + for identity in ["root", "worker-0", "worker-1", "worker-2", "worker-3"] { + assert_eq!( + declared_id(&root.join(format!("agents/h/{identity}/agent.kdl"))).as_deref(), + Some(format!("h.{identity}").as_str()), + "{identity} must be migrated after the resume" + ); + } + assert!( + !root.join(".st2/migrate-ids-incomplete").exists(), + "a completed resume clears the marker" + ); + assert_ne!(generation(root), before); +} + +#[test] +fn resume_without_a_marker_refuses() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + + let output = st2( + root, + &bin, + &[ + "catalog", + "migrate-ids", + "--host", + "h", + "--resume", + "--json", + ], + ); + assert!(!output.status.success()); + assert!( + stderr(&output).contains("nothing to resume"), + "{}", + stderr(&output) + ); +} + +#[test] +fn an_unexplained_archive_entry_refuses_because_a_hidden_subject_could_lose_its_bytes() { + let temporary = tempfile::tempdir().unwrap(); + let (catalog, bin) = fixture(&temporary); + let root = catalog.as_path(); + fs::create_dir_all(&bin).unwrap(); + live(root, "h", "root", ""); + // An archived directory with no tombstone: `observe` reports it, and migration must not freeze + // bytes that this unreadable subject may already own. + write( + root, + ".st2/archive/h/mystery/agent.kdl", + "agent \"mystery\" {\n host \"h\"\n argv \"true\"\n}\n", + ); + + let output = st2(root, &bin, &["catalog", "migrate-ids", "--host", "h", "--json"]); + assert!(!output.status.success()); + let message = stderr(&output); + assert!( + message.contains("unexplained state") && message.contains("no readable tombstone"), + "expected the archive refusal:\n{message}" + ); + assert_eq!(declared_id(&root.join("agents/h/root/agent.kdl")), None); +}