From 7d5fdc37d825516833e49296e25fcc5fe48af4b4 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 17 Sep 2026 01:39:19 +0000 Subject: [PATCH 1/3] fix(hosts): give each host its own bundle journal and lock Co-authored-by: Zack Jackson --- .../src/agents/host_bundle/component_set.rs | 24 ++- .../src/agents/host_bundle/control.rs | 21 +- .../src/agents/host_bundle/doctor.rs | 72 +++---- .../src/agents/host_bundle/runtime.rs | 3 +- .../src/agents/host_bundle/tests.rs | 163 ++++++++++++++- .../src/agents/host_bundle/writer.rs | 187 ++++++++++++++---- 6 files changed, 388 insertions(+), 82 deletions(-) diff --git a/crates/tracedecay-agent-hosts/src/agents/host_bundle/component_set.rs b/crates/tracedecay-agent-hosts/src/agents/host_bundle/component_set.rs index b5b19c22d2..b57c878fb7 100644 --- a/crates/tracedecay-agent-hosts/src/agents/host_bundle/component_set.rs +++ b/crates/tracedecay-agent-hosts/src/agents/host_bundle/component_set.rs @@ -34,7 +34,7 @@ use super::{ HostBundleJournalEntryV1, HostBundleLifecycleOpV1, HostBundleManifestV1, HostBundleReceiptArtifactV1, HostBundleRollbackBoundaryV1, HostBundleVerificationAdapterV1, HostComponentSetJournalComponentV1, HostComponentSetJournalStateV1, HostComponentSetJournalV1, - HostComponentSetReceiptV1, HostComponentV1, HostKindV1, + HostComponentSetReceiptV1, HostComponentV1, HostKindV1, stock_host_kinds, }; /// Public component-set lifecycle façade over the capability-rooted writer. @@ -56,9 +56,19 @@ impl<'a> HostComponentSetTransactionV1<'a> { &mut self, registration: &mut R, ) -> Result<(), HostBundleError> { + let Some(journal) = self.writer.load_component_set_journal()? else { + for host in stock_host_kinds() { + if self.writer.load_journal_for(host)?.is_some() { + self.writer.recover_interrupted_operation(host)?; + } + } + return Ok(()); + }; + let host = journal.host; + self.writer.ensure_host_lock(host)?; self.writer - .recover_component_set_operation(None, registration)?; - self.writer.recover_interrupted_operation() + .recover_component_set_operation(Some(host), registration)?; + self.writer.recover_interrupted_operation(host) } /// Recover only `host`'s pending component-set journal. Other hosts' @@ -69,9 +79,10 @@ impl<'a> HostComponentSetTransactionV1<'a> { host: HostKindV1, registration: &mut R, ) -> Result<(), HostBundleError> { + self.writer.ensure_host_lock(host)?; self.writer .recover_component_set_operation(Some(host), registration)?; - self.writer.recover_interrupted_operation() + self.writer.recover_interrupted_operation(host) } pub fn preview( @@ -84,7 +95,7 @@ impl<'a> HostComponentSetTransactionV1<'a> { // Only this host's own pending journal blocks the preview. A wedged // transaction for an unrelated host mutates a disjoint path space and // is not a reason to refuse work here. - if self.writer.load_journal()?.is_some() + if self.writer.load_journal_for(component_set.host)?.is_some() || self .writer .load_component_set_journal_for(component_set.host)? @@ -238,7 +249,8 @@ impl HostBundleWriterV1 { registration: &mut R, ) -> Result { validate_component_set_request(component_set, request)?; - if self.load_journal()?.is_some() { + self.ensure_host_lock(component_set.host)?; + if self.load_journal_for(component_set.host)?.is_some() { return Err(host_bundle_recovery_required!()); } // Never clobber this host's own outstanding journal: it is the only diff --git a/crates/tracedecay-agent-hosts/src/agents/host_bundle/control.rs b/crates/tracedecay-agent-hosts/src/agents/host_bundle/control.rs index 6f372f5393..e54417e7aa 100644 --- a/crates/tracedecay-agent-hosts/src/agents/host_bundle/control.rs +++ b/crates/tracedecay-agent-hosts/src/agents/host_bundle/control.rs @@ -23,6 +23,11 @@ use super::{ }; pub(super) const HOST_BUNDLE_CONTROL_DIR: &str = ".tracedecay-host-bundle-v1"; +/// Legacy shared single-component journal. One file per lifecycle root meant +/// recovering host Y rolled back host X, and a wedged journal blocked every +/// other host. Journals are host-scoped now; this name is still read (and +/// retired) so a journal left by an older binary is recovered rather than +/// orphaned. pub(super) const HOST_BUNDLE_JOURNAL_FILE: &str = "journal.v1.json"; /// Legacy shared component-set journal name. One journal per lifecycle root /// meant an interrupted transaction for any host blocked every other host. @@ -33,6 +38,9 @@ pub(super) const HOST_COMPONENT_SET_STAGE_DIR: &str = "component-set-staging"; /// Set-aside directory for journals an operator explicitly abandoned with /// `tracedecay host-bundle recover --quarantine --yes`. Backups stay in place. pub(super) const HOST_BUNDLE_QUARANTINE_DIR: &str = "quarantine"; +/// Retired lifecycle-root lock. Hosts do not share a write target, so each +/// host owns `writer.{slug}.v1.lock`. This name is not acquired; a new binary +/// must not recreate it or independent hosts serialize again. pub(super) const HOST_BUNDLE_LOCK_FILE: &str = "writer.v1.lock"; pub(super) const MAX_CONTROL_FILE_BYTES: usize = 256 * 1024; @@ -561,12 +569,21 @@ pub(super) fn receipt_file(host: HostKindV1, component: HostComponentV1) -> Stri /// reason to refuse Y. `first_party_host_artifact_prefixes_are_disjoint` /// pins that premise as a test, so a future host that violates it fails the /// suite rather than silently widening the blast radius. The receipt namespace -/// is already host-scoped (`receipt_file`), and the single writer lock still -/// serializes all mutation within a lifecycle root. +/// is already host-scoped (`receipt_file`). The writer lock is host-scoped +/// too (`writer_lock_file`): one host's in-flight mutation is a real +/// invariant, a second host's is not. pub(super) fn component_set_journal_file(host: HostKindV1) -> String { format!("component-set-journal.{}.v1.json", host.descriptor().slug()) } +pub(super) fn journal_file(host: HostKindV1) -> String { + format!("journal.{}.v1.json", host.descriptor().slug()) +} + +pub(super) fn writer_lock_file(host: HostKindV1) -> String { + format!("writer.{}.v1.lock", host.descriptor().slug()) +} + pub(super) fn component_set_receipt_file(operation_id: [u8; 16]) -> String { format!( "component-set-receipt.{}.v1.json", diff --git a/crates/tracedecay-agent-hosts/src/agents/host_bundle/doctor.rs b/crates/tracedecay-agent-hosts/src/agents/host_bundle/doctor.rs index 41317a951b..1da0e26b2b 100644 --- a/crates/tracedecay-agent-hosts/src/agents/host_bundle/doctor.rs +++ b/crates/tracedecay-agent-hosts/src/agents/host_bundle/doctor.rs @@ -10,7 +10,7 @@ use tracedecay_host_integration::host_bundle_storage_failure; use super::control::{ HOST_BUNDLE_CONTROL_DIR, HOST_BUNDLE_JOURNAL_FILE, HOST_COMPONENT_SET_JOURNAL_FILE, - MAX_CONTROL_FILE_BYTES, component_set_journal_file, component_slug, receipt_file, + MAX_CONTROL_FILE_BYTES, component_set_journal_file, component_slug, journal_file, receipt_file, receipt_identity_from_file_name, validate_component_set_journal, validate_journal, validate_receipt, }; @@ -365,44 +365,50 @@ pub fn inspect_installed_host_bundle_components_at( repair_action: component_repair_action, }); } - let journal_path = control_root.join(HOST_BUNDLE_JOURNAL_FILE); - if journal_path.exists() { - let journal = fs::read(&journal_path) - .ok() - .filter(|bytes| !bytes.is_empty() && bytes.len() <= MAX_CONTROL_FILE_BYTES) - .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) - .filter(|journal| validate_journal(journal).is_ok()); - match journal { - Some(journal) => { - if let Some(component) = components.iter_mut().find(|component| { - component.host == Some(journal.host) - && component.component == Some(journal.component) - }) { - component.state = HostBundleComponentDoctorStateV1::Repairable; - component.repair_action = repair_action( - journal.host, - journal.component, - HostBundleComponentDoctorStateV1::Repairable, - HostBundleRegistrationStateV1::Current, - ); - } else { - components.push(HostBundleComponentDoctorResultV1 { - receipt_path: journal_path.clone(), - host: Some(journal.host), - component: Some(journal.component), - state: HostBundleComponentDoctorStateV1::Repairable, - registration: None, - artifacts: Vec::new(), - repair_action: repair_action( + // Single-component journals are host-scoped; the legacy shared name is + // still inspected so a journal left by an older binary stays visible. + let journal_paths = std::iter::once(HOST_BUNDLE_JOURNAL_FILE.to_string()) + .chain(stock_host_kinds().into_iter().map(journal_file)) + .map(|file| control_root.join(file)); + for journal_path in journal_paths { + if journal_path.exists() { + let journal = fs::read(&journal_path) + .ok() + .filter(|bytes| !bytes.is_empty() && bytes.len() <= MAX_CONTROL_FILE_BYTES) + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) + .filter(|journal| validate_journal(journal).is_ok()); + match journal { + Some(journal) => { + if let Some(component) = components.iter_mut().find(|component| { + component.host == Some(journal.host) + && component.component == Some(journal.component) + }) { + component.state = HostBundleComponentDoctorStateV1::Repairable; + component.repair_action = repair_action( journal.host, journal.component, HostBundleComponentDoctorStateV1::Repairable, HostBundleRegistrationStateV1::Current, - ), - }); + ); + } else { + components.push(HostBundleComponentDoctorResultV1 { + receipt_path: journal_path.clone(), + host: Some(journal.host), + component: Some(journal.component), + state: HostBundleComponentDoctorStateV1::Repairable, + registration: None, + artifacts: Vec::new(), + repair_action: repair_action( + journal.host, + journal.component, + HostBundleComponentDoctorStateV1::Repairable, + HostBundleRegistrationStateV1::Current, + ), + }); + } } + None => components.push(corrupt_component_result(journal_path, None, None)), } - None => components.push(corrupt_component_result(journal_path, None, None)), } } // Component-set journals are host-scoped; the legacy shared name is still diff --git a/crates/tracedecay-agent-hosts/src/agents/host_bundle/runtime.rs b/crates/tracedecay-agent-hosts/src/agents/host_bundle/runtime.rs index 9175f79912..7ae6e52018 100644 --- a/crates/tracedecay-agent-hosts/src/agents/host_bundle/runtime.rs +++ b/crates/tracedecay-agent-hosts/src/agents/host_bundle/runtime.rs @@ -104,7 +104,8 @@ where contents: &[HostBundleArtifactContentV1], ) -> Result { self.verifier.verify_manifest(manifest)?; - self.storage.recover_lifecycle()?; + // execute_lifecycle recovers this manifest's host. Recovering every + // host here would roll back an unrelated host's journal. self.storage .execute_lifecycle(manifest, request, contents, &self.verifier) } diff --git a/crates/tracedecay-agent-hosts/src/agents/host_bundle/tests.rs b/crates/tracedecay-agent-hosts/src/agents/host_bundle/tests.rs index b1cdd620cc..c6e9331e01 100644 --- a/crates/tracedecay-agent-hosts/src/agents/host_bundle/tests.rs +++ b/crates/tracedecay-agent-hosts/src/agents/host_bundle/tests.rs @@ -5,9 +5,10 @@ use sha2::{Digest, Sha256}; use tracedecay_host_integration::host_bundle_storage_failure; use super::control::{ - HOST_BUNDLE_CONTROL_DIR, HOST_COMPONENT_SET_JOURNAL_FILE, component_set_journal_file, - expected_ownership_marker, host_bundle_backup_receipt_file, host_bundle_restore_receipt_file, - receipt_file, validate_component_set_journal, + HOST_BUNDLE_CONTROL_DIR, HOST_BUNDLE_JOURNAL_FILE, HOST_BUNDLE_LOCK_FILE, + HOST_COMPONENT_SET_JOURNAL_FILE, component_set_journal_file, expected_ownership_marker, + host_bundle_backup_receipt_file, host_bundle_restore_receipt_file, journal_file, receipt_file, + validate_component_set_journal, writer_lock_file, }; use super::doctor::doctor_artifact_state; use super::planner::plan_artifact_action; @@ -896,6 +897,160 @@ fn host_scoped_component_set(host: HostKindV1, slug: &str, tag: &[u8]) -> HostCo } } +/// Defect: one `writer.v1.lock` and one `journal.v1.json` still serialized +/// every host after component-set journals were split. A writer that has +/// already admitted OpenCode must not stop Codex, and must not roll back +/// OpenCode's legacy single-component journal. +#[test] +fn a_host_lock_does_not_exclude_an_unrelated_host() { + let root = tempfile::tempdir().unwrap(); + let opencode = host_scoped_component_set(HostKindV1::OpenCode, "opencode", b"v1"); + let opencode_request = + component_set_request(HostKindV1::OpenCode, HostBundleLifecycleOpV1::Install, 61); + let mut holder = HostBundleWriterV1::open(root.path()).unwrap(); + HostComponentSetTransactionV1::new(&mut holder) + .execute( + &opencode, + &opencode_request, + &ComponentSetVerifier::from_set(&opencode), + &mut ArtifactOnlyTestRegistration, + ) + .expect("opencode install admits the host lock"); + assert!( + root.path() + .join(HOST_BUNDLE_CONTROL_DIR) + .join(writer_lock_file(HostKindV1::OpenCode)) + .is_file() + ); + assert!( + !root + .path() + .join(HOST_BUNDLE_CONTROL_DIR) + .join(HOST_BUNDLE_LOCK_FILE) + .exists(), + "the retired lifecycle-root lock must not be recreated" + ); + + let codex = host_scoped_component_set(HostKindV1::Codex, "codex", b"v1"); + let codex_request = + component_set_request(HostKindV1::Codex, HostBundleLifecycleOpV1::Install, 62); + let mut other = HostBundleWriterV1::open(root.path()).unwrap(); + HostComponentSetTransactionV1::new(&mut other) + .execute( + &codex, + &codex_request, + &ComponentSetVerifier::from_set(&codex), + &mut ArtifactOnlyTestRegistration, + ) + .expect("codex must not wait on opencode's writer lock"); + assert_eq!( + fs::read(root.path().join("codex/core.json")).unwrap(), + b"v1" + ); + + let contended = host_scoped_component_set(HostKindV1::OpenCode, "opencode", b"v2"); + let contended_request = + component_set_request(HostKindV1::OpenCode, HostBundleLifecycleOpV1::Repair, 63); + assert!( + matches!( + HostComponentSetTransactionV1::new(&mut other) + .execute( + &contended, + &contended_request, + &ComponentSetVerifier::from_set(&contended), + &mut ArtifactOnlyTestRegistration, + ) + .err(), + Some(HostBundleError::RecoveryRequired(_)) + ), + "the same host still has exactly one writer" + ); +} + +/// Defect: `journal.v1.json` was still one file for every host. Recovering or +/// installing Codex must not roll back an OpenCode journal left by an older +/// binary. OpenCode's own recovery retires that legacy name. +#[test] +fn a_legacy_single_component_journal_is_attributed_to_its_own_host() { + let root = tempfile::tempdir().unwrap(); + let artifact = root.path().join("opencode/core.json"); + fs::create_dir_all(artifact.parent().unwrap()).unwrap(); + fs::write(&artifact, b"opencode-bytes").unwrap(); + let digest: [u8; 32] = Sha256::digest(b"opencode-bytes").into(); + let journal = HostBundleJournalV1 { + schema_version: HOST_BUNDLE_RECEIPT_SCHEMA_VERSION, + operation_id: [71; 16], + host: HostKindV1::OpenCode, + component: HostComponentV1::Core, + operation: HostBundleLifecycleOpV1::Install, + manifest_digest: digest, + state: HostBundleJournalStateV1::Prepared, + previous_receipt: None, + entries: vec![HostBundleJournalEntryV1 { + relative_path: "opencode/core.json".to_string(), + backup_name: None, + backup_created: false, + wrote_new: true, + installed_digest: Some(digest), + }], + }; + let control = root.path().join(HOST_BUNDLE_CONTROL_DIR); + fs::create_dir_all(&control).unwrap(); + fs::write( + control.join(HOST_BUNDLE_JOURNAL_FILE), + serde_json::to_vec(&journal).unwrap(), + ) + .unwrap(); + + let codex = host_scoped_component_set(HostKindV1::Codex, "codex", b"v1"); + let codex_request = + component_set_request(HostKindV1::Codex, HostBundleLifecycleOpV1::Install, 72); + let mut writer = HostBundleWriterV1::open(root.path()).unwrap(); + HostComponentSetTransactionV1::new(&mut writer) + .execute( + &codex, + &codex_request, + &ComponentSetVerifier::from_set(&codex), + &mut ArtifactOnlyTestRegistration, + ) + .expect("codex install must not refuse on opencode's legacy journal"); + assert_eq!(fs::read(&artifact).unwrap(), b"opencode-bytes"); + assert!( + control.join(HOST_BUNDLE_JOURNAL_FILE).is_file(), + "codex must not retire opencode's legacy journal" + ); + + let doctor = inspect_installed_host_bundle_components_at( + root.path(), + root.path(), + &CurrentRegistration, + crate::agents::TEST_GENERATOR_COMMIT, + ) + .unwrap(); + assert!( + doctor.components.iter().any(|component| { + component.host == Some(HostKindV1::OpenCode) + && component.component == Some(HostComponentV1::Core) + && component.state == HostBundleComponentDoctorStateV1::Repairable + }), + "doctor still reports the legacy single-component journal" + ); + + HostComponentSetTransactionV1::new(&mut writer) + .recover_host(HostKindV1::OpenCode, &mut ArtifactOnlyTestRegistration) + .expect("opencode recovery owns the legacy journal"); + assert!( + !artifact.exists(), + "opencode recovery rolls its own interrupted install back" + ); + assert!(!control.join(HOST_BUNDLE_JOURNAL_FILE).exists()); + assert!(!control.join(journal_file(HostKindV1::OpenCode)).exists()); + assert_eq!( + fs::read(root.path().join("codex/core.json")).unwrap(), + b"v1" + ); +} + /// Defect: one shared journal per lifecycle root meant a wedged opencode /// repair blocked codex, cursor, cline, roo-code, kilo, kiro, and kimi in /// the same `tracedecay reinstall`. Journals are host-scoped now, and the @@ -1352,7 +1507,7 @@ fn feedback_switch_apply_restore_and_aggregate_receipt_share_writer_recovery() { &[], ) .unwrap(); - let writer = switch.into_lifecycle().into_storage(); + let mut writer = switch.into_lifecycle().into_storage(); let aggregate = writer .publish_feedback_component_set_receipt(&previous, &restore.restore_receipt) .unwrap(); diff --git a/crates/tracedecay-agent-hosts/src/agents/host_bundle/writer.rs b/crates/tracedecay-agent-hosts/src/agents/host_bundle/writer.rs index 4def178000..6056811051 100644 --- a/crates/tracedecay-agent-hosts/src/agents/host_bundle/writer.rs +++ b/crates/tracedecay-agent-hosts/src/agents/host_bundle/writer.rs @@ -21,9 +21,9 @@ use super::control::{ HOST_BUNDLE_QUARANTINE_DIR, HOST_COMPONENT_SET_JOURNAL_FILE, HOST_COMPONENT_SET_STAGE_DIR, MAX_CONTROL_FILE_BYTES, backup_name, component_set_journal_file, component_set_receipt_file, host_bundle_backup_receipt_file, host_bundle_restore_receipt_file, host_bundle_snapshot_name, - is_safe_component, latest_host_component_set_receipt_at, receipt_file, validate_backup_receipt, - validate_component_set_journal, validate_component_set_receipt, validate_journal, - validate_receipt, validate_restore_receipt, + is_safe_component, journal_file, latest_host_component_set_receipt_at, receipt_file, + validate_backup_receipt, validate_component_set_journal, validate_component_set_receipt, + validate_journal, validate_receipt, validate_restore_receipt, writer_lock_file, }; use super::model::{HostBundleExecutionRequestV1, HostBundleLifecycleStorageV1}; use super::planner::{ @@ -43,23 +43,41 @@ use super::{ static HOST_BUNDLE_TEMP_NONCE: AtomicU64 = AtomicU64::new(1); +/// Exclusive owner of one host's mutable bundle state. +/// +/// The lock is released when the writer switches hosts or is dropped. A +/// second host does not share this file: their artifact trees, journals, and +/// receipts are already disjoint. +struct HostWriterLock { + host: HostKindV1, + file: fs::File, +} + +impl Drop for HostWriterLock { + fn drop(&mut self) { + if let Err(error) = self.file.unlock() { + tracing::warn!( + error = %error, + host = ?self.host, + "host bundle writer lock could not be released" + ); + } + } +} + /// Atomic, capability-rooted host-bundle writer. Every descendant directory /// is opened without following symlinks; files are staged, fsynced, renamed, /// and followed by a directory sync before receipt publication. +/// +/// Opening does not take a lifecycle-root lock and does not recover another +/// host's journal. Mutation acquires `writer.{slug}.v1.lock` for the host +/// being written and holds it until the writer switches hosts or drops. pub struct HostBundleWriterV1 { pub(super) root_path: PathBuf, pub(super) lifecycle_root_path: PathBuf, root: Dir, control: Dir, - _writer_lock: fs::File, -} - -impl Drop for HostBundleWriterV1 { - fn drop(&mut self) { - if let Err(error) = self._writer_lock.unlock() { - tracing::warn!(error = %error, "host bundle writer lock could not be released"); - } - } + host_lock: Option, } impl HostBundleWriterV1 { @@ -81,26 +99,55 @@ impl HostBundleWriterV1 { let lifecycle_root = Dir::open_ambient_dir(&lifecycle_root_path, ambient_authority()) .map_err(|_| HostBundleError::UnsafeInstallPath)?; let control = open_or_create_nofollow_dir(&lifecycle_root, HOST_BUNDLE_CONTROL_DIR)?; - let writer_lock = open_writer_lock(&control)?; - let mut writer = Self { + Ok(Self { root_path, lifecycle_root_path, root, control, - _writer_lock: writer_lock, - }; - writer.recover_interrupted_operation()?; - Ok(writer) + host_lock: None, + }) + } + + /// Acquire this host's writer lock, releasing any other host's lock first. + /// + /// Same-host re-entry on this writer is a no-op. A different process that + /// already holds the host lock is refused; that is the one shared writer + /// this host actually needs. Independent hosts are not refused. + pub(super) fn ensure_host_lock(&mut self, host: HostKindV1) -> Result<(), HostBundleError> { + if self + .host_lock + .as_ref() + .is_some_and(|lock| lock.host == host) + { + return Ok(()); + } + self.host_lock = None; + if writer_lock_file(host) == HOST_BUNDLE_LOCK_FILE { + return Err(HostBundleError::UnsafeInstallPath); + } + self.host_lock = Some(open_host_writer_lock(&self.control, host)?); + Ok(()) } /// Recover by rolling an incomplete transaction back from its immutable /// backups. A receipt matching the journal operation is a durable commit /// marker and is never rolled back after a crash between receipt/journal /// cleanup. - pub fn recover_interrupted_operation(&mut self) -> Result<(), HostBundleError> { - let Some(journal) = self.load_journal()? else { + pub fn recover_interrupted_operation( + &mut self, + host: HostKindV1, + ) -> Result<(), HostBundleError> { + self.ensure_host_lock(host)?; + self.recover_host_journal_locked(host) + } + + fn recover_host_journal_locked(&mut self, host: HostKindV1) -> Result<(), HostBundleError> { + let Some(journal) = self.load_journal_for(host)? else { return Ok(()); }; + if journal.host != host { + return Err(HostBundleError::ReceiptCorrupted); + } validate_journal(&journal)?; if let Some(receipt) = self.load_receipt(journal.host, journal.component)? @@ -110,7 +157,7 @@ impl HostBundleWriterV1 { && receipt.manifest_digest == journal.manifest_digest }) { - self.remove_control_file(HOST_BUNDLE_JOURNAL_FILE)?; + self.remove_journal(host)?; if receipt.rollback_boundary == HostBundleRollbackBoundaryV1::Passed { self.cleanup_unreferenced_backup_dir(journal.operation_id)?; } @@ -169,7 +216,7 @@ impl HostBundleWriterV1 { Some(receipt) => self.write_receipt(&receipt)?, None => self.remove_receipt(journal.host, journal.component)?, } - self.remove_control_file(HOST_BUNDLE_JOURNAL_FILE)?; + self.remove_journal(host)?; self.cleanup_unreferenced_backup_dir(journal.operation_id) } @@ -186,6 +233,7 @@ impl HostBundleWriterV1 { if request.operation_id == [0; 16] { return Err(HostBundleError::InvalidManifest); } + self.ensure_host_lock(manifest.host)?; verifier.verify_manifest(manifest)?; let content_by_path = validate_artifact_contents(manifest, request, contents)?; // Scoped to this manifest's own host: another host's pending @@ -196,7 +244,7 @@ impl HostBundleWriterV1 { { return Err(host_bundle_recovery_required!()); } - self.recover_interrupted_operation()?; + self.recover_host_journal_locked(manifest.host)?; let previous_receipt = self.load_receipt(manifest.host, manifest.component)?; let manifest_digest = manifest.canonical_digest()?; if let Some(receipt) = previous_receipt.as_ref() @@ -361,7 +409,7 @@ impl HostBundleWriterV1 { self.write_receipt(&receipt)?; journal.state = HostBundleJournalStateV1::Committed; self.write_journal(&journal)?; - self.remove_control_file(HOST_BUNDLE_JOURNAL_FILE)?; + self.remove_journal(manifest.host)?; if receipt.rollback_boundary == HostBundleRollbackBoundaryV1::Passed { self.cleanup_unreferenced_backup_dir(request.operation_id)?; } @@ -569,14 +617,37 @@ impl HostBundleWriterV1 { self.remove_control_file(&component_set_receipt_file(operation_id)) } - pub(super) fn load_journal(&self) -> Result, HostBundleError> { - read_control_json(&self.control, HOST_BUNDLE_JOURNAL_FILE)? + fn read_journal_file( + &self, + file_name: &str, + ) -> Result, HostBundleError> { + read_control_json(&self.control, file_name)? .map(|bytes| { serde_json::from_slice(&bytes).map_err(|_| HostBundleError::ReceiptCorrupted) }) .transpose() } + /// Load the pending single-component journal for one host. + /// + /// A journal written by an older binary lives under the shared legacy name + /// and carries its own `host` field, so it is attributed to exactly one + /// host. Recovering a different host must not see it. + pub(super) fn load_journal_for( + &self, + host: HostKindV1, + ) -> Result, HostBundleError> { + if let Some(journal) = self.read_journal_file(&journal_file(host))? { + if journal.host != host { + return Err(HostBundleError::ReceiptCorrupted); + } + return Ok(Some(journal)); + } + Ok(self + .read_journal_file(HOST_BUNDLE_JOURNAL_FILE)? + .filter(|journal| journal.host == host)) + } + fn read_component_set_journal_file( &self, file_name: &str, @@ -651,7 +722,29 @@ impl HostBundleWriterV1 { fn write_journal(&self, journal: &HostBundleJournalV1) -> Result<(), HostBundleError> { validate_journal(journal)?; let bytes = serde_json::to_vec(journal).map_err(|_| HostBundleError::ReceiptCorrupted)?; - atomic_write_nofollow(&self.control, HOST_BUNDLE_JOURNAL_FILE, &bytes, true) + atomic_write_nofollow(&self.control, &journal_file(journal.host), &bytes, true)?; + // A journal written by an older binary lives under the shared legacy + // name. Once its host-scoped successor is durable, retire it so the + // legacy file can never shadow or double-recover this transaction. + // Never unlink a legacy journal that belongs to a different host. + if self + .read_journal_file(HOST_BUNDLE_JOURNAL_FILE)? + .is_some_and(|legacy| legacy.host == journal.host) + { + self.remove_control_file(HOST_BUNDLE_JOURNAL_FILE)?; + } + Ok(()) + } + + fn remove_journal(&self, host: HostKindV1) -> Result<(), HostBundleError> { + self.remove_control_file(&journal_file(host))?; + if self + .read_journal_file(HOST_BUNDLE_JOURNAL_FILE)? + .is_some_and(|legacy| legacy.host == host) + { + self.remove_control_file(HOST_BUNDLE_JOURNAL_FILE)?; + } + Ok(()) } #[hotpath::measure(label = "hosts.agent.host_bundle.component_set_journal_persist")] @@ -701,6 +794,7 @@ impl HostBundleWriterV1 { host: HostKindV1, now_unix: u64, ) -> Result, HostBundleError> { + self.ensure_host_lock(host)?; let mut moved = None; for file in [ component_set_journal_file(host), @@ -840,7 +934,7 @@ impl HostBundleWriterV1 { /// the same operation id returns the existing receipt after revalidation. /// A missing, edited, or foreign artifact fails before receipt publication. pub fn backup_component( - &self, + &mut self, manifest: &HostBundleManifestV1, operation_id: [u8; 16], explicit_confirmation: bool, @@ -852,6 +946,7 @@ impl HostBundleWriterV1 { if !explicit_confirmation { return Err(HostBundleError::ConfirmationRequired); } + self.ensure_host_lock(manifest.host)?; manifest.validate_structure()?; verifier.verify_manifest(manifest)?; if let Some(receipt) = self.load_backup_receipt(operation_id)? { @@ -978,7 +1073,7 @@ impl HostBundleWriterV1 { } pub fn publish_feedback_component_set_receipt( - &self, + &mut self, manifest: &HostBundleManifestV1, component_receipt: &HostBundleInstallReceiptV1, ) -> Result { @@ -988,6 +1083,7 @@ impl HostBundleWriterV1 { { return Err(HostBundleError::WrongTarget); } + self.ensure_host_lock(component_receipt.host)?; let previous = latest_host_component_set_receipt_at( &self.lifecycle_root_path, component_receipt.host, @@ -1130,7 +1226,14 @@ impl HostBundleWriterV1 { impl HostBundleLifecycleStorageV1 for HostBundleWriterV1 { fn recover_lifecycle(&mut self) -> Result<(), HostBundleError> { - self.recover_interrupted_operation() + // Explicit recover-all. Each host owns its journal, so each recovery + // takes only that host's lock and releases it before the next host. + for host in stock_host_kinds() { + if self.load_journal_for(host)?.is_some() { + self.recover_interrupted_operation(host)?; + } + } + Ok(()) } fn execute_lifecycle( @@ -1175,9 +1278,14 @@ fn open_or_create_nofollow_dir(parent: &Dir, name: &str) -> Result Ok(directory), Err(error) if error.kind() == io::ErrorKind::NotFound => { - parent - .create_dir(name) - .map_err(|_| host_bundle_storage_failure!())?; + match parent.create_dir(name) { + Ok(()) => {} + // Two hosts may create a shared parent (`backups/`, `.config/`) + // at once. The directory is a namespace, not a shared state + // object; the loser retries the open instead of failing. + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {} + Err(_) => return Err(host_bundle_storage_failure!()), + } parent .open_dir_nofollow(name) .map_err(|_| HostBundleError::UnsafeInstallPath) @@ -1186,7 +1294,14 @@ fn open_or_create_nofollow_dir(parent: &Dir, name: &str) -> Result Result { +fn open_host_writer_lock( + control: &Dir, + host: HostKindV1, +) -> Result { + let name = writer_lock_file(host); + if !is_safe_component(&name) { + return Err(HostBundleError::UnsafeInstallPath); + } let mut options = CapOpenOptions::new(); options .read(true) @@ -1194,12 +1309,12 @@ fn open_writer_lock(control: &Dir) -> Result { .create(true) .follow(FollowSymlinks::No); let file = control - .open_with(HOST_BUNDLE_LOCK_FILE, &options) + .open_with(&name, &options) .map_err(|_| HostBundleError::UnsafeInstallPath)? .into_std(); file.try_lock_exclusive() .map_err(|_| host_bundle_recovery_required!())?; - Ok(file) + Ok(HostWriterLock { host, file }) } pub(super) fn read_regular_nofollow( From 4f8bf8e9836042092214bfd34d6c772643840b17 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 17 Sep 2026 01:39:23 +0000 Subject: [PATCH 2/3] fix(automation): store skill usage in per-skill files Co-authored-by: Zack Jackson --- .../src/automation/managed_skills.rs | 7 +- .../src/automation/outcomes_test.rs | 2 + .../src/automation/skill_usage.rs | 189 ++++++---- .../src/automation/skill_usage/analytics.rs | 44 +-- .../automation/skill_usage/recommendations.rs | 2 + .../src/automation/skill_usage/store.rs | 356 ++++++++++++++++++ .../tests/agent_suite/managed_skills_test.rs | 7 +- 7 files changed, 505 insertions(+), 102 deletions(-) create mode 100644 crates/tracedecay-automation-runtime/src/automation/skill_usage/store.rs diff --git a/crates/tracedecay-automation-runtime/src/automation/managed_skills.rs b/crates/tracedecay-automation-runtime/src/automation/managed_skills.rs index 0d06ed1787..32a2521ad2 100644 --- a/crates/tracedecay-automation-runtime/src/automation/managed_skills.rs +++ b/crates/tracedecay-automation-runtime/src/automation/managed_skills.rs @@ -1252,9 +1252,10 @@ mod transaction_tests { let profile = temp.path(); let original = skill("ledger-skill", "# Ledger"); save_managed_skill(profile, &original).await.unwrap(); - let ledger_path = super::super::skill_usage::skill_usage_ledger_path(profile); - std::fs::remove_file(&ledger_path).unwrap(); - std::fs::create_dir(&ledger_path).unwrap(); + let record_path = + super::super::skill_usage::skill_usage_record_path(profile, "ledger-skill"); + std::fs::remove_file(&record_path).unwrap(); + std::fs::create_dir(&record_path).unwrap(); let pinned = set_managed_skill_pinned(profile, "ledger-skill", true) .await diff --git a/crates/tracedecay-automation-runtime/src/automation/outcomes_test.rs b/crates/tracedecay-automation-runtime/src/automation/outcomes_test.rs index 9bbb700309..a053939e9e 100644 --- a/crates/tracedecay-automation-runtime/src/automation/outcomes_test.rs +++ b/crates/tracedecay-automation-runtime/src/automation/outcomes_test.rs @@ -1,6 +1,7 @@ use super::super::skill_usage::SkillUsageRecord; use super::*; use crate::automation::AutomationRunControl; +use std::collections::BTreeSet; use std::sync::Arc; static OUTCOME_PERSISTENCE_DB_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); @@ -33,6 +34,7 @@ fn summary(skill_id: &str) -> SkillUsageRecord { activated_at: None, view_count_at_activation: None, use_count_at_activation: None, + imported_analytics_events: BTreeSet::new(), } } diff --git a/crates/tracedecay-automation-runtime/src/automation/skill_usage.rs b/crates/tracedecay-automation-runtime/src/automation/skill_usage.rs index 7ba880037d..020746e0a0 100644 --- a/crates/tracedecay-automation-runtime/src/automation/skill_usage.rs +++ b/crates/tracedecay-automation-runtime/src/automation/skill_usage.rs @@ -5,12 +5,13 @@ use serde::{Deserialize, Serialize}; use super::config_error; use super::managed_skills::{ManagedSkill, ManagedSkillSource, ManagedSkillState}; -use tracedecay_domain::errors::{Result, TraceDecayError}; +use tracedecay_domain::errors::Result; use tracedecay_runtime_core::tracedecay::current_timestamp; mod analytics; mod overlap; mod recommendations; +mod store; pub use crate::ports::session_store::AnalyticsEventRecord; pub use analytics::analytics_import_key_for_request; @@ -72,6 +73,11 @@ pub struct SkillUsageRecord { pub view_count_at_activation: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub use_count_at_activation: Option, + /// Import keys that already counted toward this skill. They are not a + /// store-wide set: each key names this skill, so another skill must not + /// share the write. + #[serde(default, skip_serializing_if = "BTreeSet::is_empty")] + pub imported_analytics_events: BTreeSet, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -139,6 +145,7 @@ impl SkillUsageRecord { activated_at: None, view_count_at_activation: None, use_count_at_activation: None, + imported_analytics_events: BTreeSet::new(), } } @@ -188,58 +195,39 @@ pub fn skill_usage_ledger_path(profile_root: &Path) -> PathBuf { .join(SKILL_USAGE_LEDGER_FILENAME) } +pub fn skill_usage_record_path(profile_root: &Path, skill_id: &str) -> PathBuf { + store::skill_usage_record_path(profile_root, skill_id) +} + #[hotpath::measure(label = "automation.skill_usage.load", future = true)] pub async fn load_skill_usage_ledger(profile_root: &Path) -> Result { - let path = skill_usage_ledger_path(profile_root); - let bytes = match tokio::fs::read(&path).await { - Ok(bytes) => bytes, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => { - return Ok(SkillUsageLedger::default()); - } - Err(e) => { - return Err(config_error(format!( - "failed to read skill usage ledger '{}': {e}", - path.display() - ))); - } - }; - serde_json::from_slice(&bytes).map_err(|e| { - config_error(format!( - "failed to parse skill usage ledger '{}': {e}", - path.display() - )) - }) + store::load_ledger(profile_root).await } #[hotpath::measure(label = "automation.skill_usage.save", future = true)] pub async fn save_skill_usage_ledger(profile_root: &Path, ledger: &SkillUsageLedger) -> Result<()> { - let path = skill_usage_ledger_path(profile_root); - if let Some(parent) = path.parent() { - tokio::fs::create_dir_all(parent).await.map_err(|e| { - config_error(format!( - "failed to create skill usage ledger directory '{}': {e}", - parent.display() - )) - })?; + // Split the snapshot. Do not rewrite an aggregate file, and do not delete + // a skill file that this snapshot does not mention. + for record in ledger.records.values() { + let owned = record.clone(); + let skill_id = owned.skill_id.clone(); + let first_seen_at = owned.first_seen_at; + store::update_record(profile_root, &skill_id, first_seen_at, move |slot| { + *slot = owned; + }) + .await?; } - let bytes = serde_json::to_vec_pretty(ledger).map_err(TraceDecayError::from)?; - tokio::fs::write(&path, bytes).await.map_err(|e| { - config_error(format!( - "failed to write skill usage ledger '{}': {e}", - path.display() - )) - }) + Ok(()) } pub async fn sync_skill_usage_metadata(profile_root: &Path, skill: &ManagedSkill) -> Result<()> { - let mut ledger = load_skill_usage_ledger(profile_root).await?; + let skill = skill.clone(); let skill_id = skill.metadata.id.clone(); - let record = ledger - .records - .entry(skill_id.clone()) - .or_insert_with(|| SkillUsageRecord::new(skill_id, 0)); - record.merge_skill_metadata(skill); - save_skill_usage_ledger(profile_root, &ledger).await + store::update_record(profile_root, &skill_id, 0, move |record| { + record.merge_skill_metadata(&skill); + }) + .await + .map(|_| ()) } pub async fn record_skill_usage_event( @@ -248,18 +236,14 @@ pub async fn record_skill_usage_event( skill: Option<&ManagedSkill>, ) -> Result { let skill_id = ledger_skill_id(&event.skill_name)?; - let mut ledger = load_skill_usage_ledger(profile_root).await?; - let record = ledger - .records - .entry(skill_id.clone()) - .or_insert_with(|| SkillUsageRecord::new(skill_id, event.timestamp)); - if let Some(skill) = skill { - record.merge_skill_metadata(skill); - } - record.record(&event); - let updated = record.clone(); - save_skill_usage_ledger(profile_root, &ledger).await?; - Ok(updated) + let skill = skill.cloned(); + store::update_record(profile_root, &skill_id, event.timestamp, move |record| { + if let Some(skill) = skill.as_ref() { + record.merge_skill_metadata(skill); + } + record.record(&event); + }) + .await } pub async fn record_skill_usage( @@ -273,13 +257,16 @@ pub async fn record_skill_usage( ) -> Result { let skill_id = skill.metadata.id.clone(); let timestamp = current_timestamp(); - let mut ledger = load_skill_usage_ledger(profile_root).await?; - let updated = { - let record = ledger - .records - .entry(skill_id.clone()) - .or_insert_with(|| SkillUsageRecord::new(skill_id, timestamp)); - record.merge_skill_metadata(skill); + let skill = skill.clone(); + let import_key = metadata + .as_ref() + .and_then(|metadata| metadata.get("imported_analytics_event_key")) + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|key| !key.is_empty()) + .map(str::to_string); + store::update_record(profile_root, &skill_id, timestamp, move |record| { + record.merge_skill_metadata(&skill); record.record(&SkillUsageEvent { skill_name: skill.metadata.id.clone(), action, @@ -291,21 +278,11 @@ pub async fn record_skill_usage( insert_sorted_unique(&mut record.targets, target); } } - record.clone() - }; - if let Some(import_key) = metadata - .as_ref() - .and_then(|metadata| metadata.get("imported_analytics_event_key")) - .and_then(serde_json::Value::as_str) - .map(str::trim) - .filter(|key| !key.is_empty()) - { - ledger - .imported_analytics_events - .insert(import_key.to_string()); - } - save_skill_usage_ledger(profile_root, &ledger).await?; - Ok(updated) + if let Some(import_key) = import_key { + record.imported_analytics_events.insert(import_key); + } + }) + .await } pub async fn load_skill_usage_records( @@ -410,3 +387,63 @@ fn insert_sorted_unique(values: &mut Vec, value: String) { fn max_optional(existing: Option, timestamp: i64) -> i64 { existing.map_or(timestamp, |current| current.max(timestamp)) } + +#[cfg(test)] +mod tests { + use super::{SkillUsageAction, SkillUsageEvent, record_skill_usage_event}; + + #[tokio::test] + async fn recording_one_skill_does_not_rewrite_another_skills_file() { + let root = tempfile::tempdir().unwrap(); + record_skill_usage_event( + root.path(), + SkillUsageEvent { + skill_name: "skill-a".to_string(), + action: SkillUsageAction::Use, + timestamp: 10, + target: None, + }, + None, + ) + .await + .unwrap(); + record_skill_usage_event( + root.path(), + SkillUsageEvent { + skill_name: "skill-b".to_string(), + action: SkillUsageAction::View, + timestamp: 11, + target: None, + }, + None, + ) + .await + .unwrap(); + let before = super::load_skill_usage_ledger(root.path()).await.unwrap(); + assert_eq!(before.records["skill-b"].view_count, 1); + + record_skill_usage_event( + root.path(), + SkillUsageEvent { + skill_name: "skill-a".to_string(), + action: SkillUsageAction::Use, + timestamp: 12, + target: None, + }, + None, + ) + .await + .unwrap(); + + let after = super::load_skill_usage_ledger(root.path()).await.unwrap(); + assert_eq!(after.records["skill-a"].use_count, 2); + assert_eq!( + after.records["skill-b"].view_count, 1, + "skill B's last-view must survive skill A's write" + ); + assert!( + !super::skill_usage_ledger_path(root.path()).exists(), + "independent skills must not share skill_usage.json" + ); + } +} diff --git a/crates/tracedecay-automation-runtime/src/automation/skill_usage/analytics.rs b/crates/tracedecay-automation-runtime/src/automation/skill_usage/analytics.rs index f1f591ecb6..910b2650b8 100644 --- a/crates/tracedecay-automation-runtime/src/automation/skill_usage/analytics.rs +++ b/crates/tracedecay-automation-runtime/src/automation/skill_usage/analytics.rs @@ -7,16 +7,13 @@ use crate::ports::session_store::{ use tracedecay_automation::analytics::{UsageKind, infer_usage_events}; use tracedecay_domain::errors::Result; -use super::{ - SkillUsageAction, SkillUsageEvent, SkillUsageRecord, config_error, ledger_skill_id, - load_skill_usage_ledger, save_skill_usage_ledger, -}; +use super::store; +use super::{SkillUsageAction, SkillUsageEvent, SkillUsageRecord, config_error, ledger_skill_id}; pub async fn ingest_analytics_events( profile_root: &Path, events: &[AnalyticsEventRecord], ) -> Result> { - let mut ledger = load_skill_usage_ledger(profile_root).await?; let mut touched = Vec::new(); let mut seen = BTreeSet::new(); for event in events { @@ -33,25 +30,30 @@ pub async fn ingest_analytics_events( if !seen.insert(dedupe.clone()) { continue; } - if !ledger.imported_analytics_events.insert(dedupe) { + let skill_name = usage.name.clone(); + let provider = event.provider.clone(); + let timestamp = event.timestamp; + let Some(record) = store::record_imported_event( + profile_root, + &skill_id, + timestamp, + dedupe, + move |record| { + record.record(&SkillUsageEvent { + skill_name, + action, + timestamp, + target: Some(provider), + }); + }, + ) + .await? + else { continue; - } - let record = ledger - .records - .entry(skill_id.clone()) - .or_insert_with(|| SkillUsageRecord::new(skill_id, event.timestamp)); - record.record(&SkillUsageEvent { - skill_name: usage.name, - action, - timestamp: event.timestamp, - target: Some(event.provider.clone()), - }); - touched.push(record.clone()); + }; + touched.push(record); } } - if !touched.is_empty() { - save_skill_usage_ledger(profile_root, &ledger).await?; - } Ok(touched) } diff --git a/crates/tracedecay-automation-runtime/src/automation/skill_usage/recommendations.rs b/crates/tracedecay-automation-runtime/src/automation/skill_usage/recommendations.rs index 808cb7c73e..2ea965f2d9 100644 --- a/crates/tracedecay-automation-runtime/src/automation/skill_usage/recommendations.rs +++ b/crates/tracedecay-automation-runtime/src/automation/skill_usage/recommendations.rs @@ -269,6 +269,7 @@ mod tests { use super::super::super::outcomes::SKILL_ACTIVATION_WINDOW_SECS; use super::super::SkillUsageRecord; use super::*; + use std::collections::BTreeSet; /// An activated, active, automation-authored skill with recent views but /// no uses since activation: kept by the plain activity heuristics, so any @@ -295,6 +296,7 @@ mod tests { activated_at: Some(now - SKILL_ACTIVATION_WINDOW_SECS - 1), view_count_at_activation: Some(0), use_count_at_activation: Some(0), + imported_analytics_events: BTreeSet::new(), } } diff --git a/crates/tracedecay-automation-runtime/src/automation/skill_usage/store.rs b/crates/tracedecay-automation-runtime/src/automation/skill_usage/store.rs new file mode 100644 index 0000000000..98702e0020 --- /dev/null +++ b/crates/tracedecay-automation-runtime/src/automation/skill_usage/store.rs @@ -0,0 +1,356 @@ +//! Per-skill usage files. Independent skills must not share a write target. +//! +//! The legacy `skill_usage.json` map is read once and split. After that, each +//! skill owns `skill_usage/.json`. Import-dedupe keys that +//! already name a skill live on that skill's record. The leftover legacy key +//! set is drained to a read-only file and never written by a new event. + +use std::collections::BTreeSet; +use std::fs::{self, File, OpenOptions}; +use std::io::{self, Write}; +use std::path::{Path, PathBuf}; + +use fs2::FileExt; +use serde::Serialize; + +use super::{SkillUsageLedger, SkillUsageRecord, config_error, skill_usage_ledger_path}; +use tracedecay_domain::errors::{Result, TraceDecayError}; + +const SKILL_USAGE_DIR: &str = "skill_usage"; +const LEGACY_IMPORTS_FILE: &str = "legacy-imported-events.json"; +const MIGRATE_LOCK_FILE: &str = ".migrate.lock"; + +pub(super) fn skill_usage_dir(profile_root: &Path) -> PathBuf { + profile_root.join("agent_managed").join(SKILL_USAGE_DIR) +} + +pub(super) fn skill_usage_record_path(profile_root: &Path, skill_id: &str) -> PathBuf { + skill_usage_dir(profile_root).join(record_file_name(skill_id)) +} + +pub(super) async fn load_ledger(profile_root: &Path) -> Result { + let root = profile_root.to_path_buf(); + tokio::task::spawn_blocking(move || load_ledger_sync(&root)) + .await + .map_err(|error| config_error(format!("skill usage load task failed: {error}")))? +} + +pub(super) async fn update_record( + profile_root: &Path, + skill_id: &str, + seed_timestamp: i64, + mutate: impl FnOnce(&mut SkillUsageRecord) + Send + 'static, +) -> Result { + let root = profile_root.to_path_buf(); + let skill_id = skill_id.to_string(); + tokio::task::spawn_blocking(move || { + migrate_legacy(&root)?; + with_skill_lock( + &root, + &skill_id, + |record| { + mutate(record); + record.skill_id = skill_id.clone(); + Ok(true) + }, + seed_timestamp, + ) + }) + .await + .map_err(|error| config_error(format!("skill usage update task failed: {error}")))? +} + +pub(super) async fn record_imported_event( + profile_root: &Path, + skill_id: &str, + seed_timestamp: i64, + import_key: String, + mutate: impl FnOnce(&mut SkillUsageRecord) + Send + 'static, +) -> Result> { + let root = profile_root.to_path_buf(); + let skill_id = skill_id.to_string(); + tokio::task::spawn_blocking(move || { + migrate_legacy(&root)?; + let legacy = read_legacy_imports(&root)?; + let mut applied = false; + let record = with_skill_lock( + &root, + &skill_id, + |record| { + if record.imported_analytics_events.contains(&import_key) + || legacy.contains(&import_key) + { + return Ok(false); + } + record.imported_analytics_events.insert(import_key.clone()); + mutate(record); + applied = true; + Ok(true) + }, + seed_timestamp, + )?; + Ok(applied.then_some(record)) + }) + .await + .map_err(|error| config_error(format!("skill usage import task failed: {error}")))? +} + +fn load_ledger_sync(profile_root: &Path) -> Result { + migrate_legacy(profile_root)?; + let mut ledger = SkillUsageLedger::default(); + let directory = skill_usage_dir(profile_root); + let entries = match fs::read_dir(&directory) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + ledger.imported_analytics_events = read_legacy_imports(profile_root)?; + return Ok(ledger); + } + Err(error) => { + return Err(config_error(format!( + "failed to read skill usage directory '{}': {error}", + directory.display() + ))); + } + }; + for entry in entries { + let entry = entry.map_err(|error| { + config_error(format!( + "failed to read skill usage directory '{}': {error}", + directory.display() + )) + })?; + let name = entry.file_name(); + let Some(name) = name.to_str() else { + continue; + }; + if !is_skill_record_file(name) { + continue; + } + let record = read_record(&entry.path())?; + ledger + .imported_analytics_events + .extend(record.imported_analytics_events.iter().cloned()); + ledger.records.insert(record.skill_id.clone(), record); + } + ledger + .imported_analytics_events + .extend(read_legacy_imports(profile_root)?); + Ok(ledger) +} + +fn with_skill_lock( + profile_root: &Path, + skill_id: &str, + mutate: impl FnOnce(&mut SkillUsageRecord) -> Result, + seed_timestamp: i64, +) -> Result { + let directory = skill_usage_dir(profile_root); + fs::create_dir_all(&directory).map_err(|error| { + config_error(format!( + "failed to create skill usage directory '{}': {error}", + directory.display() + )) + })?; + let lock_path = directory.join(format!("{}.lock", record_stem(skill_id))); + let lock = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .open(&lock_path) + .map_err(|error| { + config_error(format!( + "failed to open skill usage lock '{}': {error}", + lock_path.display() + )) + })?; + lock.lock_exclusive().map_err(|error| { + config_error(format!( + "failed to lock skill usage record '{}': {error}", + skill_id + )) + })?; + let path = skill_usage_record_path(profile_root, skill_id); + let existing = read_record_if_present(&path)?; + let mut record = existing + .clone() + .unwrap_or_else(|| SkillUsageRecord::new(skill_id.to_string(), seed_timestamp)); + record.skill_id = skill_id.to_string(); + if mutate(&mut record)? { + write_json(&path, &record)?; + } + let _ = lock.unlock(); + Ok(if path.exists() { + read_record(&path)? + } else { + record + }) +} + +fn migrate_legacy(profile_root: &Path) -> Result<()> { + let legacy_path = skill_usage_ledger_path(profile_root); + if !legacy_path.exists() { + return Ok(()); + } + let directory = skill_usage_dir(profile_root); + fs::create_dir_all(&directory).map_err(|error| { + config_error(format!( + "failed to create skill usage directory '{}': {error}", + directory.display() + )) + })?; + let lock_path = directory.join(MIGRATE_LOCK_FILE); + let lock = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .open(&lock_path) + .map_err(|error| { + config_error(format!( + "failed to open skill usage migration lock '{}': {error}", + lock_path.display() + )) + })?; + lock.lock_exclusive() + .map_err(|error| config_error(format!("failed to lock skill usage migration: {error}")))?; + let result = migrate_legacy_locked(profile_root, &legacy_path); + let _ = lock.unlock(); + result +} + +fn migrate_legacy_locked(profile_root: &Path, legacy_path: &Path) -> Result<()> { + if !legacy_path.exists() { + return Ok(()); + } + let ledger = read_legacy_ledger(legacy_path)?; + for (skill_id, record) in ledger.records { + let path = skill_usage_record_path(profile_root, &skill_id); + if path.exists() { + continue; + } + write_json(&path, &record)?; + } + let imports_path = skill_usage_dir(profile_root).join(LEGACY_IMPORTS_FILE); + if !imports_path.exists() && !ledger.imported_analytics_events.is_empty() { + write_json(&imports_path, &ledger.imported_analytics_events)?; + } + match fs::remove_file(legacy_path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(config_error(format!( + "failed to retire skill usage ledger '{}': {error}", + legacy_path.display() + ))), + } +} + +fn read_legacy_ledger(path: &Path) -> Result { + let bytes = fs::read(path).map_err(|error| { + config_error(format!( + "failed to read skill usage ledger '{}': {error}", + path.display() + )) + })?; + serde_json::from_slice(&bytes).map_err(|error| { + config_error(format!( + "failed to parse skill usage ledger '{}': {error}", + path.display() + )) + }) +} + +fn read_legacy_imports(profile_root: &Path) -> Result> { + let path = skill_usage_dir(profile_root).join(LEGACY_IMPORTS_FILE); + let bytes = match fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(BTreeSet::new()), + Err(error) => { + return Err(config_error(format!( + "failed to read legacy skill usage imports '{}': {error}", + path.display() + ))); + } + }; + serde_json::from_slice(&bytes).map_err(|error| { + config_error(format!( + "failed to parse legacy skill usage imports '{}': {error}", + path.display() + )) + }) +} + +fn read_record_if_present(path: &Path) -> Result> { + match fs::read(path) { + Ok(bytes) => serde_json::from_slice(&bytes).map(Some).map_err(|error| { + config_error(format!( + "failed to parse skill usage record '{}': {error}", + path.display() + )) + }), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(config_error(format!( + "failed to read skill usage record '{}': {error}", + path.display() + ))), + } +} + +fn read_record(path: &Path) -> Result { + read_record_if_present(path)?.ok_or_else(|| { + config_error(format!( + "skill usage record disappeared '{}'", + path.display() + )) + }) +} + +fn write_json(path: &Path, value: &impl Serialize) -> Result<()> { + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).map_err(|error| { + config_error(format!( + "failed to create skill usage directory '{}': {error}", + parent.display() + )) + })?; + } + let bytes = serde_json::to_vec_pretty(value).map_err(TraceDecayError::from)?; + let temporary = path.with_extension("json.tmp"); + let mut file = File::create(&temporary).map_err(|error| { + config_error(format!( + "failed to write skill usage record '{}': {error}", + temporary.display() + )) + })?; + file.write_all(&bytes).map_err(|error| { + config_error(format!( + "failed to write skill usage record '{}': {error}", + temporary.display() + )) + })?; + file.sync_all().map_err(|error| { + config_error(format!( + "failed to sync skill usage record '{}': {error}", + temporary.display() + )) + })?; + fs::rename(&temporary, path).map_err(|error| { + config_error(format!( + "failed to publish skill usage record '{}': {error}", + path.display() + )) + }) +} + +fn record_file_name(skill_id: &str) -> String { + format!("{}.json", record_stem(skill_id)) +} + +fn record_stem(skill_id: &str) -> String { + hex::encode(skill_id.as_bytes()) +} + +fn is_skill_record_file(name: &str) -> bool { + let Some(stem) = name.strip_suffix(".json") else { + return false; + }; + !stem.is_empty() && stem.len() % 2 == 0 && stem.bytes().all(|byte| byte.is_ascii_hexdigit()) +} diff --git a/crates/tracedecay/tests/agent_suite/managed_skills_test.rs b/crates/tracedecay/tests/agent_suite/managed_skills_test.rs index cd5a9e47c8..f36bba590c 100644 --- a/crates/tracedecay/tests/agent_suite/managed_skills_test.rs +++ b/crates/tracedecay/tests/agent_suite/managed_skills_test.rs @@ -8,7 +8,7 @@ use tracedecay_automation_runtime::automation::managed_skills::{ use tracedecay_automation_runtime::automation::skill_usage::{ AnalyticsEventRecord, SkillUsageAction, SkillUsageEvent, ingest_analytics_events, load_skill_usage_records, record_skill_usage, record_skill_usage_event, - skill_usage_ledger_path, summarize_skill_usage, summarize_skill_usage_for, + skill_usage_record_path, summarize_skill_usage, summarize_skill_usage_for, }; fn draft() -> ManagedSkillDraft { @@ -746,7 +746,10 @@ async fn managed_skill_usage_ledger_records_views_uses_and_patches() { .await .unwrap(); - assert!(skill_usage_ledger_path(&profile_root).is_file()); + assert!( + skill_usage_record_path(&profile_root, "repo-hygiene").is_file(), + "each skill owns its usage file; writers must not share skill_usage.json" + ); let records = load_skill_usage_records(&profile_root, None).await.unwrap(); assert_eq!(records.len(), 1); assert_eq!(records[0].skill_id, "repo-hygiene"); From b85d94201fbee908cb0a62572c73da243dff4112 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 17 Sep 2026 02:03:01 +0000 Subject: [PATCH 3/3] fix(code-index): identify publication pointer by content digest Co-authored-by: Zack Jackson --- .../code_index_scheduler/publication_store.rs | 16 +++------------- 1 file changed, 3 insertions(+), 13 deletions(-) diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs index f983b22e3e..97b9b5366e 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs @@ -1086,19 +1086,9 @@ impl DaemonCodeIndexPublicationStoreV1 { } let mtime = metadata.modified().ok(); let size = metadata.len(); - { - let memo = self - .pointer_memo - .lock() - .unwrap_or_else(PoisonError::into_inner); - if let Some(memo) = memo.as_ref() - && memo.size == size - && memo.mtime.is_some() - && memo.mtime == mtime - { - return Ok(Some(memo.pointer.clone())); - } - } + // Size and mtime are not a content identity. A same-length pointer + // rewrite in one timestamp tick (generation file, state digest) must + // not satisfy the compare-and-swap of a different incumbent. let bytes = std::fs::read(&self.active_path).map_err(Self::unavailable)?; let digest = Self::state_digest(&bytes); {