diff --git a/src-tauri/src/pi_config/document.rs b/src-tauri/src/pi_config/document.rs index 55dfd01b2..1ede570dd 100644 --- a/src-tauri/src/pi_config/document.rs +++ b/src-tauri/src/pi_config/document.rs @@ -20,6 +20,7 @@ use std::sync::{Arc, LazyLock, Mutex, MutexGuard}; use super::shared_file::{ compare_exchange_shared_file_bytes, delete_shared_file, read_shared_file, + sync_shared_file_parent, }; const MAX_PI_MODELS_BYTES: u64 = 8 * 1024 * 1024; @@ -35,6 +36,9 @@ static PI_JSON_TRAILING_COMMAS: LazyLock = LazyLock::new(|| { }); static PATH_LOCKS: LazyLock>>>> = LazyLock::new(|| Mutex::new(HashMap::new())); +#[cfg(test)] +static BEFORE_PROVIDER_VERIFY: LazyLock>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); fn path_lock(path: &Path) -> Result>, AppError> { let mut locks = PATH_LOCKS @@ -379,23 +383,7 @@ fn apply_pi_provider_patch_checked( for _ in 0..MAX_MUTATION_ATTEMPTS { let before = read_models_bytes(path)?; if let Some(expected) = expected { - let observed = provider_values_from_bytes(path, before.as_deref(), expected.keys())?; - if let Some((provider_key, expected_value)) = - expected.iter().find(|(provider_key, expected_value)| { - observed.get(*provider_key) != Some(*expected_value) - }) - { - return Err(AppError::Conflict(format!( - "Pi provider key '{provider_key}' changed since directory/catalog preflight \ - (expected {}, observed {})", - provider_value_label(expected_value), - provider_value_label( - observed - .get(provider_key) - .expect("every requested provider key is observed") - ) - ))); - } + ensure_provider_values_match(path, before.as_deref(), expected)?; } if before.is_none() && patch.values().all(Option::is_none) { return Ok(false); @@ -460,6 +448,30 @@ fn provider_values_from_bytes<'a>( .collect()) } +fn ensure_provider_values_match( + path: &Path, + bytes: Option<&[u8]>, + expected: &IndexMap>, +) -> Result<(), AppError> { + let observed = provider_values_from_bytes(path, bytes, expected.keys())?; + if let Some((provider_key, expected_value)) = expected + .iter() + .find(|(provider_key, expected_value)| observed.get(*provider_key) != Some(*expected_value)) + { + return Err(AppError::Conflict(format!( + "Pi provider key '{provider_key}' changed since directory/catalog preflight \ + (expected {}, observed {})", + provider_value_label(expected_value), + provider_value_label( + observed + .get(provider_key) + .expect("every requested provider key is observed") + ) + ))); + } + Ok(()) +} + fn provider_value_label(value: &Option) -> &'static str { if value.is_some() { "present" @@ -499,7 +511,24 @@ impl PiProviderPatchReceipt { /// concurrent edit of an owned key turns compensation into an explicit /// conflict instead of being overwritten. pub(crate) fn rollback(&self) -> Result<(), AppError> { - apply_pi_provider_patch_checked(&self.path, Some(&self.attempted), &self.before.values)?; + if let Err(error) = + apply_pi_provider_patch_checked(&self.path, Some(&self.attempted), &self.before.values) + { + let observed = + snapshot_pi_provider_values(&self.path, self.before.values.keys().cloned())?; + if observed.values != self.before.values { + return Err(error); + } + // A namespace mutation may have committed before its durability + // barrier reported failure. Re-observe the exact semantic state + // and retry the parent sync before declaring compensation done. + sync_shared_file_parent(&self.path).map_err(|sync_error| { + AppError::Config(format!( + "Pi provider rollback reached the previous exact-key state but could not \ + confirm directory durability ({error}; retry={sync_error})" + )) + })?; + } remove_new_empty_document(&self.path, &self.before) } } @@ -520,6 +549,103 @@ pub(crate) fn snapshot_pi_provider_values( }) } +/// Revalidate a previously captured exact-key set without mutating the +/// document. This is the ownership-claim barrier used when runtime takeover is +/// disabled and therefore has no gateway projection write of its own. +pub(crate) fn verify_pi_provider_values( + path: &Path, + expected: &IndexMap>, +) -> Result<(), AppError> { + let lock = path_lock(path)?; + let _guard = lock_path(&lock)?; + #[cfg(test)] + if let Some(replacement) = BEFORE_PROVIDER_VERIFY + .lock() + .map_err(|error| AppError::Lock(error.to_string()))? + .remove(path) + { + fs::write(path, replacement).map_err(|error| AppError::io(path, error))?; + } + let bytes = read_models_bytes(path)?; + ensure_provider_values_match(path, bytes.as_deref(), expected) +} + +#[cfg(test)] +pub(crate) fn replace_before_next_pi_provider_verify(path: &Path, bytes: &[u8]) { + BEFORE_PROVIDER_VERIFY + .lock() + .expect("Pi provider verify hook lock") + .insert(path.to_path_buf(), bytes.to_vec()); +} + +fn attempted_provider_values( + before: &PiProviderValuesSnapshot, + patch: &IndexMap>, +) -> IndexMap> { + before + .values + .iter() + .map(|(provider_key, previous)| { + ( + provider_key.clone(), + patch + .get(provider_key) + .cloned() + .unwrap_or_else(|| previous.clone()), + ) + }) + .collect() +} + +fn failed_publish_after_conditional_compensation( + path: &Path, + before: &PiProviderValuesSnapshot, + attempted: &IndexMap>, + publish_error: AppError, +) -> AppError { + if matches!(&publish_error, AppError::Conflict(_)) { + return publish_error; + } + let observed = match snapshot_pi_provider_values(path, before.values.keys().cloned()) { + Ok(observed) => observed, + Err(observe_error) => { + return AppError::Config(format!( + "Pi provider publication failed ({publish_error}) and its commit state could not \ + be inspected for compensation: {observe_error}" + )); + } + }; + if observed.values == before.values { + if let Err(sync_error) = sync_shared_file_parent(path) { + return AppError::Config(format!( + "Pi provider publication failed ({publish_error}); the previous exact-key state \ + is visible but directory durability could not be confirmed: {sync_error}" + )); + } + return publish_error; + } + if observed.values != *attempted { + return AppError::Config(format!( + "Pi provider publication failed ({publish_error}); the live exact-key state was \ + superseded before conditional compensation and was preserved" + )); + } + + let receipt = PiProviderPatchReceipt { + path: path.to_path_buf(), + before: before.clone(), + attempted: attempted.clone(), + attempted_file_existed: observed.file_existed, + }; + match receipt.rollback() { + Ok(()) => publish_error, + Err(rollback_error) => AppError::Config(format!( + "Pi provider publication failed ({publish_error}) after the attempted exact-key state \ + became live; conditional compensation also failed: {rollback_error}" + )), + } +} + /// Publish an exact-key patch only if every preflighted provider value still /// matches. Whole-file CAS retries preserve unrelated edits, but re-check the /// provider precondition before every retry. @@ -536,20 +662,15 @@ pub(crate) fn apply_pi_provider_patch_with_receipt( "Pi provider patch contains a key which was not preflighted".to_string(), )); } - let changed = apply_pi_provider_patch_checked(path, Some(&before.values), patch)?; - let attempted = before - .values - .iter() - .map(|(provider_key, previous)| { - ( - provider_key.clone(), - patch - .get(provider_key) - .cloned() - .unwrap_or_else(|| previous.clone()), - ) - }) - .collect(); + let attempted = attempted_provider_values(before, patch); + let changed = match apply_pi_provider_patch_checked(path, Some(&before.values), patch) { + Ok(changed) => changed, + Err(error) => { + return Err(failed_publish_after_conditional_compensation( + path, before, &attempted, error, + )); + } + }; Ok(PiProviderPatchReceipt { path: path.to_path_buf(), before: before.clone(), @@ -569,17 +690,37 @@ fn remove_new_empty_document( return Ok(()); } - let current = read_shared_file(path, MAX_PI_MODELS_BYTES, "Pi models file")?; let canonical_empty = serialize_models_mutation(path, None, &|_| Ok(()))?; - if current.bytes.as_deref() == Some(canonical_empty.as_slice()) { - delete_shared_file( + let mut last_error = None; + for _ in 0..MAX_MUTATION_ATTEMPTS { + let current = read_shared_file(path, MAX_PI_MODELS_BYTES, "Pi models file")?; + match current.bytes.as_deref() { + None => { + if last_error.is_some() { + sync_shared_file_parent(path)?; + } + return Ok(()); + } + Some(bytes) if bytes != canonical_empty.as_slice() => { + // A concurrent writer added unrelated content after our key + // rollback. File ownership therefore belongs to that writer. + return Ok(()); + } + Some(_) => {} + } + match delete_shared_file( path, ¤t.revision, MAX_PI_MODELS_BYTES, "Pi models file", - )?; + ) { + Ok(_) => return Ok(()), + Err(error) => last_error = Some(error), + } } - Ok(()) + Err(last_error.unwrap_or_else(|| { + AppError::Config("Pi empty models document cleanup did not make progress".to_string()) + })) } #[cfg(test)] @@ -755,6 +896,86 @@ mod tests { assert_eq!(document["providers"]["managed"]["api"], "external"); } + #[cfg(unix)] + #[test] + fn create_parent_sync_failure_conditionally_removes_the_committed_provider() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("models.json"); + let before = + snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight"); + crate::pi_config::shared_file::fail_next_parent_sync_for_test(&path); + + let error = apply_pi_provider_patch_with_receipt( + &path, + &before, + &IndexMap::from([( + "managed".to_string(), + Some(serde_json::json!({"api": "attempted"})), + )]), + ) + .expect_err("durability failure must remain visible"); + assert!(error.to_string().contains("injected")); + assert!( + !path.exists(), + "a failed create must not leave a committed provider or empty shadow file" + ); + } + + #[cfg(unix)] + #[test] + fn replace_parent_sync_failure_conditionally_restores_the_previous_provider() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("models.json"); + fs::write( + &path, + r#"{"root":"preserved","providers":{"managed":{"api":"before"}}}"#, + ) + .expect("seed"); + let before = + snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight"); + crate::pi_config::shared_file::fail_next_parent_sync_for_test(&path); + + apply_pi_provider_patch_with_receipt( + &path, + &before, + &IndexMap::from([( + "managed".to_string(), + Some(serde_json::json!({"api": "attempted"})), + )]), + ) + .expect_err("durability failure must remain visible"); + let restored: Value = + serde_json::from_slice(&fs::read(&path).expect("restored document")).expect("parse"); + assert_eq!(restored["root"], "preserved"); + assert_eq!(restored["providers"]["managed"]["api"], "before"); + } + + #[cfg(unix)] + #[test] + fn remove_key_parent_sync_failure_conditionally_restores_the_deleted_provider() { + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("models.json"); + fs::write( + &path, + r#"{"providers":{"managed":{"api":"before"},"external":{"api":"keep"}}}"#, + ) + .expect("seed"); + let before = + snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight"); + crate::pi_config::shared_file::fail_next_parent_sync_for_test(&path); + + apply_pi_provider_patch_with_receipt( + &path, + &before, + &IndexMap::from([("managed".to_string(), None)]), + ) + .expect_err("durability failure must remain visible"); + let restored: Value = + serde_json::from_slice(&fs::read(&path).expect("restored document")).expect("parse"); + assert_eq!(restored["providers"]["managed"]["api"], "before"); + assert_eq!(restored["providers"]["external"]["api"], "keep"); + } + #[test] fn external_rename_during_patch_is_reparsed_before_owned_fields_change() { let temp = tempfile::tempdir().expect("tempdir"); diff --git a/src-tauri/src/pi_config/shared_file.rs b/src-tauri/src/pi_config/shared_file.rs index 0f1029315..7b3500046 100644 --- a/src-tauri/src/pi_config/shared_file.rs +++ b/src-tauri/src/pi_config/shared_file.rs @@ -26,6 +26,9 @@ static BEFORE_COMPARE_EXCHANGE: LazyLock> = #[cfg(test)] static BEFORE_ROLLBACK_EXCHANGE: LazyLock> = LazyLock::new(|| Mutex::new(HashMap::new())); +#[cfg(all(test, unix))] +static FAIL_NEXT_PARENT_SYNC: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct SharedFileSnapshot { @@ -132,6 +135,20 @@ pub(crate) fn compare_exchange_shared_file_bytes( ) } +/// Retry durability for a namespace state which was observed after a +/// compare/exchange returned an error. The caller is still responsible for +/// verifying the exact live value before treating the mutation as committed +/// or compensated. +pub(crate) fn sync_shared_file_parent(path: &Path) -> Result<(), AppError> { + let parent = path.parent().ok_or_else(|| { + AppError::InvalidInput(format!( + "Pi shared-file path has no parent: {}", + path.display() + )) + })?; + sync_parent(parent) +} + fn compare_exchange_under_lock( path: &Path, expected: Option<&[u8]>, @@ -405,6 +422,13 @@ fn rename_error( #[cfg(unix)] fn sync_parent(parent: &Path) -> Result<(), AppError> { + #[cfg(test)] + if fail_parent_sync_for_test(parent)? { + return Err(AppError::io( + parent, + std::io::Error::other("injected Pi parent-directory sync failure"), + )); + } File::open(parent) .and_then(|directory| directory.sync_all()) .map_err(|error| AppError::io(parent, error)) @@ -611,6 +635,36 @@ pub(crate) fn replace_before_next_compare_exchange(path: &Path, bytes: &[u8]) { .insert(path.to_path_buf(), bytes.to_vec()); } +#[cfg(all(test, unix))] +pub(crate) fn fail_next_parent_sync_for_test(path: &Path) { + let parent = path + .parent() + .expect("a shared-file test path must have a parent") + .to_path_buf(); + let mut failures = FAIL_NEXT_PARENT_SYNC + .lock() + .expect("parent sync test hook lock"); + failures + .entry(parent) + .and_modify(|remaining| *remaining = remaining.saturating_add(1)) + .or_insert(1); +} + +#[cfg(all(test, unix))] +fn fail_parent_sync_for_test(parent: &Path) -> Result { + let mut failures = FAIL_NEXT_PARENT_SYNC + .lock() + .map_err(|error| AppError::Config(format!("Pi sync test hook is poisoned: {error}")))?; + let Some(remaining) = failures.get_mut(parent) else { + return Ok(false); + }; + *remaining = remaining.saturating_sub(1); + if *remaining == 0 { + failures.remove(parent); + } + Ok(true) +} + #[cfg(test)] fn run_file_replacement_hook( hooks: &Mutex, diff --git a/src-tauri/src/proxy/pi_runtime.rs b/src-tauri/src/proxy/pi_runtime.rs index 757c79ba3..abb84cbc9 100644 --- a/src-tauri/src/proxy/pi_runtime.rs +++ b/src-tauri/src/proxy/pi_runtime.rs @@ -58,6 +58,12 @@ pub(crate) struct PiRuntimeSnapshot { pub(crate) server_generation: u64, pub(crate) catalog_epoch: u64, gateway_token: GatewayToken, + /// Exact native values which made this immutable runtime reachable. + /// + /// A fenced runtime may only be re-published while these values still + /// match `models.json`; retaining the projection alongside the routes + /// avoids reconstructing ownership from mutable database/settings state. + native_projection: IndexMap>, app_config: AppProxyConfig, providers: HashMap, failover_ids: Vec, @@ -366,6 +372,7 @@ pub(crate) fn build_pi_runtime( server_generation, catalog_epoch, gateway_token, + native_projection: projection_patch.clone(), app_config, providers, failover_ids, @@ -715,6 +722,18 @@ impl PiRuntimeStore { .map(|snapshot| snapshot.gateway_token.clone()) } + /// Return the exact native projection paired with the fenced runtime. + /// This remains available while admission is at an odd epoch so recovery + /// can prove that re-publication would still describe Pi's live file. + pub(crate) fn retained_native_projection(&self) -> Option>> { + self.publication + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .current + .as_ref() + .map(|snapshot| snapshot.native_projection.clone()) + } + pub(crate) async fn admission_guard( self: &Arc, server_generation: u64, diff --git a/src-tauri/src/services/pi_catalog.rs b/src-tauri/src/services/pi_catalog.rs index ced01527c..0d99f04ca 100644 --- a/src-tauri/src/services/pi_catalog.rs +++ b/src-tauri/src/services/pi_catalog.rs @@ -1256,6 +1256,27 @@ mod tests { use serde_json::json; use std::sync::Arc; + struct TestHome(Option); + + impl TestHome { + fn install(path: &std::path::Path) -> Result { + let previous = std::env::var_os("CC_SWITCH_TEST_HOME"); + std::env::set_var("CC_SWITCH_TEST_HOME", path); + crate::settings::reload_settings()?; + Ok(Self(previous)) + } + } + + impl Drop for TestHome { + fn drop(&mut self) { + match self.0.take() { + Some(value) => std::env::set_var("CC_SWITCH_TEST_HOME", value), + None => std::env::remove_var("CC_SWITCH_TEST_HOME"), + } + let _ = crate::settings::reload_settings(); + } + } + fn managed_input(id: &str, base_url: &str) -> ProviderMutationInput { ProviderMutationInput { id: id.to_string(), @@ -1303,6 +1324,107 @@ mod tests { Ok(config) } + fn configure_pi_directory(path: &std::path::Path) -> Result<(), AppError> { + let mut app_settings = crate::settings::get_settings(); + app_settings.pi_config_dir = Some(path.to_string_lossy().into_owned()); + app_settings.pi_takeover_enabled = false; + crate::settings::update_settings(app_settings) + } + + #[test] + #[serial_test::serial] + #[cfg(unix)] + fn create_durability_failure_compensates_database_and_native_projection() -> Result<(), AppError> + { + let temp = tempfile::tempdir().expect("tempdir"); + let _home = TestHome::install(temp.path())?; + let pi_dir = temp.path().join("pi-agent"); + configure_pi_directory(&pi_dir)?; + let state = AppState::new(Arc::new(Database::memory()?)); + let models_path = pi_dir.join("models.json"); + crate::pi_config::shared_file::fail_next_parent_sync_for_test(&models_path); + + let error = PiCatalogCoordinator::apply( + &state, + PiCatalogMutation::CreateProvider { + input: managed_input("ambiguous-create", "https://create.example/v1"), + provider_key: "ambiguous-native".to_string(), + activate_if_first: false, + }, + ) + .expect_err("directory durability failure must fail the service operation"); + assert!(error.to_string().contains("injected")); + assert!(state + .db + .get_provider_aggregate(PI_APP, "ambiguous-create")? + .is_none()); + assert!(state.db.get_pi_projection("ambiguous-create")?.is_none()); + assert!( + !models_path.exists(), + "the failed create must not leave a provider or empty shadow file" + ); + Ok(()) + } + + #[test] + #[serial_test::serial] + fn native_import_revalidates_exact_values_before_claiming_success() -> Result<(), AppError> { + let temp = tempfile::tempdir().expect("tempdir"); + let _home = TestHome::install(temp.path())?; + let pi_dir = temp.path().join("pi-agent"); + std::fs::create_dir_all(&pi_dir).expect("Pi directory"); + configure_pi_directory(&pi_dir)?; + let state = AppState::new(Arc::new(Database::memory()?)); + let models_path = pi_dir.join("models.json"); + let original = json!({ + "name": "Native import", + "api": "openai-responses", + "baseUrl": "https://original.example/v1", + "apiKey": "literal-key", + "models": [{"id": "model-a", "name": "Model A"}] + }); + std::fs::write( + &models_path, + serde_json::to_vec_pretty(&json!({"providers": {"native": original}})) + .expect("serialize original"), + ) + .expect("write original"); + let fingerprint = inspect_pi_native_entry(&models_path, "native", &BTreeMap::new())? + .expect("native entry") + .diagnostic + .fingerprint; + let external = serde_json::to_vec_pretty(&json!({ + "providers": { + "native": { + "name": "External replacement", + "api": "openai-responses", + "baseUrl": "https://external.example/v1", + "apiKey": "external-key", + "models": [{"id": "external-model"}] + } + } + })) + .expect("serialize external"); + crate::pi_config::document::replace_before_next_pi_provider_verify(&models_path, &external); + + PiCatalogCoordinator::apply( + &state, + PiCatalogMutation::ImportNative { + provider_key: "native".to_string(), + expected_fingerprint: fingerprint, + }, + ) + .expect_err("an external edit before the final barrier must reject import"); + assert!(state.db.get_provider_aggregate(PI_APP, "native")?.is_none()); + assert!(state.db.get_pi_projection("native")?.is_none()); + assert_eq!( + std::fs::read(&models_path).expect("external native file"), + external, + "the external writer remains authoritative" + ); + Ok(()) + } + #[test] fn portable_sql_import_claims_only_an_absent_identity_key_and_is_idempotent( ) -> Result<(), AppError> { diff --git a/src-tauri/src/services/proxy.rs b/src-tauri/src/services/proxy.rs index 7cfb31d6b..3fee77315 100644 --- a/src-tauri/src/services/proxy.rs +++ b/src-tauri/src/services/proxy.rs @@ -629,6 +629,13 @@ impl ProxyService { } if !crate::settings::pi_takeover_enabled() { self.pi_runtime.close(catalog_epoch).await?; + if let Some(expected_native) = expected_native { + let models_path = crate::pi_config::native::get_pi_models_path()?; + crate::pi_config::document::verify_pi_provider_values( + &models_path, + &expected_native.values, + )?; + } return Ok(Vec::new()); } let listener = self @@ -702,6 +709,36 @@ impl ProxyService { Ok(build.direct_only_provider_ids) } + /// Re-open a fenced runtime only while Pi's live exact-key projection + /// still matches the immutable snapshot which produced that runtime. + /// + /// Rebuilding this witness from the mutable database or settings would + /// turn a failed compensation into stale admission, so the projection is + /// retained inside `PiRuntimeSnapshot`. + async fn republish_current_pi_runtime_if_native_matches( + &self, + catalog_epoch: u64, + ) -> Result { + let Some(expected_projection) = self.pi_runtime.retained_native_projection() else { + self.pi_runtime.close(catalog_epoch).await?; + return Ok(false); + }; + let verification = crate::pi_config::native::get_pi_models_path().and_then(|models_path| { + crate::pi_config::document::verify_pi_provider_values( + &models_path, + &expected_projection, + ) + }); + if let Err(error) = verification { + self.pi_runtime.close(catalog_epoch).await?; + return Err(AppError::Conflict(format!( + "Pi native projection changed while runtime admission was fenced; admission \ + remains closed: {error}" + ))); + } + self.pi_runtime.republish_current(catalog_epoch).await + } + #[cfg(test)] pub(crate) fn fail_next_pi_reconcile_for_test(&self) { self.fail_next_pi_reconcile.store(true, Ordering::Release); @@ -973,7 +1010,9 @@ impl ProxyService { let old_direct_receipt = match self.restore_pi_direct_projection_at(&old_models_path) { Ok(receipt) => receipt, Err(error) => { - let _ = self.pi_runtime.republish_current(epoch).await; + let _ = self + .republish_current_pi_runtime_if_native_matches(epoch) + .await; return Err(error); } }; @@ -1019,7 +1058,9 @@ impl ProxyService { .await .is_ok() } else { - self.pi_runtime.republish_current(epoch).await.is_ok() + self.republish_current_pi_runtime_if_native_matches(epoch) + .await + .unwrap_or(false) }; return Err(AppError::Config(format!( "failed to reconcile Pi prompts in the new directory: {error}; rollback: settings={settings_restored}, prompts={prompts_restored}, skills={skills_restored}, gateway={runtime_restored}" @@ -1048,7 +1089,9 @@ impl ProxyService { .await .is_ok() } else { - self.pi_runtime.republish_current(epoch).await.is_ok() + self.republish_current_pi_runtime_if_native_matches(epoch) + .await + .unwrap_or(false) }; return Err(AppError::Config(format!( "failed to reconcile Pi Skills in the new directory: {error}; rollback: settings={settings_restored}, prompts={prompts_restored}, skills={skills_restored}, gateway={runtime_restored}" @@ -1078,7 +1121,9 @@ impl ProxyService { .await .is_ok() } else { - self.pi_runtime.republish_current(epoch).await.is_ok() + self.republish_current_pi_runtime_if_native_matches(epoch) + .await + .unwrap_or(false) }; return Err(AppError::Config(format!( "failed to publish Pi in the new native directory: {error}; rollback: settings={settings_restored}, prompts={prompts_restored}, skills={skills_restored}, old_gateway={old_gateway_restored}" @@ -1096,7 +1141,9 @@ impl ProxyService { let direct_receipt = match self.restore_pi_direct_projection() { Ok(receipt) => receipt, Err(error) => { - let _ = self.pi_runtime.republish_current(epoch).await; + let _ = self + .republish_current_pi_runtime_if_native_matches(epoch) + .await; return Err(error); } }; @@ -1761,8 +1808,10 @@ impl ProxyService { if let Err(error) = self.reconcile_pi_runtime().await { if was_enabled { let epoch = self.pi_runtime.begin_mutation().await; - let restored = self.pi_runtime.republish_current(epoch).await; - return Err(if restored.is_ok() { + let restored = self + .republish_current_pi_runtime_if_native_matches(epoch) + .await; + return Err(if matches!(restored, Ok(true)) { format!( "failed to refresh Pi gateway catalog; the previous runtime remains active: {error}" ) @@ -1814,7 +1863,9 @@ impl ProxyService { let direct_receipt = match self.restore_pi_direct_projection() { Ok(receipt) => receipt, Err(error) => { - let _ = self.pi_runtime.republish_current(epoch).await; + let _ = self + .republish_current_pi_runtime_if_native_matches(epoch) + .await; return Err(format!( "failed to restore Pi's direct native projection: {error}" )); @@ -5234,6 +5285,112 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn fenced_pi_runtime_is_not_republished_after_external_native_drift() { + let home = TempHome::new(); + crate::settings::reload_settings().expect("reload isolated settings"); + let pi_dir = home.dir.path().join("pi"); + std::fs::create_dir_all(&pi_dir).expect("Pi directory"); + let mut settings = crate::settings::get_settings(); + settings.pi_config_dir = Some(pi_dir.to_string_lossy().into_owned()); + settings.pi_takeover_enabled = true; + crate::settings::update_settings(settings).expect("Pi settings"); + + let db = Arc::new(Database::memory().expect("database")); + let direct = json!({ + "name": "Managed Pi", + "api": "openai-responses", + "baseUrl": "https://managed.example/v1", + "apiKey": "managed-key", + "models": [{"id": "model-a", "name": "Model A"}] + }); + db.create_pi_catalog_provider( + NewProviderAggregate::from_input( + AppType::Pi.as_str(), + ProviderMutationInput { + id: "managed-pi".to_string(), + name: "Managed Pi".to_string(), + settings_config: direct.clone(), + website_url: None, + category: None, + created_at: None, + sort_index: Some(0), + notes: None, + meta: None, + icon: Some("pi".to_string()), + icon_color: None, + in_failover_queue: false, + }, + ) + .expect("aggregate"), + "managed-pi", + ) + .expect("managed provider"); + let models_path = pi_dir.join("models.json"); + std::fs::write( + &models_path, + serde_json::to_vec_pretty(&json!({"providers": {"managed-pi": direct}})) + .expect("direct models"), + ) + .expect("seed direct models"); + + let service = ProxyService::new(db); + *service + .pi_listener + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(PiListenerIdentity { + server_generation: 41, + gateway_origin: url::Url::parse("http://127.0.0.1:15721/").expect("gateway origin"), + }); + service + .reconcile_pi_runtime() + .await + .expect("publish initial runtime"); + assert_eq!( + service + .get_takeover_status() + .await + .expect("active status") + .pi_operational_state, + PiTakeoverOperationalState::Active + ); + + let epoch = service.begin_pi_catalog_mutation().await; + let external = serde_json::to_vec_pretty(&json!({ + "providers": { + "managed-pi": { + "name": "External", + "api": "openai-responses", + "baseUrl": "https://external.example/v1", + "apiKey": "external-key", + "models": [{"id": "external-model"}] + } + } + })) + .expect("external models"); + std::fs::write(&models_path, &external).expect("external edit"); + + assert!(matches!( + service + .republish_current_pi_runtime_if_native_matches(epoch) + .await, + Err(AppError::Conflict(_)) + )); + assert_eq!( + service + .get_takeover_status() + .await + .expect("degraded status") + .pi_operational_state, + PiTakeoverOperationalState::Degraded + ); + assert_eq!( + std::fs::read(&models_path).expect("external file remains"), + external + ); + } + fn seed_codex_model_template() { let codex_dir = crate::codex_config::get_codex_config_dir(); std::fs::create_dir_all(&codex_dir).expect("create codex dir");