feat: schema-level ranking profile definitions

Add support for defining ranking profiles in schema YAML, allowing
deployments to override builtin profiles with deployment-specific
signal names and tuning parameters.

- Add override_register() to ProfileRegistry for clean builtin replacement
- Add with_profiles() to TidalDbBuilder to thread schema profiles
- Parse profiles section in config.rs with full sort/strategy/agg support
- Validate that profile signal references exist in schema at startup
- Change load_schema() to return (Schema, Vec<RankingProfile>)

This closes the gap where the builtin for_you profile referenced
signals (view, like, share) that don't exist in deployment schemas,
causing all feed scores to normalize to 1.0.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Cargo.lock

@@ -3325,6 +3325,7 @@ dependencies = [
  "serde_json",
  "serde_yaml",
  "subtle",
+ "tempfile",
  "thiserror 2.0.18",
  "tidaldb",
  "tokio",

tidal-server/Cargo.toml

@@ -23,3 +23,6 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tidaldb = { path = "../tidal" }
+
+[dev-dependencies]
+tempfile = "3"

tidal-server/src/config.rs

@@ -3,13 +3,17 @@ use std::path::Path;
 use std::time::Duration;
 
 use serde::Deserialize;
+use tidaldb::ranking::profile::{
+    Boost, CandidateStrategy, DiversitySpec, Exclude, Gate, Penalty, ProfileDecay, RankingProfile,
+    SignalAgg, Sort,
+};
 use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, TextFieldType, Window};
 
 use crate::error::{Result, ServerError};
 
 const DEFAULT_SCHEMA_YAML: &str = include_str!("../config/default-schema.yaml");
 
-pub fn load_schema(path: Option<&Path>) -> Result<Schema> {
+pub fn load_schema(path: Option<&Path>) -> Result<(Schema, Vec<RankingProfile>)> {
     let raw = read_config(path, DEFAULT_SCHEMA_YAML)?;
     let spec: SchemaSpec = serde_yaml::from_str(&raw)
         .map_err(|e| ServerError::SchemaConfig(format!("parse schema yaml: {e}")))?;
@@ -20,14 +24,17 @@ pub fn load_schema(path: Option<&Path>) -> Result<Schema> {
         ));
     }
 
+    // Collect signal names for profile validation.
+    let signal_names: Vec<String> = spec.signals.iter().map(|s| s.name.clone()).collect();
+
     let mut builder = SchemaBuilder::new();
-    for signal in spec.signals {
+    for signal in &spec.signals {
         let mut sig = builder.signal(
             &signal.name,
             parse_entity_kind(&signal.entity)?,
             signal.decay.to_decay_spec()?,
         );
-        if let Some(windows) = signal.windows {
+        if let Some(ref windows) = signal.windows {
             let parsed: Result<Vec<Window>> = windows.iter().map(|w| parse_window(w)).collect();
             sig = sig.windows(&parsed?);
         }
@@ -37,13 +44,13 @@ pub fn load_schema(path: Option<&Path>) -> Result<Schema> {
         let _ = sig.add();
     }
 
-    if let Some(text_fields) = spec.text_fields {
+    if let Some(ref text_fields) = spec.text_fields {
         for field in text_fields {
             builder.text_field(&field.name, parse_text_field_type(&field.kind)?);
         }
     }
 
-    if let Some(embeddings) = spec.embedding_slots {
+    if let Some(ref embeddings) = spec.embedding_slots {
         for slot in embeddings {
             builder.embedding_slot(
                 &slot.name,
@@ -53,7 +60,22 @@ pub fn load_schema(path: Option<&Path>) -> Result<Schema> {
         }
     }
 
-    builder.build().map_err(ServerError::SchemaBuild)
+    let schema = builder.build().map_err(ServerError::SchemaBuild)?;
+
+    // Parse profiles and validate signal references.
+    let profiles = if let Some(profile_specs) = spec.profiles {
+        let mut parsed = Vec::with_capacity(profile_specs.len());
+        for spec in profile_specs {
+            let profile = spec.to_ranking_profile()?;
+            validate_profile_signals(&profile, &signal_names)?;
+            parsed.push(profile);
+        }
+        parsed
+    } else {
+        Vec::new()
+    };
+
+    Ok((schema, profiles))
 }
 
 fn read_config(path: Option<&Path>, fallback: &str) -> Result<String> {
@@ -70,6 +92,8 @@ struct SchemaSpec {
     text_fields: Option<Vec<TextFieldSpec>>,
     #[serde(default)]
     embedding_slots: Option<Vec<EmbeddingSpec>>,
+    #[serde(default)]
+    profiles: Option<Vec<ProfileSpec>>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -171,3 +195,598 @@ fn parse_text_field_type(input: &str) -> Result<TextFieldType> {
         ))),
     }
 }
+
+// ── Profile YAML types ──────────────────────────────────────────────────────
+
+#[derive(Debug, Deserialize)]
+struct ProfileSpec {
+    name: String,
+    #[serde(default = "default_version")]
+    version: u32,
+    #[serde(default)]
+    candidate_strategy: Option<CandidateStrategySpec>,
+    #[serde(default)]
+    sort: Option<SortSpec>,
+    #[serde(default)]
+    boosts: Vec<BoostSpec>,
+    #[serde(default)]
+    decay: Option<ProfileDecaySpec>,
+    #[serde(default)]
+    gates: Vec<GateSpec>,
+    #[serde(default)]
+    penalties: Vec<PenaltySpec>,
+    #[serde(default)]
+    excludes: Vec<ExcludeSpec>,
+    #[serde(default)]
+    diversity: Option<DiversitySpecConfig>,
+    #[serde(default)]
+    exploration: f64,
+}
+
+fn default_version() -> u32 {
+    1
+}
+
+#[derive(Debug, Deserialize)]
+struct BoostSpec {
+    signal: String,
+    agg: String,
+    window: String,
+    weight: f64,
+}
+
+#[derive(Debug, Deserialize)]
+struct GateSpec {
+    signal: String,
+    agg: String,
+    window: String,
+    min_threshold: f64,
+}
+
+#[derive(Debug, Deserialize)]
+struct PenaltySpec {
+    signal: String,
+    agg: String,
+    window: String,
+    weight: f64,
+}
+
+#[derive(Debug, Deserialize)]
+struct ExcludeSpec {
+    signal: String,
+    agg: String,
+    window: String,
+    above: f64,
+}
+
+#[derive(Debug, Deserialize)]
+struct ProfileDecaySpec {
+    signal: String,
+    half_life_secs: u64,
+    weight: f64,
+}
+
+#[derive(Debug, Deserialize)]
+struct DiversitySpecConfig {
+    #[serde(default)]
+    max_per_creator: Option<usize>,
+    #[serde(default)]
+    format_mix_max_fraction: Option<f64>,
+}
+
+/// Sort can be a simple string (`"trending"`, `"new"`) or a map with parameters.
+#[derive(Debug, Deserialize)]
+#[serde(untagged)]
+enum SortSpec {
+    Simple(String),
+    Parameterized(std::collections::HashMap<String, serde_yaml::Value>),
+}
+
+/// Candidate strategy can be a simple string or a map with parameters.
+#[derive(Debug, Deserialize)]
+#[serde(untagged)]
+enum CandidateStrategySpec {
+    Simple(String),
+    Parameterized(std::collections::HashMap<String, serde_yaml::Value>),
+}
+
+// ── Profile conversion ──────────────────────────────────────────────────────
+
+impl ProfileSpec {
+    fn to_ranking_profile(&self) -> Result<RankingProfile> {
+        let candidate_strategy = match &self.candidate_strategy {
+            Some(spec) => parse_candidate_strategy(spec)?,
+            None => CandidateStrategy::Scan {
+                sort_field: "created_at".into(),
+            },
+        };
+
+        let sort = match &self.sort {
+            Some(spec) => Some(parse_sort(spec)?),
+            None => None,
+        };
+
+        let boosts = self
+            .boosts
+            .iter()
+            .map(|b| {
+                Ok(Boost {
+                    signal: b.signal.clone(),
+                    agg: parse_agg(&b.agg)?,
+                    window: parse_window(&b.window)?,
+                    weight: b.weight,
+                })
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        let gates = self
+            .gates
+            .iter()
+            .map(|g| {
+                Ok(Gate {
+                    signal: g.signal.clone(),
+                    agg: parse_agg(&g.agg)?,
+                    window: parse_window(&g.window)?,
+                    min_threshold: g.min_threshold,
+                })
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        let penalties = self
+            .penalties
+            .iter()
+            .map(|p| {
+                Ok(Penalty {
+                    signal: p.signal.clone(),
+                    agg: parse_agg(&p.agg)?,
+                    window: parse_window(&p.window)?,
+                    weight: p.weight,
+                })
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        let excludes = self
+            .excludes
+            .iter()
+            .map(|e| {
+                Ok(Exclude {
+                    signal: e.signal.clone(),
+                    agg: parse_agg(&e.agg)?,
+                    window: parse_window(&e.window)?,
+                    above: e.above,
+                })
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        let decay = self.decay.as_ref().map(|d| ProfileDecay {
+            signal: d.signal.clone(),
+            half_life_secs: d.half_life_secs,
+            weight: d.weight,
+        });
+
+        let diversity = match &self.diversity {
+            Some(d) => DiversitySpec {
+                max_per_creator: d.max_per_creator,
+                format_mix_max_fraction: d.format_mix_max_fraction,
+            },
+            None => DiversitySpec::default(),
+        };
+
+        Ok(RankingProfile {
+            name: self.name.clone(),
+            version: self.version,
+            candidate_strategy,
+            boosts,
+            decay,
+            gates,
+            penalties,
+            excludes,
+            diversity,
+            exploration: self.exploration,
+            sort,
+            is_builtin: false,
+        })
+    }
+}
+
+fn parse_agg(input: &str) -> Result<SignalAgg> {
+    match input.trim().to_lowercase().as_str() {
+        "value" => Ok(SignalAgg::Value),
+        "velocity" => Ok(SignalAgg::Velocity),
+        "decay_score" => Ok(SignalAgg::DecayScore),
+        "ratio" => Ok(SignalAgg::Ratio),
+        "relative_velocity" => Ok(SignalAgg::RelativeVelocity),
+        other => Err(ServerError::SchemaConfig(format!(
+            "unknown signal aggregation '{other}'"
+        ))),
+    }
+}
+
+fn parse_sort(spec: &SortSpec) -> Result<Sort> {
+    match spec {
+        SortSpec::Simple(s) => match s.trim().to_lowercase().as_str() {
+            "trending" => Ok(Sort::Trending),
+            "rising" => Ok(Sort::Rising),
+            "controversial" => Ok(Sort::Controversial),
+            "hidden_gems" => Ok(Sort::HiddenGems),
+            "shuffle" => Ok(Sort::Shuffle),
+            "new" => Ok(Sort::New),
+            "most_followed" => Ok(Sort::MostFollowed),
+            "creator_engagement_rate" => Ok(Sort::CreatorEngagementRate),
+            "alphabetical_asc" => Ok(Sort::AlphabeticalAsc),
+            "alphabetical_desc" => Ok(Sort::AlphabeticalDesc),
+            "shortest" => Ok(Sort::Shortest),
+            "longest" => Ok(Sort::Longest),
+            "live_viewer_count" => Ok(Sort::LiveViewerCount),
+            "date_saved" => Ok(Sort::DateSaved),
+            other => Err(ServerError::SchemaConfig(format!(
+                "unknown sort mode '{other}'"
+            ))),
+        },
+        SortSpec::Parameterized(map) => {
+            if let Some(val) = map.get("hot") {
+                let gravity = extract_f64(val, "gravity")?.unwrap_or(1.8);
+                Ok(Sort::Hot { gravity })
+            } else if let Some(val) = map.get("top_window") {
+                let window = extract_window(val, "window")?;
+                Ok(Sort::TopWindow { window })
+            } else if let Some(val) = map.get("most_viewed") {
+                let window = extract_window(val, "window")?;
+                Ok(Sort::MostViewed { window })
+            } else if let Some(val) = map.get("most_liked") {
+                let window = extract_window(val, "window")?;
+                Ok(Sort::MostLiked { window })
+            } else if let Some(val) = map.get("most_commented") {
+                let window = extract_window(val, "window")?;
+                Ok(Sort::MostCommented { window })
+            } else if let Some(val) = map.get("most_shared") {
+                let window = extract_window(val, "window")?;
+                Ok(Sort::MostShared { window })
+            } else {
+                Err(ServerError::SchemaConfig(format!(
+                    "unknown parameterized sort: {map:?}"
+                )))
+            }
+        }
+    }
+}
+
+fn parse_candidate_strategy(spec: &CandidateStrategySpec) -> Result<CandidateStrategy> {
+    match spec {
+        CandidateStrategySpec::Simple(s) => match s.trim().to_lowercase().as_str() {
+            "scan" => Ok(CandidateStrategy::Scan {
+                sort_field: "created_at".into(),
+            }),
+            "relationship" => Ok(CandidateStrategy::Relationship),
+            "hybrid" => Ok(CandidateStrategy::Hybrid),
+            "cohort_trending" => Ok(CandidateStrategy::CohortTrending),
+            other => Err(ServerError::SchemaConfig(format!(
+                "unknown candidate strategy '{other}'"
+            ))),
+        },
+        CandidateStrategySpec::Parameterized(map) => {
+            if let Some(val) = map.get("ann") {
+                let slot = extract_string(val, "slot")?;
+                let limit = extract_u64(val, "limit")?.unwrap_or(100) as usize;
+                Ok(CandidateStrategy::Ann { slot, limit })
+            } else if let Some(val) = map.get("signal_ranked") {
+                let signal = extract_string(val, "signal")?;
+                let window_str = extract_string_field(val, "window")?;
+                let window = parse_window(&window_str)?;
+                Ok(CandidateStrategy::SignalRanked { signal, window })
+            } else if let Some(val) = map.get("scan") {
+                let sort_field =
+                    extract_string_field(val, "sort_field").unwrap_or_else(|_| "created_at".into());
+                Ok(CandidateStrategy::Scan { sort_field })
+            } else {
+                Err(ServerError::SchemaConfig(format!(
+                    "unknown parameterized candidate strategy: {map:?}"
+                )))
+            }
+        }
+    }
+}
+
+// ── YAML value extraction helpers ───────────────────────────────────────────
+
+fn extract_f64(val: &serde_yaml::Value, field: &str) -> Result<Option<f64>> {
+    match val {
+        serde_yaml::Value::Mapping(map) => {
+            if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) {
+                v.as_f64()
+                    .or_else(|| v.as_i64().map(|i| i as f64))
+                    .map(Some)
+                    .ok_or_else(|| ServerError::SchemaConfig(format!("'{field}' must be a number")))
+            } else {
+                Ok(None)
+            }
+        }
+        _ => Ok(None),
+    }
+}
+
+fn extract_u64(val: &serde_yaml::Value, field: &str) -> Result<Option<u64>> {
+    match val {
+        serde_yaml::Value::Mapping(map) => {
+            if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) {
+                v.as_u64().map(Some).ok_or_else(|| {
+                    ServerError::SchemaConfig(format!("'{field}' must be a positive integer"))
+                })
+            } else {
+                Ok(None)
+            }
+        }
+        _ => Ok(None),
+    }
+}
+
+fn extract_string(val: &serde_yaml::Value, field: &str) -> Result<String> {
+    extract_string_field(val, field)
+}
+
+fn extract_string_field(val: &serde_yaml::Value, field: &str) -> Result<String> {
+    match val {
+        serde_yaml::Value::Mapping(map) => {
+            if let Some(v) = map.get(serde_yaml::Value::String(field.to_owned())) {
+                v.as_str()
+                    .map(|s| s.to_owned())
+                    .ok_or_else(|| ServerError::SchemaConfig(format!("'{field}' must be a string")))
+            } else {
+                Err(ServerError::SchemaConfig(format!(
+                    "missing required field '{field}'"
+                )))
+            }
+        }
+        _ => Err(ServerError::SchemaConfig(format!(
+            "expected mapping with field '{field}'"
+        ))),
+    }
+}
+
+fn extract_window(val: &serde_yaml::Value, field: &str) -> Result<Window> {
+    let s = extract_string_field(val, field)?;
+    parse_window(&s)
+}
+
+// ── Signal validation ───────────────────────────────────────────────────────
+
+fn validate_profile_signals(profile: &RankingProfile, signal_names: &[String]) -> Result<()> {
+    let check = |signal: &str, context: &str| -> Result<()> {
+        if !signal_names.iter().any(|s| s == signal) {
+            Err(ServerError::SchemaConfig(format!(
+                "profile '{}': {} references unknown signal '{signal}'",
+                profile.name, context
+            )))
+        } else {
+            Ok(())
+        }
+    };
+
+    for boost in &profile.boosts {
+        check(&boost.signal, "boost")?;
+    }
+    for gate in &profile.gates {
+        check(&gate.signal, "gate")?;
+    }
+    for penalty in &profile.penalties {
+        check(&penalty.signal, "penalty")?;
+    }
+    for exclude in &profile.excludes {
+        check(&exclude.signal, "exclude")?;
+    }
+    if let Some(ref decay) = profile.decay {
+        check(&decay.signal, "decay")?;
+    }
+    Ok(())
+}
+
+// ── Tests ───────────────────────────────────────────────────────────────────
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_schema_with_profiles() {
+        let yaml = r#"
+signals:
+  - name: chat
+    entity: item
+    decay:
+      exponential:
+        half_life_seconds: 1209600
+    windows: [24h, 7d]
+  - name: favorite
+    entity: item
+    decay:
+      exponential:
+        half_life_seconds: 2592000
+    windows: [7d, 30d]
+
+profiles:
+  - name: for_you
+    sort:
+      hot:
+        gravity: 1.5
+    boosts:
+      - signal: chat
+        agg: decay_score
+        window: all_time
+        weight: 1.5
+      - signal: favorite
+        agg: decay_score
+        window: all_time
+        weight: 2.5
+    diversity:
+      max_per_creator: 2
+    exploration: 0.1
+"#;
+        let spec: SchemaSpec = serde_yaml::from_str(yaml).unwrap();
+        assert!(spec.profiles.is_some());
+        let profiles = spec.profiles.unwrap();
+        assert_eq!(profiles.len(), 1);
+        let profile = profiles[0].to_ranking_profile().unwrap();
+        assert_eq!(profile.name, "for_you");
+        assert_eq!(profile.boosts.len(), 2);
+        assert!(!profile.is_builtin);
+        assert!((profile.exploration - 0.1).abs() < f64::EPSILON);
+        assert!(
+            matches!(profile.sort, Some(Sort::Hot { gravity }) if (gravity - 1.5).abs() < f64::EPSILON)
+        );
+        assert_eq!(profile.diversity.max_per_creator, Some(2));
+    }
+
+    #[test]
+    fn parse_simple_sort_modes() {
+        let cases = vec![
+            ("trending", true),
+            ("new", true),
+            ("shuffle", true),
+            ("rising", true),
+            ("controversial", true),
+            ("hidden_gems", true),
+            ("bad_sort", false),
+        ];
+        for (input, should_ok) in cases {
+            let result = parse_sort(&SortSpec::Simple(input.into()));
+            assert_eq!(result.is_ok(), should_ok, "sort '{input}'");
+        }
+    }
+
+    #[test]
+    fn parse_parameterized_sort() {
+        let mut map = std::collections::HashMap::new();
+        let mut inner = serde_yaml::Mapping::new();
+        inner.insert(
+            serde_yaml::Value::String("window".into()),
+            serde_yaml::Value::String("7d".into()),
+        );
+        map.insert("top_window".into(), serde_yaml::Value::Mapping(inner));
+        let result = parse_sort(&SortSpec::Parameterized(map)).unwrap();
+        assert!(matches!(
+            result,
+            Sort::TopWindow {
+                window: Window::SevenDays
+            }
+        ));
+    }
+
+    #[test]
+    fn parse_agg_variants() {
+        assert!(matches!(parse_agg("value").unwrap(), SignalAgg::Value));
+        assert!(matches!(
+            parse_agg("velocity").unwrap(),
+            SignalAgg::Velocity
+        ));
+        assert!(matches!(
+            parse_agg("decay_score").unwrap(),
+            SignalAgg::DecayScore
+        ));
+        assert!(matches!(parse_agg("ratio").unwrap(), SignalAgg::Ratio));
+        assert!(matches!(
+            parse_agg("relative_velocity").unwrap(),
+            SignalAgg::RelativeVelocity
+        ));
+        assert!(parse_agg("unknown").is_err());
+    }
+
+    #[test]
+    fn parse_candidate_strategy_variants() {
+        let scan = parse_candidate_strategy(&CandidateStrategySpec::Simple("scan".into())).unwrap();
+        assert!(matches!(scan, CandidateStrategy::Scan { .. }));
+
+        let rel = parse_candidate_strategy(&CandidateStrategySpec::Simple("relationship".into()))
+            .unwrap();
+        assert!(matches!(rel, CandidateStrategy::Relationship));
+    }
+
+    #[test]
+    fn unknown_signal_in_profile_rejected() {
+        let profile = RankingProfile {
+            name: "test".into(),
+            version: 1,
+            candidate_strategy: CandidateStrategy::Scan {
+                sort_field: "created_at".into(),
+            },
+            boosts: vec![Boost {
+                signal: "nonexistent".into(),
+                agg: SignalAgg::Value,
+                window: Window::AllTime,
+                weight: 1.0,
+            }],
+            decay: None,
+            gates: vec![],
+            penalties: vec![],
+            excludes: vec![],
+            diversity: DiversitySpec::default(),
+            exploration: 0.0,
+            sort: None,
+            is_builtin: false,
+        };
+        let signal_names = vec!["chat".into(), "favorite".into()];
+        let result = validate_profile_signals(&profile, &signal_names);
+        assert!(result.is_err());
+        let err = result.unwrap_err().to_string();
+        assert!(err.contains("nonexistent"), "error: {err}");
+    }
+
+    #[test]
+    fn load_schema_returns_tuple() {
+        let (schema, profiles) = load_schema(None).unwrap();
+        // Default schema has no profiles section.
+        assert!(profiles.is_empty());
+        assert!(schema.signals().next().is_some());
+    }
+
+    #[test]
+    fn schema_with_no_profiles_section() {
+        let yaml = r#"
+signals:
+  - name: view
+    entity: item
+    decay:
+      exponential:
+        half_life_seconds: 604800
+    windows: [24h, 7d]
+"#;
+        let tmp = tempfile::NamedTempFile::new().unwrap();
+        std::fs::write(tmp.path(), yaml).unwrap();
+        let (_, profiles) = load_schema(Some(tmp.path())).unwrap();
+        assert!(profiles.is_empty());
+    }
+
+    #[test]
+    fn profile_defaults() {
+        let yaml = r#"
+signals:
+  - name: view
+    entity: item
+    decay:
+      exponential:
+        half_life_seconds: 604800
+    windows: [24h, 7d]
+
+profiles:
+  - name: minimal
+    boosts:
+      - signal: view
+        agg: value
+        window: all_time
+        weight: 1.0
+"#;
+        let tmp = tempfile::NamedTempFile::new().unwrap();
+        std::fs::write(tmp.path(), yaml).unwrap();
+        let (_, profiles) = load_schema(Some(tmp.path())).unwrap();
+        assert_eq!(profiles.len(), 1);
+        let p = &profiles[0];
+        assert_eq!(p.version, 1);
+        assert!(!p.is_builtin);
+        assert!(p.sort.is_none());
+        assert!((p.exploration).abs() < f64::EPSILON);
+        assert!(matches!(
+            p.candidate_strategy,
+            CandidateStrategy::Scan { .. }
+        ));
+    }
+}

tidal-server/src/main.rs

@@ -76,9 +76,11 @@ fn init_tracing() {
 }
 
 async fn run_standalone(args: StandaloneArgs) -> Result<()> {
-    let schema = load_schema(args.schema.as_deref())?;
+    let (schema, profiles) = load_schema(args.schema.as_deref())?;
 
-    let mut builder = TidalDb::builder().with_schema(schema.clone());
+    let mut builder = TidalDb::builder()
+        .with_schema(schema.clone())
+        .with_profiles(profiles);
     if let Some(dir) = args.data_dir {
         builder = builder.with_data_dir(dir);
     } else {

tidal-server/tests/middleware.rs

@@ -17,10 +17,11 @@ use tower::ServiceExt;
 fn make_state() -> Arc<ServerState> {
     // Use the default schema so ranking profiles are available — same setup as
     // `run_standalone` in main.rs.
-    let schema = tidal_server::config::load_schema(None).unwrap();
+    let (schema, profiles) = tidal_server::config::load_schema(None).unwrap();
     let db = TidalDb::builder()
         .ephemeral()
         .with_schema(schema)
+        .with_profiles(profiles)
         .open()
         .unwrap();
     Arc::new(ServerState::new(db))

tidal/src/db/builder.rs

@@ -10,6 +10,7 @@ use super::metrics::MetricsState;
 use super::paths::Paths;
 
 use crate::load::RateLimiterConfig;
+use crate::ranking::profile::RankingProfile;
 use crate::schema::Schema;
 
 /// Fluent builder for constructing a [`TidalDb`] instance.
@@ -55,6 +56,8 @@ pub struct TidalDbBuilder {
     /// When set and `NodeRole::Follower`, the receiver is started automatically.
     /// When set and `NodeRole::Leader`, the WAL shipper is spawned automatically.
     transport: Option<std::sync::Arc<dyn crate::replication::Transport>>,
+    /// Schema-defined profiles that override matching builtins.
+    profiles: Vec<RankingProfile>,
 }
 
 impl TidalDbBuilder {
@@ -67,6 +70,7 @@ impl TidalDbBuilder {
             schema: None,
             rate_limiter_config: None,
             transport: None,
+            profiles: Vec::new(),
         }
     }
 
@@ -84,6 +88,18 @@ impl TidalDbBuilder {
         self
     }
 
+    /// Attach schema-defined ranking profiles that override matching builtins.
+    ///
+    /// Profiles provided here are registered via [`override_register`] after
+    /// built-in profiles during [`open`](Self::open). This allows schema YAML
+    /// to replace built-in profile definitions with deployment-specific signal
+    /// names and tuning parameters.
+    #[must_use]
+    pub fn with_profiles(mut self, profiles: Vec<RankingProfile>) -> Self {
+        self.profiles = profiles;
+        self
+    }
+
     /// Configure per-agent session rate limiting.
     ///
     /// When set, each `(agent_id, session_id)` pair gets a token bucket with
@@ -328,7 +344,7 @@ impl TidalDbBuilder {
 
         if let Some(schema) = self.schema {
             // Wire in storage, WAL, signal ledger, and M2 indexes.
-            let result = TidalDb::open_with_schema(&self.config, schema)?;
+            let result = TidalDb::open_with_schema(&self.config, schema, self.profiles)?;
 
             // ── Fix B: Schema fingerprint persistence ───────────────────
             // Persistent mode only. Compute a BLAKE3 hash of the schema's

tidal/src/db/builder_tests.rs

@@ -28,6 +28,7 @@ fn builder_persistent_requires_data_dir() {
         schema: None,
         rate_limiter_config: None,
         transport: None,
+        profiles: Vec::new(),
     };
     let result = builder.validate();
     assert!(result.is_err());

tidal/src/db/open.rs

@@ -9,6 +9,7 @@ use crate::entities::{
     CreatorItemsBitmap, HardNegIndex, InteractionLedger, PreferenceVectors, UserStateIndex,
 };
 use crate::ranking::builtins::register_builtins;
+use crate::ranking::profile::RankingProfile;
 use crate::ranking::registry::ProfileRegistry;
 use crate::schema::{DurabilityError, EntityId, Schema, TidalError, Timestamp};
 use crate::signals::{NoopWalWriter, SignalLedger, SignalTypeId};
@@ -61,6 +62,7 @@ impl super::TidalDb {
     pub(crate) fn open_with_schema(
         config: &super::Config,
         schema: Schema,
+        schema_profiles: Vec<RankingProfile>,
     ) -> crate::Result<OpenResult> {
         let last_seq = Arc::new(AtomicU64::new(0));
 
@@ -70,6 +72,17 @@ impl super::TidalDb {
             TidalError::internal("open", format!("failed to register builtin profiles: {e}"))
         })?;
 
+        // Register schema-defined profiles, overriding matching builtins.
+        for profile in schema_profiles {
+            let name = profile.name.clone();
+            profile_registry.override_register(profile).map_err(|e| {
+                TidalError::internal(
+                    "open",
+                    format!("failed to register schema profile '{name}': {e}"),
+                )
+            })?;
+        }
+
         // Initialize M2 indexes (empty -- populated as items are written).
         let category_index = BitmapIndex::new("category");
         let format_index = BitmapIndex::new("format");

tidal/src/ranking/registry.rs

@@ -115,6 +115,46 @@ impl ProfileRegistry {
         Ok(())
     }
 
+    /// Override-register a profile, replacing all existing versions for that name.
+    ///
+    /// Clears any previously registered versions (including builtins) and inserts
+    /// the new profile as the sole version. Applies the same validation as
+    /// [`register`](Self::register) (name format, exploration bounds, gate
+    /// thresholds) but skips version monotonicity since all prior versions are
+    /// removed.
+    ///
+    /// This is used by schema-defined profiles to cleanly replace built-in
+    /// defaults without version gymnastics.
+    ///
+    /// # Errors
+    ///
+    /// Same as [`register`](Self::register) except `VersionConflict` is impossible.
+    pub fn override_register(&mut self, profile: RankingProfile) -> Result<(), ProfileError> {
+        if !is_valid_name(&profile.name) {
+            return Err(ProfileError::InvalidName(profile.name));
+        }
+
+        if !(0.0..=0.5).contains(&profile.exploration) {
+            return Err(ProfileError::ExplorationOutOfRange(profile.exploration));
+        }
+
+        for gate in &profile.gates {
+            if matches!(gate.agg, SignalAgg::DecayScore | SignalAgg::Ratio) {
+                if !(0.0..=1.0).contains(&gate.min_threshold) {
+                    return Err(ProfileError::GateThresholdOutOfRange(gate.min_threshold));
+                }
+            } else if gate.min_threshold < 0.0 {
+                return Err(ProfileError::GateThresholdOutOfRange(gate.min_threshold));
+            }
+        }
+
+        // Clear all existing versions for this name, then insert the new one.
+        let versions = self.profiles.entry(profile.name.clone()).or_default();
+        versions.clear();
+        versions.insert(profile.version, profile);
+        Ok(())
+    }
+
     /// Get the latest version of a named profile.
     ///
     /// # Errors
@@ -289,4 +329,68 @@ mod tests {
         let result = registry.get("nonexistent");
         assert!(matches!(result, Err(ProfileError::NotFound(_))));
     }
+
+    #[test]
+    fn override_register_replaces_builtin() {
+        let mut registry = ProfileRegistry::new();
+        let mut builtin = minimal_profile("for_you", 1);
+        builtin.is_builtin = true;
+        registry.register(builtin).unwrap();
+
+        // Override with a schema-defined version.
+        let mut override_profile = minimal_profile("for_you", 1);
+        override_profile.exploration = 0.1;
+        registry.override_register(override_profile).unwrap();
+
+        let profile = registry.get("for_you").unwrap();
+        assert!(!profile.is_builtin);
+        assert!((profile.exploration - 0.1).abs() < f64::EPSILON);
+    }
+
+    #[test]
+    fn override_register_clears_all_versions() {
+        let mut registry = ProfileRegistry::new();
+        registry.register(minimal_profile("trending", 1)).unwrap();
+        registry.register(minimal_profile("trending", 2)).unwrap();
+
+        // Override clears both v1 and v2.
+        let override_profile = minimal_profile("trending", 1);
+        registry.override_register(override_profile).unwrap();
+
+        // Only v1 exists now.
+        assert!(registry.get_version("trending", 2).is_err());
+        assert!(registry.get_version("trending", 1).is_ok());
+    }
+
+    #[test]
+    fn override_register_validates_constraints() {
+        let mut registry = ProfileRegistry::new();
+
+        // Invalid name.
+        let result = registry.override_register(minimal_profile("BAD NAME", 1));
+        assert!(matches!(result, Err(ProfileError::InvalidName(_))));
+
+        // Exploration out of range.
+        let mut profile = minimal_profile("bad_explore", 1);
+        profile.exploration = 0.9;
+        let result = registry.override_register(profile);
+        assert!(matches!(
+            result,
+            Err(ProfileError::ExplorationOutOfRange(_))
+        ));
+
+        // Gate threshold out of range.
+        let mut profile = minimal_profile("bad_gate", 1);
+        profile.gates.push(Gate {
+            signal: "view".into(),
+            agg: SignalAgg::DecayScore,
+            window: Window::AllTime,
+            min_threshold: 1.5,
+        });
+        let result = registry.override_register(profile);
+        assert!(matches!(
+            result,
+            Err(ProfileError::GateThresholdOutOfRange(_))
+        ));
+    }
 }