Phase D: AI-safe views — capability-surface projections over base data

Implements the llms3.com "AI-safe views" pattern: a named projection
that exposes only whitelisted columns, with optional row filter and
per-column redactions. AI agents (or Phase 13 roles) bind to the view;
they can never accidentally see PII even if they write raw SQL.

Schema (shared::types):
- AiView { name, base_dataset, columns: Vec<String>, row_filter,
           column_redactions: HashMap<String, Redaction>, ... }
- Redaction enum: Null | Hash | Mask { keep_prefix, keep_suffix }

Catalog (catalogd::registry):
- put_view validates base dataset exists + columns non-empty
- Persists JSON at _catalog/views/{name}.json (sanitized name)
- rebuild() loads views alongside dataset manifests on startup

Query layer (queryd::context):
- build_context registers every AiView as a DataFusion view object
- Constructed SELECT applies whitelist projection, WHERE filter, and
  redaction expressions per column
  - Mask: substr(prefix) + repeat('*', mid_len) + substr(suffix)
  - Hash: digest(value, 'sha256')
  - Null: CAST(NULL AS VARCHAR) AS col
- DataFusion handles JOINs/aggregates over the view natively — it's a
  real view, not a query rewrite

HTTP (catalogd::service):
- POST /catalog/views (create)
- GET  /catalog/views (list)
- GET  /catalog/views/{name} (full def)
- DELETE /catalog/views/{name}

End-to-end test on candidates (100K rows, 15 columns):

  candidates_safe view:
    columns: candidate_id, first_name, city, state, vertical,
             skills, years_experience, status
    row_filter: status != 'blocked'
    redaction: candidate_id mask(prefix=3, suffix=2)

  SELECT * FROM candidates_safe LIMIT 5
    -> 8 columns only, candidate_id shown as "CAN******01"
       (PII fields email/phone/last_name absent from result)

  SELECT email FROM candidates_safe
    -> fails (column not in projection)

  SELECT email FROM candidates
    -> succeeds (raw table still accessible by name —
       Phase 13 access control is the gate, not the view itself)

Survives restart — view definitions reload from object storage.

Limits / not in MVP:
- View CANNOT shadow base table by name (DataFusion treats them as
  separate identifiers; access control must restrict raw-table access)
- row_filter is treated as trusted SQL — operators must validate
  before persisting; only authenticated admin path should call put_view
- Redaction expressions assume column is castable to VARCHAR; numeric
  redactions could be misleading (a Hash on Int64 returns a hex string
  that won't equi-join with another hash on the same value type)

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

crates/catalogd/src/registry.rs

@@ -1,6 +1,6 @@
 use shared::types::{
-    DatasetId, DatasetManifest, ObjectRef, SchemaFingerprint,
-    ColumnMeta, Lineage, FreshnessContract, RefreshPolicy, Sensitivity,
+    AiView, ColumnMeta, DatasetId, DatasetManifest, FreshnessContract, Lineage, ObjectRef,
+    RefreshPolicy, SchemaFingerprint, Sensitivity,
 };
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -9,6 +9,14 @@ use tokio::sync::RwLock;
 use storaged::ops;
 use object_store::ObjectStore;
 
+/// Make a view name safe for use as an object storage key.
+/// Allows letters, digits, `_`, `-`, `.`. Anything else becomes `_`.
+fn sanitize_view_name(name: &str) -> String {
+    name.chars()
+        .map(|c| if c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.' { c } else { '_' })
+        .collect()
+}
+
 #[derive(Debug, Clone, Default, serde::Serialize)]
 pub struct MigrateBucketsReport {
     pub refs_examined: usize,
@@ -34,11 +42,14 @@ pub struct MetadataUpdate {
 }
 
 const MANIFEST_PREFIX: &str = "_catalog/manifests";
+const VIEW_PREFIX: &str = "_catalog/views";
 
 /// In-memory dataset registry backed by manifest persistence in object storage.
+/// Also tracks AiViews (Phase D) — safe projections over base datasets.
 #[derive(Clone)]
 pub struct Registry {
     datasets: Arc<RwLock<HashMap<DatasetId, DatasetManifest>>>,
+    views: Arc<RwLock<HashMap<String, AiView>>>,
     store: Arc<dyn ObjectStore>,
 }
 
@@ -46,11 +57,12 @@ impl Registry {
     pub fn new(store: Arc<dyn ObjectStore>) -> Self {
         Self {
             datasets: Arc::new(RwLock::new(HashMap::new())),
+            views: Arc::new(RwLock::new(HashMap::new())),
             store,
         }
     }
 
-    /// Rebuild in-memory index from persisted manifests on startup.
+    /// Rebuild in-memory index from persisted manifests + views on startup.
     pub async fn rebuild(&self) -> Result<usize, String> {
         let keys = ops::list(&self.store, Some(MANIFEST_PREFIX)).await?;
         let mut datasets = self.datasets.write().await;
@@ -63,6 +75,29 @@ impl Registry {
         }
         let count = datasets.len();
         tracing::info!("catalog rebuilt: {count} datasets loaded");
+
+        // Phase D: load AiView definitions alongside manifests.
+        let view_keys = ops::list(&self.store, Some(VIEW_PREFIX)).await.unwrap_or_default();
+        let mut views = self.views.write().await;
+        views.clear();
+        for key in &view_keys {
+            if !key.ends_with(".json") { continue; }
+            let data = match ops::get(&self.store, key).await {
+                Ok(d) => d,
+                Err(e) => {
+                    tracing::warn!("view '{key}': read failed: {e}");
+                    continue;
+                }
+            };
+            match serde_json::from_slice::<AiView>(&data) {
+                Ok(view) => { views.insert(view.name.clone(), view); }
+                Err(e) => tracing::warn!("view '{key}': parse failed: {e}"),
+            }
+        }
+        if !views.is_empty() {
+            tracing::info!("catalog: {} views loaded", views.len());
+        }
+
         Ok(count)
     }
 
@@ -354,6 +389,55 @@ impl Registry {
         Ok(report)
     }
 
+    // --- Phase D: AI-safe views ---
+
+    /// Create or replace a named view. Validates that the base dataset
+    /// exists and that the column whitelist is non-empty. Persists to
+    /// `_catalog/views/{name}.json`.
+    pub async fn put_view(&self, mut view: AiView) -> Result<AiView, String> {
+        if view.name.is_empty() {
+            return Err("view name is empty".into());
+        }
+        if view.columns.is_empty() {
+            return Err("view must whitelist at least one column".into());
+        }
+        // Base dataset must exist (read-side will fail anyway, fail-fast here
+        // so operators don't end up with dangling views).
+        if self.get_by_name(&view.base_dataset).await.is_none() {
+            return Err(format!(
+                "base dataset '{}' not found in catalog",
+                view.base_dataset
+            ));
+        }
+        if view.created_at.timestamp() == 0 {
+            view.created_at = chrono::Utc::now();
+        }
+
+        let key = format!("{VIEW_PREFIX}/{}.json", sanitize_view_name(&view.name));
+        let json = serde_json::to_vec_pretty(&view).map_err(|e| e.to_string())?;
+        ops::put(&self.store, &key, json.into()).await?;
+
+        let mut views = self.views.write().await;
+        views.insert(view.name.clone(), view.clone());
+        tracing::info!("view registered: {} over '{}'", view.name, view.base_dataset);
+        Ok(view)
+    }
+
+    pub async fn get_view(&self, name: &str) -> Option<AiView> {
+        self.views.read().await.get(name).cloned()
+    }
+
+    pub async fn list_views(&self) -> Vec<AiView> {
+        self.views.read().await.values().cloned().collect()
+    }
+
+    pub async fn delete_view(&self, name: &str) -> Result<(), String> {
+        let key = format!("{VIEW_PREFIX}/{}.json", sanitize_view_name(name));
+        ops::delete(&self.store, &key).await?;
+        self.views.write().await.remove(name);
+        Ok(())
+    }
+
     /// List datasets whose `embedding_stale_since` is set — they need a refresh.
     pub async fn stale_datasets(&self) -> Vec<DatasetManifest> {
         let datasets = self.datasets.read().await;