Merge remote-tracking branch 'origin/main' into test/enrich-prd-pipeline

# Conflicts:
#	auditor/audit.ts
#	auditor/checks/inference.ts
#	auditor/checks/kb_query.ts
#	auditor/claim_parser.ts
#	tests/real-world/scrum_master_pipeline.ts

Cargo.lock

@@ -46,7 +46,6 @@ dependencies = [
 name = "aibridge"
 version = "0.1.0"
 dependencies = [
- "async-trait",
  "axum",
  "reqwest",
  "serde",
@@ -8920,7 +8919,6 @@ dependencies = [
  "object_store",
  "parquet 55.2.0",
  "queryd",
- "reqwest",
  "serde",
  "serde_json",
  "sha2",

config/routing.toml

@@ -1,61 +0,0 @@
-# Phase 40: Routing Engine Configuration
-#
-# Human-editable rules for model → provider routing.
-# Matching order: first match wins.
-
-[[rule]]
-model_pattern = "gpt-4*"
-provider = "openrouter"
-max_tokens = 4096
-temperature = 0.7
-
-[[rule]]
-model_pattern = "claude*"
-provider = "claude"
-max_tokens = 4096
-temperature = 0.7
-
-[[rule]]
-model_pattern = "gemini*"
-provider = "gemini"
-max_tokens = 8192
-temperature = 0.9
-
-[[rule]]
-model_pattern = "qwen3.5*"
-provider = "ollama"
-max_tokens = 4096
-temperature = 0.3
-
-[[rule]]
-model_pattern = "qwen3*"
-provider = "ollama"
-max_tokens = 2048
-temperature = 0.3
-
-[[rule]]
-model_pattern = "gpt-oss*"
-provider = "ollama"
-temperature = 0.1
-
-[[rule]]
-model_pattern = "*"
-provider = "ollama"
-temperature = 0.5
-
-# Fallback chain: if primary fails, try these in order
-fallback = ["ollama", "openrouter"]
-
-# Cost gating (tokens = cents per 1M)
-[cost]
-ollama = 0
-openrouter = 15
-claude = 15
-gemini = 0
-
-# Daily budget per provider (cents)
-[daily_budget]
-ollama = 0
-openrouter = 1000
-claude = 500
-gemini = 0

crates/aibridge/Cargo.toml

@@ -11,4 +11,3 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 tracing = { workspace = true }
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
-async-trait = "0.1"

crates/aibridge/src/lib.rs

@@ -1,8 +1,5 @@
 pub mod client;
 pub mod context;
 pub mod continuation;
-pub mod provider;
-pub mod providers;
-pub mod routing;
 pub mod service;
 pub mod tree_split;

crates/aibridge/src/provider.rs

@@ -1,39 +0,0 @@
-use async_trait::async_trait;
-use crate::client::{GenerateRequest, GenerateResponse, EmbedRequest, EmbedResponse};
-
-#[async_trait]
-pub trait ProviderAdapter: Send + Sync {
-    /// Name for routing (ollama, openrouter, etc.)
-    fn name(&self) -> &str;
-
-    /// Chat completion — returns text, model, token counts
-    async fn chat(&self, req: GenerateRequest) -> Result<GenerateResponse, String>;
-
-    /// Embeddings — returns vectors, model, dimensions
-    async fn embed(&self, req: EmbedRequest) -> Result<EmbedResponse, String>;
-
-    /// Unload model from VRAM (optional, no-op if not supported)
-    async fn unload(&self, _model: &str) -> Result<(), String> {
-        Ok(())
-    }
-
-    /// Health check
-    async fn health(&self) -> Result<serde_json::Value, String>;
-}
-
-/// Routing key extracted from model name.
-/// - "qwen3.5:latest" → "ollama"
-/// - "openrouter/anthropic/claude-3.5-sonnet" → "openrouter"
-/// - "gpt-4o" → "ollama" (default)
-pub fn provider_key(model: &str) -> &'static str {
-    let lower = model.to_lowercase();
-    if lower.starts_with("openrouter/") {
-        "openrouter"
-    } else if lower.starts_with("gemini") {
-        "gemini"
-    } else if lower.starts_with("claude") {
-        "claude"
-    } else {
-        "ollama"  // default: local Ollama
-    }
-}

crates/aibridge/src/providers/mod.rs

@@ -1,2 +0,0 @@
-pub mod ollama;
-pub mod openrouter;

crates/aibridge/src/providers/ollama.rs

@@ -1,37 +0,0 @@
-use async_trait::async_trait;
-use crate::client::{AiClient, GenerateRequest, GenerateResponse, EmbedRequest, EmbedResponse};
-use crate::provider::ProviderAdapter;
-
-pub struct OllamaAdapter {
-    client: AiClient,
-}
-
-impl OllamaAdapter {
-    pub fn new(client: AiClient) -> Self {
-        Self { client }
-    }
-}
-
-#[async_trait]
-impl ProviderAdapter for OllamaAdapter {
-    fn name(&self) -> &str {
-        "ollama"
-    }
-
-    async fn chat(&self, req: GenerateRequest) -> Result<GenerateResponse, String> {
-        self.client.generate(req).await
-    }
-
-    async fn embed(&self, req: EmbedRequest) -> Result<EmbedResponse, String> {
-        self.client.embed(req).await
-    }
-
-    async fn unload(&self, model: &str) -> Result<(), String> {
-        let _: serde_json::Value = self.client.unload_model(model).await?;
-        Ok(())
-    }
-
-    async fn health(&self) -> Result<serde_json::Value, String> {
-        self.client.health().await
-    }
-}

crates/aibridge/src/providers/openrouter.rs

@@ -1,150 +0,0 @@
-use async_trait::async_trait;
-use reqwest::Client;
-use serde::{Deserialize, Serialize};
-use std::time::Duration;
-
-use crate::client::{GenerateRequest, GenerateResponse, EmbedRequest, EmbedResponse};
-use crate::provider::ProviderAdapter;
-
-pub struct OpenRouterAdapter {
-    client: Client,
-    base_url: String,
-    api_key: String,
-    default_model: String,
-}
-
-#[derive(Serialize)]
-struct OpenRouterChatRequest {
-    model: String,
-    messages: Vec<OpenRouterMessage>,
-    temperature: Option<f64>,
-    max_tokens: Option<u32>,
-}
-
-#[derive(Serialize)]
-struct OpenRouterMessage {
-    role: String,
-    content: String,
-}
-
-#[derive(Deserialize)]
-struct OpenRouterChatResponse {
-    choices: Vec<OpenRouterChoice>,
-    usage: OpenRouterUsage,
-    model: String,
-}
-
-#[derive(Deserialize)]
-struct OpenRouterChoice {
-    message: OpenRouterMessageOut,
-}
-
-#[derive(Deserialize)]
-struct OpenRouterMessageOut {
-    role: String,
-    content: String,
-}
-
-#[derive(Deserialize)]
-struct OpenRouterUsage {
-    prompt_tokens: Option<u32>,
-    completion_tokens: Option<u32>,
-    total_tokens: Option<u32>,
-}
-
-impl OpenRouterAdapter {
-    pub fn new(base_url: &str, api_key: String, default_model: &str) -> Self {
-        let client = Client::builder()
-            .timeout(Duration::from_secs(180))
-            .build()
-            .expect("failed to build HTTP client");
-        Self {
-            client,
-            base_url: base_url.trim_end_matches('/').to_string(),
-            api_key,
-            default_model: default_model.to_string(),
-        }
-    }
-
-    fn chat_model(&self, model: &str) -> String {
-        // Strip "openrouter/" prefix if present
-        let m = model.trim_start_matches("openrouter/");
-        if m.is_empty() || m == model {
-            self.default_model.clone()
-        } else {
-            m.to_string()
-        }
-    }
-
-    fn to_openrouter_messages(req: &GenerateRequest) -> Vec<OpenRouterMessage> {
-        let mut out = vec![];
-        if let Some(sys) = &req.system {
-            out.push(OpenRouterMessage { role: "system".into(), content: sys.clone() });
-        }
-        out.push(OpenRouterMessage {
-            role: "user".into(),
-            content: req.prompt.clone(),
-        });
-        out
-    }
-}
-
-#[async_trait]
-impl ProviderAdapter for OpenRouterAdapter {
-    fn name(&self) -> &str {
-        "openrouter"
-    }
-
-    async fn chat(&self, req: GenerateRequest) -> Result<GenerateResponse, String> {
-        let model = self.chat_model(req.model.as_deref().unwrap_or(""));
-        let or_req = OpenRouterChatRequest {
-            model: model.clone(),
-            messages: OpenRouterAdapter::to_openrouter_messages(&req),
-            temperature: req.temperature,
-            max_tokens: req.max_tokens,
-        };
-
-        let resp = self.client
-            .post(format!("{}/chat/completions", self.base_url))
-            .header("Authorization", format!("Bearer {}", self.api_key))
-            .header("Content-Type", "application/json")
-            .json(&or_req)
-            .send()
-            .await
-            .map_err(|e| format!("openrouter request failed: {e}"))?;
-
-        let status = resp.status();
-        let body = resp.text().await.unwrap_or_default();
-
-        if !status.is_success() {
-            return Err(format!("openrouter error ({}): {}", status, body));
-        }
-
-        let or_resp: OpenRouterChatResponse = serde_json::from_str(&body)
-            .map_err(|e| format!("openrouter parse error: {e}"))?;
-
-        let choice = or_resp.choices.into_iter().next()
-            .ok_or("no completion choice returned")?;
-        let usage = or_resp.usage;
-
-        Ok(GenerateResponse {
-            text: choice.message.content,
-            model: or_resp.model,
-            tokens_evaluated: usage.prompt_tokens.map(|n| n as u64),
-            tokens_generated: usage.completion_tokens.map(|n| n as u64),
-        })
-    }
-
-    async fn embed(&self, _req: EmbedRequest) -> Result<EmbedResponse, String> {
-        Err("openrouter: embed not implemented".into())
-    }
-
-    async fn health(&self) -> Result<serde_json::Value, String> {
-        // OpenRouter doesn't have a dedicated health endpoint,
-        // so we just return a healthy response if the client works.
-        Ok(serde_json::json!({
-            "status": "ok",
-            "provider": "openrouter",
-        }))
-    }
-}

crates/aibridge/src/routing.rs

@@ -1,94 +0,0 @@
-use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct RoutingRule {
-    pub model_pattern: String,
-    pub provider: String,
-    pub max_tokens: Option<u32>,
-    pub temperature: Option<f64>,
-}
-
-#[derive(Clone, Default)]
-pub struct RoutingEngine {
-    rules: Vec<RoutingRule>,
-    fallback_chain: Vec<String>,
-}
-
-impl RoutingEngine {
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    pub fn with_rules(mut self, rules: Vec<RoutingRule>) -> Self {
-        self.rules = rules;
-        self
-    }
-
-    pub fn with_fallback(mut self, chain: Vec<String>) -> Self {
-        self.fallback_chain = chain;
-        self
-    }
-
-    pub fn route(&self, model: &str) -> RouteDecision {
-        let lower = model.to_lowercase();
-
-        for rule in &self.rules {
-            if glob_match(&rule.model_pattern.to_lowercase(), &lower) {
-                return RouteDecision {
-                    provider: rule.provider.clone(),
-                    model: model.to_string(),
-                    max_tokens: rule.max_tokens,
-                    temperature: rule.temperature,
-                };
-            }
-        }
-
-        if let Some(first) = self.fallback_chain.first() {
-            RouteDecision {
-                provider: first.clone(),
-                model: model.to_string(),
-                max_tokens: None,
-                temperature: None,
-            }
-        } else {
-            RouteDecision {
-                provider: "ollama".to_string(),
-                model: model.to_string(),
-                max_tokens: None,
-                temperature: None,
-            }
-        }
-    }
-}
-
-#[derive(Clone, Debug)]
-pub struct RouteDecision {
-    pub provider: String,
-    pub model: String,
-    pub max_tokens: Option<u32>,
-    pub temperature: Option<f64>,
-}
-
-fn glob_match(pattern: &str, name: &str) -> bool {
-    if pattern.contains('*') {
-        let parts: Vec<&str> = pattern.split('*').collect();
-        if parts.len() == 2 {
-            return name.starts_with(parts[0]) && name.ends_with(parts[1]);
-        } else if parts.len() == 1 {
-            return name.starts_with(parts[0]) || name.ends_with(parts[1]);
-        }
-    }
-    pattern == name
-}
-
-impl Default for RoutingRule {
-    fn default() -> Self {
-        Self {
-            model_pattern: "*".to_string(),
-            provider: "ollama".to_string(),
-            max_tokens: None,
-            temperature: None,
-        }
-    }
-}

crates/catalogd/src/service.rs

@@ -31,10 +31,6 @@ pub fn router(registry: Registry) -> Router {
         // Phase 17: model profiles
         .route("/profiles", post(create_profile).get(list_profiles))
         .route("/profiles/{id}", get(get_profile).delete(delete_profile))
-        // Phase 41: profile-type routes
-        .route("/profiles/retrieval", get(list_retrieval_profiles))
-        .route("/profiles/memory", get(list_memory_profiles))
-        .route("/profiles/observer", get(list_observer_profiles))
         .with_state(registry)
 }
 
@@ -402,9 +398,6 @@ struct CreateProfileRequest {
     /// indexes to. Defaults to Parquet+HNSW.
     #[serde(default)]
     vector_backend: shared::types::VectorBackend,
-    /// Phase 41: Profile type for routing + activation behavior.
-    #[serde(default)]
-    profile_type: shared::types::ProfileType,
 }
 
 fn default_embed_model_req() -> String { "nomic-embed-text".to_string() }
@@ -424,7 +417,6 @@ async fn create_profile(
         created_by: req.created_by,
         bucket: req.bucket,
         vector_backend: req.vector_backend,
-        profile_type: req.profile_type,
     };
     match registry.put_profile(profile).await {
         Ok(p) => Ok((StatusCode::CREATED, Json(p))),
@@ -436,24 +428,6 @@ async fn list_profiles(State(registry): State<Registry>) -> impl IntoResponse {
     Json(registry.list_profiles().await)
 }
 
-async fn list_retrieval_profiles(State(registry): State<Registry>) -> impl IntoResponse {
-    let all = registry.list_profiles().await;
-    let retrieval: Vec<_> = all.into_iter().filter(|p| p.profile_type == shared::types::ProfileType::Retrieval).collect();
-    Json(retrieval)
-}
-
-async fn list_memory_profiles(State(registry): State<Registry>) -> impl IntoResponse {
-    let all = registry.list_profiles().await;
-    let memory: Vec<_> = all.into_iter().filter(|p| p.profile_type == shared::types::ProfileType::Memory).collect();
-    Json(memory)
-}
-
-async fn list_observer_profiles(State(registry): State<Registry>) -> impl IntoResponse {
-    let all = registry.list_profiles().await;
-    let observer: Vec<_> = all.into_iter().filter(|p| p.profile_type == shared::types::ProfileType::Observer).collect();
-    Json(observer)
-}
-
 async fn get_profile(
     State(registry): State<Registry>,
     Path(id): Path<String>,

crates/gateway/src/v1/mod.rs

@@ -24,7 +24,7 @@ use axum::{
     Json,
 };
 use serde::{Deserialize, Serialize};
-use std::{collections::HashMap, sync::Arc};
+use std::sync::Arc;
 use tokio::sync::RwLock;
 
 #[derive(Clone)]
@@ -46,16 +46,6 @@ pub struct Usage {
     pub prompt_tokens: u64,
     pub completion_tokens: u64,
     pub total_tokens: u64,
-    #[serde(default)]
-    pub by_provider: std::collections::HashMap<String, ProviderUsage>,
-}
-
-#[derive(Default, Clone, Serialize)]
-pub struct ProviderUsage {
-    pub requests: u64,
-    pub prompt_tokens: u64,
-    pub completion_tokens: u64,
-    pub total_tokens: u64,
 }
 
 pub fn router(state: V1State) -> Router {
@@ -145,22 +135,18 @@ async fn chat(
     let start_time = chrono::Utc::now();
     let start_instant = std::time::Instant::now();
 
-    let (resp, used_provider) = match provider.as_str() {
+    let resp = match provider.as_str() {
-        "ollama" | "local" | "" => {
+        "ollama" | "local" | "" => ollama::chat(&state.ai_client, &req)
-            let r = ollama::chat(&state.ai_client, &req)
+            .await
-                .await
+            .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama local: {e}")))?,
-                .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama local: {e}")))?;
-            (r, "ollama".to_string())
-        }
         "ollama_cloud" | "cloud" => {
             let key = state.ollama_cloud_key.as_deref().ok_or((
                 StatusCode::SERVICE_UNAVAILABLE,
-                "OLLAMA_CLOUD_KEY not configured".to_string(),
+                "OLLAMA_CLOUD_KEY not configured — set env or populate /root/llm_team_config.json".to_string(),
             ))?;
-            let r = ollama_cloud::chat(key, &req)
+            ollama_cloud::chat(key, &req)
                 .await
-                .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama cloud: {e}")))?;
+                .map_err(|e| (StatusCode::BAD_GATEWAY, format!("ollama cloud: {e}")))?
-            (r, "ollama_cloud".to_string())
         }
         other => {
             return Err((
@@ -182,7 +168,7 @@ async fn chat(
             .map(|c| c.message.content.clone())
             .unwrap_or_default();
         lf.emit_chat(langfuse_trace::ChatTrace {
-            provider: used_provider.clone(),
+            provider: provider.clone(),
             model: resp.model.clone(),
             input: req.messages.clone(),
             output,
@@ -197,19 +183,12 @@ async fn chat(
         });
     }
 
-    // Phase 40: per-provider usage tracking
     {
         let mut u = state.usage.write().await;
         u.requests += 1;
         u.prompt_tokens += resp.usage.prompt_tokens as u64;
         u.completion_tokens += resp.usage.completion_tokens as u64;
         u.total_tokens += resp.usage.total_tokens as u64;
-
-        let provider_usage = u.by_provider.entry(used_provider).or_default();
-        provider_usage.requests += 1;
-        provider_usage.prompt_tokens += resp.usage.prompt_tokens as u64;
-        provider_usage.completion_tokens += resp.usage.completion_tokens as u64;
-        provider_usage.total_tokens += resp.usage.total_tokens as u64;
     }
 
     Ok(Json(resp))

crates/shared/src/types.rs

@@ -269,20 +269,6 @@ pub struct ModelProfile {
     /// / 5M+ vector workloads.
     #[serde(default)]
     pub vector_backend: VectorBackend,
-    /// Phase 41: Profile type for routing + activation behavior.
-    /// Default Execution for backward compatibility.
-    #[serde(default)]
-    pub profile_type: ProfileType,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, Default)]
-#[serde(rename_all = "snake_case")]
-pub enum ProfileType {
-    #[default]
-    Execution,
-    Retrieval,
-    Memory,
-    Observer,
 }
 
 fn default_embed_model() -> String { "nomic-embed-text".to_string() }

crates/vectord/src/jobs.rs

@@ -8,13 +8,6 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::RwLock;
 
-#[derive(Debug, Clone, Serialize)]
-#[serde(rename_all = "lowercase")]
-pub enum JobType {
-    Embed,
-    ProfileActivation,
-}
-
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "lowercase")]
 pub enum JobStatus {
@@ -26,61 +19,24 @@ pub enum JobStatus {
 #[derive(Debug, Clone, Serialize)]
 pub struct Job {
     pub id: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub job_type: Option<JobType>,
     pub status: JobStatus,
-    pub index_name: Option<String>,
+    pub index_name: String,
-    pub profile_id: Option<String>,
     pub total_chunks: usize,
+    /// How many chunks have been embedded so far.
+    /// Also serialized as "processed" for backward-compat with monitoring scripts.
+    pub embedded_chunks: usize,
     #[serde(rename = "processed")]
     pub processed_alias: usize,
+    pub total: usize,
     pub progress_pct: f32,
     pub storage_key: Option<String>,
     pub error: Option<String>,
     pub started_at: String,
     pub completed_at: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none")]
+    pub chunks_per_sec: f32,
-    pub result: Option<serde_json::Value>,
-}
-
-impl Job {
-    pub fn new_embed(index_name: &str, total_chunks: usize) -> Self {
-        Self {
-            id: format!("job-{}", chrono::Utc::now().timestamp_millis()),
-            job_type: Some(JobType::Embed),
-            status: JobStatus::Running,
-            index_name: Some(index_name.to_string()),
-            profile_id: None,
-            total_chunks,
-            processed_alias: 0,
-            progress_pct: 0.0,
-            storage_key: None,
-            error: None,
-            started_at: chrono::Utc::now().to_rfc3339(),
-            completed_at: None,
-            result: None,
-        }
-    }
-
-    pub fn new_profile_activation(profile_id: &str) -> Self {
-        Self {
-            id: format!("job-{}", chrono::Utc::now().timestamp_millis()),
-            job_type: Some(JobType::ProfileActivation),
-            status: JobStatus::Running,
-            index_name: None,
-            profile_id: Some(profile_id.to_string()),
-            total_chunks: 0,
-            processed_alias: 0,
-            progress_pct: 0.0,
-            storage_key: None,
-            error: None,
-            started_at: chrono::Utc::now().to_rfc3339(),
-            completed_at: None,
-            result: None,
-        }
-    }
 }
 
+/// Shared progress tracker that background tasks update.
 #[derive(Clone)]
 pub struct JobTracker {
     jobs: Arc<RwLock<HashMap<String, Job>>>,
@@ -93,42 +49,57 @@ impl JobTracker {
         }
     }
 
-    pub async fn create_embed(&self, index_name: &str, total_chunks: usize) -> String {
+    /// Register a new job. Returns the job ID.
-        let job = Job::new_embed(index_name, total_chunks);
+    pub async fn create(&self, index_name: &str, total_chunks: usize) -> String {
-        let id = job.id.clone();
+        let id = format!("job-{}", chrono::Utc::now().timestamp_millis());
+        let job = Job {
+            id: id.clone(),
+            status: JobStatus::Running,
+            index_name: index_name.to_string(),
+            total_chunks,
+            embedded_chunks: 0,
+            processed_alias: 0,
+            total: total_chunks,
+            progress_pct: 0.0,
+            storage_key: None,
+            error: None,
+            started_at: chrono::Utc::now().to_rfc3339(),
+            completed_at: None,
+            chunks_per_sec: 0.0,
+        };
         self.jobs.write().await.insert(id.clone(), job);
         id
     }
 
-    pub async fn create_profile_activation(&self, profile_id: &str) -> String {
+    /// Update progress.
-        let job = Job::new_profile_activation(profile_id);
+    pub async fn update_progress(&self, id: &str, embedded: usize, rate: f32) {
-        let id = job.id.clone();
-        self.jobs.write().await.insert(id.clone(), job);
-        id
-    }
-
-    pub async fn update_embed_progress(&self, id: &str, embedded: usize, _rate: f32) {
         let mut jobs = self.jobs.write().await;
         if let Some(job) = jobs.get_mut(id) {
-            job.processed_alias = embedded;
+            job.embedded_chunks = embedded;
+            job.processed_alias = embedded;  // keep alias in sync
             job.progress_pct = if job.total_chunks > 0 {
                 (embedded as f32 / job.total_chunks as f32) * 100.0
             } else {
                 0.0
             };
+            job.chunks_per_sec = rate;
         }
     }
 
-    pub async fn complete(&self, id: &str, result: Option<serde_json::Value>) {
+    /// Mark job as completed.
+    pub async fn complete(&self, id: &str, storage_key: String) {
         let mut jobs = self.jobs.write().await;
         if let Some(job) = jobs.get_mut(id) {
             job.status = JobStatus::Completed;
+            job.embedded_chunks = job.total_chunks;
+            job.processed_alias = job.total_chunks;
             job.progress_pct = 100.0;
+            job.storage_key = Some(storage_key);
             job.completed_at = Some(chrono::Utc::now().to_rfc3339());
-            job.result = result;
         }
     }
 
+    /// Mark job as failed.
     pub async fn fail(&self, id: &str, error: String) {
         let mut jobs = self.jobs.write().await;
         if let Some(job) = jobs.get_mut(id) {
@@ -138,11 +109,13 @@ impl JobTracker {
         }
     }
 
+    /// Get job status.
     pub async fn get(&self, id: &str) -> Option<Job> {
         self.jobs.read().await.get(id).cloned()
     }
 
+    /// List all jobs.
     pub async fn list(&self) -> Vec<Job> {
         self.jobs.read().await.values().cloned().collect()
     }
-}
+}

crates/vectord/src/service.rs

@@ -7,7 +7,6 @@ use axum::{
 };
 use object_store::ObjectStore;
 use serde::{Deserialize, Serialize};
-use serde_json::json;
 use std::sync::Arc;
 
 use aibridge::client::{AiClient, EmbedRequest, GenerateRequest};
@@ -197,7 +196,7 @@ async fn create_index(
     let bucket = req.bucket.clone().unwrap_or_else(|| "primary".to_string());
 
     // Create job and return immediately
-    let job_id = state.job_tracker.create_embed(&index_name, n_chunks).await;
+    let job_id = state.job_tracker.create(&index_name, n_chunks).await;
     tracing::info!("job {job_id}: indexing '{}' — {} docs → {} chunks (background)", index_name, n_docs, n_chunks);
 
     // Spawn supervised dual-pipeline embedding
@@ -241,7 +240,7 @@ async fn create_index(
                 };
                 let _ = registry.register(meta).await;
 
-                tracker.complete(&jid, Some(json!({ "storage_key": key }))).await;
+                tracker.complete(&jid, key).await;
                 tracing::info!("job {jid}: completed — {n_chunks} chunks in {elapsed:.0}s ({rate:.0}/sec)");
             }
             Err(e) => {
@@ -483,7 +482,7 @@ async fn _run_embedding_job_legacy(
         // Update progress
         let elapsed = start.elapsed().as_secs_f32();
         let rate = if elapsed > 0.0 { all_vectors.len() as f32 / elapsed } else { 0.0 };
-        tracker.update_embed_progress(job_id, all_vectors.len(), rate).await;
+        tracker.update_progress(job_id, all_vectors.len(), rate).await;
 
         // Log every 100 batches
         if (i + 1) % 100 == 0 {
@@ -1372,218 +1371,239 @@ async fn activate_profile(
     State(state): State<VectorState>,
     Path(profile_id): Path<String>,
 ) -> impl IntoResponse {
-    tracing::info!("[activate_profile] START profile_id={}", profile_id);
+    let t0 = std::time::Instant::now();
+
     let profile = match state.catalog.get_profile(&profile_id).await {
         Some(p) => p,
         None => return Err((StatusCode::NOT_FOUND, format!("profile not found: {profile_id}"))),
     };
 
-    let job_id = state.job_tracker.create_profile_activation(&profile_id).await;
+    let mut warmed = Vec::new();
-    let job_id_for_response = job_id.clone();
+    let mut failures = Vec::new();
-    let tracker = state.job_tracker.clone();
+    let mut total_vectors = 0usize;
-    let catalog = state.catalog.clone();
-    let index_registry = state.index_registry.clone();
-    let bucket_registry = state.bucket_registry.clone();
-    let lance = state.lance.clone();
-    let embedding_cache = state.embedding_cache.clone();
-    let hnsw_store = state.hnsw_store.clone();
-    let promotion_registry = state.promotion_registry.clone();
-    let ai_client = state.ai_client.clone();
-    let active_profile = state.active_profile.clone();
-    let profile_name = profile.ollama_name.clone();
-    let profile_id_clone = profile.id.clone();
-    let profile_bucket = profile.bucket.clone();
-    let profile_bound = profile.bound_datasets.clone();
-    let profile_hnsw = profile.hnsw_config.clone();
-    let profile_backend = profile.vector_backend.clone();
-    let profile_full = profile.clone();
 
-    tokio::spawn(async move {
+    // Phase 17 / C: VRAM-aware swap. If another profile currently holds
-        let t0 = std::time::Instant::now();
+    // the GPU and uses a DIFFERENT Ollama model than the one being
-        let mut warmed = Vec::new();
+    // activated, unload it first (keep_alive=0). Same-model activations
-        let mut failures = Vec::new();
+    // skip the unload — no point churning a model that's already loaded.
-        let mut total_vectors = 0usize;
+    let previous_slot = {
-        let job_id = job_id;
+        let guard = state.active_profile.read().await;
+        guard.clone()
+    };
+    if let Some(prev) = &previous_slot {
+        if prev.ollama_name != profile.ollama_name {
+            match state.ai_client.unload_model(&prev.ollama_name).await {
+                Ok(_) => tracing::info!(
+                    "profile swap: unloaded '{}' ({} -> {})",
+                    prev.ollama_name, prev.profile_id, profile.id,
+                ),
+                Err(e) => failures.push(format!(
+                    "unload previous model '{}': {e}", prev.ollama_name,
+                )),
+            }
+        }
+    }
 
-        let previous_slot = {
+    // Federation layer 2: if this profile declares its own bucket and
-            let guard = active_profile.read().await;
+    // that bucket isn't registered yet, auto-provision it under the
-            guard.clone()
+    // configured profile_root. This is the moment a "dormant" profile
-        };
+    // becomes live — its bucket exists and is readable/writable.
-        if let Some(prev) = &previous_slot {
+    if let Some(bucket_name) = profile.bucket.clone() {
-            if prev.ollama_name != profile_name {
+        if !state.bucket_registry.contains(&bucket_name) {
-                match ai_client.unload_model(&prev.ollama_name).await {
+            let root = format!(
-                    Ok(_) => tracing::info!(
+                "{}/{}",
-                        "profile swap: unloaded '{}' ({} -> {})",
+                state.bucket_registry.profile_root().trim_end_matches('/'),
-                        prev.ollama_name, prev.profile_id, profile_id_clone,
+                bucket_name.replace(':', "_"),
-                    ),
+            );
-                    Err(e) => failures.push(format!("unload previous model '{}': {e}", prev.ollama_name)),
+            let bc = shared::config::BucketConfig {
+                name: bucket_name.clone(),
+                backend: "local".to_string(),
+                root: Some(root.clone()),
+                bucket: None,
+                region: None,
+                endpoint: None,
+                secret_ref: None,
+            };
+            match state.bucket_registry.add_bucket(bc).await {
+                Ok(info) => {
+                    tracing::info!(
+                        "profile '{}' activated bucket '{}' (root={}, reachable={})",
+                        profile.id, bucket_name, root, info.reachable,
+                    );
+                }
+                Err(e) => {
+                    failures.push(format!(
+                        "auto-provision bucket '{}': {}", bucket_name, e,
+                    ));
                 }
             }
         }
+    }
 
-        if let Some(bucket_name) = profile_bucket.clone() {
+    let all_indexes = state.index_registry.list(None, None).await;
-            if !bucket_registry.contains(&bucket_name) {
+    let use_lance = profile.vector_backend == shared::types::VectorBackend::Lance;
-                let root = format!(
+
-                    "{}/{}",
+    for binding in &profile.bound_datasets {
-                    bucket_registry.profile_root().trim_end_matches('/'),
+        let matched: Vec<_> = all_indexes
-                    bucket_name.replace(':', "_"),
+            .iter()
-                );
+            .filter(|m| &m.source == binding)
-                let bc = shared::config::BucketConfig {
+            .collect();
-                    name: bucket_name.clone(),
+        if matched.is_empty() {
-                    backend: "local".to_string(),
+            failures.push(format!(
-                    root: Some(root.clone()),
+                "no vector index found for binding '{}'", binding,
-                    bucket: None,
+            ));
-                    region: None,
+            continue;
-                    endpoint: None,
+        }
-                    secret_ref: None,
+        for meta in matched {
-                };
+            if use_lance {
-                match bucket_registry.add_bucket(bc).await {
+                // --- Lance activation path ---
-                    Ok(info) => {
+                // Ensure a Lance dataset exists for this index. If it
-                        tracing::info!(
+                // doesn't, auto-migrate from the Parquet blob. Then
-                            "profile '{}' activated bucket '{}' (root={}, reachable={})",
+                // ensure an IVF_PQ index is built.
-                            profile_id_clone, bucket_name, root, info.reachable,
+                let bucket = meta.bucket.clone();
-                        );
+                let lance_store = match state.lance.store_for_new(&meta.index_name, &bucket).await {
+                    Ok(s) => s,
+                    Err(e) => {
+                        failures.push(format!("{}: lance store init: {e}", meta.index_name));
+                        continue;
                     }
-                    Err(e) => failures.push(format!("auto-provision bucket '{}': {}", bucket_name, e)),
+                };
-                }
+                let count = lance_store.count().await.unwrap_or(0);
-            }
+                if count == 0 {
-        }
+                    // Auto-migrate from existing Parquet.
-
+                    let pq_store = match state.bucket_registry.get(&bucket) {
-        let all_indexes = index_registry.list(None, None).await;
-        let use_lance = profile_backend == shared::types::VectorBackend::Lance;
-
-        for binding in &profile_bound {
-            let matched: Vec<_> = all_indexes
-                .iter()
-                .filter(|m| &m.source == binding)
-                .collect();
-            if matched.is_empty() {
-                failures.push(format!("no vector index found for binding '{}'", binding));
-                continue;
-            }
-            for meta in matched {
-                if use_lance {
-                    let bucket = meta.bucket.clone();
-                    let lance_store = match lance.store_for_new(&meta.index_name, &bucket).await {
                         Ok(s) => s,
-                        Err(e) => { failures.push(format!("{}: lance store init: {e}", meta.index_name)); continue; }
+                        Err(e) => { failures.push(format!("{}: bucket: {e}", meta.index_name)); continue; }
                     };
-                    let count = lance_store.count().await.unwrap_or(0);
+                    match storaged::ops::get(&pq_store, &meta.storage_key).await {
-                    if count == 0 {
+                        Ok(bytes) => {
-                        let pq_store = match bucket_registry.get(&bucket) {
+                            let build_t = std::time::Instant::now();
-                            Ok(s) => s,
+                            match lance_store.migrate_from_parquet_bytes(&bytes).await {
-                            Err(e) => { failures.push(format!("{}: bucket: {e}", meta.index_name)); continue; }
+                                Ok(ms) => {
-                        };
+                                    total_vectors += ms.rows_written;
-                        match storaged::ops::get(&pq_store, &meta.storage_key).await {
+                                    tracing::info!(
-                            Ok(bytes) => {
+                                        "lance auto-migrate '{}': {} rows in {:.2}s",
-                                let build_t = std::time::Instant::now();
+                                        meta.index_name, ms.rows_written, ms.duration_secs,
-                                match lance_store.migrate_from_parquet_bytes(&bytes).await {
+                                    );
-                                    Ok(ms) => {
+                                    warmed.push(WarmedIndex {
-                                        total_vectors += ms.rows_written;
+                                        index_name: meta.index_name.clone(),
-                                        tracing::info!("lance auto-migrate '{}': {} rows in {:.2}s", meta.index_name, ms.rows_written, ms.duration_secs);
+                                        source: meta.source.clone(),
-                                        warmed.push(WarmedIndex {
+                                        vectors: ms.rows_written,
-                                            index_name: meta.index_name.clone(),
+                                        hnsw_build_secs: build_t.elapsed().as_secs_f32(),
-                                            source: meta.source.clone(),
+                                    });
-                                            vectors: ms.rows_written,
-                                            hnsw_build_secs: build_t.elapsed().as_secs_f32(),
-                                        });
-                                    }
-                                    Err(e) => failures.push(format!("{}: lance migrate: {e}", meta.index_name)),
                                 }
+                                Err(e) => failures.push(format!("{}: lance migrate: {e}", meta.index_name)),
                             }
-                            Err(e) => failures.push(format!("{}: read parquet: {e}", meta.index_name)),
                         }
-                    } else {
+                        Err(e) => failures.push(format!("{}: read parquet: {e}", meta.index_name)),
-                        total_vectors += count;
+                    }
+                } else {
+                    total_vectors += count;
+                    warmed.push(WarmedIndex {
+                        index_name: meta.index_name.clone(),
+                        source: meta.source.clone(),
+                        vectors: count,
+                        hnsw_build_secs: 0.0,
+                    });
+                }
+                // Ensure IVF_PQ vector index exists.
+                if !lance_store.has_vector_index().await.unwrap_or(false) {
+                    match lance_store.build_index(316, 8, 48).await {
+                        Ok(ix) => tracing::info!(
+                            "lance auto-index '{}': IVF_PQ built in {:.1}s",
+                            meta.index_name, ix.build_time_secs,
+                        ),
+                        Err(e) => failures.push(format!("{}: lance IVF_PQ build: {e}", meta.index_name)),
+                    }
+                }
+                // Ensure scalar btree on doc_id for O(log N) random fetch.
+                if !lance_store.has_scalar_index("doc_id").await.unwrap_or(false) {
+                    match lance_store.build_scalar_index("doc_id").await {
+                        Ok(ix) => tracing::info!(
+                            "lance auto-index '{}': doc_id btree built in {:.2}s",
+                            meta.index_name, ix.build_time_secs,
+                        ),
+                        Err(e) => failures.push(format!("{}: lance doc_id btree: {e}", meta.index_name)),
+                    }
+                }
+            } else {
+                // --- Parquet + HNSW activation path (existing) ---
+                let embeddings = match state.embedding_cache.get_or_load(&meta.index_name).await {
+                    Ok(arc) => arc,
+                    Err(e) => {
+                        failures.push(format!("{}: load failed: {}", meta.index_name, e));
+                        continue;
+                    }
+                };
+                total_vectors += embeddings.len();
+
+                let profile_default = trial::HnswConfig {
+                    ef_construction: profile.hnsw_config.ef_construction,
+                    ef_search: profile.hnsw_config.ef_search,
+                    seed: profile.hnsw_config.seed,
+                };
+                let cfg = state
+                    .promotion_registry
+                    .config_or(&meta.index_name, profile_default)
+                    .await;
+                let build_t = std::time::Instant::now();
+                match state
+                    .hnsw_store
+                    .build_index_with_config(&meta.index_name, (*embeddings).clone(), &cfg)
+                    .await
+                {
+                    Ok(_) => {
                         warmed.push(WarmedIndex {
                             index_name: meta.index_name.clone(),
                             source: meta.source.clone(),
-                            vectors: count,
+                            vectors: embeddings.len(),
-                            hnsw_build_secs: 0.0,
+                            hnsw_build_secs: build_t.elapsed().as_secs_f32(),
                         });
                     }
-                    if !lance_store.has_vector_index().await.unwrap_or(false) {
+                    Err(e) => {
-                        match lance_store.build_index(316, 8, 48).await {
+                        failures.push(format!("{}: HNSW build failed: {}", meta.index_name, e));
-                            Ok(ix) => tracing::info!("lance auto-index '{}': IVF_PQ built in {:.1}s", meta.index_name, ix.build_time_secs),
-                            Err(e) => failures.push(format!("{}: lance IVF_PQ build: {e}", meta.index_name)),
-                        }
-                    }
-                    if !lance_store.has_scalar_index("doc_id").await.unwrap_or(false) {
-                        match lance_store.build_scalar_index("doc_id").await {
-                            Ok(ix) => tracing::info!("lance auto-index '{}': doc_id btree built in {:.2}s", meta.index_name, ix.build_time_secs),
-                            Err(e) => failures.push(format!("{}: lance doc_id btree: {e}", meta.index_name)),
-                        }
-                    }
-                } else {
-                    let embeddings = match embedding_cache.get_or_load(&meta.index_name).await {
-                        Ok(arc) => arc,
-                        Err(e) => { failures.push(format!("{}: load failed: {}", meta.index_name, e)); continue; }
-                    };
-                    total_vectors += embeddings.len();
-
-                    let profile_default = trial::HnswConfig {
-                        ef_construction: profile_hnsw.ef_construction,
-                        ef_search: profile_hnsw.ef_search,
-                        seed: profile_hnsw.seed,
-                    };
-                    let cfg = promotion_registry
-                        .config_or(&meta.index_name, profile_default)
-                        .await;
-                    let build_t = std::time::Instant::now();
-                    match hnsw_store
-                        .build_index_with_config(&meta.index_name, (*embeddings).clone(), &cfg)
-                        .await
-                    {
-                        Ok(_) => {
-                            warmed.push(WarmedIndex {
-                                index_name: meta.index_name.clone(),
-                                source: meta.source.clone(),
-                                vectors: embeddings.len(),
-                                hnsw_build_secs: build_t.elapsed().as_secs_f32(),
-                            });
-                        }
-                        Err(e) => failures.push(format!("{}: HNSW build failed: {}", meta.index_name, e)),
                     }
                 }
             }
         }
+    }
 
-        let mut model_preloaded = false;
+    // Preload the new profile's Ollama model proactively. Same-model
-        match ai_client.preload_model(&profile_name).await {
+    // re-activations are cheap (Ollama no-ops if already loaded).
-            Ok(_) => {
+    let mut model_preloaded = false;
-                model_preloaded = true;
+    match state.ai_client.preload_model(&profile.ollama_name).await {
-                tracing::info!("profile '{}' preloaded ollama model '{}'", profile_id_clone, profile_name);
+        Ok(_) => {
-            }
+            model_preloaded = true;
-            Err(e) => failures.push(format!("preload ollama model '{}': {e}", profile_name)),
+            tracing::info!("profile '{}' preloaded ollama model '{}'",
+                profile.id, profile.ollama_name);
         }
+        Err(e) => failures.push(format!(
+            "preload ollama model '{}': {e}", profile.ollama_name,
+        )),
+    }
 
-        {
+    // Take the GPU slot.
-            let mut guard = active_profile.write().await;
+    {
-            *guard = Some(ActiveProfileSlot {
+        let mut guard = state.active_profile.write().await;
-                profile_id: profile_id_clone.clone(),
+        *guard = Some(ActiveProfileSlot {
-                ollama_name: profile_name.clone(),
+            profile_id: profile.id.clone(),
-                activated_at: chrono::Utc::now(),
+            ollama_name: profile.ollama_name.clone(),
-            });
+            activated_at: chrono::Utc::now(),
-        }
+        });
+    }
 
-        let result = serde_json::to_value(ActivateReport {
+    let report = ActivateReport {
-            profile_id: profile_id_clone,
+        profile_id: profile.id,
-            ollama_name: profile_name,
+        ollama_name: profile.ollama_name,
-            indexes_warmed: warmed,
+        indexes_warmed: warmed,
-            failures,
+        failures,
-            total_vectors,
+        total_vectors,
-            duration_secs: t0.elapsed().as_secs_f32(),
+        duration_secs: t0.elapsed().as_secs_f32(),
-            model_preloaded,
+        model_preloaded,
-            previous_profile: previous_slot.map(|s| s.profile_id),
+        previous_profile: previous_slot.map(|s| s.profile_id),
-        }).ok();
+    };
 
-        tracker.complete(&job_id, result).await;
+    Ok(Json(report))
-    });
-
-    Ok(Json(json!({
-        "job_id": job_id_for_response,
-        "message": format!("profile activation started — poll /vectors/jobs/{} for progress", job_id_for_response),
-    })))
 }
 
 /// Unload this profile's model and clear the active slot. No-op if the

crates/vectord/src/supervisor.rs

@@ -170,7 +170,7 @@ pub async fn run_supervised(
                         // Update job tracker
                         let elapsed = start_time.elapsed().as_secs_f32();
                         let rate = if elapsed > 0.0 { done as f32 / elapsed } else { 0.0 };
-                        tracker.update_embed_progress(&jid, done, rate).await;
+                        tracker.update_progress(&jid, done, rate).await;
 
                         // Update checkpoint
                         let mut ckpt = ckpt.write().await;

docs/PHASES.md

@@ -324,31 +324,6 @@
   - Mem0-style upsert: `/seed` with `append=true` (default) routes through `upsert_entry`, which decides ADD / UPDATE / NOOP on (operation, day, city, state). Same-day re-seed merges names (union, stable order) instead of duplicating the row. Identical re-seed is a no-op. Different-day same-operation is a fresh ADD. Playbook_id stays stable on UPDATE so prior citations remain valid.
   - Letta-style hot cache: `PlaybookMemory` now holds a `geo_index: HashMap<(city_lower, state_upper), Vec<entry_idx>>` rebuilt on every mutation. Geo-filtered boost queries skip the full scan and hit the O(1) key lookup. At 1.9K entries the full scan was sub-ms; the index scales the same path to 100K+ without code changes.
   - `UpsertOutcome` enum reported back to callers — `{mode: added|updated|noop, playbook_id, merged_names?}` + `entries_after`.
-- [x] **Phase 37: Hot-swap async** (2026-04-22)
-  - Extended `JobTracker` with `JobType::ProfileActivation` + `Embed` enum variants
-  - Made `activate_profile` return immediately with `job_id`, work runs in background via `tokio::spawn`
-  - Background jobs tracked via `POST /vectors/jobs/{id}` + `GET /vectors/jobs`
-- [x] **Phase 38: Universal API Skeleton** (2026-04-23)
-  - `/v1/chat` — OpenAI-compatible POST, forwards to local Ollama or Ollama Cloud
-  - `/v1/usage` — returns `{requests, prompt_tokens, completion_tokens, total_tokens}`
-  - `/v1/sessions` — returns `{data: [], note: "Phase 38: stateless"}`
-  - Langfuse trace integration (fire-and-forget, Phase 40 early)
-  - 12 unit tests green, curl gates pass
-- [x] **Phase 39: Provider Adapter Refactor** (2026-04-23)
-  - `ProviderAdapter` trait with `chat()` + `embed()` + `unload()` + `health()`
-  - `OllamaAdapter` — wraps existing AiClient
-  - `OpenRouterAdapter` — HTTP client to openrouter.ai
-  - `provider_key()` routing by model prefix (openrouter/* → OpenRouter)
-- [x] **Phase 40: Routing & Policy Engine** (2026-04-23)
-  - `RoutingEngine` with `RouteDecision` in aibridge::routing
-  - `config/routing.toml` — rules by model_pattern, fallback chain, cost gating
-  - Per-provider usage tracking: Usage.by_provider
-  - 12 gateway tests green, curl gates pass
-- [x] **Phase 41: Profile System Expansion** (2026-04-23)
-  - `ProfileType` enum: Execution, Retrieval, Memory, Observer
-  - Per-type endpoints: `/profiles/retrieval`, `/profiles/memory`, `/profiles/observer`
-  - `profile_type` field on ModelProfile
-  - Guard fix: automated scrumaudit.py finds real issues
 - [ ] Fine-tuned domain models (Phase 25+)
 - [ ] Multi-node query distribution (only if ceilings bite)
 

tests/multi-agent/scenarios/manifest.json

@@ -1,126 +1,126 @@
 {
   "count": 20,
-  "seed": 42,
+  "seed": 1337,
   "scenarios": [
     {
-      "file": "scen_000_Heritage_Foods_Indianapolis.json",
+      "file": "scen_000_Great_Lakes_Mfg_Cincinnati.json",
-      "client": "Heritage Foods",
-      "city": "Indianapolis",
-      "events": 5
-    },
-    {
-      "file": "scen_001_Great_Lakes_Mfg_Madison.json",
       "client": "Great Lakes Mfg",
-      "city": "Madison",
+      "city": "Cincinnati",
-      "events": 2
-    },
-    {
-      "file": "scen_002_Vanguard_Components_Lexington.json",
-      "client": "Vanguard Components",
-      "city": "Lexington",
-      "events": 2
-    },
-    {
-      "file": "scen_003_Cornerstone_Fabrication_Fort_Wayne.json",
-      "client": "Cornerstone Fabrication",
-      "city": "Fort Wayne",
       "events": 4
     },
     {
-      "file": "scen_004_Horizon_Supply_Louisville.json",
+      "file": "scen_001_Parallel_Machining_Joliet.json",
-      "client": "Horizon Supply",
+      "client": "Parallel Machining",
-      "city": "Louisville",
+      "city": "Joliet",
-      "events": 3
-    },
-    {
-      "file": "scen_005_Summit_Industrial_Akron.json",
-      "client": "Summit Industrial",
-      "city": "Akron",
       "events": 2
     },
     {
-      "file": "scen_006_Centennial_Packaging_Flint.json",
+      "file": "scen_002_Summit_Industrial_Cincinnati.json",
-      "client": "Centennial Packaging",
+      "client": "Summit Industrial",
-      "city": "Flint",
+      "city": "Cincinnati",
       "events": 3
     },
     {
-      "file": "scen_007_Pioneer_Assembly_Grand_Rapids.json",
+      "file": "scen_003_Pioneer_Assembly_Chicago.json",
       "client": "Pioneer Assembly",
-      "city": "Grand Rapids",
+      "city": "Chicago",
       "events": 1
     },
     {
-      "file": "scen_008_Cornerstone_Fabrication_Grand_Rapids.json",
+      "file": "scen_004_Midway_Distribution_Columbus.json",
-      "client": "Cornerstone Fabrication",
-      "city": "Grand Rapids",
-      "events": 3
-    },
-    {
-      "file": "scen_009_Midway_Distribution_Fort_Wayne.json",
       "client": "Midway Distribution",
-      "city": "Fort Wayne",
+      "city": "Columbus",
+      "events": 2
+    },
+    {
+      "file": "scen_005_Apex_Warehouse_Cleveland.json",
+      "client": "Apex Warehouse",
+      "city": "Cleveland",
       "events": 3
     },
     {
-      "file": "scen_010_Keystone_Plastics_Lexington.json",
+      "file": "scen_006_Pioneer_Assembly_Flint.json",
-      "client": "Keystone Plastics",
+      "client": "Pioneer Assembly",
-      "city": "Lexington",
+      "city": "Flint",
       "events": 5
     },
     {
-      "file": "scen_011_Cornerstone_Fabrication_Toledo.json",
+      "file": "scen_007_Riverfront_Steel_Toledo.json",
-      "client": "Cornerstone Fabrication",
+      "client": "Riverfront Steel",
       "city": "Toledo",
       "events": 3
     },
     {
-      "file": "scen_012_Heritage_Foods_Gary.json",
+      "file": "scen_008_Northland_Logistics_Indianapolis.json",
-      "client": "Heritage Foods",
+      "client": "Northland Logistics",
-      "city": "Gary",
+      "city": "Indianapolis",
-      "events": 3
-    },
-    {
-      "file": "scen_013_Riverfront_Steel_Columbus.json",
-      "client": "Riverfront Steel",
-      "city": "Columbus",
-      "events": 3
-    },
-    {
-      "file": "scen_014_Keystone_Plastics_Cincinnati.json",
-      "client": "Keystone Plastics",
-      "city": "Cincinnati",
-      "events": 2
-    },
-    {
-      "file": "scen_015_Beacon_Freight_Detroit.json",
-      "client": "Beacon Freight",
-      "city": "Detroit",
       "events": 4
     },
     {
-      "file": "scen_016_Parallel_Machining_Grand_Rapids.json",
+      "file": "scen_009_Parallel_Machining_Flint.json",
       "client": "Parallel Machining",
-      "city": "Grand Rapids",
+      "city": "Flint",
       "events": 3
     },
     {
-      "file": "scen_017_Parallel_Machining_Gary.json",
+      "file": "scen_010_Northland_Logistics_Chicago.json",
-      "client": "Parallel Machining",
+      "client": "Northland Logistics",
-      "city": "Gary",
+      "city": "Chicago",
-      "events": 3
-    },
-    {
-      "file": "scen_018_Cornerstone_Fabrication_Louisville.json",
-      "client": "Cornerstone Fabrication",
-      "city": "Louisville",
-      "events": 5
-    },
-    {
-      "file": "scen_019_Summit_Industrial_Kansas_City.json",
-      "client": "Summit Industrial",
-      "city": "Kansas City",
       "events": 2
+    },
+    {
+      "file": "scen_011_Heritage_Foods_Flint.json",
+      "client": "Heritage Foods",
+      "city": "Flint",
+      "events": 3
+    },
+    {
+      "file": "scen_012_Parallel_Machining_Kansas_City.json",
+      "client": "Parallel Machining",
+      "city": "Kansas City",
+      "events": 3
+    },
+    {
+      "file": "scen_013_Horizon_Supply_Flint.json",
+      "client": "Horizon Supply",
+      "city": "Flint",
+      "events": 3
+    },
+    {
+      "file": "scen_014_Midway_Distribution_Indianapolis.json",
+      "client": "Midway Distribution",
+      "city": "Indianapolis",
+      "events": 4
+    },
+    {
+      "file": "scen_015_Cornerstone_Fabrication_Kansas_City.json",
+      "client": "Cornerstone Fabrication",
+      "city": "Kansas City",
+      "events": 4
+    },
+    {
+      "file": "scen_016_Riverfront_Steel_Columbus.json",
+      "client": "Riverfront Steel",
+      "city": "Columbus",
+      "events": 4
+    },
+    {
+      "file": "scen_017_Summit_Industrial_Detroit.json",
+      "client": "Summit Industrial",
+      "city": "Detroit",
+      "events": 2
+    },
+    {
+      "file": "scen_018_Heritage_Foods_Cincinnati.json",
+      "client": "Heritage Foods",
+      "city": "Cincinnati",
+      "events": 4
+    },
+    {
+      "file": "scen_019_Midway_Distribution_Chicago.json",
+      "client": "Midway Distribution",
+      "city": "Chicago",
+      "events": 3
     }
   ]
 }

tests/real-world/nine_consecutive_audits.ts

@@ -1,6 +1,6 @@
 // Nine-consecutive audit runner — empirical test of the predictive-
-// compounding property. Runs the audit pipeline 9 times against the
+// compounding property. Pushes 9 empty commits to the current branch,
-// same PR (each time with a new diff from Gitea), captures the
+// waits for each audit to complete on the new SHA, captures the
 // verdict + audit_lessons state after each run, and reports whether
 // the KB stabilizes or drifts.
 //
@@ -20,31 +20,49 @@
 //
 // Run: bun run tests/real-world/nine_consecutive_audits.ts
 
-import { readFile, writeFile } from "node:fs/promises";
+import { readFile } from "node:fs/promises";
-import { join } from "node:path";
 import { aggregate } from "../../auditor/kb_index.ts";
-import { getPrSnapshot } from "../../auditor/gitea.ts";
-import { auditPr } from "../../auditor/audit.ts";
 
 const REPO = "/home/profit/lakehouse";
 const AUDIT_LESSONS = `${REPO}/data/_kb/audit_lessons.jsonl`;
 const VERDICTS_DIR = `${REPO}/data/_auditor/verdicts`;
 const POLL_INTERVAL_MS = 5_000;
+const AUDIT_TIMEOUT_MS = 180_000;
 const RUNS = Number(process.env.LH_AUDIT_RUNS ?? 9);
 const TARGET_PR = Number(process.env.LH_AUDIT_PR ?? 8);
-const SKIP_INFERENCE = process.env.LH_AUDITOR_SKIP_INFERENCE !== "0";
-const RESET_KB = process.env.LH_RESET_KB === "1";
 
-async function waitForVerdict(prNum: number, sha: string, deadlineMs: number): Promise<any> {
+async function sh(cmd: string): Promise<{ stdout: string; stderr: string; code: number }> {
+  const p = Bun.spawn(["bash", "-lc", cmd], { cwd: REPO, stdout: "pipe", stderr: "pipe" });
+  const [stdout, stderr] = await Promise.all([new Response(p.stdout).text(), new Response(p.stderr).text()]);
+  const code = await p.exited;
+  return { stdout, stderr, code };
+}
+
+async function getHeadSha(): Promise<string> {
+  const r = await sh("git rev-parse HEAD");
+  return r.stdout.trim();
+}
+
+async function pushEmptyCommit(n: number): Promise<string> {
+  const msg = `test: nine-consecutive audit run ${n}/${RUNS} (compounding probe)`;
+  await sh(`GIT_AUTHOR_NAME=profit GIT_AUTHOR_EMAIL=profit@lakehouse GIT_COMMITTER_NAME=profit GIT_COMMITTER_EMAIL=profit@lakehouse git commit --allow-empty -m "${msg}"`);
+  const sha = await getHeadSha();
+  const pushCmd = `PAT="dead60d1160a02f81d241197d5d18f4608794fb2"; git -c credential.helper='!f() { echo "username=profit"; echo "password='$PAT'"; }; f' push origin HEAD 2>&1`;
+  const pr = await sh(pushCmd);
+  if (pr.code !== 0) throw new Error(`push failed: ${pr.stderr || pr.stdout}`);
+  return sha;
+}
+
+async function waitForVerdict(sha: string, deadlineMs: number): Promise<any> {
   const short = sha.slice(0, 12);
-  const path = join(VERDICTS_DIR, `${prNum}-${short}.json`);
+  const path = `${VERDICTS_DIR}/${TARGET_PR}-${short}.json`;
   const start = Date.now();
   while (Date.now() - start < deadlineMs) {
     try {
       const raw = await readFile(path, "utf8");
       return JSON.parse(raw);
     } catch { /* not yet */ }
-    await Bun.sleep(POLL_INTERVAL_MS);
+    await new Promise(r => setTimeout(r, POLL_INTERVAL_MS));
   }
   throw new Error(`no verdict file after ${deadlineMs}ms: ${path}`);
 }
@@ -55,14 +73,10 @@ async function captureAggState(): Promise<{ sig_count: number; max_count: number
     scopeFn: (r) => (r?.pr_number !== undefined ? `pr-${r.pr_number}` : undefined),
   });
   const list = Array.from(agg.values()).sort((a, b) => b.count - a.count);
-  const recurring = list.filter(r => r.count >= 2);
-  const recurringMaxCount = recurring.length > 0 ? Math.max(...recurring.map(a => a.count)) : 0;
-  const recurringMaxConf = recurring.length > 0 ? Math.max(...recurring.map(a => a.confidence)) : 0;
   return {
     sig_count: list.length,
     max_count: list[0]?.count ?? 0,
-    max_confidence: recurringMaxConf,
+    max_confidence: list.reduce((m, a) => Math.max(m, a.confidence), 0),
-    recurring_max_count: recurringMaxCount,
     top3: list.slice(0, 3).map(a => ({
       sig: a.signature,
       count: a.count,
@@ -86,25 +100,12 @@ interface RunRecord {
   kb_sig_count_after: number;
   kb_max_count_after: number;
   kb_max_confidence_after: number;
-  kb_recurring_max_count: number;
 }
 
 async function main() {
   console.log(`[nine] target PR: #${TARGET_PR}`);
   console.log(`[nine] runs:      ${RUNS}`);
-  console.log(`[nine] skip_inference: ${SKIP_INFERENCE}`);
-  console.log(`[nine] reset_kb:  ${RESET_KB}`);
   console.log(`[nine] audit_lessons.jsonl: ${AUDIT_LESSONS}`);
-
-  if (RESET_KB) {
-    console.log("[nine] clearing audit_lessons.jsonl for clean test...");
-    await writeFile(AUDIT_LESSONS, "");
-  }
-  console.log("");
-
-  const pr = await getPrSnapshot(TARGET_PR);
-  console.log(`[nine] PR #${pr.number}: "${pr.title}" (head=${pr.head_sha.slice(0, 12)})`);
-  console.log(`[nine] files in diff: ${pr.files.length}`);
   console.log("");
 
   const baseline = await captureAggState();
@@ -115,18 +116,13 @@ async function main() {
   for (let n = 1; n <= RUNS; n++) {
     const t0 = Date.now();
     console.log(`─── run ${n}/${RUNS} ───`);
-
+    const sha = await pushEmptyCommit(n);
-    const verdict = await auditPr(pr, {
+    console.log(`  pushed ${sha.slice(0, 12)}`);
-      dry_run: true,
+    const verdict = await waitForVerdict(sha, AUDIT_TIMEOUT_MS);
-      skip_dynamic: true,
-      skip_inference: SKIP_INFERENCE,
-    });
-
-    console.log(`  sha ${verdict.head_sha.slice(0, 12)}`);
     const after = await captureAggState();
     const rec: RunRecord = {
       run: n,
-      sha: verdict.head_sha.slice(0, 12),
+      sha: sha.slice(0, 12),
       verdict_overall: String(verdict.overall),
       findings_total: Number(verdict.metrics?.findings_total ?? 0),
       findings_block: Number(verdict.metrics?.findings_block ?? 0),
@@ -138,11 +134,10 @@ async function main() {
       kb_sig_count_after: after.sig_count,
       kb_max_count_after: after.max_count,
       kb_max_confidence_after: after.max_confidence,
-      kb_recurring_max_count: after.recurring_max_count,
     };
     records.push(rec);
     console.log(`  verdict=${rec.verdict_overall} findings=${rec.findings_total} (b=${rec.findings_block} w=${rec.findings_warn})`);
-    console.log(`  kb after: sig=${rec.kb_sig_count_after} max_count=${rec.kb_max_count_after} recurring_max=${rec.kb_recurring_max_count} max_conf=${rec.kb_max_confidence_after.toFixed(2)}`);
+    console.log(`  kb after: sig=${rec.kb_sig_count_after} max_count=${rec.kb_max_count_after} max_conf=${rec.kb_max_confidence_after.toFixed(2)}`);
     console.log(`  elapsed: ${((Date.now() - t0) / 1000).toFixed(1)}s`);
     console.log("");
   }
@@ -158,10 +153,10 @@ async function main() {
   console.log("");
   console.log("═══ COMPOUNDING PROPERTY ═══");
   const sigDelta = records[records.length - 1].kb_sig_count_after - baseline.sig_count;
+  const maxCount = records[records.length - 1].kb_max_count_after;
   const maxConf = records[records.length - 1].kb_max_confidence_after;
-  const recurringMax = records[records.length - 1].kb_recurring_max_count;
   console.log(`  signatures added over ${RUNS} runs: ${sigDelta}`);
-  console.log(`  max recurring count after run ${RUNS}: ${recurringMax} (same-PR recurrences per signature)`);
+  console.log(`  max count after run ${RUNS}:         ${maxCount} (same-PR recurrences per signature)`);
   console.log(`  max confidence after run ${RUNS}:    ${maxConf.toFixed(2)} (expect LOW — same-PR should not inflate)`);
 
   const verdictSet = new Set(records.map(r => r.verdict_overall));
@@ -171,10 +166,10 @@ async function main() {
     console.log(`  verdict oscillated across runs: ${[...verdictSet].join(" | ")} ✗`);
   }
 
-  if (maxConf < 0.6 && recurringMax < 5) {
+  if (maxConf < 0.3) {
     console.log(`  confidence policy holding: same-PR noise stays below escalation threshold ✓`);
   } else {
-    console.log(`  ⚠ cross-cutting pattern detected (conf=${maxConf.toFixed(2)}, recurring=${recurringMax}) — kb_index policy escalated`);
+    console.log(`  ⚠ confidence escalated above 0.3 on same-PR noise — kb_index policy needs tightening`);
   }
 
   const jsonOut = `${REPO}/tests/real-world/runs/nine_consecutive_${Date.now().toString(36)}.json`;
@@ -183,4 +178,4 @@ async function main() {
   console.log(`  report: ${jsonOut}`);
 }
 
-main().catch(e => { console.error("[nine] fatal:", e); process.exit(1); });
+main().catch(e => { console.error("[nine] fatal:", e); process.exit(1); });