admin/CC-Switch
1
0
Fork 0
mirror of https://github.com/farion1231/cc-switch.git synced 2026-08-04 11:43:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(pi): bind native and prompt recovery to revisions

Browse Source
This commit is contained in:
SaladDay
2026-08-03 04:43:44 +00:00
parent c94983d5fa
commit 0b609158d7
10 changed files with 778 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/database/dao/prompts.rs
+142 -68
View File
@@ -6,51 +6,106 @@ use crate::database::{lock_conn, Database};
use crate::error::AppError;
use crate::prompt::Prompt;
use indexmap::IndexMap;
use rusqlite::params;
use rusqlite::{params, Connection, Transaction};
fn query_prompts(conn: &Connection, app_type: &str) -> Result<IndexMap<String, Prompt>, AppError> {
let mut stmt = conn
.prepare(
"SELECT id, name, content, description, enabled, created_at, updated_at
FROM prompts WHERE app_type = ?1
ORDER BY created_at ASC, id ASC",
)
.map_err(|e| AppError::Database(e.to_string()))?;
let prompt_iter = stmt
.query_map(params![app_type], |row| {
let id: String = row.get(0)?;
let name: String = row.get(1)?;
let content: String = row.get(2)?;
let description: Option<String> = row.get(3)?;
let enabled: bool = row.get(4)?;
let created_at: Option<i64> = row.get(5)?;
let updated_at: Option<i64> = row.get(6)?;
Ok((
id.clone(),
Prompt {
id,
name,
content,
description,
enabled,
created_at,
updated_at,
},
))
})
.map_err(|e| AppError::Database(e.to_string()))?;
let mut prompts = IndexMap::new();
for prompt_res in prompt_iter {
let (id, prompt) = prompt_res.map_err(|e| AppError::Database(e.to_string()))?;
prompts.insert(id, prompt);
}
Ok(prompts)
}
fn validate_prompt_selection(prompts: &IndexMap<String, Prompt>) -> Result<(), AppError> {
if prompts.values().filter(|prompt| prompt.enabled).count() > 1 {
return Err(AppError::InvalidInput(
"at most one prompt may be enabled for an app".to_string(),
));
}
Ok(())
}
fn replace_prompt_rows(
transaction: &Transaction<'_>,
app_type: &str,
prompts: &IndexMap<String, Prompt>,
) -> Result<(), AppError> {
transaction
.execute("DELETE FROM prompts WHERE app_type = ?1", [app_type])
.map_err(|error| AppError::Database(error.to_string()))?;
let mut statement = transaction
.prepare(
"INSERT OR REPLACE INTO prompts (
id, app_type, name, content, description, enabled, created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
)
.map_err(|error| AppError::Database(error.to_string()))?;
for prompt in prompts.values() {
statement
.execute(params![
prompt.id,
app_type,
prompt.name,
prompt.content,
prompt.description,
prompt.enabled,
prompt.created_at,
prompt.updated_at,
])
.map_err(|error| AppError::Database(error.to_string()))?;
}
Ok(())
}
fn prompt_libraries_equal(
left: &IndexMap<String, Prompt>,
right: &IndexMap<String, Prompt>,
) -> bool {
left.len() == right.len()
&& left
.iter()
.all(|(id, prompt)| right.get(id) == Some(prompt))
}
impl Database {
/// 获取指定应用类型的所有提示词
pub fn get_prompts(&self, app_type: &str) -> Result<IndexMap<String, Prompt>, AppError> {
let conn = lock_conn!(self.conn);
let mut stmt = conn
.prepare(
"SELECT id, name, content, description, enabled, created_at, updated_at
FROM prompts WHERE app_type = ?1
ORDER BY created_at ASC, id ASC",
)
.map_err(|e| AppError::Database(e.to_string()))?;
let prompt_iter = stmt
.query_map(params![app_type], |row| {
let id: String = row.get(0)?;
let name: String = row.get(1)?;
let content: String = row.get(2)?;
let description: Option<String> = row.get(3)?;
let enabled: bool = row.get(4)?;
let created_at: Option<i64> = row.get(5)?;
let updated_at: Option<i64> = row.get(6)?;
Ok((
id.clone(),
Prompt {
id,
name,
content,
description,
enabled,
created_at,
updated_at,
},
))
})
.map_err(|e| AppError::Database(e.to_string()))?;
let mut prompts = IndexMap::new();
for prompt_res in prompt_iter {
let (id, prompt) = prompt_res.map_err(|e| AppError::Database(e.to_string()))?;
prompts.insert(id, prompt);
}
Ok(prompts)
query_prompts(&conn, app_type)
}
/// 保存提示词
@@ -85,41 +140,60 @@ impl Database {
app_type: &str,
prompts: &IndexMap<String, Prompt>,
) -> Result<(), AppError> {
if prompts.values().filter(|prompt| prompt.enabled).count() > 1 {
return Err(AppError::InvalidInput(
"at most one prompt may be enabled for an app".to_string(),
));
}
validate_prompt_selection(prompts)?;
let mut conn = lock_conn!(self.conn);
let transaction = conn
.transaction()
.map_err(|error| AppError::Database(error.to_string()))?;
replace_prompt_rows(&transaction, app_type, prompts)?;
transaction
.execute("DELETE FROM prompts WHERE app_type = ?1", [app_type])
.commit()
.map_err(|error| AppError::Database(error.to_string()))
}
/// Atomically publish a complete prompt library only while its full
/// before-image still matches. This is the database half of Pi's
/// native-file/portable-library compare-and-swap boundary.
pub(crate) fn compare_exchange_prompt_selection(
&self,
app_type: &str,
expected: &IndexMap<String, Prompt>,
replacement: &IndexMap<String, Prompt>,
) -> Result<(), AppError> {
validate_prompt_selection(replacement)?;
self.compare_exchange_prompt_selection_unchecked(app_type, expected, replacement)
}
/// Restore a captured before-image only if the database still contains the
/// exact attempted projection. The before-image may predate the current
/// single-selection invariant, so compensation must preserve it byte for
/// byte instead of refusing to restore legacy rows.
pub(crate) fn restore_prompt_selection_if_attempted(
&self,
app_type: &str,
attempted: &IndexMap<String, Prompt>,
before: &IndexMap<String, Prompt>,
) -> Result<(), AppError> {
self.compare_exchange_prompt_selection_unchecked(app_type, attempted, before)
}
fn compare_exchange_prompt_selection_unchecked(
&self,
app_type: &str,
expected: &IndexMap<String, Prompt>,
replacement: &IndexMap<String, Prompt>,
) -> Result<(), AppError> {
let mut conn = lock_conn!(self.conn);
let transaction = conn
.transaction()
.map_err(|error| AppError::Database(error.to_string()))?;
{
let mut statement = transaction
.prepare(
"INSERT OR REPLACE INTO prompts (
id, app_type, name, content, description, enabled, created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
)
.map_err(|error| AppError::Database(error.to_string()))?;
for prompt in prompts.values() {
statement
.execute(params![
prompt.id,
app_type,
prompt.name,
prompt.content,
prompt.description,
prompt.enabled,
prompt.created_at,
prompt.updated_at,
])
.map_err(|error| AppError::Database(error.to_string()))?;
}
let observed = query_prompts(&transaction, app_type)?;
if !prompt_libraries_equal(&observed, expected) {
return Err(AppError::Conflict(format!(
"{app_type} prompt library changed since it was read"
)));
}
replace_prompt_rows(&transaction, app_type, replacement)?;
transaction
.commit()
.map_err(|error| AppError::Database(error.to_string()))
src-tauri/src/pi_config/document.rs
+155 -15
View File
@@ -13,6 +13,7 @@ use jsonc_parser::cst::{
use jsonc_parser::ParseOptions;
use regex::Regex;
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
@@ -318,6 +319,10 @@ pub(super) fn read_pi_models_document(path: &Path) -> Result<PiModelsDocument, A
parse_pi_models_document(path, bytes.as_deref())
}
pub(super) fn pi_raw_provider_fingerprint(raw_source: &str) -> String {
format!("sha256:{:x}", Sha256::digest(raw_source.as_bytes()))
}
fn serialize_models_mutation(
path: &Path,
before: Option<&[u8]>,
@@ -366,12 +371,13 @@ pub(crate) fn apply_pi_provider_patch(
path: &Path,
patch: &IndexMap<String, Option<Value>>,
) -> Result<(), AppError> {
apply_pi_provider_patch_checked(path, None, patch).map(|_| ())
apply_pi_provider_patch_checked(path, None, None, patch).map(|_| ())
}
fn apply_pi_provider_patch_checked(
path: &Path,
expected: Option<&IndexMap<String, Option<Value>>>,
expected_fingerprints: Option<&IndexMap<String, String>>,
patch: &IndexMap<String, Option<Value>>,
) -> Result<bool, AppError> {
let lock = path_lock(path)?;
@@ -385,6 +391,9 @@ fn apply_pi_provider_patch_checked(
if let Some(expected) = expected {
ensure_provider_values_match(path, before.as_deref(), expected)?;
}
if let Some(expected_fingerprints) = expected_fingerprints {
ensure_provider_fingerprints_match(path, before.as_deref(), expected_fingerprints)?;
}
if before.is_none() && patch.values().all(Option::is_none) {
return Ok(false);
}
@@ -472,6 +481,28 @@ fn ensure_provider_values_match(
Ok(())
}
fn ensure_provider_fingerprints_match(
path: &Path,
bytes: Option<&[u8]>,
expected: &IndexMap<String, String>,
) -> Result<(), AppError> {
let document = parse_pi_models_document(path, bytes)?;
for (provider_key, expected_fingerprint) in expected {
let observed = document
.providers()
.get(provider_key)
.map(|entry| pi_raw_provider_fingerprint(&entry.raw_source));
if observed.as_deref() != Some(expected_fingerprint) {
return Err(AppError::Conflict(format!(
"Pi native provider '{provider_key}' changed since inspection \
(expected raw fingerprint {expected_fingerprint}, observed {})",
observed.as_deref().unwrap_or("missing")
)));
}
}
Ok(())
}
fn provider_value_label(value: &Option<Value>) -> &'static str {
if value.is_some() {
"present"
@@ -511,9 +542,12 @@ 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> {
if let Err(error) =
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),
None,
&self.before.values,
) {
let observed =
snapshot_pi_provider_values(&self.path, self.before.values.keys().cloned())?;
if observed.values != self.before.values {
@@ -559,6 +593,30 @@ pub(crate) fn verify_pi_provider_values(
let lock = path_lock(path)?;
let _guard = lock_path(&lock)?;
#[cfg(test)]
run_before_provider_verify_hook(path)?;
let bytes = read_models_bytes(path)?;
ensure_provider_values_match(path, bytes.as_deref(), expected)
}
/// Revalidate semantic exact-key values and raw entry fingerprints under one
/// path lock. Native import uses the stronger raw barrier because its ownership
/// token is the fingerprint returned by public inspection.
pub(crate) fn verify_pi_provider_preconditions(
path: &Path,
expected_values: &IndexMap<String, Option<Value>>,
expected_fingerprints: &IndexMap<String, String>,
) -> Result<(), AppError> {
let lock = path_lock(path)?;
let _guard = lock_path(&lock)?;
#[cfg(test)]
run_before_provider_verify_hook(path)?;
let bytes = read_models_bytes(path)?;
ensure_provider_values_match(path, bytes.as_deref(), expected_values)?;
ensure_provider_fingerprints_match(path, bytes.as_deref(), expected_fingerprints)
}
#[cfg(test)]
fn run_before_provider_verify_hook(path: &Path) -> Result<(), AppError> {
if let Some(replacement) = BEFORE_PROVIDER_VERIFY
.lock()
.map_err(|error| AppError::Lock(error.to_string()))?
@@ -566,8 +624,7 @@ pub(crate) fn verify_pi_provider_values(
{
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)
Ok(())
}
#[cfg(test)]
@@ -603,9 +660,7 @@ fn failed_publish_after_conditional_compensation(
attempted: &IndexMap<String, Option<Value>>,
publish_error: AppError,
) -> AppError {
if matches!(&publish_error, AppError::Conflict(_)) {
return publish_error;
}
let retryable_conflict = matches!(&publish_error, AppError::Conflict(_));
let observed = match snapshot_pi_provider_values(path, before.values.keys().cloned()) {
Ok(observed) => observed,
Err(observe_error) => {
@@ -616,15 +671,23 @@ fn failed_publish_after_conditional_compensation(
}
};
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}"
));
if !retryable_conflict {
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 {
if retryable_conflict {
// A normal compare/exchange conflict must remain retryable. Its
// canonical value belongs to the concurrent writer; any displaced
// recovery artifact is already named in the lower-level error.
return publish_error;
}
return AppError::Config(format!(
"Pi provider publication failed ({publish_error}); the live exact-key state was \
superseded before conditional compensation and was preserved"
@@ -653,6 +716,17 @@ pub(crate) fn apply_pi_provider_patch_with_receipt(
path: &Path,
before: &PiProviderValuesSnapshot,
patch: &IndexMap<String, Option<Value>>,
) -> Result<PiProviderPatchReceipt, AppError> {
apply_pi_provider_patch_with_receipt_and_fingerprints(path, before, None, patch)
}
/// Publish an exact-key patch while atomically binding selected entries to the
/// raw inspection fingerprints which authorized an ownership claim.
pub(crate) fn apply_pi_provider_patch_with_receipt_and_fingerprints(
path: &Path,
before: &PiProviderValuesSnapshot,
expected_fingerprints: Option<&IndexMap<String, String>>,
patch: &IndexMap<String, Option<Value>>,
) -> Result<PiProviderPatchReceipt, AppError> {
if patch
.keys()
@@ -663,7 +737,12 @@ pub(crate) fn apply_pi_provider_patch_with_receipt(
));
}
let attempted = attempted_provider_values(before, patch);
let changed = match apply_pi_provider_patch_checked(path, Some(&before.values), patch) {
let changed = match apply_pi_provider_patch_checked(
path,
Some(&before.values),
expected_fingerprints,
patch,
) {
Ok(changed) => changed,
Err(error) => {
return Err(failed_publish_after_conditional_compensation(
@@ -976,6 +1055,67 @@ mod tests {
assert_eq!(restored["providers"]["external"]["api"], "keep");
}
#[test]
fn failed_conflict_recovery_after_replace_compensates_attempted_provider_state() {
let temp = tempfile::tempdir().expect("tempdir");
let path = temp.path().join("models.json");
fs::write(&path, br#"{"providers":{"managed":{"api":"before"}}}"#).expect("seed");
let before =
snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight");
let external = br#"{"providers":{"managed":{"api":"external"}}}"#;
crate::pi_config::shared_file::replace_before_next_compare_exchange(&path, external);
crate::pi_config::shared_file::fail_next_rollback_restore_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("an uncertain conflict must remain an error");
assert!(matches!(error, AppError::Conflict(_)));
let restored = read_pi_models_document(&path).expect("restored document");
assert_eq!(restored.providers()["managed"].value["api"], "before");
assert!(
fs::read_dir(temp.path())
.expect("recovery directory")
.filter_map(Result::ok)
.any(|entry| fs::read(entry.path()).ok().as_deref() == Some(external)),
"the displaced external version must remain recoverable"
);
}
#[test]
fn failed_conflict_recovery_after_delete_compensates_attempted_absence() {
let temp = tempfile::tempdir().expect("tempdir");
let path = temp.path().join("models.json");
fs::write(&path, br#"{"providers":{"managed":{"api":"before"}}}"#).expect("seed");
let before =
snapshot_pi_provider_values(&path, ["managed".to_string()]).expect("preflight");
let external = br#"{"providers":{"managed":{"api":"external"}}}"#;
crate::pi_config::shared_file::replace_before_next_compare_exchange(&path, external);
crate::pi_config::shared_file::fail_next_rollback_restore_for_test(&path);
let error = apply_pi_provider_patch_with_receipt(
&path,
&before,
&IndexMap::from([("managed".to_string(), None)]),
)
.expect_err("an uncertain delete conflict must remain an error");
assert!(matches!(error, AppError::Conflict(_)));
let restored = read_pi_models_document(&path).expect("restored document");
assert_eq!(restored.providers()["managed"].value["api"], "before");
assert!(
fs::read_dir(temp.path())
.expect("recovery directory")
.filter_map(Result::ok)
.any(|entry| fs::read(entry.path()).ok().as_deref() == Some(external)),
"the quarantined external version must remain recoverable"
);
}
#[test]
fn external_rename_during_patch_is_reparsed_before_owned_fields_change() {
let temp = tempfile::tempdir().expect("tempdir");
src-tauri/src/pi_config/gateway.rs
+73 -3
View File
@@ -10,6 +10,7 @@ use super::composer::{
PiComposedHeader, PiComposedNativeModel, PiComposerStatus, PiNativeComposition,
};
use http::{HeaderMap, HeaderName, HeaderValue};
use std::fmt;
use url::Url;
/// Headers owned by HTTP framing, the proxy hop, or gateway-generated request
@@ -153,11 +154,17 @@ pub(crate) struct PiGatewayAssessment {
pub plans: Vec<CandidateHeaderPlan>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[derive(Clone, PartialEq, Eq)]
struct DeferredHeaderValue {
raw: String,
}
impl fmt::Debug for DeferredHeaderValue {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("DeferredHeaderValue(<redacted>)")
}
}
impl DeferredHeaderValue {
fn new(raw: impl Into<String>) -> Self {
Self { raw: raw.into() }
@@ -183,7 +190,7 @@ impl DeferredHeaderValue {
}
}
#[derive(Debug, Clone)]
#[derive(Clone)]
pub(crate) struct CandidateHeaderPlan {
family: PiGatewayApiFamily,
endpoint: Url,
@@ -194,6 +201,24 @@ pub(crate) struct CandidateHeaderPlan {
protocol_identity_predictable: bool,
}
impl fmt::Debug for CandidateHeaderPlan {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CandidateHeaderPlan")
.field("family", &self.family)
.field("endpoint", &self.endpoint)
.field("credential", &"<redacted>")
.field("auth_header", &self.auth_header)
.field("provider_headers", &self.provider_headers)
.field("model_headers", &self.model_headers)
.field(
"protocol_identity_predictable",
&self.protocol_identity_predictable,
)
.finish()
}
}
#[derive(Debug, Clone)]
struct PlannedHeader {
name: HeaderName,
@@ -202,7 +227,7 @@ struct PlannedHeader {
class: ConfiguredHeaderClass,
}
#[derive(Debug, Clone)]
#[derive(Clone)]
pub(crate) struct MaterializedCandidate {
pub endpoint: Url,
pub headers: HeaderMap,
@@ -211,6 +236,24 @@ pub(crate) struct MaterializedCandidate {
protocol_identity_predictable: bool,
}
impl fmt::Debug for MaterializedCandidate {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
let header_names = self.headers.keys().collect::<Vec<_>>();
let protocol_header_names = self.protocol_headers.keys().collect::<Vec<_>>();
formatter
.debug_struct("MaterializedCandidate")
.field("endpoint", &self.endpoint)
.field("header_names", &header_names)
.field("family", &self.family)
.field("protocol_header_names", &protocol_header_names)
.field(
"protocol_identity_predictable",
&self.protocol_identity_predictable,
)
.finish()
}
}
pub(crate) trait DeferredValueResolver {
fn resolve(&self, expression: &str) -> Option<String>;
}
@@ -773,6 +816,33 @@ mod tests {
);
}
#[test]
fn sensitive_gateway_debug_output_redacts_credentials_and_header_values() {
let credential = "sk-debug-credential-never-log";
let header_secret = "debug-header-value-never-log";
let composition = composed(json!({
"api": "openai-responses",
"baseUrl": "https://example.test/v1",
"apiKey": credential,
"headers": {"x-private": header_secret},
"models": [{"id": "m"}]
}));
let assessment = assess_composition(&composition);
assert_eq!(assessment.capability, PiGatewayCapability::Proxyable);
let assessment_debug = format!("{assessment:?}");
assert!(!assessment_debug.contains(credential));
assert!(!assessment_debug.contains(header_secret));
let materialized = assessment.plans[0]
.materialize(&|_expression: &str| None)
.expect("literal plan materializes");
let materialized_debug = format!("{materialized:?}");
assert!(!materialized_debug.contains(credential));
assert!(!materialized_debug.contains(header_secret));
assert!(materialized_debug.contains("authorization"));
assert!(materialized_debug.contains("x-private"));
}
#[test]
fn unknown_api_is_composed_but_direct_only() {
let composition = composed(json!({
src-tauri/src/pi_config/native.rs
+2 -7
View File
@@ -8,7 +8,7 @@
use super::composer::{
compose_explicit_custom_catalog, PiComposerReasonCode, PiComposerStatus, PiNativeComposition,
};
use super::document::{read_pi_models_document, PiRawProviderEntry};
use super::document::{pi_raw_provider_fingerprint, read_pi_models_document, PiRawProviderEntry};
use super::gateway::{
assess_composition, PiGatewayAssessment, PiGatewayCapability, PiGatewayReasonCode,
};
@@ -23,7 +23,6 @@ use super::raw_schema::{
use crate::config::get_home_dir;
use crate::error::AppError;
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashSet};
use std::path::{Path, PathBuf};
use url::Url;
@@ -292,7 +291,7 @@ fn analyze_native_entry(
.get("name")
.and_then(Value::as_str)
.map(ToOwned::to_owned),
fingerprint: fingerprint(&entry.raw_source),
fingerprint: pi_raw_provider_fingerprint(&entry.raw_source),
kind,
raw_validity,
managed_assessment: managed.assessment,
@@ -711,10 +710,6 @@ fn escape_json_pointer(value: &str) -> String {
value.replace('~', "~0").replace('/', "~1")
}
fn fingerprint(raw_source: &str) -> String {
format!("sha256:{:x}", Sha256::digest(raw_source.as_bytes()))
}
#[cfg(test)]
mod tests {
use super::*;
src-tauri/src/pi_config/shared_file.rs
+46 -7
View File
@@ -26,6 +26,9 @@ static BEFORE_COMPARE_EXCHANGE: LazyLock<Mutex<CompareExchangeHooks>> =
#[cfg(test)]
static BEFORE_ROLLBACK_EXCHANGE: LazyLock<Mutex<CompareExchangeHooks>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
#[cfg(test)]
static FAIL_NEXT_ROLLBACK_RESTORE: LazyLock<Mutex<HashMap<PathBuf, usize>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
#[cfg(all(test, unix))]
static FAIL_NEXT_PARENT_SYNC: LazyLock<Mutex<HashMap<PathBuf, usize>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
@@ -251,13 +254,15 @@ fn replace_existing_if_equal(
}
run_before_rollback_exchange_hook(path)?;
let proposed_or_raced = restore_displaced(&displaced, path).map_err(|error| {
AppError::Conflict(format!(
"{label} changed during atomic replacement and could not be restored; \
let proposed_or_raced = run_before_rollback_restore_hook(path)
.and_then(|()| restore_displaced(&displaced, path))
.map_err(|error| {
AppError::Conflict(format!(
"{label} changed during atomic replacement and could not be restored; \
the displaced bytes remain at {}: {error}",
displaced.display()
))
})?;
displaced.display()
))
})?;
let proposed_bytes = read_regular_bytes(&proposed_or_raced, max_bytes, label)?;
if proposed_bytes.as_deref() == Some(replacement) {
fs::remove_file(&proposed_or_raced)
@@ -311,7 +316,8 @@ fn delete_existing_if_equal(
return Ok(snapshot(None));
}
match rename_noreplace(&quarantine, path) {
match run_before_rollback_restore_hook(path).and_then(|()| rename_noreplace(&quarantine, path))
{
Ok(()) => {
sync_parent(
path.parent()
@@ -700,6 +706,39 @@ fn replace_before_next_rollback_exchange(path: &Path, bytes: &[u8]) {
.insert(path.to_path_buf(), bytes.to_vec());
}
#[cfg(test)]
pub(crate) fn fail_next_rollback_restore_for_test(path: &Path) {
let mut failures = FAIL_NEXT_ROLLBACK_RESTORE
.lock()
.expect("rollback restore test hook lock");
failures
.entry(path.to_path_buf())
.and_modify(|remaining| *remaining = remaining.saturating_add(1))
.or_insert(1);
}
#[cfg(test)]
fn run_before_rollback_restore_hook(path: &Path) -> std::io::Result<()> {
let mut failures = FAIL_NEXT_ROLLBACK_RESTORE
.lock()
.map_err(|error| std::io::Error::other(error.to_string()))?;
let Some(remaining) = failures.get_mut(path) else {
return Ok(());
};
*remaining = remaining.saturating_sub(1);
if *remaining == 0 {
failures.remove(path);
}
Err(std::io::Error::other(
"injected Pi rollback-restore failure",
))
}
#[cfg(not(test))]
fn run_before_rollback_restore_hook(_path: &Path) -> std::io::Result<()> {
Ok(())
}
#[cfg(test)]
fn run_before_rollback_exchange_hook(path: &Path) -> Result<(), AppError> {
run_file_replacement_hook(&BEFORE_ROLLBACK_EXCHANGE, path)
src-tauri/src/prompt.rs
+1 -1
View File
@@ -1,6 +1,6 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Prompt {
pub id: String,
pub name: String,
src-tauri/src/proxy/pi_runtime.rs
+55 -5
View File
@@ -21,6 +21,7 @@ use indexmap::IndexMap;
use serde_json::{Map, Value};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap};
use std::fmt;
use std::io::Read;
use std::process::{Command, Stdio};
use std::sync::{mpsc, Arc, RwLock};
@@ -52,8 +53,25 @@ struct PiRouteBinding {
provider_id: String,
}
#[derive(Clone)]
struct PiNativeProjectionWitness(IndexMap<String, Option<Value>>);
impl fmt::Debug for PiNativeProjectionWitness {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
let keys = self
.0
.iter()
.map(|(key, value)| (key, if value.is_some() { "present" } else { "absent" }))
.collect::<Vec<_>>();
formatter
.debug_tuple("PiNativeProjectionWitness")
.field(&keys)
.finish()
}
}
/// One immutable catalog matching a successfully published native projection.
#[derive(Debug, Clone)]
#[derive(Clone)]
pub(crate) struct PiRuntimeSnapshot {
pub(crate) server_generation: u64,
pub(crate) catalog_epoch: u64,
@@ -63,13 +81,28 @@ pub(crate) struct PiRuntimeSnapshot {
/// 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<String, Option<Value>>,
native_projection: PiNativeProjectionWitness,
app_config: AppProxyConfig,
providers: HashMap<String, PiRuntimeProvider>,
failover_ids: Vec<String>,
routes: HashMap<String, PiRouteBinding>,
}
impl fmt::Debug for PiRuntimeSnapshot {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("PiRuntimeSnapshot")
.field("server_generation", &self.server_generation)
.field("catalog_epoch", &self.catalog_epoch)
.field("gateway_token", &self.gateway_token)
.field("native_projection", &self.native_projection)
.field("provider_count", &self.providers.len())
.field("failover_ids", &self.failover_ids)
.field("route_count", &self.routes.len())
.finish_non_exhaustive()
}
}
#[derive(Debug, Clone)]
pub(crate) struct PiRequestCandidate {
pub(crate) provider_id: String,
@@ -95,7 +128,6 @@ pub(crate) struct PiMaterializedAttempt {
pub(crate) url: Url,
}
#[derive(Debug)]
pub(crate) struct PiRuntimeBuild {
pub(crate) snapshot: Arc<PiRuntimeSnapshot>,
/// Exact keys only. A direct-only provider deliberately keeps its original
@@ -372,7 +404,7 @@ pub(crate) fn build_pi_runtime(
server_generation,
catalog_epoch,
gateway_token,
native_projection: projection_patch.clone(),
native_projection: PiNativeProjectionWitness(projection_patch.clone()),
app_config,
providers,
failover_ids,
@@ -731,7 +763,7 @@ impl PiRuntimeStore {
.unwrap_or_else(std::sync::PoisonError::into_inner)
.current
.as_ref()
.map(|snapshot| snapshot.native_projection.clone())
.map(|snapshot| snapshot.native_projection.0.clone())
}
pub(crate) async fn admission_guard(
@@ -1154,6 +1186,24 @@ mod tests {
use super::*;
use serde_json::json;
#[test]
fn native_projection_debug_exposes_only_key_presence() {
let secret = "runtime-native-secret-never-log";
let witness = PiNativeProjectionWitness(IndexMap::from([
(
"managed".to_string(),
Some(json!({"apiKey": secret, "headers": {"x-private": secret}})),
),
("removed".to_string(), None),
]));
let debug = format!("{witness:?}");
assert!(!debug.contains(secret));
assert!(debug.contains("managed"));
assert!(debug.contains("present"));
assert!(debug.contains("removed"));
assert!(debug.contains("absent"));
}
#[test]
fn environment_resolution_matches_vendored_transport_oracle() {
std::env::set_var("PI_RUNTIME_TEST_VALUE", "environment-secret");
src-tauri/src/services/pi_catalog.rs
+88 -2
View File
@@ -95,6 +95,8 @@ pub(crate) struct PiCatalogMutationResult {
native_defaults_receipt: Option<PiNativeDefaultsReceipt>,
#[serde(skip)]
native_patch_receipt: Option<PiProviderPatchReceipt>,
#[serde(skip)]
native_fingerprint_preconditions: IndexMap<String, String>,
}
pub(crate) struct PiCatalogCoordinator;
@@ -241,6 +243,11 @@ impl PiCatalogCoordinator {
.ok()
.and_then(|result| result.native_patch_receipt.as_ref())
.cloned();
let native_fingerprint_preconditions = result
.as_ref()
.ok()
.map(|result| result.native_fingerprint_preconditions.clone())
.unwrap_or_default();
let mut expected_native = snapshot.native.clone();
if let Some(receipt) = native_patch_receipt.as_ref() {
for (provider_key, attempted) in receipt.attempted_values() {
@@ -252,9 +259,11 @@ impl PiCatalogCoordinator {
let reconcile = futures::executor::block_on(
state
.proxy_service
.reconcile_pi_runtime_at_epoch_with_native_precondition(
.reconcile_pi_runtime_at_epoch_with_native_claim_precondition(
catalog_epoch,
Some(&expected_native),
(!native_fingerprint_preconditions.is_empty())
.then_some(&native_fingerprint_preconditions),
),
);
match (result, reconcile) {
@@ -892,7 +901,12 @@ impl PiCatalogCoordinator {
NewProviderAggregate::from_input(PI_APP, input)?,
provider_key,
)?;
Ok(success(Some(provider_key.to_string())))
Ok(
success(Some(provider_key.to_string())).with_native_fingerprint_precondition(
provider_key.to_string(),
inspection.diagnostic.fingerprint,
),
)
}
fn set_default(
@@ -1227,6 +1241,7 @@ fn success(provider_id: Option<String>) -> PiCatalogMutationResult {
provider_id,
native_defaults_receipt: None,
native_patch_receipt: None,
native_fingerprint_preconditions: IndexMap::new(),
}
}
@@ -1240,6 +1255,16 @@ impl PiCatalogMutationResult {
self.native_patch_receipt = receipt;
self
}
fn with_native_fingerprint_precondition(
mut self,
provider_key: String,
fingerprint: String,
) -> Self {
self.native_fingerprint_preconditions
.insert(provider_key, fingerprint);
self
}
}
fn authority_error(authority: PiCatalogAuthority, message: impl std::fmt::Display) -> AppError {
@@ -1425,6 +1450,67 @@ mod tests {
Ok(())
}
#[test]
#[serial_test::serial]
fn native_import_revalidates_raw_fingerprint_after_semantically_equivalent_edit(
) -> 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 = br#"{
"providers": {
"native": {
"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, original).expect("write original");
let fingerprint = inspect_pi_native_entry(&models_path, "native", &BTreeMap::new())?
.expect("native entry")
.diagnostic
.fingerprint;
let external = br#"{
"providers": {
"native": {
// raw ownership changed while parsed values stayed identical
"name": "Native import",
"api": "openai-responses",
"baseUrl": "https://original.example/v1",
"apiKey": "literal-key",
"models": [
{"id": "model-a", "name": "Model A"},
],
},
},
}"#;
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("a raw-only external edit must reject the ownership claim");
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,
"raw external ownership must remain intact"
);
Ok(())
}
#[test]
fn portable_sql_import_claims_only_an_absent_identity_key_and_is_idempotent(
) -> Result<(), AppError> {
src-tauri/src/services/prompt.rs
+172 -28
View File
@@ -96,6 +96,11 @@ impl PromptService {
pub fn delete_prompt(state: &AppState, app: AppType, id: &str) -> Result<(), AppError> {
if matches!(app, AppType::Pi) {
let _switch_guard = futures::executor::block_on(
state
.proxy_service
.lock_switch_for_app(AppType::Pi.as_str()),
);
let guard = lock_instruction_files()?;
let prompts = state.db.get_prompts(AppType::Pi.as_str())?;
let snapshot =
@@ -108,7 +113,13 @@ impl PromptService {
"无法删除 Pi 当前生效的提示词".to_string(),
));
}
return state.db.delete_prompt(AppType::Pi.as_str(), id);
let mut after = prompts.clone();
after.shift_remove(id);
return state.db.compare_exchange_prompt_selection(
AppType::Pi.as_str(),
&prompts,
&after,
);
}
let prompts = state.db.get_prompts(app.as_str())?;
@@ -200,6 +211,11 @@ impl PromptService {
pub fn import_from_file(state: &AppState, app: AppType) -> Result<String, AppError> {
if matches!(app, AppType::Pi) {
let _switch_guard = futures::executor::block_on(
state
.proxy_service
.lock_switch_for_app(AppType::Pi.as_str()),
);
let guard = lock_instruction_files()?;
let snapshot =
PiPromptFileService::read_under_guard(&guard, PiPromptFileKind::GlobalContext)?;
@@ -258,24 +274,29 @@ impl PromptService {
state: &AppState,
app: AppType,
) -> Result<usize, AppError> {
// 幂等性保护：该应用已有提示词则跳过
let existing = state.db.get_prompts(app.as_str())?;
if !existing.is_empty() {
return Ok(0);
}
if matches!(app, AppType::Pi) {
let _switch_guard = futures::executor::block_on(
state
.proxy_service
.lock_switch_for_app(AppType::Pi.as_str()),
);
let guard = lock_instruction_files()?;
let existing = state.db.get_prompts(app.as_str())?;
if !existing.is_empty() {
return Ok(0);
}
let snapshot =
PiPromptFileService::read_under_guard(&guard, PiPromptFileKind::GlobalContext)?;
if !snapshot.exists {
return Ok(0);
}
let timestamp = get_unix_timestamp()?;
state.db.save_prompt(
app.as_str(),
&Prompt {
id: format!("auto-imported-{timestamp}"),
let mut after = existing.clone();
let id = format!("auto-imported-{timestamp}");
after.insert(
id.clone(),
Prompt {
id,
name: format!(
"Auto-imported Prompt {}",
chrono::Local::now().format("%Y-%m-%d %H:%M")
@@ -286,10 +307,19 @@ impl PromptService {
created_at: Some(timestamp),
updated_at: Some(timestamp),
},
)?;
);
state
.db
.compare_exchange_prompt_selection(app.as_str(), &existing, &after)?;
return Ok(1);
}
// 幂等性保护：该应用已有提示词则跳过
let existing = state.db.get_prompts(app.as_str())?;
if !existing.is_empty() {
return Ok(0);
}
let file_path = prompt_file_path(&app)?;
// 检查文件是否存在
@@ -344,6 +374,11 @@ impl PromptService {
/// counterpart is added. A missing file disables every row. The file is
/// never created, replaced, or deleted by portable reconciliation.
pub(crate) fn reconcile_pi_portable_import(state: &AppState) -> Result<(), AppError> {
let _switch_guard = futures::executor::block_on(
state
.proxy_service
.lock_switch_for_app(AppType::Pi.as_str()),
);
let guard = lock_instruction_files()?;
Self::reconcile_pi_native_under_guard(state.db.as_ref(), &guard)
}
@@ -365,15 +400,15 @@ impl PromptService {
continue;
}
db.save_prompt_selection(AppType::Pi.as_str(), &prompts)?;
db.compare_exchange_prompt_selection(AppType::Pi.as_str(), &original, &prompts)?;
#[cfg(test)]
replace_pi_agents_after_reconcile_save_for_test(&snapshot.path)?;
replace_pi_agents_after_reconcile_save_for_test(db, &snapshot.path)?;
let verified =
match PiPromptFileService::read_under_guard(guard, PiPromptFileKind::GlobalContext)
{
Ok(verified) => verified,
Err(error) => {
restore_pi_library_after_failed_reconcile(db, &original, &error)?;
restore_pi_library_after_failed_reconcile(db, &original, &prompts, &error)?;
return Err(error);
}
};
@@ -383,6 +418,7 @@ impl PromptService {
restore_pi_library_after_failed_reconcile(
db,
&original,
&prompts,
&AppError::Conflict(
"Pi AGENTS.md changed after prompt-library publication".to_string(),
),
@@ -427,6 +463,11 @@ impl PromptService {
}
fn upsert_pi_prompt(state: &AppState, prompt: Prompt) -> Result<(), AppError> {
let _switch_guard = futures::executor::block_on(
state
.proxy_service
.lock_switch_for_app(AppType::Pi.as_str()),
);
let guard = lock_instruction_files()?;
let before = state.db.get_prompts(AppType::Pi.as_str())?;
let snapshot =
@@ -454,9 +495,10 @@ impl PromptService {
&snapshot.revision,
&prompt.content,
)?;
if let Err(error) = state
.db
.save_prompt_selection(AppType::Pi.as_str(), &prompts)
if let Err(error) =
state
.db
.compare_exchange_prompt_selection(AppType::Pi.as_str(), &before, &prompts)
{
restore_pi_prompt_file(&guard, &published, &snapshot)?;
return Err(error);
@@ -471,9 +513,10 @@ impl PromptService {
PiPromptFileKind::GlobalContext,
&snapshot.revision,
)?;
if let Err(error) = state
.db
.save_prompt_selection(AppType::Pi.as_str(), &prompts)
if let Err(error) =
state
.db
.compare_exchange_prompt_selection(AppType::Pi.as_str(), &before, &prompts)
{
if removed {
let missing = PiPromptFileService::read_under_guard(
@@ -494,10 +537,15 @@ impl PromptService {
state
.db
.save_prompt_selection(AppType::Pi.as_str(), &prompts)
.compare_exchange_prompt_selection(AppType::Pi.as_str(), &before, &prompts)
}
fn enable_pi_prompt(state: &AppState, id: &str) -> Result<(), AppError> {
let _switch_guard = futures::executor::block_on(
state
.proxy_service
.lock_switch_for_app(AppType::Pi.as_str()),
);
let guard = lock_instruction_files()?;
let before = state.db.get_prompts(AppType::Pi.as_str())?;
let target = before
@@ -521,7 +569,11 @@ impl PromptService {
for prompt in after.values_mut() {
prompt.enabled = prompt.id == id;
}
if let Err(error) = state.db.save_prompt_selection(AppType::Pi.as_str(), &after) {
if let Err(error) =
state
.db
.compare_exchange_prompt_selection(AppType::Pi.as_str(), &before, &after)
{
restore_pi_prompt_file(&guard, &published, &snapshot)?;
return Err(error);
}
@@ -534,7 +586,8 @@ impl PromptService {
) -> Result<String, AppError> {
let timestamp = get_unix_timestamp()?;
let id = format!("imported-{timestamp}");
let mut prompts = state.db.get_prompts(AppType::Pi.as_str())?;
let before = state.db.get_prompts(AppType::Pi.as_str())?;
let mut prompts = before.clone();
for prompt in prompts.values_mut() {
prompt.enabled = false;
}
@@ -559,7 +612,7 @@ impl PromptService {
);
state
.db
.save_prompt_selection(AppType::Pi.as_str(), &prompts)?;
.compare_exchange_prompt_selection(AppType::Pi.as_str(), &before, &prompts)?;
Ok(id)
}
}
@@ -670,13 +723,15 @@ fn build_pi_reconciled_library(
fn restore_pi_library_after_failed_reconcile(
db: &Database,
original: &IndexMap<String, Prompt>,
attempted: &IndexMap<String, Prompt>,
cause: &AppError,
) -> Result<(), AppError> {
db.save_prompt_selection(AppType::Pi.as_str(), original)
db.restore_prompt_selection_if_attempted(AppType::Pi.as_str(), attempted, original)
.map_err(|restore_error| {
AppError::Config(format!(
"Pi prompt reconciliation lost its native revision ({cause}) and failed to \
restore the previous portable library: {restore_error}"
restore the previous portable library without overwriting a newer database \
revision: {restore_error}"
))
})
}
@@ -687,7 +742,17 @@ static PI_RECONCILE_AFTER_SAVE_REPLACEMENTS: std::sync::LazyLock<
> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::VecDeque::new()));
#[cfg(test)]
fn replace_pi_agents_after_reconcile_save_for_test(path: &str) -> Result<(), AppError> {
fn replace_pi_agents_after_reconcile_save_for_test(
db: &Database,
path: &str,
) -> Result<(), AppError> {
if let Some(replacement) = PI_RECONCILE_AFTER_SAVE_DB_REPLACEMENTS
.lock()
.map_err(|error| AppError::Lock(error.to_string()))?
.pop_front()
{
db.save_prompt_selection(AppType::Pi.as_str(), &replacement)?;
}
let replacement = PI_RECONCILE_AFTER_SAVE_REPLACEMENTS
.lock()
.map_err(|error| AppError::Lock(error.to_string()))?
@@ -698,6 +763,19 @@ fn replace_pi_agents_after_reconcile_save_for_test(path: &str) -> Result<(), App
Ok(())
}
#[cfg(test)]
static PI_RECONCILE_AFTER_SAVE_DB_REPLACEMENTS: std::sync::LazyLock<
std::sync::Mutex<std::collections::VecDeque<IndexMap<String, Prompt>>>,
> = std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::VecDeque::new()));
#[cfg(test)]
fn replace_pi_library_after_next_reconcile_save_for_test(replacement: IndexMap<String, Prompt>) {
PI_RECONCILE_AFTER_SAVE_DB_REPLACEMENTS
.lock()
.expect("Pi reconcile DB hook lock")
.push_back(replacement);
}
#[cfg(test)]
fn replace_pi_agents_after_each_reconcile_save_for_test(
replacements: impl IntoIterator<Item = &'static str>,
@@ -1192,4 +1270,70 @@ mod tests {
"native-3"
);
}
#[test]
#[serial]
fn reconcile_compensation_preserves_a_concurrent_portable_library() {
let temp = tempfile::tempdir().expect("tempdir");
let _restore = EnvRestore::set("PI_CODING_AGENT_DIR", temp.path());
std::fs::write(temp.path().join("AGENTS.md"), "native-0").expect("seed AGENTS.md");
let state = AppState::new(Arc::new(Database::memory().expect("database")));
state
.db
.save_prompt(
AppType::Pi.as_str(),
&prompt("before", "portable-before", true, 1),
)
.expect("seed original library");
let imported = IndexMap::from([(
"restored-import".to_string(),
prompt("restored-import", "portable-restored", true, 2),
)]);
replace_pi_library_after_next_reconcile_save_for_test(imported.clone());
replace_pi_agents_after_each_reconcile_save_for_test(["native-1"]);
let error = PromptService::reconcile_pi_library(&state)
.expect_err("stale compensation must not overwrite a concurrent import");
assert!(
error.to_string().contains("newer database revision"),
"the conflict must explain why compensation stopped: {error}"
);
assert_eq!(
state
.db
.get_prompts(AppType::Pi.as_str())
.expect("preserved imported library"),
imported
);
assert_eq!(
std::fs::read_to_string(temp.path().join("AGENTS.md")).expect("latest native"),
"native-1"
);
}
#[test]
fn prompt_selection_compensation_can_restore_an_exact_legacy_before_image() {
let db = Database::memory().expect("database");
db.save_prompt(AppType::Pi.as_str(), &prompt("legacy-a", "a", true, 1))
.expect("first legacy row");
db.save_prompt(AppType::Pi.as_str(), &prompt("legacy-b", "b", true, 2))
.expect("second legacy row");
let before = db
.get_prompts(AppType::Pi.as_str())
.expect("legacy before-image");
let mut attempted = before.clone();
attempted.get_mut("legacy-a").expect("first row").enabled = false;
db.compare_exchange_prompt_selection(AppType::Pi.as_str(), &before, &attempted)
.expect("publish valid projection");
db.restore_prompt_selection_if_attempted(AppType::Pi.as_str(), &attempted, &before)
.expect("restore exact legacy image");
assert_eq!(
db.get_prompts(AppType::Pi.as_str())
.expect("restored library"),
before
);
}
}
src-tauri/src/services/proxy.rs
+44 -5
View File
@@ -620,6 +620,20 @@ impl ProxyService {
&self,
catalog_epoch: u64,
expected_native: Option<&crate::pi_config::document::PiProviderValuesSnapshot>,
) -> Result<Vec<String>, AppError> {
self.reconcile_pi_runtime_at_epoch_with_native_claim_precondition(
catalog_epoch,
expected_native,
None,
)
.await
}
pub(crate) async fn reconcile_pi_runtime_at_epoch_with_native_claim_precondition(
&self,
catalog_epoch: u64,
expected_native: Option<&crate::pi_config::document::PiProviderValuesSnapshot>,
expected_fingerprints: Option<&indexmap::IndexMap<String, String>>,
) -> Result<Vec<String>, AppError> {
#[cfg(test)]
if self.fail_next_pi_reconcile.swap(false, Ordering::AcqRel) {
@@ -631,10 +645,19 @@ impl ProxyService {
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,
)?;
match expected_fingerprints {
Some(expected_fingerprints) => {
crate::pi_config::document::verify_pi_provider_preconditions(
&models_path,
&expected_native.values,
expected_fingerprints,
)?;
}
None => crate::pi_config::document::verify_pi_provider_values(
&models_path,
&expected_native.values,
)?,
}
}
return Ok(Vec::new());
}
@@ -660,6 +683,21 @@ impl ProxyService {
)?;
let models_path = crate::pi_config::native::get_pi_models_path()?;
let native_receipt = if build.projection_patch.is_empty() {
if let Some(expected_native) = expected_native {
match expected_fingerprints {
Some(expected_fingerprints) => {
crate::pi_config::document::verify_pi_provider_preconditions(
&models_path,
&expected_native.values,
expected_fingerprints,
)?;
}
None => crate::pi_config::document::verify_pi_provider_values(
&models_path,
&expected_native.values,
)?,
}
}
None
} else {
let expected = match expected_native {
@@ -690,9 +728,10 @@ impl ProxyService {
)?,
};
Some(
crate::pi_config::document::apply_pi_provider_patch_with_receipt(
crate::pi_config::document::apply_pi_provider_patch_with_receipt_and_fingerprints(
&models_path,
&expected,
expected_fingerprints,
&build.projection_patch,
)?,
)