Phase 17: Model profiles + scoped search — the LLM-brain keystone

Implements PRD invariant 9 ("every reader gets its own profile") and
completes the multi-model substrate vision. Local models (or agents)
bind to a named set of datasets; activation pre-loads their vector
indexes into memory; search enforces scope.

Schema (shared::types):
- ModelProfile { id, ollama_name, description, bound_datasets,
                 hnsw_config, embed_model, created_at, created_by }
- ProfileHnswConfig mirrors vectord::trial::HnswConfig to avoid a
  cross-crate dep cycle. Default (ec=80, es=30) matches the Phase 15
  trial winner.
- bound_datasets can reference raw dataset names OR AiView names
  (both register as DataFusion tables with the same name, so mixing
  raw tables and PII-redacted views composes naturally)

Catalog (catalogd::registry):
- put_profile validates id is a slug (alphanumeric + -_ only) and
  every binding resolves to an existing dataset or view
- Persistence at _catalog/profiles/{id}.json, loaded on rebuild
- get_profile / list_profiles / delete_profile

HTTP endpoints:
- POST /catalog/profiles  (create/update)
- GET  /catalog/profiles  (list)
- GET/DELETE /catalog/profiles/{id}
- POST /vectors/profile/{id}/activate  (HNSW hot-load)
- POST /vectors/profile/{id}/search    (scope-enforced)

Activation (vectord::service::activate_profile):
- For each bound dataset, find vector indexes with matching source
- Pre-load embeddings into EmbeddingCache
- Build HNSW with profile's config
- Report warmed indexes + per-binding failures + duration
- Failures on individual bindings don't abort — "substrate keeps
  working" per ADR-017

Scoped search (vectord::service::profile_scoped_search):
- Look up profile, verify index.source ∈ profile.bound_datasets
- Returns 403 with allowed bindings list if out-of-scope
- Uses HNSW if index is warm, brute-force cosine otherwise (graceful
  degradation — no "must activate first" friction)

Bug fix surfaced during testing: vectord::refresh::try_update_index_meta
was a no-op for first-time indexes, so threat_intel_v1 and
kb_team_runs_v1 (both built via refresh after Phase C shipped) didn't
show up in the index registry. Now it auto-infers the source from the
index name convention (`{source}_vN`) and registers new metadata with
reasonable defaults.

End-to-end verified:
- Created security-analyst profile bound to [threat_intel]
- POST /vectors/profile/security-analyst/activate → warmed
  threat_intel_v1 (54 vectors) in 156ms, HNSW built
- Within-scope search: method=hnsw, returned relevant IP indicators
- Out-of-scope: tried to search resumes_100k_v2 (source=candidates)
  → 403 "profile 'security-analyst' is not bound to 'candidates' —
    allowed bindings: [\"threat_intel\"]"
- staffing-recruiter profile created bound to candidates + placements;
  search without activation fell through to brute_force (graceful)

Deferred (Phase 17 followups):
- VRAM-aware activation (unload-then-load via Ollama keep_alive=0)
  — Ollama already handles this; we don't need to reinvent
- Model-identity in audit trail — Phase 13 has role-based audit;
  adding model_id is ~20 LOC when we want it
- Profile bucket pre-load (profile:user bucket mount) — Phase 17.5

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

crates/catalogd/src/registry.rs

@@ -1,6 +1,6 @@
 use shared::types::{
-    AiView, ColumnMeta, DatasetId, DatasetManifest, FreshnessContract, Lineage, ObjectRef,
-    RefreshPolicy, SchemaFingerprint, Sensitivity, Tombstone,
+    AiView, ColumnMeta, DatasetId, DatasetManifest, FreshnessContract, Lineage, ModelProfile,
+    ObjectRef, RefreshPolicy, SchemaFingerprint, Sensitivity, Tombstone,
 };
 
 use crate::tombstones::TombstoneStore;
@@ -45,6 +45,7 @@ pub struct MetadataUpdate {
 
 const MANIFEST_PREFIX: &str = "_catalog/manifests";
 const VIEW_PREFIX: &str = "_catalog/views";
+const PROFILE_PREFIX: &str = "_catalog/profiles";
 
 /// In-memory dataset registry backed by manifest persistence in object storage.
 /// Also tracks AiViews (Phase D) — safe projections over base datasets.
@@ -53,6 +54,7 @@ const VIEW_PREFIX: &str = "_catalog/views";
 pub struct Registry {
     datasets: Arc<RwLock<HashMap<DatasetId, DatasetManifest>>>,
     views: Arc<RwLock<HashMap<String, AiView>>>,
+    profiles: Arc<RwLock<HashMap<String, ModelProfile>>>,
     tombstones: TombstoneStore,
     store: Arc<dyn ObjectStore>,
 }
@@ -62,6 +64,7 @@ impl Registry {
         Self {
             datasets: Arc::new(RwLock::new(HashMap::new())),
             views: Arc::new(RwLock::new(HashMap::new())),
+            profiles: Arc::new(RwLock::new(HashMap::new())),
             tombstones: TombstoneStore::new(store.clone()),
             store,
         }
@@ -136,9 +139,79 @@ impl Registry {
             tracing::info!("catalog: {} views loaded", views.len());
         }
 
+        // Phase 17: load model profiles.
+        let profile_keys = ops::list(&self.store, Some(PROFILE_PREFIX)).await.unwrap_or_default();
+        let mut profiles = self.profiles.write().await;
+        profiles.clear();
+        for key in &profile_keys {
+            if !key.ends_with(".json") { continue; }
+            let data = match ops::get(&self.store, key).await {
+                Ok(d) => d,
+                Err(e) => { tracing::warn!("profile '{key}': read failed: {e}"); continue; }
+            };
+            match serde_json::from_slice::<ModelProfile>(&data) {
+                Ok(p) => { profiles.insert(p.id.clone(), p); }
+                Err(e) => tracing::warn!("profile '{key}': parse failed: {e}"),
+            }
+        }
+        if !profiles.is_empty() {
+            tracing::info!("catalog: {} model profiles loaded", profiles.len());
+        }
+
         Ok(count)
     }
 
+    // --- Phase 17: Model profiles ---
+
+    /// Create or replace a model profile. Validates id slug + ensures
+    /// every bound_dataset exists (as either a raw dataset or an AiView).
+    pub async fn put_profile(&self, profile: ModelProfile) -> Result<ModelProfile, String> {
+        if profile.id.is_empty() {
+            return Err("profile id is empty".into());
+        }
+        if !profile.id.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') {
+            return Err(format!(
+                "profile id '{}' must be alphanumeric + '-' or '_' only", profile.id,
+            ));
+        }
+        for binding in &profile.bound_datasets {
+            let exists = self.get_by_name(binding).await.is_some()
+                || self.get_view(binding).await.is_some();
+            if !exists {
+                return Err(format!(
+                    "bound dataset '{}' not found as dataset or view", binding,
+                ));
+            }
+        }
+
+        let key = format!("{PROFILE_PREFIX}/{}.json", profile.id);
+        let json = serde_json::to_vec_pretty(&profile).map_err(|e| e.to_string())?;
+        ops::put(&self.store, &key, json.into()).await?;
+
+        let mut profiles = self.profiles.write().await;
+        profiles.insert(profile.id.clone(), profile.clone());
+        tracing::info!(
+            "profile registered: {} -> ollama={} bindings={:?}",
+            profile.id, profile.ollama_name, profile.bound_datasets,
+        );
+        Ok(profile)
+    }
+
+    pub async fn get_profile(&self, id: &str) -> Option<ModelProfile> {
+        self.profiles.read().await.get(id).cloned()
+    }
+
+    pub async fn list_profiles(&self) -> Vec<ModelProfile> {
+        self.profiles.read().await.values().cloned().collect()
+    }
+
+    pub async fn delete_profile(&self, id: &str) -> Result<(), String> {
+        let key = format!("{PROFILE_PREFIX}/{id}.json");
+        ops::delete(&self.store, &key).await?;
+        self.profiles.write().await.remove(id);
+        Ok(())
+    }
+
     /// Register a new dataset. Persists manifest to storage before updating memory.
     pub async fn register(
         &self,