From 8b77f013d15836ab7c2b48913510107365933b77 Mon Sep 17 00:00:00 2001 From: Tom Tang <4220945+shiba4life@users.noreply.github.com> Date: Mon, 30 Mar 2026 10:22:27 -0700 Subject: [PATCH 1/2] feat: add field-level interest category classification to schema service The schema service now assigns interest categories (e.g. "Photography", "Cooking", "Running") to canonical fields via LLM classification. Categories are propagated to schemas via field_interest_categories HashMap, following the same pattern as field_data_classifications. - Add interest_category to CanonicalField (serde-compatible, backward compat) - Add INTEREST_CATEGORIES vocabulary as single source of truth - Add classify_interest_category_with_llm() and LLM prompt - Add field_interest_categories to Schema struct + deserializer - Propagate via apply_canonical_interest_categories() in add_schema/expand_schema - Add run_interest_category_backfill() for migrating existing fields - Interest category is best-effort (non-blocking on failure) Co-Authored-By: Claude Opus 4.6 (1M context) --- src/llm_registry/prompts/classification.rs | 76 +++++++++ src/schema/types/declarative_schemas.rs | 12 +- src/schema_service/classify.rs | 171 ++++++++++++++++++++- src/schema_service/state.rs | 77 +++++++++- src/schema_service/state_expansion.rs | 17 +- src/schema_service/state_fields.rs | 86 +++++++++++ src/schema_service/types.rs | 8 +- 7 files changed, 438 insertions(+), 9 deletions(-) diff --git a/src/llm_registry/prompts/classification.rs b/src/llm_registry/prompts/classification.rs index 4eb068353..7e8b98684 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,28 @@ 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..353b888a1 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,149 @@ 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 +316,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..ed78eccad 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,60 @@ 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 +472,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 From 6bf2ac27fc2f17a85a77a69737d2f0d6a3e07c7e Mon Sep 17 00:00:00 2001 From: Tom Tang <4220945+shiba4life@users.noreply.github.com> Date: Mon, 30 Mar 2026 10:27:12 -0700 Subject: [PATCH 2/2] style: fix formatting Co-Authored-By: Claude Opus 4.6 (1M context) --- src/llm_registry/prompts/classification.rs | 3 ++- src/schema_service/classify.rs | 12 +++++++----- src/schema_service/state_fields.rs | 3 +-- 3 files changed, 10 insertions(+), 8 deletions(-) diff --git a/src/llm_registry/prompts/classification.rs b/src/llm_registry/prompts/classification.rs index 7e8b98684..2851bb3cf 100644 --- a/src/llm_registry/prompts/classification.rs +++ b/src/llm_registry/prompts/classification.rs @@ -115,7 +115,8 @@ mod tests { #[test] fn interest_prompt_contains_field_name_and_description() { - let prompt = build_interest_category_prompt("photo_album", "the album containing the photo"); + 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")); diff --git a/src/schema_service/classify.rs b/src/schema_service/classify.rs index 353b888a1..81ec5608e 100644 --- a/src/schema_service/classify.rs +++ b/src/schema_service/classify.rs @@ -148,7 +148,12 @@ pub async fn classify_interest_category_with_llm( )) .no_proxy() .build() - .map_err(|e| format!("Failed to create HTTP client for interest classification: {}", e))?; + .map_err(|e| { + format!( + "Failed to create HTTP client for interest classification: {}", + e + ) + })?; let request_body = serde_json::json!({ "model": models::ANTHROPIC_HAIKU, @@ -247,10 +252,7 @@ pub async fn classify_interest_category_with_llm( /// 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 { +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) => { diff --git a/src/schema_service/state_fields.rs b/src/schema_service/state_fields.rs index ed78eccad..418cafc25 100644 --- a/src/schema_service/state_fields.rs +++ b/src/schema_service/state_fields.rs @@ -383,8 +383,7 @@ impl SchemaServiceState { let mut backfilled = 0usize; for (field_name, description) in &fields_to_backfill { - let category = - super::classify::infer_interest_category(field_name, description).await; + 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() {