use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use asupersync::Cx;
use chrono::{DateTime, Duration, Utc};
use crate::core::search::{
SearchDedupMode, SearchFusionWeights, SearchHit, SearchOptions, SearchSourceMode,
resolved_search_fusion_weights, run_search_with_read_connection_seeded_with_cx,
search_hit_meets_relevance_floor, sort_search_hits_by_score_order,
};
use crate::db::{DbConnection, StoredAuditEntry, StoredFeedbackEvent, audit_actions};
use crate::models::MemoryScope;
use crate::obs::audit_events::query_hash as audit_query_hash;
use crate::runtime::determinism::Deterministic;
use crate::search::SpeedMode;
pub const PACK_ITEM_EVIDENCE_SCHEMA_V1: &str = "ee.outcome.pack_item_evidence.v1";
pub const DEFAULT_LABEL_WINDOW_MINUTES: u32 = 30;
const LABEL_SET_HASH_DOMAIN: &str = "ee.shadow.label_set.v1";
const FRESHNESS_HALF_LIFE_DAYS: f64 = 90.0;
const DENSE_BASE_WEIGHT: f64 = 1.0;
const WEAK_BASE_WEIGHT: f64 = 0.5;
const CANCELLATION_CHUNK: usize = 256;
const PACK_METADATA_QUERY_CHAR_CAP: usize = 2048;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LabelExtractionConfig {
pub label_window_minutes: u32,
}
impl Default for LabelExtractionConfig {
fn default() -> Self {
Self {
label_window_minutes: DEFAULT_LABEL_WINDOW_MINUTES,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LabelSource {
PackItemOutcome,
SearchWindowAssociation,
}
impl LabelSource {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::PackItemOutcome => "dense",
Self::SearchWindowAssociation => "weak",
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct LabeledTriple {
pub query: String,
pub memory_id: String,
pub signal: String,
pub base_weight: f64,
pub weight: f64,
pub age_days: f64,
pub source: LabelSource,
pub feedback_event_id: String,
pub pack_record_id: Option<String>,
pub audit_row_id: Option<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct LabelExtractionReport {
pub triples: Vec<LabeledTriple>,
pub distinct_queries: usize,
pub memory_event_count: usize,
pub dense_count: usize,
pub weak_count: usize,
pub dense_unresolvable: usize,
pub weak_unreplayable: usize,
pub weak_unmatched: usize,
pub label_set_hash: String,
}
#[derive(Debug)]
pub enum ShadowTuningError {
Cancelled(asupersync::CancelReason),
Storage { message: String },
}
impl std::fmt::Display for ShadowTuningError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Cancelled(reason) => {
write!(f, "shadow-tuning label extraction cancelled: {reason:?}")
}
Self::Storage { message } => {
write!(f, "shadow-tuning storage error: {message}")
}
}
}
}
impl std::error::Error for ShadowTuningError {}
fn shadow_checkpoint(cx: &Cx) -> Result<(), ShadowTuningError> {
cx.checkpoint().map_err(|_| {
ShadowTuningError::Cancelled(cx.cancel_reason().unwrap_or_else(|| {
crate::core::outcome::attributed_cancel_reason(
cx,
asupersync::CancelKind::User,
"shadow-tuning label extraction cancelled without a recorded reason",
)
}))
})
}
fn storage_error(context: &str, error: &dyn std::fmt::Display) -> ShadowTuningError {
ShadowTuningError::Storage {
message: format!("{context}: {error}"),
}
}
pub fn extract_labeled_triples(
cx: &Cx,
connection: &DbConnection,
workspace_id: &str,
config: &LabelExtractionConfig,
as_of: DateTime<Utc>,
) -> Result<LabelExtractionReport, ShadowTuningError> {
shadow_checkpoint(cx)?;
let events = connection
.list_feedback_events(workspace_id)
.map_err(|error| storage_error("list feedback events", &error))?;
shadow_checkpoint(cx)?;
let returned_mem_audits: Vec<StoredAuditEntry> = connection
.list_audit_by_action(audit_actions::SEARCH_RETURNED_MEM, None)
.map_err(|error| storage_error("list search.returned_mem audit rows", &error))?
.into_iter()
.filter(|row| row.workspace_id.as_deref() == Some(workspace_id))
.collect();
shadow_checkpoint(cx)?;
let metadata = connection
.list_recent_pack_record_metadata_for_workspace(workspace_id, u32::MAX)
.map_err(|error| storage_error("list pack record metadata", &error))?;
let mut pack_queries: BTreeMap<String, String> = BTreeMap::new();
let mut query_text_by_hash: BTreeMap<String, String> = BTreeMap::new();
for (index, meta) in metadata.into_iter().enumerate() {
if index % CANCELLATION_CHUNK == 0 {
shadow_checkpoint(cx)?;
}
let query = if meta.query.chars().count() >= PACK_METADATA_QUERY_CHAR_CAP {
connection
.get_pack_record(&meta.id)
.map_err(|error| storage_error("load full pack record", &error))?
.map_or(meta.query, |record| record.query)
} else {
meta.query
};
let hash = audit_query_hash(&query);
match query_text_by_hash.get(&hash) {
Some(existing) if existing <= &query => {}
_ => {
query_text_by_hash.insert(hash, query.clone());
}
}
pack_queries.insert(meta.id, query);
}
join_labeled_triples(
cx,
&events,
&returned_mem_audits,
&pack_queries,
&query_text_by_hash,
config,
as_of,
)
}
enum DenseLinkage {
NotDense,
Unresolvable,
Linked {
pack_id: String,
},
}
fn parse_pack_item_linkage(evidence_json: Option<&str>) -> DenseLinkage {
let Some(raw) = evidence_json else {
return DenseLinkage::NotDense;
};
let Ok(value) = serde_json::from_str::<serde_json::Value>(raw) else {
return DenseLinkage::NotDense;
};
if value.get("schema").and_then(serde_json::Value::as_str) != Some(PACK_ITEM_EVIDENCE_SCHEMA_V1)
{
return DenseLinkage::NotDense;
}
match value.get("packId").and_then(serde_json::Value::as_str) {
Some(pack_id) if !pack_id.trim().is_empty() => DenseLinkage::Linked {
pack_id: pack_id.to_owned(),
},
_ => DenseLinkage::Unresolvable,
}
}
struct ParsedReturnedMem {
id: String,
memory_id: String,
timestamp: DateTime<Utc>,
query_hash: Option<String>,
}
fn parse_returned_mem_rows(
rows: &[StoredAuditEntry],
) -> Result<Vec<ParsedReturnedMem>, ShadowTuningError> {
let mut parsed = Vec::with_capacity(rows.len());
for row in rows {
if row.action != audit_actions::SEARCH_RETURNED_MEM {
continue;
}
let Some(memory_id) = row.target_id.as_deref() else {
continue;
};
let timestamp = DateTime::parse_from_rfc3339(&row.timestamp)
.map_err(|error| {
storage_error(
&format!("audit row {} has an unparsable timestamp", row.id),
&error,
)
})?
.with_timezone(&Utc);
let query_hash = row
.details
.as_deref()
.and_then(|details| serde_json::from_str::<serde_json::Value>(details).ok())
.and_then(|value| {
value
.get("queryHash")
.and_then(serde_json::Value::as_str)
.map(str::to_owned)
});
parsed.push(ParsedReturnedMem {
id: row.id.clone(),
memory_id: memory_id.to_owned(),
timestamp,
query_hash,
});
}
Ok(parsed)
}
#[allow(clippy::too_many_lines)]
pub fn join_labeled_triples(
cx: &Cx,
events: &[StoredFeedbackEvent],
returned_mem_audits: &[StoredAuditEntry],
pack_queries: &BTreeMap<String, String>,
query_text_by_hash: &BTreeMap<String, String>,
config: &LabelExtractionConfig,
as_of: DateTime<Utc>,
) -> Result<LabelExtractionReport, ShadowTuningError> {
shadow_checkpoint(cx)?;
let window = Duration::minutes(i64::from(config.label_window_minutes));
let parsed_audits = parse_returned_mem_rows(returned_mem_audits)?;
let mut audits_by_memory: BTreeMap<&str, Vec<&ParsedReturnedMem>> = BTreeMap::new();
for audit in &parsed_audits {
audits_by_memory
.entry(audit.memory_id.as_str())
.or_default()
.push(audit);
}
let mut triples: Vec<LabeledTriple> = Vec::new();
let mut memory_event_count = 0_usize;
let mut dense_unresolvable = 0_usize;
let mut weak_unreplayable = 0_usize;
let mut weak_unmatched = 0_usize;
for (index, event) in events.iter().enumerate() {
if index % CANCELLATION_CHUNK == 0 {
shadow_checkpoint(cx)?;
}
if event.target_type != "memory" {
continue;
}
memory_event_count += 1;
let created_at = DateTime::parse_from_rfc3339(&event.created_at)
.map_err(|error| {
storage_error(
&format!("feedback event {} has an unparsable created_at", event.id),
&error,
)
})?
.with_timezone(&Utc);
let age_days = age_in_days(created_at, as_of);
let freshness = (-age_days / FRESHNESS_HALF_LIFE_DAYS).exp2();
match parse_pack_item_linkage(event.evidence_json.as_deref()) {
DenseLinkage::Linked { pack_id } => match pack_queries.get(&pack_id) {
Some(query) => triples.push(LabeledTriple {
query: query.clone(),
memory_id: event.target_id.clone(),
signal: event.signal.clone(),
base_weight: DENSE_BASE_WEIGHT,
weight: DENSE_BASE_WEIGHT * freshness,
age_days,
source: LabelSource::PackItemOutcome,
feedback_event_id: event.id.clone(),
pack_record_id: Some(pack_id),
audit_row_id: None,
}),
None => dense_unresolvable += 1,
},
DenseLinkage::Unresolvable => dense_unresolvable += 1,
DenseLinkage::NotDense => {
let nearest = audits_by_memory
.get(event.target_id.as_str())
.into_iter()
.flatten()
.filter(|audit| {
audit.timestamp <= created_at && created_at - audit.timestamp <= window
})
.max_by(|a, b| a.timestamp.cmp(&b.timestamp).then_with(|| a.id.cmp(&b.id)));
match nearest {
None => weak_unmatched += 1,
Some(audit) => {
let resolved = audit
.query_hash
.as_deref()
.and_then(|hash| query_text_by_hash.get(hash));
match resolved {
None => weak_unreplayable += 1,
Some(query) => triples.push(LabeledTriple {
query: query.clone(),
memory_id: event.target_id.clone(),
signal: event.signal.clone(),
base_weight: WEAK_BASE_WEIGHT,
weight: WEAK_BASE_WEIGHT * freshness,
age_days,
source: LabelSource::SearchWindowAssociation,
feedback_event_id: event.id.clone(),
pack_record_id: None,
audit_row_id: Some(audit.id.clone()),
}),
}
}
}
}
}
}
shadow_checkpoint(cx)?;
triples.sort_by(|a, b| {
a.query
.cmp(&b.query)
.then_with(|| a.memory_id.cmp(&b.memory_id))
.then_with(|| a.feedback_event_id.cmp(&b.feedback_event_id))
});
let distinct_queries = triples
.iter()
.map(|triple| triple.query.as_str())
.collect::<BTreeSet<&str>>()
.len();
let dense_count = triples
.iter()
.filter(|triple| triple.source == LabelSource::PackItemOutcome)
.count();
let weak_count = triples.len() - dense_count;
let label_set_hash = label_set_hash(&triples);
Ok(LabelExtractionReport {
triples,
distinct_queries,
memory_event_count,
dense_count,
weak_count,
dense_unresolvable,
weak_unreplayable,
weak_unmatched,
label_set_hash,
})
}
fn age_in_days(created_at: DateTime<Utc>, as_of: DateTime<Utc>) -> f64 {
let seconds = (as_of - created_at).num_seconds().max(0);
#[allow(clippy::cast_precision_loss)]
let seconds = seconds as f64;
seconds / 86_400.0
}
fn append_len_prefixed(out: &mut Vec<u8>, bytes: &[u8]) {
let len = u32::try_from(bytes.len()).unwrap_or(u32::MAX);
out.extend_from_slice(&len.to_be_bytes());
out.extend_from_slice(bytes);
}
fn label_set_hash(triples: &[LabeledTriple]) -> String {
let mut input = Vec::new();
append_len_prefixed(&mut input, LABEL_SET_HASH_DOMAIN.as_bytes());
input.extend_from_slice(
&u32::try_from(triples.len())
.unwrap_or(u32::MAX)
.to_be_bytes(),
);
for triple in triples {
append_len_prefixed(&mut input, triple.query.as_bytes());
append_len_prefixed(&mut input, triple.memory_id.as_bytes());
append_len_prefixed(&mut input, triple.signal.as_bytes());
append_len_prefixed(&mut input, triple.source.as_str().as_bytes());
append_len_prefixed(&mut input, triple.feedback_event_id.as_bytes());
append_len_prefixed(
&mut input,
triple.pack_record_id.as_deref().unwrap_or("").as_bytes(),
);
append_len_prefixed(
&mut input,
triple.audit_row_id.as_deref().unwrap_or("").as_bytes(),
);
input.extend_from_slice(&triple.base_weight.to_bits().to_be_bytes());
input.extend_from_slice(&triple.weight.to_bits().to_be_bytes());
input.extend_from_slice(&triple.age_days.to_bits().to_be_bytes());
}
format!("blake3:{}", blake3::hash(&input).to_hex())
}
const FUSION_LEXICAL_CLAMP: (f32, f32) = (0.2, 0.7);
const FUSION_SEMANTIC_CLAMP: (f32, f32) = (0.2, 0.7);
const FUSION_GRAPH_CLAMP: (f32, f32) = (0.0, 0.3);
const FUSION_GRID_OFFSETS: [f32; 4] = [-0.10, -0.05, 0.05, 0.10];
const DESCENT_MAX_ROUNDS: u32 = 2;
const DESCENT_INITIAL_STEP: f32 = 0.025;
pub const REPLAY_POOL_LIMIT: u32 = 200;
const EVALUATION_HASH_DOMAIN: &str = "ee.shadow.retrieval_tuning_evaluation.v1";
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct TuningWeights {
pub lexical: f32,
pub semantic: f32,
pub graph: f32,
}
impl TuningWeights {
#[must_use]
pub fn compiled_defaults() -> Self {
Self::from_fusion(SearchFusionWeights::default())
}
#[must_use]
pub fn incumbent_for_workspace(workspace_path: &Path) -> Self {
Self::from_fusion(resolved_search_fusion_weights(workspace_path))
}
fn from_fusion(weights: SearchFusionWeights) -> Self {
Self {
lexical: weights.lexical,
semantic: weights.semantic,
graph: weights.graph,
}
}
fn to_fusion(self) -> SearchFusionWeights {
SearchFusionWeights {
lexical: self.lexical,
semantic: self.semantic,
graph: self.graph,
}
}
fn clamped(self) -> Self {
Self {
lexical: self
.lexical
.clamp(FUSION_LEXICAL_CLAMP.0, FUSION_LEXICAL_CLAMP.1),
semantic: self
.semantic
.clamp(FUSION_SEMANTIC_CLAMP.0, FUSION_SEMANTIC_CLAMP.1),
graph: self.graph.clamp(FUSION_GRAPH_CLAMP.0, FUSION_GRAPH_CLAMP.1),
}
}
fn key(self) -> (u32, u32, u32) {
(
self.lexical.to_bits(),
self.semantic.to_bits(),
self.graph.to_bits(),
)
}
}
#[derive(Debug, Clone)]
pub struct QueryReplay {
pub query: String,
pub hits: Vec<SearchHit>,
}
#[derive(Debug, Clone)]
pub struct ReplayCollection {
pub replays: Vec<QueryReplay>,
pub unrecoverable_hits: usize,
}
pub async fn collect_query_replays_with_cx(
cx: &Cx,
read_connection: &crate::db::DbConnection,
workspace_path: &Path,
database_path: &Path,
queries: &BTreeSet<String>,
as_of: DateTime<Utc>,
) -> Result<ReplayCollection, ShadowTuningError> {
let determinism = Deterministic::from_seed(0);
let mut replays = Vec::with_capacity(queries.len());
for query in queries {
shadow_checkpoint(cx)?;
let options = SearchOptions {
workspace_path: workspace_path.to_path_buf(),
database_path: Some(database_path.to_path_buf()),
index_dir: None,
query: query.clone(),
limit: REPLAY_POOL_LIMIT,
speed: SpeedMode::Default,
explain: false,
as_of: Some(as_of),
include_tombstoned: false,
include_expired: false,
include_future: false,
include_stale: false,
relevance_floor: Some(0.0),
dedup_mode: SearchDedupMode::DocId,
source_mode: SearchSourceMode::Hybrid,
strict_source_mode: false,
memory_scope: MemoryScope::default(),
strict_scope: false,
};
let report = run_search_with_read_connection_seeded_with_cx(
cx,
&options,
read_connection,
determinism.shared_child("search.rerank"),
)
.await
.map_err(|error| storage_error("replay search failed", &error))?;
replays.push(QueryReplay {
query: query.clone(),
hits: report.results,
});
}
Ok(ReplayCollection {
replays,
unrecoverable_hits: 0,
})
}
fn signal_gain(signal: &str) -> Option<f64> {
match signal {
"helpful" | "confirmation" => Some(1.0),
"harmful" | "contradiction" => Some(-2.0),
_ => None,
}
}
struct QueryLabelGains {
norm: f64,
gains: Vec<(String, f64)>,
}
struct GroupedLabels {
by_query: BTreeMap<String, QueryLabelGains>,
labels_unmapped_signal: usize,
queries_without_gain: usize,
}
fn group_label_gains(labels: &[LabeledTriple]) -> GroupedLabels {
let mut raw: BTreeMap<String, Vec<(String, f64)>> = BTreeMap::new();
let mut labels_unmapped_signal = 0_usize;
for label in labels {
let Some(gain) = signal_gain(&label.signal) else {
labels_unmapped_signal += 1;
continue;
};
raw.entry(label.query.clone())
.or_default()
.push((label.memory_id.clone(), label.weight * gain));
}
let mut by_query = BTreeMap::new();
let mut queries_without_gain = 0_usize;
for (query, gains) in raw {
let denom: f64 = gains.iter().map(|(_, value)| value.abs()).sum();
if denom <= f64::EPSILON {
queries_without_gain += 1;
continue;
}
by_query.insert(
query,
QueryLabelGains {
norm: 1.0 / denom,
gains,
},
);
}
GroupedLabels {
by_query,
labels_unmapped_signal,
queries_without_gain,
}
}
#[allow(clippy::cast_possible_truncation)]
fn ranked_pool(hits: &[SearchHit], weights: SearchFusionWeights) -> BTreeMap<String, usize> {
let mut reranked = hits
.iter()
.filter(|hit| hit.source == crate::core::search::ScoreSource::Reranked)
.cloned()
.collect::<Vec<_>>();
reranked.retain(|hit| search_hit_meets_relevance_floor(hit, None));
sort_search_hits_by_score_order(&mut reranked);
let mut lexical = hits
.iter()
.filter(|hit| hit.source != crate::core::search::ScoreSource::Reranked)
.filter_map(|hit| {
hit.lexical_score
.filter(|score| score.is_finite())
.map(|score| crate::search::ScoredResult {
doc_id: hit.doc_id.clone().into(),
score,
source: crate::search::ScoreSource::Lexical,
index: None,
fast_score: None,
quality_score: None,
lexical_score: Some(score),
rerank_score: None,
explanation: None,
metadata: None,
})
})
.collect::<Vec<_>>();
lexical.sort_by(|left, right| {
right
.score
.total_cmp(&left.score)
.then_with(|| left.doc_id.cmp(&right.doc_id))
});
let mut semantic = hits
.iter()
.filter(|hit| hit.source != crate::core::search::ScoreSource::Reranked)
.filter_map(|hit| {
hit.quality_score
.or(hit.fast_score)
.filter(|score| score.is_finite())
.map(|score| (hit.doc_id.clone(), score))
})
.collect::<Vec<_>>();
semantic.sort_by(|left, right| {
right
.1
.total_cmp(&left.1)
.then_with(|| left.0.cmp(&right.0))
});
let semantic = semantic
.into_iter()
.enumerate()
.map(
|(index, (doc_id, score))| frankensearch::core::types::VectorHit {
index: u32::try_from(index).unwrap_or(u32::MAX),
score,
doc_id: doc_id.into(),
},
)
.collect::<Vec<_>>();
let (lexical_weight, semantic_weight) = weights.upstream_rrf_weights();
let config = frankensearch::RrfConfig {
lexical_weight,
semantic_weight,
..frankensearch::RrfConfig::default()
};
let by_id = hits
.iter()
.map(|hit| (hit.doc_id.as_str(), hit))
.collect::<BTreeMap<_, _>>();
let fused_ids = frankensearch::rrf_fuse(&lexical, &semantic, hits.len(), 0, &config)
.into_iter()
.filter_map(|fused| {
let doc_id = fused.doc_id.to_string();
let mut hit = (*by_id.get(doc_id.as_str())?).clone();
hit.score = fused.rrf_score as f32;
hit.source = crate::core::search::ScoreSource::Hybrid;
search_hit_meets_relevance_floor(&hit, None).then_some(hit.doc_id)
})
.collect::<Vec<_>>();
reranked
.into_iter()
.map(|hit| hit.doc_id)
.chain(fused_ids)
.enumerate()
.map(|(index, doc_id)| (doc_id, index + 1))
.collect()
}
fn score_candidate(
replays: &BTreeMap<&str, &QueryReplay>,
grouped: &GroupedLabels,
weights: SearchFusionWeights,
) -> f64 {
let mut total = 0.0_f64;
for (query, label_gains) in &grouped.by_query {
let Some(replay) = replays.get(query.as_str()) else {
continue;
};
let ranks = ranked_pool(&replay.hits, weights);
let mut query_score = 0.0_f64;
for (memory_id, weighted_gain) in &label_gains.gains {
if let Some(rank) = ranks.get(memory_id) {
#[allow(clippy::cast_precision_loss)]
let discount = (1.0 + *rank as f64).log2();
query_score += weighted_gain / discount;
}
}
total += label_gains.norm * query_score;
}
total
}
#[must_use]
pub fn score_fusion_candidate(
replays: &[QueryReplay],
labels: &[LabeledTriple],
weights: TuningWeights,
) -> f64 {
let replays_by_query: BTreeMap<&str, &QueryReplay> = replays
.iter()
.map(|replay| (replay.query.as_str(), replay))
.collect();
let grouped = group_label_gains(labels);
score_candidate(&replays_by_query, &grouped, weights.to_fusion())
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct CandidateScore {
pub weights: TuningWeights,
pub score: f64,
pub origin: &'static str,
}
#[derive(Debug, Clone, PartialEq)]
pub struct TuningEvaluation {
pub incumbent: CandidateScore,
pub candidates: Vec<CandidateScore>,
pub winner: Option<CandidateScore>,
pub relative_margin: Option<f64>,
pub queries_scored: usize,
pub queries_without_gain: usize,
pub labels_unmapped_signal: usize,
pub labels_total: usize,
pub graph_axis_degenerate: bool,
pub evaluation_hash: String,
}
fn enumerate_grid(incumbent: TuningWeights) -> Vec<TuningWeights> {
let mut seen = BTreeSet::new();
let mut vectors = Vec::new();
let mut push = |candidate: TuningWeights, vectors: &mut Vec<TuningWeights>| {
if seen.insert(candidate.key()) {
vectors.push(candidate);
}
};
push(incumbent.clamped(), &mut vectors);
for axis in 0..3_usize {
for offset in FUSION_GRID_OFFSETS {
let mut candidate = incumbent;
match axis {
0 => candidate.lexical += offset,
1 => candidate.semantic += offset,
_ => candidate.graph += offset,
}
push(candidate.clamped(), &mut vectors);
}
}
vectors
}
fn descent_neighbors(center: TuningWeights, step: f32) -> Vec<TuningWeights> {
let mut neighbors = Vec::with_capacity(6);
for axis in 0..3_usize {
for direction in [-1.0_f32, 1.0] {
let mut candidate = center;
let delta = step * direction;
match axis {
0 => candidate.lexical += delta,
1 => candidate.semantic += delta,
_ => candidate.graph += delta,
}
neighbors.push(candidate.clamped());
}
}
neighbors
}
fn evaluation_hash(
incumbent: &CandidateScore,
candidates: &[CandidateScore],
labels_total: usize,
) -> String {
let mut input = Vec::new();
append_len_prefixed(&mut input, EVALUATION_HASH_DOMAIN.as_bytes());
input.extend_from_slice(
&u32::try_from(labels_total)
.unwrap_or(u32::MAX)
.to_be_bytes(),
);
let push_candidate = |candidate: &CandidateScore, input: &mut Vec<u8>| {
append_len_prefixed(input, candidate.origin.as_bytes());
input.extend_from_slice(&candidate.weights.lexical.to_bits().to_be_bytes());
input.extend_from_slice(&candidate.weights.semantic.to_bits().to_be_bytes());
input.extend_from_slice(&candidate.weights.graph.to_bits().to_be_bytes());
input.extend_from_slice(&candidate.score.to_bits().to_be_bytes());
};
push_candidate(incumbent, &mut input);
for candidate in candidates {
push_candidate(candidate, &mut input);
}
format!("blake3:{}", blake3::hash(&input).to_hex())
}
pub fn evaluate_fusion_candidates(
cx: &Cx,
replays: &[QueryReplay],
labels: &[LabeledTriple],
incumbent: TuningWeights,
) -> Result<TuningEvaluation, ShadowTuningError> {
shadow_checkpoint(cx)?;
let replays_by_query: BTreeMap<&str, &QueryReplay> = replays
.iter()
.map(|replay| (replay.query.as_str(), replay))
.collect();
let grouped = group_label_gains(labels);
let incumbent = incumbent.clamped();
let incumbent_score = CandidateScore {
weights: incumbent,
score: score_candidate(&replays_by_query, &grouped, incumbent.to_fusion()),
origin: "incumbent",
};
let mut evaluated: BTreeSet<(u32, u32, u32)> = BTreeSet::new();
evaluated.insert(incumbent.key());
let mut candidates: Vec<CandidateScore> = Vec::new();
for weights in enumerate_grid(incumbent) {
if !evaluated.insert(weights.key()) {
continue;
}
shadow_checkpoint(cx)?;
candidates.push(CandidateScore {
weights,
score: score_candidate(&replays_by_query, &grouped, weights.to_fusion()),
origin: "grid",
});
}
let mut best = candidates
.iter()
.copied()
.fold(incumbent_score, |best, candidate| {
if candidate.score > best.score {
candidate
} else {
best
}
});
for round in 0..DESCENT_MAX_ROUNDS {
let step = DESCENT_INITIAL_STEP / 2.0_f32.powi(i32::try_from(round).unwrap_or(0));
let mut improved = false;
for weights in descent_neighbors(best.weights, step) {
if !evaluated.insert(weights.key()) {
continue;
}
shadow_checkpoint(cx)?;
let candidate = CandidateScore {
weights,
score: score_candidate(&replays_by_query, &grouped, weights.to_fusion()),
origin: "descent",
};
candidates.push(candidate);
if candidate.score > best.score {
best = candidate;
improved = true;
}
}
if !improved {
break;
}
}
let winner = (best.origin != "incumbent" && best.score > incumbent_score.score).then_some(best);
let relative_margin = winner.and_then(|winner| {
(incumbent_score.score.abs() > f64::EPSILON)
.then(|| (winner.score - incumbent_score.score) / incumbent_score.score.abs())
});
let hash = evaluation_hash(&incumbent_score, &candidates, labels.len());
Ok(TuningEvaluation {
incumbent: incumbent_score,
candidates,
winner,
relative_margin,
queries_scored: grouped.by_query.len(),
queries_without_gain: grouped.queries_without_gain,
labels_unmapped_signal: grouped.labels_unmapped_signal,
labels_total: labels.len(),
graph_axis_degenerate: true,
evaluation_hash: hash,
})
}
pub const RETRIEVAL_TUNING_REPORT_SCHEMA_V1: &str = "ee.shadow.retrieval_tuning_report.v1";
pub const RETRIEVAL_TUNING_POLICY_ID: &str = "candidate.retrieval.outcome_tuned_weights";
pub const INSUFFICIENT_OUTCOME_EVIDENCE_CODE: &str = "insufficient_outcome_evidence";
const REPORT_HASH_DOMAIN: &str = "ee.shadow.retrieval_tuning_report.hash.v1";
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct RetrievalTuningGateConfig {
pub min_triples: usize,
pub min_queries: usize,
pub promote_margin: f64,
}
impl Default for RetrievalTuningGateConfig {
fn default() -> Self {
Self {
min_triples: 50,
min_queries: 15,
promote_margin: 0.03,
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct RetrievalTuningReport {
pub db_generation: u64,
pub labels: LabelExtractionReport,
pub abstained: bool,
pub abstention_reason: Option<&'static str>,
pub evaluation: Option<TuningEvaluation>,
pub promotable: bool,
pub report_hash: String,
}
pub fn assemble_retrieval_tuning_report(
labels: LabelExtractionReport,
evaluation: Option<TuningEvaluation>,
db_generation: u64,
gate: &RetrievalTuningGateConfig,
) -> Result<RetrievalTuningReport, ShadowTuningError> {
let abstained =
labels.triples.len() < gate.min_triples || labels.distinct_queries < gate.min_queries;
if abstained != evaluation.is_none() {
return Err(ShadowTuningError::Storage {
message: format!(
"evidence gate and evaluation presence disagree (abstained={abstained}, evaluation={})",
if evaluation.is_some() {
"present"
} else {
"absent"
}
),
});
}
let promotable = !abstained
&& evaluation.as_ref().is_some_and(|evaluation| {
evaluation.winner.is_some()
&& evaluation
.relative_margin
.is_some_and(|margin| margin >= gate.promote_margin)
});
let mut report = RetrievalTuningReport {
db_generation,
labels,
abstained,
abstention_reason: abstained.then_some(INSUFFICIENT_OUTCOME_EVIDENCE_CODE),
evaluation,
promotable,
report_hash: String::new(),
};
let canonical = retrieval_tuning_report_json_value(&report, false).to_string();
let mut input = Vec::new();
append_len_prefixed(&mut input, REPORT_HASH_DOMAIN.as_bytes());
append_len_prefixed(&mut input, canonical.as_bytes());
report.report_hash = format!("blake3:{}", blake3::hash(&input).to_hex());
Ok(report)
}
pub async fn run_retrieval_tuning_with_cx(
cx: &Cx,
connection: &DbConnection,
workspace_path: &Path,
database_path: &Path,
workspace_id: &str,
as_of: DateTime<Utc>,
extraction: &LabelExtractionConfig,
gate: &RetrievalTuningGateConfig,
) -> Result<RetrievalTuningReport, ShadowTuningError> {
let labels = extract_labeled_triples(cx, connection, workspace_id, extraction, as_of)?;
let db_generation = connection
.get_workspace_generation(workspace_id)
.map_err(|error| storage_error("read workspace generation", &error))?
.unwrap_or(0);
if labels.triples.len() < gate.min_triples || labels.distinct_queries < gate.min_queries {
return assemble_retrieval_tuning_report(labels, None, db_generation, gate);
}
let queries: BTreeSet<String> = labels
.triples
.iter()
.map(|triple| triple.query.clone())
.collect();
let replays = collect_query_replays_with_cx(
cx,
connection,
workspace_path,
database_path,
&queries,
as_of,
)
.await?;
let incumbent = TuningWeights::incumbent_for_workspace(workspace_path);
let evaluation = evaluate_fusion_candidates(cx, &replays.replays, &labels.triples, incumbent)?;
assemble_retrieval_tuning_report(labels, Some(evaluation), db_generation, gate)
}
pub fn run_retrieval_tuning(
workspace_path: &Path,
database_path: &Path,
workspace_id: &str,
as_of: DateTime<Utc>,
extraction: &LabelExtractionConfig,
gate: &RetrievalTuningGateConfig,
) -> Result<RetrievalTuningReport, ShadowTuningError> {
crate::core::run_cli_with_cx(std::time::Duration::from_secs(600), |cx| async move {
let connection =
DbConnection::open_file(database_path).map_err(|error| ShadowTuningError::Storage {
message: format!("open workspace database: {error}"),
})?;
run_retrieval_tuning_with_cx(
&cx,
&connection,
workspace_path,
database_path,
workspace_id,
as_of,
extraction,
gate,
)
.await
})
.map_err(|error| ShadowTuningError::Storage {
message: format!("start shadow-tuning runtime: {error}"),
})?
}
fn weights_json(weights: TuningWeights) -> serde_json::Value {
serde_json::json!({
"lexical": f64::from(weights.lexical),
"semantic": f64::from(weights.semantic),
"graph": f64::from(weights.graph),
})
}
fn candidate_json(candidate: &CandidateScore) -> serde_json::Value {
serde_json::json!({
"weights": weights_json(candidate.weights),
"score": candidate.score,
"origin": candidate.origin,
})
}
fn retrieval_tuning_report_json_value(
report: &RetrievalTuningReport,
include_hash: bool,
) -> serde_json::Value {
let labels = &report.labels;
#[allow(clippy::cast_precision_loss)]
let dense_share = if labels.triples.is_empty() {
0.0
} else {
labels.dense_count as f64 / labels.triples.len() as f64
};
let incumbent = report
.evaluation
.as_ref()
.map(|evaluation| candidate_json(&evaluation.incumbent));
let candidates = report.evaluation.as_ref().map(|evaluation| {
evaluation
.candidates
.iter()
.map(candidate_json)
.collect::<Vec<_>>()
});
let winner = report.evaluation.as_ref().and_then(|evaluation| {
evaluation.winner.map(|winner| {
let mut value = candidate_json(&winner);
if let Some(object) = value.as_object_mut() {
object.insert(
"relativeMargin".to_owned(),
evaluation
.relative_margin
.map_or(serde_json::Value::Null, serde_json::Value::from),
);
}
value
})
});
let mut value = serde_json::json!({
"schema": RETRIEVAL_TUNING_REPORT_SCHEMA_V1,
"policyId": RETRIEVAL_TUNING_POLICY_ID,
"dbGeneration": report.db_generation,
"labelSet": {
"triples": labels.triples.len(),
"distinctQueries": labels.distinct_queries,
"hash": labels.label_set_hash,
"denseShare": dense_share,
"denseUnresolvable": labels.dense_unresolvable,
"weakUnreplayable": labels.weak_unreplayable,
"weakUnmatched": labels.weak_unmatched,
},
"abstained": report.abstained,
"abstentionReason": report.abstention_reason,
"incumbent": incumbent,
"candidates": candidates,
"winner": winner,
"promotable": report.promotable,
"diagnostics": report.evaluation.as_ref().map(|evaluation| {
serde_json::json!({
"evaluationHash": evaluation.evaluation_hash,
"graphAxisDegenerate": evaluation.graph_axis_degenerate,
"queriesScored": evaluation.queries_scored,
"queriesWithoutGain": evaluation.queries_without_gain,
"labelsUnmappedSignal": evaluation.labels_unmapped_signal,
})
}),
});
if include_hash {
if let Some(object) = value.as_object_mut() {
object.insert(
"reportHash".to_owned(),
serde_json::Value::from(report.report_hash.clone()),
);
}
}
value
}
#[must_use]
pub fn render_retrieval_tuning_report_json(report: &RetrievalTuningReport) -> String {
retrieval_tuning_report_json_value(report, true).to_string()
}
pub const RETRIEVAL_TUNING_REPORT_FILENAME: &str = "retrieval_tuning_report.json";
pub const PROMOTE_RETRIEVAL_WEIGHTS_AUDIT_ACTION: &str = "shadow.promote_retrieval_weights";
pub const DEMOTE_RETRIEVAL_WEIGHTS_AUDIT_ACTION: &str = "shadow.demote_retrieval_weights";
#[must_use]
pub fn shadow_report_path(workspace_path: &Path) -> std::path::PathBuf {
workspace_path
.join(".ee")
.join("shadow")
.join(RETRIEVAL_TUNING_REPORT_FILENAME)
}
pub fn persist_retrieval_tuning_report(
workspace_path: &Path,
report: &RetrievalTuningReport,
) -> Result<std::path::PathBuf, ShadowTuningError> {
let path = shadow_report_path(workspace_path);
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.map_err(|error| storage_error("create shadow report directory", &error))?;
}
std::fs::write(&path, render_retrieval_tuning_report_json(report))
.map_err(|error| storage_error("write shadow report", &error))?;
Ok(path)
}
#[derive(Debug, Clone, PartialEq)]
pub enum PromoteRefusal {
ReportMissing { path: String },
ReportInvalid { reason: String },
Abstained,
NotPromotable { relative_margin: Option<f64> },
StaleGeneration { report: u64, current: u64 },
NoPriorPromotion,
}
impl PromoteRefusal {
#[must_use]
pub fn message(&self) -> String {
match self {
Self::ReportMissing { path } => {
format!("no persisted tuning report at {path}; run ee shadow run first")
}
Self::ReportInvalid { reason } => format!("persisted tuning report unusable: {reason}"),
Self::Abstained => {
"the persisted report abstained (insufficient outcome evidence); nothing to promote"
.to_owned()
}
Self::NotPromotable { relative_margin } => format!(
"the persisted report is not promotable (relative margin {:?} below the gate or no strict winner)",
relative_margin
),
Self::StaleGeneration { report, current } => format!(
"the persisted report was evaluated at db generation {report} but the workspace is now at {current}; re-run ee shadow run"
),
Self::NoPriorPromotion => {
"no prior promotion audit found for this workspace; nothing to demote".to_owned()
}
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct OverlayChange {
pub applied: bool,
pub prior_config: String,
pub new_config: String,
pub diff: Vec<String>,
pub report_hash: Option<String>,
}
fn workspace_config_path(workspace_path: &Path) -> std::path::PathBuf {
workspace_path.join(".ee").join("config.toml")
}
fn read_config_bytes(workspace_path: &Path) -> Result<(String, bool), ShadowTuningError> {
let path = workspace_config_path(workspace_path);
match std::fs::read_to_string(&path) {
Ok(contents) => Ok((contents, true)),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok((String::new(), false)),
Err(error) => Err(storage_error("read workspace config.toml", &error)),
}
}
#[allow(clippy::type_complexity)]
pub fn promote_retrieval_weights(
workspace_path: &Path,
connection: &DbConnection,
workspace_id: &str,
dry_run: bool,
) -> Result<Result<OverlayChange, PromoteRefusal>, ShadowTuningError> {
let report_path = shadow_report_path(workspace_path);
let raw = match std::fs::read_to_string(&report_path) {
Ok(raw) => raw,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
return Ok(Err(PromoteRefusal::ReportMissing {
path: report_path.display().to_string(),
}));
}
Err(error) => return Err(storage_error("read shadow report", &error)),
};
let report: serde_json::Value = match serde_json::from_str(&raw) {
Ok(value) => value,
Err(error) => {
return Ok(Err(PromoteRefusal::ReportInvalid {
reason: format!("not valid JSON: {error}"),
}));
}
};
if report.get("schema").and_then(serde_json::Value::as_str)
!= Some(RETRIEVAL_TUNING_REPORT_SCHEMA_V1)
{
return Ok(Err(PromoteRefusal::ReportInvalid {
reason: "schema is not ee.shadow.retrieval_tuning_report.v1".to_owned(),
}));
}
if report.get("abstained").and_then(serde_json::Value::as_bool) == Some(true) {
return Ok(Err(PromoteRefusal::Abstained));
}
if report
.get("promotable")
.and_then(serde_json::Value::as_bool)
!= Some(true)
{
let relative_margin = report
.pointer("/winner/relativeMargin")
.and_then(serde_json::Value::as_f64);
return Ok(Err(PromoteRefusal::NotPromotable { relative_margin }));
}
let report_generation = report
.get("dbGeneration")
.and_then(serde_json::Value::as_u64)
.unwrap_or(0);
let current_generation = connection
.get_workspace_generation(workspace_id)
.map_err(|error| storage_error("read workspace generation", &error))?
.unwrap_or(0);
if report_generation != current_generation {
return Ok(Err(PromoteRefusal::StaleGeneration {
report: report_generation,
current: current_generation,
}));
}
let Some(weights) = report.pointer("/winner/weights") else {
return Ok(Err(PromoteRefusal::ReportInvalid {
reason: "promotable report has no winner weights".to_owned(),
}));
};
let (Some(lexical), Some(semantic), Some(graph)) = (
weights.get("lexical").and_then(serde_json::Value::as_f64),
weights.get("semantic").and_then(serde_json::Value::as_f64),
weights.get("graph").and_then(serde_json::Value::as_f64),
) else {
return Ok(Err(PromoteRefusal::ReportInvalid {
reason: "winner weights are not numeric".to_owned(),
}));
};
let report_hash = report
.get("reportHash")
.and_then(serde_json::Value::as_str)
.map(str::to_owned);
let (prior_config, prior_existed) = read_config_bytes(workspace_path)?;
let mut document = prior_config
.parse::<toml_edit::DocumentMut>()
.map_err(|error| storage_error("parse workspace config.toml", &error))?;
let mut diff = Vec::new();
{
let search_item = document.entry("search").or_insert(toml_edit::table());
let Some(search) = search_item.as_table_like_mut() else {
return Ok(Err(PromoteRefusal::ReportInvalid {
reason: "[search] in config.toml is not a table".to_owned(),
}));
};
for (key, value) in [
("lexical_weight", lexical),
("semantic_weight", semantic),
("graph_weight", graph),
] {
let prior = search
.get(key)
.and_then(toml_edit::Item::as_value)
.map(std::string::ToString::to_string);
diff.push(format!(
"search.{key}: {} -> {value}",
prior.as_deref().unwrap_or("(unset)")
));
search.insert(key, toml_edit::value(value));
}
}
let new_config = document.to_string();
let change = OverlayChange {
applied: !dry_run,
prior_config: prior_config.clone(),
new_config: new_config.clone(),
diff,
report_hash: report_hash.clone(),
};
if dry_run {
return Ok(Ok(change));
}
std::fs::write(workspace_config_path(workspace_path), &new_config)
.map_err(|error| storage_error("write workspace config.toml", &error))?;
let details = serde_json::json!({
"schema": "ee.shadow.retrieval_weights_promotion.v1",
"policyId": RETRIEVAL_TUNING_POLICY_ID,
"reportHash": report_hash,
"priorExisted": prior_existed,
"priorConfigToml": prior_config,
"newValues": { "lexical": lexical, "semantic": semantic, "graph": graph },
})
.to_string();
connection
.insert_audit(
&crate::db::generate_audit_id(),
&crate::db::CreateAuditInput {
workspace_id: Some(workspace_id.to_owned()),
actor: None,
action: PROMOTE_RETRIEVAL_WEIGHTS_AUDIT_ACTION.to_owned(),
target_type: Some("workspace".to_owned()),
target_id: Some(workspace_id.to_owned()),
details: Some(details),
},
)
.map_err(|error| storage_error("record promotion audit", &error))?;
Ok(Ok(change))
}
#[allow(clippy::type_complexity)]
pub fn demote_retrieval_weights(
workspace_path: &Path,
connection: &DbConnection,
workspace_id: &str,
dry_run: bool,
) -> Result<Result<OverlayChange, PromoteRefusal>, ShadowTuningError> {
let audits = connection
.list_audit_by_action(PROMOTE_RETRIEVAL_WEIGHTS_AUDIT_ACTION, None)
.map_err(|error| storage_error("list promotion audits", &error))?;
let Some(promotion) = audits
.iter()
.find(|entry| entry.workspace_id.as_deref() == Some(workspace_id))
else {
return Ok(Err(PromoteRefusal::NoPriorPromotion));
};
let details: serde_json::Value = promotion
.details
.as_deref()
.and_then(|raw| serde_json::from_str(raw).ok())
.unwrap_or(serde_json::Value::Null);
let Some(prior_config) = details
.get("priorConfigToml")
.and_then(serde_json::Value::as_str)
else {
return Ok(Err(PromoteRefusal::ReportInvalid {
reason: "promotion audit carries no priorConfigToml".to_owned(),
}));
};
let (current_config, _) = read_config_bytes(workspace_path)?;
let change = OverlayChange {
applied: !dry_run,
prior_config: current_config,
new_config: prior_config.to_owned(),
diff: vec![format!(
"config.toml restored to pre-promotion bytes ({} bytes)",
prior_config.len()
)],
report_hash: details
.get("reportHash")
.and_then(serde_json::Value::as_str)
.map(str::to_owned),
};
if dry_run {
return Ok(Ok(change));
}
std::fs::write(workspace_config_path(workspace_path), prior_config)
.map_err(|error| storage_error("restore workspace config.toml", &error))?;
let demote_details = serde_json::json!({
"schema": "ee.shadow.retrieval_weights_demotion.v1",
"policyId": RETRIEVAL_TUNING_POLICY_ID,
"restoredFromAudit": promotion.id,
"restoredBytes": prior_config.len(),
})
.to_string();
connection
.insert_audit(
&crate::db::generate_audit_id(),
&crate::db::CreateAuditInput {
workspace_id: Some(workspace_id.to_owned()),
actor: None,
action: DEMOTE_RETRIEVAL_WEIGHTS_AUDIT_ACTION.to_owned(),
target_type: Some("workspace".to_owned()),
target_id: Some(workspace_id.to_owned()),
details: Some(demote_details),
},
)
.map_err(|error| storage_error("record demotion audit", &error))?;
Ok(Ok(change))
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used)]
use super::*;
use crate::db::{
CreateAuditInput, CreateFeedbackEventInput, CreateFeedbackQuarantineInput,
CreatePackRecordInput, CreateWorkspaceInput,
};
use asupersync::CancelReason;
use chrono::TimeZone;
type TestResult = Result<(), String>;
const WORKSPACE: &str = "wsp_00000000000000000000000901";
fn ts(minute: i64) -> DateTime<Utc> {
Utc.with_ymd_and_hms(2026, 8, 1, 12, 0, 0).unwrap() + Duration::minutes(minute)
}
fn event(
id: &str,
memory_id: &str,
created_at: DateTime<Utc>,
evidence_json: Option<String>,
) -> StoredFeedbackEvent {
StoredFeedbackEvent {
id: id.to_owned(),
workspace_id: WORKSPACE.to_owned(),
target_type: "memory".to_owned(),
target_id: memory_id.to_owned(),
signal: "helpful".to_owned(),
weight: 1.0,
source_type: "outcome_observed".to_owned(),
source_id: None,
reason: None,
evidence_json,
session_id: None,
applied_at: None,
created_at: created_at.to_rfc3339(),
}
}
fn returned_mem_audit(
id: &str,
memory_id: &str,
timestamp: DateTime<Utc>,
query_hash: Option<&str>,
) -> StoredAuditEntry {
let details = query_hash.map(|hash| {
serde_json::json!({
"queryHash": hash,
"rank": 1,
"score": 0.9,
"source": "semantic",
})
.to_string()
});
StoredAuditEntry {
id: id.to_owned(),
workspace_id: Some(WORKSPACE.to_owned()),
timestamp: timestamp.to_rfc3339(),
actor: None,
action: audit_actions::SEARCH_RETURNED_MEM.to_owned(),
target_type: Some("memory".to_owned()),
target_id: Some(memory_id.to_owned()),
details,
surface: "search".to_owned(),
mutation_kind: audit_actions::SEARCH_RETURNED_MEM.to_owned(),
before_hash: None,
after_hash: None,
prev_row_hash: None,
this_row_hash: None,
}
}
fn pack_item_evidence(pack_id: &str) -> String {
serde_json::json!({
"schema": PACK_ITEM_EVIDENCE_SCHEMA_V1,
"packId": pack_id,
"itemRank": 2,
})
.to_string()
}
fn join(
events: &[StoredFeedbackEvent],
audits: &[StoredAuditEntry],
pack_queries: &BTreeMap<String, String>,
query_text_by_hash: &BTreeMap<String, String>,
as_of: DateTime<Utc>,
) -> Result<LabelExtractionReport, String> {
let cx = Cx::for_testing();
join_labeled_triples(
&cx,
events,
audits,
pack_queries,
query_text_by_hash,
&LabelExtractionConfig::default(),
as_of,
)
.map_err(|error| error.to_string())
}
fn approx_eq(actual: f64, expected: f64) -> bool {
(actual - expected).abs() < 1e-9
}
#[test]
fn dense_pack_item_event_yields_weight_one_triple() -> TestResult {
let created = ts(0);
let events = [event(
"fev-1",
"mem-1",
created,
Some(pack_item_evidence("pack-1")),
)];
let pack_queries = BTreeMap::from([(
"pack-1".to_owned(),
"fix failing release workflow".to_owned(),
)]);
let report = join(&events, &[], &pack_queries, &BTreeMap::new(), created)?;
if report.triples.len() != 1 {
return Err(format!("expected one dense triple, got {report:?}"));
}
let triple = &report.triples[0];
if triple.query != "fix failing release workflow"
|| triple.memory_id != "mem-1"
|| triple.source != LabelSource::PackItemOutcome
|| triple.pack_record_id.as_deref() != Some("pack-1")
|| triple.audit_row_id.is_some()
{
return Err(format!("dense triple fields wrong: {triple:?}"));
}
if !approx_eq(triple.base_weight, 1.0) || !approx_eq(triple.weight, 1.0) {
return Err(format!("dense weight wrong: {triple:?}"));
}
if report.dense_count != 1 || report.weak_count != 0 || report.distinct_queries != 1 {
return Err(format!("report counts wrong: {report:?}"));
}
Ok(())
}
#[test]
fn dense_linkage_with_missing_pack_record_is_counted_not_guessed() -> TestResult {
let created = ts(0);
let events = [event(
"fev-1",
"mem-1",
created,
Some(pack_item_evidence("pack-gone")),
)];
let query = "weak query";
let hash = audit_query_hash(query);
let audits = [returned_mem_audit("aud-1", "mem-1", ts(-5), Some(&hash))];
let query_text_by_hash = BTreeMap::from([(hash.clone(), query.to_owned())]);
let report = join(
&events,
&audits,
&BTreeMap::new(),
&query_text_by_hash,
created,
)?;
if !report.triples.is_empty()
|| report.dense_unresolvable != 1
|| report.weak_unmatched != 0
|| report.weak_unreplayable != 0
{
return Err(format!("missing pack record mishandled: {report:?}"));
}
Ok(())
}
#[test]
fn weak_event_joins_nearest_preceding_returned_mem_within_window() -> TestResult {
let created = ts(0);
let near_query = "near query";
let far_query = "far query";
let near_hash = audit_query_hash(near_query);
let far_hash = audit_query_hash(far_query);
let events = [event("fev-1", "mem-1", created, None)];
let audits = [
returned_mem_audit("aud-far", "mem-1", ts(-25), Some(&far_hash)),
returned_mem_audit("aud-near", "mem-1", ts(-10), Some(&near_hash)),
];
let query_text_by_hash = BTreeMap::from([
(near_hash.clone(), near_query.to_owned()),
(far_hash.clone(), far_query.to_owned()),
]);
let report = join(
&events,
&audits,
&BTreeMap::new(),
&query_text_by_hash,
created,
)?;
if report.triples.len() != 1 {
return Err(format!("expected one weak triple, got {report:?}"));
}
let triple = &report.triples[0];
if triple.query != near_query
|| triple.source != LabelSource::SearchWindowAssociation
|| triple.audit_row_id.as_deref() != Some("aud-near")
|| !approx_eq(triple.base_weight, 0.5)
{
return Err(format!("weak triple fields wrong: {triple:?}"));
}
Ok(())
}
#[test]
fn weak_window_edge_is_inclusive_and_beyond_is_unmatched() -> TestResult {
let created = ts(0);
let query = "edge query";
let hash = audit_query_hash(query);
let query_text_by_hash = BTreeMap::from([(hash.clone(), query.to_owned())]);
let events = [event("fev-1", "mem-1", created, None)];
let at_edge = [returned_mem_audit("aud-1", "mem-1", ts(-30), Some(&hash))];
let report = join(
&events,
&at_edge,
&BTreeMap::new(),
&query_text_by_hash,
created,
)?;
if report.weak_count != 1 || report.weak_unmatched != 0 {
return Err(format!("window edge must be inclusive: {report:?}"));
}
let beyond = [returned_mem_audit(
"aud-1",
"mem-1",
ts(-30) - Duration::seconds(1),
Some(&hash),
)];
let report = join(
&events,
&beyond,
&BTreeMap::new(),
&query_text_by_hash,
created,
)?;
if report.weak_count != 0 || report.weak_unmatched != 1 {
return Err(format!("beyond-window audit must not label: {report:?}"));
}
Ok(())
}
#[test]
fn search_after_outcome_never_labels() -> TestResult {
let created = ts(0);
let query = "later query";
let hash = audit_query_hash(query);
let events = [event("fev-1", "mem-1", created, None)];
let audits = [returned_mem_audit("aud-1", "mem-1", ts(5), Some(&hash))];
let query_text_by_hash = BTreeMap::from([(hash.clone(), query.to_owned())]);
let report = join(
&events,
&audits,
&BTreeMap::new(),
&query_text_by_hash,
created,
)?;
if report.weak_count != 0 || report.weak_unmatched != 1 {
return Err(format!(
"an audit row after the outcome cannot have caused it: {report:?}"
));
}
Ok(())
}
#[test]
fn weak_hash_miss_is_counted_unreplayable_not_guessed() -> TestResult {
let created = ts(0);
let events = [event("fev-1", "mem-1", created, None)];
let audits = [returned_mem_audit(
"aud-1",
"mem-1",
ts(-5),
Some("blake3:0000000000000000"),
)];
let report = join(
&events,
&audits,
&BTreeMap::new(),
&BTreeMap::new(),
created,
)?;
if !report.triples.is_empty() || report.weak_unreplayable != 1 || report.weak_unmatched != 0
{
return Err(format!("hash miss must count unreplayable: {report:?}"));
}
Ok(())
}
#[test]
fn freshness_discount_halves_weight_at_ninety_days() -> TestResult {
let created = ts(0);
let as_of = created + Duration::days(90);
let events = [event(
"fev-1",
"mem-1",
created,
Some(pack_item_evidence("pack-1")),
)];
let pack_queries = BTreeMap::from([("pack-1".to_owned(), "aged query".to_owned())]);
let report = join(&events, &[], &pack_queries, &BTreeMap::new(), as_of)?;
let triple = &report.triples[0];
if !approx_eq(triple.age_days, 90.0) || !approx_eq(triple.weight, 0.5) {
return Err(format!("90-day freshness discount wrong: {triple:?}"));
}
Ok(())
}
#[test]
fn non_memory_targets_are_ignored() -> TestResult {
let created = ts(0);
let mut pack_event = event(
"fev-1",
"pack-1",
created,
Some(pack_item_evidence("pack-1")),
);
pack_event.target_type = "pack".to_owned();
let pack_queries = BTreeMap::from([("pack-1".to_owned(), "some query".to_owned())]);
let report = join(&[pack_event], &[], &pack_queries, &BTreeMap::new(), created)?;
if report.memory_event_count != 0 || !report.triples.is_empty() {
return Err(format!("non-memory targets must be ignored: {report:?}"));
}
Ok(())
}
#[test]
fn triples_are_sorted_and_hash_is_order_independent() -> TestResult {
let created = ts(0);
let pack_queries = BTreeMap::from([
("pack-a".to_owned(), "alpha query".to_owned()),
("pack-b".to_owned(), "beta query".to_owned()),
]);
let forward = [
event(
"fev-1",
"mem-1",
created,
Some(pack_item_evidence("pack-a")),
),
event(
"fev-2",
"mem-2",
created,
Some(pack_item_evidence("pack-b")),
),
];
let reversed = [forward[1].clone(), forward[0].clone()];
let report_a = join(&forward, &[], &pack_queries, &BTreeMap::new(), created)?;
let report_b = join(&reversed, &[], &pack_queries, &BTreeMap::new(), created)?;
if report_a != report_b {
return Err("label extraction must be input-order independent".to_owned());
}
let queries: Vec<&str> = report_a
.triples
.iter()
.map(|triple| triple.query.as_str())
.collect();
if queries != ["alpha query", "beta query"] {
return Err(format!("triples must sort by query: {queries:?}"));
}
if !report_a.label_set_hash.starts_with("blake3:") {
return Err(format!(
"hash must be prefixed: {}",
report_a.label_set_hash
));
}
let smaller = join(&forward[..1], &[], &pack_queries, &BTreeMap::new(), created)?;
if smaller.label_set_hash == report_a.label_set_hash {
return Err("different label sets must not share a fingerprint".to_owned());
}
Ok(())
}
#[test]
fn cancelled_cx_aborts_extraction() -> TestResult {
let cx = Cx::for_testing();
cx.set_cancel_reason(CancelReason::user("shadow tuning cancellation test"));
let created = ts(0);
let events = [event("fev-1", "mem-1", created, None)];
let outcome = join_labeled_triples(
&cx,
&events,
&[],
&BTreeMap::new(),
&BTreeMap::new(),
&LabelExtractionConfig::default(),
created,
);
match outcome {
Err(ShadowTuningError::Cancelled(_)) => Ok(()),
other => Err(format!("cancelled Cx must abort extraction: {other:?}")),
}
}
#[test]
fn extract_from_database_joins_dense_and_excludes_quarantine() -> TestResult {
let tempdir = tempfile::tempdir_in("/tmp").map_err(|error| error.to_string())?;
let database_path = tempdir.path().join("ee.db");
let connection =
DbConnection::open_file(&database_path).map_err(|error| error.to_string())?;
connection.migrate().map_err(|error| error.to_string())?;
connection
.insert_workspace(
WORKSPACE,
&CreateWorkspaceInput {
path: tempdir.path().display().to_string(),
name: None,
},
)
.map_err(|error| error.to_string())?;
connection
.insert_pack_record(
"pack_00000000000000000000001001",
&CreatePackRecordInput {
workspace_id: WORKSPACE.to_owned(),
query: "prepare release".to_owned(),
profile: "balanced".to_owned(),
max_tokens: 4000,
used_tokens: 0,
item_count: 0,
omitted_count: 0,
pack_hash:
"blake3:0000000000000000000000000000000000000000000000000000000000000000"
.to_owned(),
degraded_json: None,
created_by: None,
},
&[],
&[],
)
.map_err(|error| error.to_string())?;
connection
.insert_feedback_event(
"fb_00000000000000000000001001",
&CreateFeedbackEventInput {
workspace_id: WORKSPACE.to_owned(),
target_type: "memory".to_owned(),
target_id: "mem_00000000000000000000001001".to_owned(),
signal: "helpful".to_owned(),
weight: 1.0,
source_type: "outcome_observed".to_owned(),
source_id: None,
reason: None,
evidence_json: Some(pack_item_evidence("pack_00000000000000000000001001")),
session_id: None,
},
)
.map_err(|error| error.to_string())?;
connection
.insert_feedback_quarantine(
"fq_00000000000000000000001001",
&CreateFeedbackQuarantineInput {
workspace_id: WORKSPACE.to_owned(),
source_id: "poisoned-source".to_owned(),
target_type: "memory".to_owned(),
target_id: "mem_00000000000000000000001001".to_owned(),
signal: "harmful".to_owned(),
weight: 1.0,
source_type: "agent_inference".to_owned(),
proposed_event_id: None,
recorded_at: Utc::now().to_rfc3339(),
reason: "sprt_quarantine".to_owned(),
event_reason: None,
evidence_json: None,
session_id: None,
raw_event_hash: "blake3:quarantine-fixture".to_owned(),
},
)
.map_err(|error| error.to_string())?;
let cx = Cx::for_testing();
let report = extract_labeled_triples(
&cx,
&connection,
WORKSPACE,
&LabelExtractionConfig::default(),
Utc::now() + Duration::minutes(1),
)
.map_err(|error| error.to_string())?;
if report.triples.len() != 1 || report.dense_count != 1 {
return Err(format!("expected exactly the dense triple: {report:?}"));
}
let triple = &report.triples[0];
if triple.query != "prepare release"
|| triple.feedback_event_id != "fb_00000000000000000000001001"
|| triple.signal != "helpful"
{
return Err(format!("db dense triple wrong: {triple:?}"));
}
if report.memory_event_count != 1 {
return Err(format!(
"quarantined feedback must be invisible to the join: {report:?}"
));
}
Ok(())
}
#[test]
fn extract_from_database_resolves_weak_query_via_hash_join() -> TestResult {
let tempdir = tempfile::tempdir_in("/tmp").map_err(|error| error.to_string())?;
let database_path = tempdir.path().join("ee.db");
let connection =
DbConnection::open_file(&database_path).map_err(|error| error.to_string())?;
connection.migrate().map_err(|error| error.to_string())?;
connection
.insert_workspace(
WORKSPACE,
&CreateWorkspaceInput {
path: tempdir.path().display().to_string(),
name: None,
},
)
.map_err(|error| error.to_string())?;
let query = "hunt flaky mesh test";
connection
.insert_pack_record(
"pack_00000000000000000000001002",
&CreatePackRecordInput {
workspace_id: WORKSPACE.to_owned(),
query: query.to_owned(),
profile: "balanced".to_owned(),
max_tokens: 4000,
used_tokens: 0,
item_count: 0,
omitted_count: 0,
pack_hash:
"blake3:1111111111111111111111111111111111111111111111111111111111111111"
.to_owned(),
degraded_json: None,
created_by: None,
},
&[],
&[],
)
.map_err(|error| error.to_string())?;
connection
.insert_audit(
"audit_00000000000000000000001001",
&CreateAuditInput {
workspace_id: Some(WORKSPACE.to_owned()),
actor: None,
action: audit_actions::SEARCH_RETURNED_MEM.to_owned(),
target_type: Some("memory".to_owned()),
target_id: Some("mem_00000000000000000000001002".to_owned()),
details: Some(
serde_json::json!({
"queryHash": audit_query_hash(query),
"rank": 1,
"score": 0.8,
"source": "lexical",
})
.to_string(),
),
},
)
.map_err(|error| error.to_string())?;
connection
.insert_feedback_event(
"fb_00000000000000000000001002",
&CreateFeedbackEventInput {
workspace_id: WORKSPACE.to_owned(),
target_type: "memory".to_owned(),
target_id: "mem_00000000000000000000001002".to_owned(),
signal: "helpful".to_owned(),
weight: 1.0,
source_type: "outcome_observed".to_owned(),
source_id: None,
reason: None,
evidence_json: None,
session_id: None,
},
)
.map_err(|error| error.to_string())?;
let cx = Cx::for_testing();
let report = extract_labeled_triples(
&cx,
&connection,
WORKSPACE,
&LabelExtractionConfig::default(),
Utc::now() + Duration::minutes(1),
)
.map_err(|error| error.to_string())?;
if report.triples.len() != 1 || report.weak_count != 1 {
return Err(format!("expected exactly the weak triple: {report:?}"));
}
let triple = &report.triples[0];
if triple.query != query
|| triple.source != LabelSource::SearchWindowAssociation
|| triple.audit_row_id.as_deref() != Some("audit_00000000000000000000001001")
|| !approx_eq(triple.base_weight, 0.5)
{
return Err(format!("db weak triple wrong: {triple:?}"));
}
if report.weak_unreplayable != 0 || report.weak_unmatched != 0 {
return Err(format!("weak denominators must be clean: {report:?}"));
}
Ok(())
}
use crate::core::search::ScoreSource;
fn hybrid_hit(
doc_id: &str,
raw_score: f32,
lexical: Option<f32>,
semantic: Option<f32>,
) -> SearchHit {
SearchHit {
doc_id: doc_id.to_owned(),
score: raw_score,
source: ScoreSource::Hybrid,
fast_score: None,
quality_score: semantic,
lexical_score: lexical,
rerank_score: None,
metadata: None,
explanation: None,
}
}
fn triple(query: &str, memory_id: &str, signal: &str, weight: f64) -> LabeledTriple {
LabeledTriple {
query: query.to_owned(),
memory_id: memory_id.to_owned(),
signal: signal.to_owned(),
base_weight: weight,
weight,
age_days: 0.0,
source: LabelSource::PackItemOutcome,
feedback_event_id: format!("fev-{memory_id}"),
pack_record_id: Some("pack-1".to_owned()),
audit_row_id: None,
}
}
fn incumbent() -> TuningWeights {
TuningWeights::compiled_defaults()
}
#[test]
fn evaluator_hand_computed_metric_and_rank_flip_winner() -> TestResult {
let replays = [QueryReplay {
query: "q1".to_owned(),
hits: vec![
hybrid_hit("mem-a", 0.0255, Some(1.0), None),
hybrid_hit("mem-b", 0.020, None, Some(1.0)),
],
}];
let labels = [triple("q1", "mem-b", "helpful", 1.0)];
let cx = Cx::for_testing();
let evaluation = evaluate_fusion_candidates(&cx, &replays, &labels, incumbent())
.map_err(|error| error.to_string())?;
let expected_incumbent = 1.0 / 3.0_f64.log2();
if !approx_eq(evaluation.incumbent.score, expected_incumbent) {
return Err(format!(
"incumbent metric must be 1/log2(3): {evaluation:?}"
));
}
let Some(winner) = evaluation.winner else {
return Err(format!("lexical -0.10 vector must win: {evaluation:?}"));
};
if !approx_eq(winner.score, 1.0)
|| (f64::from(winner.weights.lexical) - 0.35).abs() > 1e-6
|| (f64::from(winner.weights.semantic) - 0.45).abs() > 1e-6
{
return Err(format!(
"winner must be lexical -0.10 at score 1.0: {winner:?}"
));
}
let Some(margin) = evaluation.relative_margin else {
return Err("winner must carry a relative margin".to_owned());
};
if margin <= 0.0 {
return Err(format!("margin must be positive: {margin}"));
}
Ok(())
}
#[test]
fn evaluator_is_deterministic_across_runs() -> TestResult {
let replays = [QueryReplay {
query: "q1".to_owned(),
hits: vec![
hybrid_hit("mem-a", 0.021, Some(1.0), None),
hybrid_hit("mem-b", 0.020, None, Some(1.0)),
hybrid_hit("mem-c", 0.015, Some(0.4), Some(0.4)),
],
}];
let labels = [
triple("q1", "mem-a", "helpful", 0.8),
triple("q1", "mem-b", "harmful", 0.5),
triple("q1", "mem-c", "confirmation", 1.0),
];
let cx = Cx::for_testing();
let first = evaluate_fusion_candidates(&cx, &replays, &labels, incumbent())
.map_err(|error| error.to_string())?;
let second = evaluate_fusion_candidates(&cx, &replays, &labels, incumbent())
.map_err(|error| error.to_string())?;
if first != second {
return Err("evaluation must be byte-identical across runs".to_owned());
}
if !first.evaluation_hash.starts_with("blake3:") {
return Err(format!("hash must be prefixed: {}", first.evaluation_hash));
}
let third = evaluate_fusion_candidates(&cx, &replays, &labels[..1], incumbent())
.map_err(|error| error.to_string())?;
if third.evaluation_hash == first.evaluation_hash {
return Err("different label sets must not share an evaluation hash".to_owned());
}
Ok(())
}
#[test]
fn candidate_grid_respects_clamps_and_dedups() -> TestResult {
let near_edge = TuningWeights {
lexical: 0.65,
semantic: 0.25,
graph: 0.25,
};
let vectors = enumerate_grid(near_edge);
for vector in &vectors {
if vector.lexical < FUSION_LEXICAL_CLAMP.0
|| vector.lexical > FUSION_LEXICAL_CLAMP.1
|| vector.semantic < FUSION_SEMANTIC_CLAMP.0
|| vector.semantic > FUSION_SEMANTIC_CLAMP.1
|| vector.graph < FUSION_GRAPH_CLAMP.0
|| vector.graph > FUSION_GRAPH_CLAMP.1
{
return Err(format!("clamp violated: {vector:?}"));
}
}
let mut keys = BTreeSet::new();
for vector in &vectors {
if !keys.insert(vector.key()) {
return Err(format!("duplicate candidate survived dedup: {vector:?}"));
}
}
if vectors != enumerate_grid(near_edge) {
return Err("grid enumeration must be deterministic".to_owned());
}
Ok(())
}
#[test]
fn unmapped_signals_and_zero_gain_queries_are_counted_not_guessed() -> TestResult {
let replays = [QueryReplay {
query: "q1".to_owned(),
hits: vec![hybrid_hit("mem-a", 0.02, Some(1.0), None)],
}];
let labels = [
triple("q1", "mem-a", "stale", 1.0),
triple("q2", "mem-a", "helpful", 0.0),
];
let cx = Cx::for_testing();
let evaluation = evaluate_fusion_candidates(&cx, &replays, &labels, incumbent())
.map_err(|error| error.to_string())?;
if evaluation.labels_unmapped_signal != 1
|| evaluation.queries_without_gain != 1
|| evaluation.queries_scored != 0
{
return Err(format!("honest counters wrong: {evaluation:?}"));
}
if !approx_eq(evaluation.incumbent.score, 0.0) || evaluation.winner.is_some() {
return Err(format!(
"no usable labels must mean zero scores and no winner: {evaluation:?}"
));
}
Ok(())
}
#[test]
fn reranked_hits_outrank_fusion_hits_under_every_candidate() -> TestResult {
let reranked = SearchHit {
doc_id: "mem-reranked".to_owned(),
score: 0.9,
source: ScoreSource::Reranked,
fast_score: None,
quality_score: None,
lexical_score: None,
rerank_score: Some(0.9),
metadata: None,
explanation: None,
};
let hits = vec![
hybrid_hit("mem-a", 0.021, Some(1.0), None),
reranked,
hybrid_hit("mem-b", 0.020, None, Some(1.0)),
];
for weights in [
TuningWeights {
lexical: 0.7,
semantic: 0.2,
graph: 0.0,
},
TuningWeights {
lexical: 0.2,
semantic: 0.7,
graph: 0.3,
},
] {
let ranks = ranked_pool(&hits, weights.to_fusion());
if ranks.get("mem-reranked") != Some(&1) {
return Err(format!(
"reranked hit must stay rank 1 under {weights:?}: {ranks:?}"
));
}
}
Ok(())
}
#[test]
fn cancelled_cx_aborts_evaluation_sweep() -> TestResult {
let cx = Cx::for_testing();
cx.set_cancel_reason(CancelReason::user("shadow tuning sweep cancellation test"));
let replays = [QueryReplay {
query: "q1".to_owned(),
hits: vec![hybrid_hit("mem-a", 0.02, Some(1.0), None)],
}];
let labels = [triple("q1", "mem-a", "helpful", 1.0)];
match evaluate_fusion_candidates(&cx, &replays, &labels, incumbent()) {
Err(ShadowTuningError::Cancelled(_)) => Ok(()),
other => Err(format!("cancelled Cx must abort the sweep: {other:?}")),
}
}
fn flip_fixture_label_report() -> Result<LabelExtractionReport, String> {
let created = ts(0);
let events = [event(
"fev-1",
"mem-b",
created,
Some(pack_item_evidence("pack-1")),
)];
let pack_queries = BTreeMap::from([("pack-1".to_owned(), "q1".to_owned())]);
join(&events, &[], &pack_queries, &BTreeMap::new(), created)
}
fn flip_fixture_replays() -> [QueryReplay; 1] {
[QueryReplay {
query: "q1".to_owned(),
hits: vec![
hybrid_hit("mem-a", 0.0255, Some(1.0), None),
hybrid_hit("mem-b", 0.020, None, Some(1.0)),
],
}]
}
#[test]
fn evidence_gate_abstains_below_thresholds() -> TestResult {
let labels = join(&[], &[], &BTreeMap::new(), &BTreeMap::new(), ts(0))?;
let report = assemble_retrieval_tuning_report(
labels,
None,
7,
&RetrievalTuningGateConfig::default(),
)
.map_err(|error| error.to_string())?;
if !report.abstained
|| report.abstention_reason != Some(INSUFFICIENT_OUTCOME_EVIDENCE_CODE)
|| report.promotable
|| report.evaluation.is_some()
{
return Err(format!("abstention shape wrong: {report:?}"));
}
let rendered: serde_json::Value =
serde_json::from_str(&render_retrieval_tuning_report_json(&report))
.map_err(|error| error.to_string())?;
if rendered["schema"] != RETRIEVAL_TUNING_REPORT_SCHEMA_V1
|| rendered["abstained"] != true
|| !rendered["winner"].is_null()
|| rendered["promotable"] != false
|| rendered["dbGeneration"] != 7
{
return Err(format!("abstention rendering wrong: {rendered}"));
}
Ok(())
}
#[test]
fn gate_pass_produces_promotable_report() -> TestResult {
let cx = Cx::for_testing();
let labels = flip_fixture_label_report()?;
let evaluation =
evaluate_fusion_candidates(&cx, &flip_fixture_replays(), &labels.triples, incumbent())
.map_err(|error| error.to_string())?;
let gate = RetrievalTuningGateConfig {
min_triples: 1,
min_queries: 1,
promote_margin: 0.03,
};
let report = assemble_retrieval_tuning_report(labels, Some(evaluation), 3, &gate)
.map_err(|error| error.to_string())?;
if report.abstained || !report.promotable || report.abstention_reason.is_some() {
return Err(format!("gate-pass shape wrong: {report:?}"));
}
let rendered: serde_json::Value =
serde_json::from_str(&render_retrieval_tuning_report_json(&report))
.map_err(|error| error.to_string())?;
if rendered["policyId"] != RETRIEVAL_TUNING_POLICY_ID
|| rendered["promotable"] != true
|| rendered["labelSet"]["triples"] != 1
{
return Err(format!("gate-pass rendering wrong: {rendered}"));
}
let margin = rendered["winner"]["relativeMargin"]
.as_f64()
.ok_or("winner must carry relativeMargin")?;
if margin <= 0.03 {
return Err(format!("relative margin must clear the gate: {margin}"));
}
if rendered["reportHash"].as_str().map(str::to_owned) != Some(report.report_hash.clone()) {
return Err("rendered reportHash must match the struct".to_owned());
}
Ok(())
}
#[test]
fn report_hash_is_deterministic_and_content_bound() -> TestResult {
let labels_a = join(&[], &[], &BTreeMap::new(), &BTreeMap::new(), ts(0))?;
let labels_b = join(&[], &[], &BTreeMap::new(), &BTreeMap::new(), ts(0))?;
let gate = RetrievalTuningGateConfig::default();
let first = assemble_retrieval_tuning_report(labels_a, None, 1, &gate)
.map_err(|error| error.to_string())?;
let second = assemble_retrieval_tuning_report(labels_b.clone(), None, 1, &gate)
.map_err(|error| error.to_string())?;
if first.report_hash != second.report_hash {
return Err("identical reports must share a hash".to_owned());
}
let generation_shifted = assemble_retrieval_tuning_report(labels_b, None, 2, &gate)
.map_err(|error| error.to_string())?;
if generation_shifted.report_hash == first.report_hash {
return Err("dbGeneration must be hash-bound".to_owned());
}
Ok(())
}
fn promotable_report_fixture() -> Result<RetrievalTuningReport, String> {
let labels = flip_fixture_label_report()?;
let cx = Cx::for_testing();
let evaluation =
evaluate_fusion_candidates(&cx, &flip_fixture_replays(), &labels.triples, incumbent())
.map_err(|error| error.to_string())?;
let gate = RetrievalTuningGateConfig {
min_triples: 1,
min_queries: 1,
promote_margin: 0.03,
};
assemble_retrieval_tuning_report(labels, Some(evaluation), 0, &gate)
.map_err(|error| error.to_string())
}
#[test]
fn promote_applies_overlay_and_demote_restores_bytes() -> TestResult {
let tempdir = tempfile::tempdir_in("/tmp").map_err(|error| error.to_string())?;
let workspace = tempdir.path();
let store_dir = workspace.join(".ee");
std::fs::create_dir_all(&store_dir).map_err(|error| error.to_string())?;
let database_path = store_dir.join("ee.db");
let connection =
DbConnection::open_file(&database_path).map_err(|error| error.to_string())?;
connection.migrate().map_err(|error| error.to_string())?;
connection
.insert_workspace(
WORKSPACE,
&CreateWorkspaceInput {
path: workspace.display().to_string(),
name: None,
},
)
.map_err(|error| error.to_string())?;
let report = promotable_report_fixture()?;
persist_retrieval_tuning_report(workspace, &report).map_err(|error| error.to_string())?;
let plan = promote_retrieval_weights(workspace, &connection, WORKSPACE, true)
.map_err(|error| error.to_string())?
.map_err(|refusal| format!("unexpected refusal: {refusal:?}"))?;
if plan.applied || workspace.join(".ee").join("config.toml").exists() {
return Err(format!("dry-run must write nothing: {plan:?}"));
}
let change = promote_retrieval_weights(workspace, &connection, WORKSPACE, false)
.map_err(|error| error.to_string())?
.map_err(|refusal| format!("unexpected refusal: {refusal:?}"))?;
let written = std::fs::read_to_string(workspace.join(".ee").join("config.toml"))
.map_err(|error| error.to_string())?;
if !written.contains("lexical_weight") || !written.contains("[search]") {
return Err(format!("overlay not written: {written}"));
}
if !change.prior_config.is_empty() {
return Err("prior config must be empty for a fresh workspace".to_owned());
}
let audits = connection
.list_audit_by_action(PROMOTE_RETRIEVAL_WEIGHTS_AUDIT_ACTION, None)
.map_err(|error| error.to_string())?;
if audits.is_empty() {
return Err("promotion must be audited".to_owned());
}
let demotion = demote_retrieval_weights(workspace, &connection, WORKSPACE, false)
.map_err(|error| error.to_string())?
.map_err(|refusal| format!("unexpected refusal: {refusal:?}"))?;
let restored = std::fs::read_to_string(workspace.join(".ee").join("config.toml"))
.map_err(|error| error.to_string())?;
if !restored.is_empty() || !demotion.new_config.is_empty() {
return Err(format!(
"demote must restore the exact prior bytes: {restored:?}"
));
}
Ok(())
}
#[test]
fn promote_refuses_missing_and_stale_reports() -> TestResult {
let tempdir = tempfile::tempdir_in("/tmp").map_err(|error| error.to_string())?;
let workspace = tempdir.path();
std::fs::create_dir_all(workspace.join(".ee")).map_err(|error| error.to_string())?;
let connection = DbConnection::open_file(workspace.join(".ee").join("ee.db"))
.map_err(|error| error.to_string())?;
connection.migrate().map_err(|error| error.to_string())?;
match promote_retrieval_weights(workspace, &connection, "ws-missing", false)
.map_err(|error| error.to_string())?
{
Err(PromoteRefusal::ReportMissing { .. }) => {}
other => return Err(format!("missing report must refuse: {other:?}")),
}
let labels = flip_fixture_label_report()?;
let cx = Cx::for_testing();
let evaluation =
evaluate_fusion_candidates(&cx, &flip_fixture_replays(), &labels.triples, incumbent())
.map_err(|error| error.to_string())?;
let gate = RetrievalTuningGateConfig {
min_triples: 1,
min_queries: 1,
promote_margin: 0.03,
};
let stale = assemble_retrieval_tuning_report(labels, Some(evaluation), 5, &gate)
.map_err(|error| error.to_string())?;
persist_retrieval_tuning_report(workspace, &stale).map_err(|error| error.to_string())?;
match promote_retrieval_weights(workspace, &connection, "ws-missing", false)
.map_err(|error| error.to_string())?
{
Err(PromoteRefusal::StaleGeneration { report: 5, .. }) => Ok(()),
other => Err(format!("stale report must refuse: {other:?}")),
}
}
#[test]
fn gate_evaluation_mismatch_is_loud() -> TestResult {
let cx = Cx::for_testing();
let labels = flip_fixture_label_report()?;
let evaluation =
evaluate_fusion_candidates(&cx, &flip_fixture_replays(), &labels.triples, incumbent())
.map_err(|error| error.to_string())?;
match assemble_retrieval_tuning_report(
labels,
Some(evaluation),
1,
&RetrievalTuningGateConfig::default(),
) {
Err(ShadowTuningError::Storage { .. }) => Ok(()),
other => Err(format!("gate/evaluation mismatch must be loud: {other:?}")),
}
}
}