diff --git a/src/llm_registry/prompts/classification.rs b/src/llm_registry/prompts/classification.rs index 4eb068353..2851bb3cf 100644 --- a/src/llm_registry/prompts/classification.rs +++ b/src/llm_registry/prompts/classification.rs @@ -25,6 +25,58 @@ Return format: {{"sensitivity_level": <0-4>, "data_domain": ""}}"# ) } +/// Valid interest categories for field-to-category mapping. +/// This is the single source of truth — used by the LLM prompt, the schema service, +/// and downstream discovery features. +pub const INTEREST_CATEGORIES: &[&str] = &[ + "Photography", + "Cooking", + "Running", + "Software Engineering", + "Music", + "Travel", + "Fitness", + "Reading", + "Gaming", + "Finance", + "Gardening", + "Art & Design", + "Parenting", + "Health & Wellness", + "Sports", + "Movies & TV", + "Science", + "Writing", + "Fashion", + "Home Improvement", + "Pets", + "Automotive", + "Productivity", + "Social Media", + "Education", +]; + +/// Build the interest category classification prompt for a single field. +/// +/// The LLM should return a JSON object with `interest_category` (one of the valid +/// categories, or null if the field doesn't map to a user interest). +pub fn build_interest_category_prompt(field_name: &str, description: &str) -> String { + let categories = INTEREST_CATEGORIES.join(", "); + format!( + r#"Classify this database field into a user interest category. Return ONLY a JSON object with one field, no explanation. + +Field name: "{field_name}" +Description: "{description}" + +Valid interest categories: {categories} + +If this field clearly relates to one of the above interests, return that category. +If this field is a structural/metadata field (like id, hash, timestamp, source, content_hash) or doesn't map to any interest, return null. + +Return format: {{"interest_category": "" | null}}"# + ) +} + #[cfg(test)] mod tests { use super::*; @@ -60,4 +112,29 @@ mod tests { assert!(prompt.contains(domain)); } } + + #[test] + fn interest_prompt_contains_field_name_and_description() { + let prompt = + build_interest_category_prompt("photo_album", "the album containing the photo"); + assert!(prompt.contains("photo_album")); + assert!(prompt.contains("the album containing the photo")); + assert!(prompt.contains("interest_category")); + } + + #[test] + fn interest_prompt_lists_all_categories() { + let prompt = build_interest_category_prompt("x", "y"); + for category in INTEREST_CATEGORIES { + assert!(prompt.contains(category), "Missing category: {}", category); + } + } + + #[test] + fn interest_categories_are_non_empty() { + assert!(!INTEREST_CATEGORIES.is_empty()); + for cat in INTEREST_CATEGORIES { + assert!(!cat.is_empty()); + } + } } diff --git a/src/schema/types/declarative_schemas.rs b/src/schema/types/declarative_schemas.rs index 06e12eb70..3d9f66d31 100644 --- a/src/schema/types/declarative_schemas.rs +++ b/src/schema/types/declarative_schemas.rs @@ -127,6 +127,8 @@ impl<'de> serde::Deserialize<'de> for DeclarativeSchemaDefinition { #[serde(default)] field_data_classifications: HashMap, #[serde(default)] + field_interest_categories: HashMap, + #[serde(default)] ref_fields: HashMap, #[serde(default)] field_types: HashMap, @@ -207,9 +209,10 @@ impl<'de> serde::Deserialize<'de> for DeclarativeSchemaDefinition { .insert(field_name, classifications); } - // Preserve field_descriptions, field_data_classifications, ref_fields, field_types and identity_hash + // Preserve field_descriptions, field_data_classifications, field_interest_categories, ref_fields, field_types and identity_hash schema.field_descriptions = helper.field_descriptions; schema.field_data_classifications = helper.field_data_classifications; + schema.field_interest_categories = helper.field_interest_categories; schema.ref_fields = helper.ref_fields; schema.field_types = helper.field_types; schema.identity_hash = helper.identity_hash; @@ -267,6 +270,11 @@ pub struct DeclarativeSchemaDefinition { /// Maps field_name -> DataClassification. Required for new fields at schema creation. #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub field_data_classifications: HashMap, + /// Interest categories for each field (e.g. "Photography", "Cooking", "Running"). + /// Assigned by the schema service from the canonical field registry. + /// Maps field_name -> interest category string. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub field_interest_categories: HashMap, /// Reference fields that point to child schemas /// Maps field_name -> child_schema_name #[serde(default, skip_serializing_if = "HashMap::is_empty")] @@ -324,6 +332,7 @@ impl PartialEq for DeclarativeSchemaDefinition { && self.field_molecule_uuids == other.field_molecule_uuids && self.field_classifications == other.field_classifications && self.field_data_classifications == other.field_data_classifications + && self.field_interest_categories == other.field_interest_categories && self.ref_fields == other.ref_fields && self.field_types == other.field_types && self.identity_hash == other.identity_hash @@ -463,6 +472,7 @@ impl DeclarativeSchemaDefinition { field_classifications: HashMap::new(), field_descriptions: HashMap::new(), field_data_classifications: HashMap::new(), + field_interest_categories: HashMap::new(), ref_fields: HashMap::new(), field_types: HashMap::new(), identity_hash: None, diff --git a/src/schema_service/classify.rs b/src/schema_service/classify.rs index eca01ed49..81ec5608e 100644 --- a/src/schema_service/classify.rs +++ b/src/schema_service/classify.rs @@ -1,8 +1,8 @@ -//! Field sensitivity classification inference. +//! Field sensitivity classification and interest category inference. //! -//! Determines the (sensitivity_level, data_domain) for new canonical fields -//! based on the field's description. The schema service is the sole authority -//! on data classification. +//! Determines the (sensitivity_level, data_domain) and interest_category for +//! new canonical fields based on the field's description. The schema service +//! is the sole authority on both data classification and interest categories. //! //! Strategy for new fields without an existing canonical match: //! 1. Caller-provided classification → use it @@ -10,7 +10,9 @@ //! 3. No fallback — returns error. Incorrect classification is worse than no schema. use crate::llm_registry::models; -use crate::llm_registry::prompts::classification::build_classification_prompt; +use crate::llm_registry::prompts::classification::{ + build_classification_prompt, build_interest_category_prompt, INTEREST_CATEGORIES, +}; use crate::schema::types::data_classification::DataClassification; /// Classify a field using LLM analysis of its description. @@ -121,6 +123,151 @@ pub async fn classify_with_llm( Ok(classification) } +/// Classify a field's interest category using LLM analysis of its description. +/// Returns `Ok(None)` if the field doesn't map to any interest category (structural fields). +/// Returns `Err` only on LLM communication failures. +pub async fn classify_interest_category_with_llm( + field_name: &str, + description: &str, +) -> Result, String> { + let api_key = std::env::var("ANTHROPIC_API_KEY").map_err(|_| { + "Schema service cannot classify interest categories: ANTHROPIC_API_KEY not set.".to_string() + })?; + if api_key.trim().is_empty() { + return Err( + "Schema service cannot classify interest categories: ANTHROPIC_API_KEY is empty" + .to_string(), + ); + } + + let prompt = build_interest_category_prompt(field_name, description); + + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs( + models::TIMEOUT_CLASSIFICATION, + )) + .no_proxy() + .build() + .map_err(|e| { + format!( + "Failed to create HTTP client for interest classification: {}", + e + ) + })?; + + let request_body = serde_json::json!({ + "model": models::ANTHROPIC_HAIKU, + "messages": [{"role": "user", "content": prompt}], + "max_tokens": models::MAX_TOKENS_CLASSIFICATION, + "temperature": models::TEMPERATURE_DETERMINISTIC + }); + + let response = client + .post(format!("{}/v1/messages", models::ANTHROPIC_API_URL)) + .header("x-api-key", &api_key) + .header("anthropic-version", models::ANTHROPIC_API_VERSION) + .header("Content-Type", "application/json") + .json(&request_body) + .send() + .await + .map_err(|e| { + format!( + "Interest category LLM call failed for field '{}': {}", + field_name, e + ) + })?; + + if !response.status().is_success() { + return Err(format!( + "Interest category LLM call returned status {} for field '{}'", + response.status(), + field_name + )); + } + + let resp: serde_json::Value = response.json().await.map_err(|e| { + format!( + "Failed to parse LLM response for interest category of field '{}': {}", + field_name, e + ) + })?; + + let text = resp + .get("content") + .and_then(|c| c.as_array()) + .and_then(|a| a.first()) + .and_then(|c| c.get("text")) + .and_then(|t| t.as_str()) + .ok_or_else(|| { + format!( + "LLM response missing content text for interest category of field '{}'", + field_name + ) + })?; + + // Parse the JSON response — try raw text first, then extract from markdown fence + let parsed: serde_json::Value = serde_json::from_str(text) + .or_else(|_| { + let trimmed = text.trim(); + let json_str = trimmed + .strip_prefix("```json") + .or_else(|| trimmed.strip_prefix("```")) + .and_then(|s| s.strip_suffix("```")) + .unwrap_or(trimmed) + .trim(); + serde_json::from_str(json_str) + }) + .map_err(|e| { + format!( + "Failed to parse LLM interest category for field '{}': {} (raw: {})", + field_name, e, text + ) + })?; + + let category = parsed + .get("interest_category") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + // Validate against known categories + let validated = category.filter(|cat| { + INTEREST_CATEGORIES + .iter() + .any(|valid| valid.eq_ignore_ascii_case(cat)) + }); + + if let Some(ref cat) = validated { + crate::log_feature!( + crate::logging::features::LogFeature::Schema, + info, + "LLM classified field '{}' interest category as '{}'", + field_name, + cat + ); + } + + Ok(validated) +} + +/// Infer interest category for a new canonical field. +/// Returns `Ok(None)` for structural fields or when the API key is missing. +/// Interest category is best-effort — missing it doesn't block schema creation. +pub async fn infer_interest_category(field_name: &str, description: &str) -> Option { + match classify_interest_category_with_llm(field_name, description).await { + Ok(category) => category, + Err(e) => { + crate::log_feature!( + crate::logging::features::LogFeature::Schema, + warn, + "Interest category classification failed for field '{}': {} (non-blocking)", + field_name, + e + ); + None + } + } +} + /// Infer classification for a new canonical field. /// Returns an error if classification cannot be determined — no silent fallbacks. /// @@ -171,4 +318,20 @@ mod tests { } } } + + #[tokio::test] + async fn infer_interest_category_returns_none_without_api_key() { + // Without ANTHROPIC_API_KEY, should return None (non-blocking) + let result = infer_interest_category("photo_album", "the album containing the photo").await; + // Either returns a valid category (if API key is set) or None + if let Some(ref cat) = result { + assert!( + INTEREST_CATEGORIES + .iter() + .any(|valid| valid.eq_ignore_ascii_case(cat)), + "Invalid category: {}", + cat + ); + } + } } diff --git a/src/schema_service/state.rs b/src/schema_service/state.rs index 53a7ebeed..0db8394d8 100644 --- a/src/schema_service/state.rs +++ b/src/schema_service/state.rs @@ -620,6 +620,80 @@ impl SchemaServiceState { Ok(state) } + /// Backfill interest categories for canonical fields that were registered + /// before interest category classification was added. Best-effort: failures + /// are logged but don't block startup. Call after construction. + /// + /// After backfilling canonical fields, propagates interest categories to + /// all existing schemas and persists the updated schemas. + pub async fn run_interest_category_backfill(&self) { + self.backfill_interest_categories().await; + + // Propagate to existing schemas + let schema_names: Vec = { + let schemas = match self.schemas.read() { + Ok(s) => s, + Err(_) => return, + }; + schemas + .iter() + .filter(|(_, s)| s.superseded_by.is_none()) + .map(|(name, _)| name.clone()) + .collect() + }; + + let mut updated = 0usize; + for name in &schema_names { + // Read canonical fields first + let canonical_fields = match self.canonical_fields.read() { + Ok(f) => f, + Err(_) => continue, + }; + + let mut schemas = match self.schemas.write() { + Ok(s) => s, + Err(_) => return, + }; + if let Some(schema) = schemas.get_mut(name) { + let before = schema.field_interest_categories.len(); + for field_name in schema.fields.clone().as_deref().unwrap_or(&[]) { + if schema.field_interest_categories.contains_key(field_name) { + continue; + } + if let Some(canonical) = canonical_fields.get(field_name) { + if let Some(ref category) = canonical.interest_category { + schema + .field_interest_categories + .insert(field_name.clone(), category.clone()); + } + } + } + if schema.field_interest_categories.len() > before { + // Persist to Sled directly (backfill doesn't need the full async persist) + match &self.storage { + SchemaStorage::Sled { schemas_tree, .. } => { + if let Ok(bytes) = serde_json::to_vec(&*schema) { + let _ = schemas_tree.insert(schema.name.as_bytes(), bytes); + } + } + #[cfg(feature = "aws-backend")] + SchemaStorage::Cloud { .. } => {} + } + updated += 1; + } + } + } + + if updated > 0 { + log_feature!( + LogFeature::Schema, + info, + "Backfill: updated interest categories on {} schemas", + updated + ); + } + } + pub async fn add_schema( &self, mut schema: Schema, @@ -1039,9 +1113,10 @@ impl SchemaServiceState { // Fails if classification cannot be determined (no ANTHROPIC_API_KEY for new fields). self.register_canonical_fields(&schema).await?; - // Propagate canonical field types and classifications to the schema + // Propagate canonical field types, classifications, and interest categories to the schema self.apply_canonical_types(&mut schema); self.apply_canonical_classifications(&mut schema); + self.apply_canonical_interest_categories(&mut schema); log_feature!( LogFeature::Schema, diff --git a/src/schema_service/state_expansion.rs b/src/schema_service/state_expansion.rs index a88e4b9da..6426a4320 100644 --- a/src/schema_service/state_expansion.rs +++ b/src/schema_service/state_expansion.rs @@ -211,6 +211,12 @@ impl SchemaServiceState { .entry(canonical.clone()) .or_insert(data_classification); } + if let Some(interest_category) = schema.field_interest_categories.remove(old_name) { + schema + .field_interest_categories + .entry(canonical.clone()) + .or_insert(interest_category); + } // Add mutation_mapper: old_name → canonical so AI mutations still work mutation_mappers .entry(old_name.clone()) @@ -300,6 +306,14 @@ impl SchemaServiceState { .or_insert_with(|| classification.clone()); } + // Merge field_interest_categories (keep existing, add new) + for (field, category) in &existing.field_interest_categories { + schema + .field_interest_categories + .entry(field.clone()) + .or_insert_with(|| category.clone()); + } + // Merge ref_fields (keep existing references) for (field, target) in &existing.ref_fields { schema @@ -359,9 +373,10 @@ impl SchemaServiceState { // Fails if classification cannot be determined (no ANTHROPIC_API_KEY for new fields). self.register_canonical_fields(schema).await?; - // Propagate canonical field types and classifications to the expanded schema + // Propagate canonical field types, classifications, and interest categories to the expanded schema self.apply_canonical_types(schema); self.apply_canonical_classifications(schema); + self.apply_canonical_interest_categories(schema); log_feature!( LogFeature::Schema, diff --git a/src/schema_service/state_fields.rs b/src/schema_service/state_fields.rs index 7af254e5c..418cafc25 100644 --- a/src/schema_service/state_fields.rs +++ b/src/schema_service/state_fields.rs @@ -107,6 +107,10 @@ impl SchemaServiceState { .await .map_err(FoldDbError::Config)?; + // Interest category is best-effort — doesn't block schema creation + let interest_category = + super::classify::infer_interest_category(field_name, &desc).await; + let embed_text = Self::build_embedding_text(field_name, &desc); let embedding = self.embedder.embed_text(&embed_text).ok(); @@ -116,6 +120,7 @@ impl SchemaServiceState { description: desc, field_type, classification: Some(classification), + interest_category, }, embedding, )); @@ -277,6 +282,7 @@ impl SchemaServiceState { description: desc, field_type: FieldValueType::Any, classification: None, + interest_category: None, } }; @@ -325,12 +331,15 @@ impl SchemaServiceState { embeddings.insert(field_name.clone(), vec); } let classification = schema.field_data_classifications.get(field_name).cloned(); + let interest_category = + schema.field_interest_categories.get(field_name).cloned(); fields.insert( field_name.clone(), CanonicalField { description: desc, field_type, classification, + interest_category, }, ); } @@ -345,6 +354,59 @@ impl SchemaServiceState { ); } + /// Backfill interest categories for existing canonical fields that don't have one. + /// Called on startup after loading canonical fields from storage. + /// Best-effort: failures are logged but don't block startup. + pub(super) async fn backfill_interest_categories(&self) { + let fields_to_backfill: Vec<(String, String)> = { + let fields = match self.canonical_fields.read() { + Ok(f) => f, + Err(_) => return, + }; + fields + .iter() + .filter(|(_, cf)| cf.interest_category.is_none()) + .map(|(name, cf)| (name.clone(), cf.description.clone())) + .collect() + }; + + if fields_to_backfill.is_empty() { + return; + } + + log_feature!( + LogFeature::Schema, + info, + "Backfilling interest categories for {} canonical fields", + fields_to_backfill.len() + ); + + let mut backfilled = 0usize; + for (field_name, description) in &fields_to_backfill { + let category = super::classify::infer_interest_category(field_name, description).await; + + if let Some(ref cat) = category { + let mut fields = match self.canonical_fields.write() { + Ok(f) => f, + Err(_) => continue, + }; + if let Some(canonical) = fields.get_mut(field_name) { + canonical.interest_category = Some(cat.clone()); + self.persist_canonical_field(field_name, canonical); + backfilled += 1; + } + } + } + + log_feature!( + LogFeature::Schema, + info, + "Backfilled interest categories: {}/{} fields classified", + backfilled, + fields_to_backfill.len() + ); + } + /// Persist a canonical field to sled storage. pub(super) fn persist_canonical_field(&self, name: &str, canonical: &CanonicalField) { match &self.storage { @@ -409,4 +471,27 @@ impl SchemaServiceState { } } } + + /// Populate a schema's `field_interest_categories` map from the canonical field registry. + /// Called after canonicalization to propagate interest categories from the registry to the schema. + /// Only fills in fields that don't already have an interest category declared. + pub(super) fn apply_canonical_interest_categories(&self, schema: &mut Schema) { + let fields = match self.canonical_fields.read() { + Ok(f) => f, + Err(_) => return, + }; + + for field_name in schema.fields.as_deref().unwrap_or(&[]) { + if schema.field_interest_categories.contains_key(field_name) { + continue; + } + if let Some(canonical) = fields.get(field_name) { + if let Some(ref category) = canonical.interest_category { + schema + .field_interest_categories + .insert(field_name.clone(), category.clone()); + } + } + } + } } diff --git a/src/schema_service/types.rs b/src/schema_service/types.rs index 131b17e56..f57d50413 100644 --- a/src/schema_service/types.rs +++ b/src/schema_service/types.rs @@ -9,7 +9,8 @@ use crate::schema::types::Schema; /// A canonical field entry in the global field registry. /// Carries description (for semantic matching), type (for enforcement), -/// and optional data classification (for sensitivity labeling). +/// optional data classification (for sensitivity labeling), and optional +/// interest category (for discovery/social features). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct CanonicalField { pub description: String, @@ -18,6 +19,11 @@ pub struct CanonicalField { /// that were registered before classification was required. #[serde(default, skip_serializing_if = "Option::is_none")] pub classification: Option, + /// Interest category for discovery (e.g. "Photography", "Cooking", "Running"). + /// Assigned by LLM at field registration time. `None` for fields that don't + /// map to a user interest (e.g. content_hash, source, id fields). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub interest_category: Option, } /// Response containing a list of available schema names