profit/lakehouse
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Phase 19 wiring + Path 1/2 work + chain integrity fixes

Browse Source 
Backend:
- crates/vectord/src/playbook_memory.rs (new): Phase 19 in-memory boost
  store with seed/rebuild/snapshot, plus temporal decay (e^-age/30 per
  playbook), persist_to_sql endpoint backing successful_playbooks_live,
  and discover_patterns endpoint for meta-index pattern aggregation
  (recurring certs/skills/archetype/reliability across similar past fills).
- DEFAULT_TOP_K_PLAYBOOKS bumped 5 → 25; old default silently missed
  most boosts when memory had > 25 entries.
- service.rs: new routes /vectors/playbook_memory/{seed,rebuild,stats,
  persist_sql,patterns}.

Bun staffing co-pilot (mcp-server/):
- /search, /match, /verify, /proof, /simulation/run, MCP tools all
  forward use_playbook_memory:true and playbook_memory_k:25 to the
  hybrid endpoint. Boost was previously dark across the entire app.
- /log no longer POSTs to /ingest/file — that endpoint REPLACES the
  dataset's object list, so single-row CSV writes were wiping all prior
  rows in successful_playbooks (sp_rows went 33→1 in one /log call).
  /log now seeds playbook_memory with canonical short text and calls
  /persist_sql to keep successful_playbooks_live in sync.
- /simulation/run cumulative end-of-week CSV write removed for the same
  reason. Per-day per-contract /seed (added in this session) is the
  accumulating feedback path now.
- search.html addWorkerInsight renders a green "Endorsed · N playbooks"
  chip with playbook citations when boost > 0.

Internal Dioxus UI (crates/ui/):
- Dashboard phase list rewritten through Phase 19 (was stuck at "Phase
  16: File Watcher" / "Phase 17: DB Connector" — both wrong).
- Removed fabricated "27ms" stat label.
- Ask tab examples + SQL default replaced with real staffing prompts
  against candidates/clients/job_orders (was referencing nonexistent
  employees/products/events).
- New Playbook tab exposes /vectors/playbook_memory/{stats,rebuild} and
  side-by-side hybrid search (boost OFF vs ON) with citations.

Tests (tests/multi-agent/):
- run_e2e_rated.ts: parallel two-agent (mistral + qwen2.5) build phase
  + verifier rating (geo, auth, persist, boost, speed → /10).
- network_proving.ts: continuous build → verify → repeat with
  staffing-recruiter profile hot-swap; geo-discrimination check.
- chain_of_custody.ts: single recruiter operation traced through every
  layer (Bun /search, direct /vectors/hybrid parity, /log, SQL,
  playbook_memory growth, profile activation, post-op boost lift).
This commit is contained in:
root
2026-04-20 06:21:13 -05:00
parent 8e3cac5812
commit 25b7e6c3a7
13 changed files with 4566 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/vectord/src/lib.rs
+1
View File
@@ -7,6 +7,7 @@ pub mod harness;
pub mod hnsw;
pub mod index_registry;
pub mod jobs;
pub mod playbook_memory;
pub mod promotion;
pub mod refresh;
pub mod store;
crates/vectord/src/playbook_memory.rs
+825
View File
@@ -0,0 +1,825 @@
//! Phase 19: Playbook memory — the feedback loop that makes the index
//! learn from real outcomes instead of just logging them.
//!
//! When an agent (multi-agent orchestrator or human operator) seals a
//! successful playbook, it lands in the `successful_playbooks` dataset.
//! Historically that was a write-only log. This module turns it into a
//! re-ranking signal:
//!
//! 1. `rebuild` reads every row of `successful_playbooks`, embeds the
//! operation+approach+context as one vector per playbook, parses
//! out the worker names from the `result` column, and stores both
//! the vectors and the (playbook → names) endorsement map in memory.
//!
//! 2. At query time, `compute_boost_for` takes a new operation text
//! (e.g. "fill: Welder x2 in Toledo, OH"), embeds it, brute-force
//! ranks past playbooks by cosine similarity, and returns a boost
//! map keyed by (city, state, worker_name) → `BoostEntry`. Each
//! entry carries its similarity score and the citing playbook_ids,
//! so explanations ("ranked higher because of 3 similar past fills
//! in Toledo") are free.
//!
//! 3. The `use_playbook_memory` flag on `/vectors/hybrid` adds those
//! boosts to matching search hits and re-sorts.
//!
//! Why brute force instead of another HNSW: `successful_playbooks` grows
//! by operators, not automation. A few thousand rows is the realistic
//! ceiling for years. Brute force at 10K × 768d is <10ms on this hardware
//! — not worth the operational cost of another indexed surface.
//!
//! Persistence: the endorsements map round-trips through
//! `_playbook_memory/state.json` in primary storage so the cache
//! survives restarts without a full rebuild.
use std::collections::HashMap;
use std::sync::Arc;
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use aibridge::client::{AiClient, EmbedRequest};
use object_store::ObjectStore;
use storaged::ops;
const STATE_KEY: &str = "_playbook_memory/state.json";
/// Maximum boost a single worker can accumulate across all similar past
/// playbooks. Prevents one very popular worker from always winning.
pub const MAX_BOOST_PER_WORKER: f32 = 0.25;
/// Default number of past playbooks to consider when ranking the current
/// operation. Bumped 5 → 25 on 2026-04-20 because at >100 entries in
/// memory the old default missed too many relevant playbooks — boost
/// silently failed even when the seeded workers were ideal matches.
/// 25 is brute-force-cheap (sub-ms) and covers most live operator memory.
pub const DEFAULT_TOP_K_PLAYBOOKS: usize = 25;
/// Half-life of a playbook's contribution to boost, in days. A playbook
/// 30 days old contributes half what a fresh one would; 60 days old, a
/// quarter; etc. Per Path 1 (deepen statistical) — stale endorsements
/// shouldn't dominate fresh signal. Recruiter trust depends on this.
pub const BOOST_HALF_LIFE_DAYS: f32 = 30.0;
/// Shape of one playbook in memory. The embedding is optional so we can
/// round-trip a cached state without re-embedding; the rebuild path
/// populates it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PlaybookEntry {
pub playbook_id: String,
pub operation: String,
pub approach: String,
pub context: String,
pub timestamp: String,
/// Parsed out of `result` (e.g. "2/2 filled → Matthew Roberts, Amy Davis").
/// Stored as raw names; matching against search results happens on
/// (city, state, name) tuples at boost time.
pub endorsed_names: Vec<String>,
/// City + state parsed out of the operation string. Kept separately
/// so boost matching doesn't re-parse on every query.
pub city: Option<String>,
pub state: Option<String>,
/// Embedding of `operation + approach + context`. Option so persisted
/// state can omit it on first load and have a later embed() fill in.
#[serde(default)]
pub embedding: Option<Vec<f32>>,
}
/// Persisted / in-memory state.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
struct PlaybookMemoryState {
entries: Vec<PlaybookEntry>,
/// Unix epoch millis when the last rebuild completed. Caller can
/// use this to gate "stale > N hours → trigger rebuild" behavior.
last_rebuilt_at: i64,
}
/// Per-worker boost payload. `citations` lets the response layer show
/// "boosted because of these past fills" without a second lookup.
#[derive(Debug, Clone, Serialize)]
pub struct BoostEntry {
pub boost: f32,
pub citations: Vec<String>, // playbook_ids that endorsed this worker
}
/// Live handle passed around the service. Clone-cheap (all state is
/// inside one Arc<RwLock>).
#[derive(Clone)]
pub struct PlaybookMemory {
state: Arc<RwLock<PlaybookMemoryState>>,
store: Arc<dyn ObjectStore>,
}
impl PlaybookMemory {
pub fn new(store: Arc<dyn ObjectStore>) -> Self {
Self {
state: Arc::new(RwLock::new(PlaybookMemoryState::default())),
store,
}
}
/// Best-effort load from primary storage. Missing = empty memory; the
/// first `/rebuild` call will hydrate it.
pub async fn load_from_storage(&self) -> Result<usize, String> {
let data = match ops::get(&self.store, STATE_KEY).await {
Ok(d) => d,
Err(_) => return Ok(0),
};
let persisted: PlaybookMemoryState = serde_json::from_slice(&data)
.map_err(|e| format!("parse playbook_memory state: {e}"))?;
let n = persisted.entries.len();
*self.state.write().await = persisted;
tracing::info!("playbook_memory: loaded {n} entries from {STATE_KEY}");
Ok(n)
}
async fn persist(&self) -> Result<(), String> {
let snapshot = self.state.read().await.clone();
let bytes = serde_json::to_vec_pretty(&snapshot).map_err(|e| e.to_string())?;
ops::put(&self.store, STATE_KEY, bytes.into()).await
}
/// Replace the full in-memory state atomically and persist.
pub async fn set_entries(&self, entries: Vec<PlaybookEntry>) -> Result<(), String> {
let mut s = self.state.write().await;
s.entries = entries;
s.last_rebuilt_at = chrono::Utc::now().timestamp_millis();
drop(s);
self.persist().await
}
pub async fn entry_count(&self) -> usize {
self.state.read().await.entries.len()
}
pub async fn snapshot(&self) -> Vec<PlaybookEntry> {
self.state.read().await.entries.clone()
}
/// Given an operation's embedding, find the top-K most similar past
/// playbooks (by cosine similarity) and return a per-worker boost map
/// keyed by (city, state, name). Worker is matched by the tuple so a
/// shared name across cities doesn't cross-pollinate.
///
/// Boost formula: each qualifying playbook contributes
/// `similarity * base_weight / n_workers` to each worker it endorsed,
/// where `base_weight` is tuned to keep the cap realistic without
/// forcing every result to saturate. Total per worker is capped at
/// `MAX_BOOST_PER_WORKER`.
pub async fn compute_boost_for(
&self,
query_embedding: &[f32],
top_k_playbooks: usize,
base_weight: f32,
) -> HashMap<(String, String, String), BoostEntry> {
let entries = self.state.read().await.entries.clone();
// Brute-force cosine. Empty / missing embeddings just skip.
let mut scored: Vec<(f32, &PlaybookEntry)> = entries
.iter()
.filter_map(|e| e.embedding.as_ref().map(|v| (cosine(query_embedding, v), e)))
.collect();
scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));
scored.truncate(top_k_playbooks.max(1));
let now = chrono::Utc::now();
let mut boosts: HashMap<(String, String, String), BoostEntry> = HashMap::new();
for (similarity, pb) in &scored {
// Negative or near-zero similarity = not actually related;
// skip so we don't inject noise when the memory is sparse.
if *similarity <= 0.05 { continue; }
let Some(city) = &pb.city else { continue; };
let Some(state) = &pb.state else { continue; };
let n_workers = pb.endorsed_names.len().max(1);
// Path 1 — temporal decay. Older playbooks weight less. Failure
// to parse the timestamp degrades to "no decay" (treat as fresh)
// rather than dropping the entry entirely; keeps backward
// compatibility with seed payloads that omitted timestamp.
let decay = chrono::DateTime::parse_from_rfc3339(&pb.timestamp)
.ok()
.map(|t| {
let age_days = (now.signed_duration_since(t.with_timezone(&chrono::Utc))
.num_seconds() as f32) / 86400.0;
if age_days <= 0.0 { 1.0 }
else { (-age_days / BOOST_HALF_LIFE_DAYS).exp() }
})
.unwrap_or(1.0);
let per_worker = similarity * base_weight * decay / (n_workers as f32);
for name in &pb.endorsed_names {
let key = (city.clone(), state.clone(), name.clone());
let entry = boosts.entry(key).or_insert(BoostEntry {
boost: 0.0,
citations: Vec::new(),
});
entry.boost = (entry.boost + per_worker).min(MAX_BOOST_PER_WORKER);
if !entry.citations.contains(&pb.playbook_id) {
entry.citations.push(pb.playbook_id.clone());
}
}
}
boosts
}
}
/// Cosine similarity — pulled out so rebuild/boost share one impl.
fn cosine(a: &[f32], b: &[f32]) -> f32 {
let (mut dot, mut na, mut nb) = (0.0_f32, 0.0_f32, 0.0_f32);
let n = a.len().min(b.len());
for i in 0..n {
dot += a[i] * b[i];
na += a[i] * a[i];
nb += b[i] * b[i];
}
if na == 0.0 || nb == 0.0 { return 0.0; }
dot / (na.sqrt() * nb.sqrt())
}
// ---------------- Pattern discovery (Path 2 — meta-index) ----------------
//
// Phase 19's boost path answers "for THIS exact city + role, which workers
// have we used before?" Pattern discovery answers a different question:
// "for queries like this one, what TRAITS have past successful fills had
// in common — even if no exact prior playbook covers this geo?"
//
// The discovered pattern surfaces signals the operator didn't query for:
// e.g. "every successful Welder fill we've seen carried OSHA-10 + lockout
// /tagout — you may want to filter on those." That's the meta-index
// dimension of the original PRD: identify things we didn't know about.
#[derive(Debug, Clone, Serialize)]
pub struct PatternReport {
pub query: String,
pub matched_playbooks: usize,
pub total_workers_examined: usize,
pub common_certifications: Vec<TraitFreq>,
pub common_skills: Vec<TraitFreq>,
pub modal_archetype: Option<String>,
pub reliability_p50: f64,
pub reliability_min: f64,
pub reliability_max: f64,
pub matched_playbook_ids: Vec<String>,
pub discovered_pattern: String,
pub duration_secs: f32,
}
#[derive(Debug, Clone, Serialize)]
pub struct TraitFreq {
pub name: String,
pub count: usize,
pub frequency: f32,
}
pub async fn discover_patterns(
memory: &PlaybookMemory,
ai_client: &AiClient,
catalog: &catalogd::registry::Registry,
buckets: &Arc<storaged::registry::BucketRegistry>,
query: &str,
top_k_playbooks: usize,
min_trait_frequency: f32,
) -> Result<PatternReport, String> {
let t0 = std::time::Instant::now();
// 1. Embed the query through the same nomic-embed-text model used
// for playbook embeddings, so cosine is meaningful.
let resp = ai_client
.embed(EmbedRequest { texts: vec![query.into()], model: None })
.await
.map_err(|e| format!("embed query: {e}"))?;
if resp.embeddings.is_empty() {
return Err("embed returned no vectors".into());
}
let qv: Vec<f32> = resp.embeddings[0].iter().map(|x| *x as f32).collect();
// 2. Find top-K most similar past playbooks (cosine over embeddings).
let entries = memory.snapshot().await;
let mut scored: Vec<(f32, &PlaybookEntry)> = entries
.iter()
.filter_map(|e| e.embedding.as_ref().map(|v| (cosine(&qv, v), e)))
.collect();
scored.sort_by(|a, b| b.0.partial_cmp(&a.0).unwrap_or(std::cmp::Ordering::Equal));
scored.truncate(top_k_playbooks);
let matched: Vec<(f32, PlaybookEntry)> = scored
.into_iter()
.filter(|(s, _)| *s > 0.05)
.map(|(s, e)| (s, e.clone()))
.collect();
if matched.is_empty() {
return Ok(PatternReport {
query: query.into(),
matched_playbooks: 0,
total_workers_examined: 0,
common_certifications: vec![],
common_skills: vec![],
modal_archetype: None,
reliability_p50: 0.0, reliability_min: 0.0, reliability_max: 0.0,
matched_playbook_ids: vec![],
discovered_pattern: "No similar past playbooks found.".into(),
duration_secs: t0.elapsed().as_secs_f32(),
});
}
// 3. Pull each endorsed worker's full profile from workers_500k.
// Restrict by (name, city, state) tuple so cross-city homonyms
// don't pollute the aggregate.
let mut conditions: Vec<String> = Vec::new();
let mut matched_ids: Vec<String> = Vec::new();
for (_, pb) in &matched {
matched_ids.push(pb.playbook_id.clone());
let (Some(city), Some(state)) = (pb.city.as_ref(), pb.state.as_ref()) else { continue };
for name in &pb.endorsed_names {
let esc = |s: &str| s.replace('\'', "''");
conditions.push(format!(
"(name = '{}' AND city = '{}' AND state = '{}')",
esc(name), esc(city), esc(state)
));
}
}
if conditions.is_empty() {
return Ok(PatternReport {
query: query.into(),
matched_playbooks: matched.len(),
total_workers_examined: 0,
common_certifications: vec![], common_skills: vec![],
modal_archetype: None, reliability_p50: 0.0,
reliability_min: 0.0, reliability_max: 0.0,
matched_playbook_ids: matched_ids,
discovered_pattern: "Matched playbooks but no endorsed names with city/state to lookup.".into(),
duration_secs: t0.elapsed().as_secs_f32(),
});
}
let sql = format!(
"SELECT name, role, city, state, certifications, skills, archetype, \
CAST(reliability AS DOUBLE) as reliability \
FROM workers_500k WHERE {} LIMIT 500",
conditions.join(" OR ")
);
let engine = queryd::context::QueryEngine::new(
catalog.clone(), buckets.clone(), queryd::cache::MemCache::new(0),
);
let batches = engine.query(&sql).await.map_err(|e| format!("worker lookup: {e}"))?;
// 4. Aggregate. Pipe-separated cert/skill lists, single-string archetype,
// numeric reliability. Frequencies are share-of-workers.
use arrow::array::{Array, AsArray};
let mut cert_counts: HashMap<String, usize> = HashMap::new();
let mut skill_counts: HashMap<String, usize> = HashMap::new();
let mut arch_counts: HashMap<String, usize> = HashMap::new();
let mut reliabilities: Vec<f64> = Vec::new();
let mut total = 0usize;
let get_string = |b: &arrow::record_batch::RecordBatch, col: &str, row: usize| -> String {
let Some(c) = b.column_by_name(col) else { return String::new(); };
if let Some(arr) = c.as_string_view_opt() {
if arr.is_null(row) { return String::new(); }
return arr.value(row).to_string();
}
if let Some(arr) = c.as_string_opt::<i32>() {
if arr.is_null(row) { return String::new(); }
return arr.value(row).to_string();
}
String::new()
};
let get_f64 = |b: &arrow::record_batch::RecordBatch, col: &str, row: usize| -> f64 {
let Some(c) = b.column_by_name(col) else { return 0.0; };
if let Some(arr) = c.as_primitive_opt::<arrow::datatypes::Float64Type>() {
if arr.is_null(row) { return 0.0; }
return arr.value(row);
}
0.0
};
for b in &batches {
for row in 0..b.num_rows() {
total += 1;
let certs = get_string(b, "certifications", row);
for c in certs.split(['|', ',']).map(|s| s.trim()).filter(|s| !s.is_empty() && *s != "none") {
*cert_counts.entry(c.to_string()).or_insert(0) += 1;
}
let skills = get_string(b, "skills", row);
for s in skills.split(['|', ',']).map(|s| s.trim()).filter(|s| !s.is_empty()) {
*skill_counts.entry(s.to_string()).or_insert(0) += 1;
}
let arch = get_string(b, "archetype", row);
if !arch.is_empty() {
*arch_counts.entry(arch).or_insert(0) += 1;
}
let rel = get_f64(b, "reliability", row);
if rel > 0.0 { reliabilities.push(rel); }
}
}
let total_f = total.max(1) as f32;
let to_freq = |m: HashMap<String, usize>, min: f32| -> Vec<TraitFreq> {
let mut v: Vec<TraitFreq> = m.into_iter()
.map(|(name, count)| TraitFreq { name, count, frequency: count as f32 / total_f })
.filter(|t| t.frequency >= min)
.collect();
v.sort_by(|a, b| b.count.cmp(&a.count));
v.truncate(8);
v
};
let common_certifications = to_freq(cert_counts, min_trait_frequency);
let common_skills = to_freq(skill_counts, min_trait_frequency);
let modal_archetype = arch_counts.into_iter()
.max_by_key(|(_, c)| *c)
.map(|(name, _)| name);
reliabilities.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let p50 = if reliabilities.is_empty() { 0.0 } else { reliabilities[reliabilities.len() / 2] };
let rmin = reliabilities.first().copied().unwrap_or(0.0);
let rmax = reliabilities.last().copied().unwrap_or(0.0);
// Build a human-readable discovered-pattern summary
let mut parts: Vec<String> = vec![
format!("Across {} similar past playbooks ({} workers examined)", matched.len(), total),
];
if !common_certifications.is_empty() {
let head: Vec<String> = common_certifications.iter().take(3)
.map(|t| format!("{} ({:.0}%)", t.name, t.frequency * 100.0)).collect();
parts.push(format!("recurring certifications: {}", head.join(", ")));
}
if !common_skills.is_empty() {
let head: Vec<String> = common_skills.iter().take(3)
.map(|t| format!("{} ({:.0}%)", t.name, t.frequency * 100.0)).collect();
parts.push(format!("recurring skills: {}", head.join(", ")));
}
if let Some(a) = &modal_archetype { parts.push(format!("archetype mostly: {a}")); }
if !reliabilities.is_empty() {
parts.push(format!("reliability median {:.2} (range {:.2}–{:.2})", p50, rmin, rmax));
}
let discovered_pattern = parts.join(" · ");
Ok(PatternReport {
query: query.into(),
matched_playbooks: matched.len(),
total_workers_examined: total,
common_certifications, common_skills,
modal_archetype, reliability_p50: p50,
reliability_min: rmin, reliability_max: rmax,
matched_playbook_ids: matched_ids,
discovered_pattern,
duration_secs: t0.elapsed().as_secs_f32(),
})
}
// ---------------- Persist memory → SQL (Path 2 foundation) ----------------
#[derive(Debug, Clone, Serialize)]
pub struct PersistReport {
pub rows_persisted: usize,
pub dataset_name: String,
pub fingerprint: String,
pub duration_secs: f32,
}
/// Dump current in-memory state to a queryable Parquet under
/// `successful_playbooks_live`. Registers fresh objects each call — safe
/// because in-memory state is the source of truth here, so REPLACING the
/// objects list reflects the real state, not destroying it.
///
/// Distinct from the existing `successful_playbooks` dataset (which is
/// read by `rebuild()`), so this never collides with operator imports of
/// historical playbook data. Recruiter-facing SQL surfaces should query
/// `successful_playbooks_live` for current operator activity.
pub async fn persist_to_sql(
memory: &PlaybookMemory,
catalog: &catalogd::registry::Registry,
) -> Result<PersistReport, String> {
use arrow::array::StringArray;
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
let t0 = std::time::Instant::now();
let entries = memory.snapshot().await;
let schema = Arc::new(Schema::new(vec![
Field::new("timestamp", DataType::Utf8, true),
Field::new("operation", DataType::Utf8, true),
Field::new("approach", DataType::Utf8, true),
Field::new("result", DataType::Utf8, true),
Field::new("context", DataType::Utf8, true),
]));
let timestamps: Vec<&str> = entries.iter().map(|e| e.timestamp.as_str()).collect();
let operations: Vec<&str> = entries.iter().map(|e| e.operation.as_str()).collect();
let approaches: Vec<&str> = entries.iter().map(|e| e.approach.as_str()).collect();
let contexts: Vec<&str> = entries.iter().map(|e| e.context.as_str()).collect();
// Result column is reconstructed from endorsed_names so SQL queries
// against successful_playbooks_live see the same shape as the original
// CSV-fed successful_playbooks ("N/N filled → Name1, Name2").
let results: Vec<String> = entries.iter().map(|e| {
if e.endorsed_names.is_empty() {
String::new()
} else {
let n = e.endorsed_names.len();
format!("{}/{} filled → {}", n, n, e.endorsed_names.join(", "))
}
}).collect();
let result_refs: Vec<&str> = results.iter().map(|s| s.as_str()).collect();
let batch = RecordBatch::try_new(schema.clone(), vec![
Arc::new(StringArray::from(timestamps)),
Arc::new(StringArray::from(operations)),
Arc::new(StringArray::from(approaches)),
Arc::new(StringArray::from(result_refs)),
Arc::new(StringArray::from(contexts)),
]).map_err(|e| format!("build record batch: {e}"))?;
let parquet_bytes = shared::arrow_helpers::record_batch_to_parquet(&batch)?;
let fp = shared::arrow_helpers::fingerprint_schema(&schema);
let key = "datasets/successful_playbooks_live.parquet";
ops::put(&memory.store, key, parquet_bytes.clone()).await?;
let obj = shared::types::ObjectRef {
bucket: "primary".into(),
key: key.into(),
size_bytes: parquet_bytes.len() as u64,
created_at: chrono::Utc::now(),
};
let manifest = catalog.register(
"successful_playbooks_live".into(),
fp.clone(),
vec![obj],
).await?;
Ok(PersistReport {
rows_persisted: entries.len(),
dataset_name: manifest.name,
fingerprint: fp.0,
duration_secs: t0.elapsed().as_secs_f32(),
})
}
// ---------------- Rebuild (the core of Phase 19) ----------------
#[derive(Debug, Clone, Serialize)]
pub struct RebuildReport {
pub rows_scanned: usize,
pub entries_built: usize,
pub total_names_endorsed: usize,
pub duration_secs: f32,
}
/// Full rebuild: scan `successful_playbooks`, extract endorsements, embed
/// each row's operation+approach+context, replace the in-memory state.
///
/// Returns the report so callers can show operators what happened.
pub async fn rebuild(
memory: &PlaybookMemory,
ai_client: &AiClient,
catalog: &catalogd::registry::Registry,
buckets: &Arc<storaged::registry::BucketRegistry>,
) -> Result<RebuildReport, String> {
let t0 = std::time::Instant::now();
// 1. Pull every row of successful_playbooks through the query engine.
let sql = "SELECT timestamp, operation, approach, result, context \
FROM successful_playbooks";
let engine = queryd::context::QueryEngine::new(
catalog.clone(),
buckets.clone(),
queryd::cache::MemCache::new(0),
);
let batches = engine
.query(sql)
.await
.map_err(|e| format!("query successful_playbooks: {e}"))?;
let mut rows: Vec<(String, String, String, String, String)> = Vec::new();
for b in &batches {
let n = b.num_rows();
let get = |col: &str, row: usize| -> String {
use arrow::array::{Array, AsArray};
let Some(c) = b.column_by_name(col) else { return String::new(); };
if let Some(arr) = c.as_string_view_opt() {
if arr.is_null(row) { return String::new(); }
return arr.value(row).to_string();
}
if let Some(arr) = c.as_string_opt::<i32>() {
if arr.is_null(row) { return String::new(); }
return arr.value(row).to_string();
}
String::new()
};
for row in 0..n {
rows.push((
get("timestamp", row),
get("operation", row),
get("approach", row),
get("result", row),
get("context", row),
));
}
}
let rows_scanned = rows.len();
// 2. For each row, build a PlaybookEntry (no embedding yet). Parse
// the operation for (city, state) and the result for names.
let mut entries: Vec<PlaybookEntry> = rows
.into_iter()
.map(|(ts, op, approach, result, ctx)| {
let (city, state) = parse_city_state(&op);
let names = parse_names(&result);
PlaybookEntry {
playbook_id: stable_id(&ts, &op),
operation: op,
approach,
context: ctx,
timestamp: ts,
endorsed_names: names,
city,
state,
embedding: None,
}
})
.collect();
// 3. Embed in one batch. Sidecar's embed handles batching internally;
// chunk here to ~64 per request to keep memory flat.
const EMBED_BATCH: usize = 64;
for chunk_start in (0..entries.len()).step_by(EMBED_BATCH) {
let end = (chunk_start + EMBED_BATCH).min(entries.len());
let texts: Vec<String> = entries[chunk_start..end]
.iter()
.map(embed_text)
.collect();
let req = EmbedRequest { texts, model: None };
let resp = ai_client
.embed(req)
.await
.map_err(|e| format!("embed batch [{chunk_start}..{end}]: {e}"))?;
for (i, v) in resp.embeddings.iter().enumerate() {
let f32v: Vec<f32> = v.iter().map(|&x| x as f32).collect();
entries[chunk_start + i].embedding = Some(f32v);
}
}
let total_names_endorsed: usize = entries.iter().map(|e| e.endorsed_names.len()).sum();
let entries_built = entries.len();
memory.set_entries(entries).await?;
Ok(RebuildReport {
rows_scanned,
entries_built,
total_names_endorsed,
duration_secs: t0.elapsed().as_secs_f32(),
})
}
fn embed_text(e: &PlaybookEntry) -> String {
// Compact one-liner per playbook. Excludes timestamp (no semantic
// signal) and includes the fills as words (they're occasionally
// meaningful — "Luis Harris" might semantically correlate with
// Spanish-speaker names in future queries).
format!(
"{} | {} | {} | fills: {}",
e.operation,
e.approach,
e.context,
e.endorsed_names.join(", "),
)
}
/// Derive a stable id from (timestamp, operation). Two playbooks with
/// identical timestamp+operation collapse to one — benign dedup.
fn stable_id(ts: &str, op: &str) -> String {
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(ts.as_bytes());
h.update(b"|");
h.update(op.as_bytes());
let bytes = h.finalize();
format!("pb-{}", hex_short(&bytes, 12))
}
fn hex_short(b: &[u8], n: usize) -> String {
let mut s = String::with_capacity(n * 2);
for byte in &b[..b.len().min(n)] {
s.push_str(&format!("{byte:02x}"));
}
s
}
/// Parse "fill: Welder x2 in Toledo, OH" → ("Toledo", "OH").
/// Returns None for malformed operations.
fn parse_city_state(op: &str) -> (Option<String>, Option<String>) {
// Split on " in " then parse "City, ST"
let after_in = match op.split(" in ").nth(1) {
Some(s) => s,
None => return (None, None),
};
let parts: Vec<&str> = after_in.splitn(2, ',').collect();
if parts.len() != 2 {
return (None, None);
}
let city = parts[0].trim().to_string();
// state might be followed by more context; take leading alpha chars
let state: String = parts[1].trim()
.chars()
.take_while(|c| c.is_ascii_alphabetic())
.collect();
if city.is_empty() || state.is_empty() {
return (None, None);
}
(Some(city), Some(state))
}
/// Parse "2/2 filled → Matthew Roberts, Amy Davis" → ["Matthew Roberts", "Amy Davis"].
fn parse_names(result: &str) -> Vec<String> {
// Everything after the arrow; split on ", ".
let after_arrow = match result.split('→').nth(1) {
Some(s) => s.trim(),
None => return Vec::new(),
};
// Strip trailing noise like "(and N more)" that some emitters add.
let cleaned = after_arrow.split(" (").next().unwrap_or(after_arrow);
cleaned
.split(',')
.map(|n| n.trim().to_string())
.filter(|n| !n.is_empty())
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_city_state_extracts_both() {
let (c, s) = parse_city_state("fill: Welder x2 in Toledo, OH");
assert_eq!(c.as_deref(), Some("Toledo"));
assert_eq!(s.as_deref(), Some("OH"));
}
#[test]
fn parse_city_state_handles_multiword_city() {
let (c, s) = parse_city_state("fill: Loader x1 in Grand Rapids, MI");
assert_eq!(c.as_deref(), Some("Grand Rapids"));
assert_eq!(s.as_deref(), Some("MI"));
}
#[test]
fn parse_city_state_malformed_returns_none() {
let (c, s) = parse_city_state("fill: something weird");
assert!(c.is_none());
assert!(s.is_none());
}
#[test]
fn parse_names_extracts_after_arrow() {
let ns = parse_names("2/2 filled → Matthew Roberts, Amy Davis");
assert_eq!(ns, vec!["Matthew Roberts".to_string(), "Amy Davis".to_string()]);
}
#[test]
fn parse_names_handles_single_fill() {
let ns = parse_names("1/1 filled → Jose Reed");
assert_eq!(ns, vec!["Jose Reed".to_string()]);
}
#[test]
fn parse_names_handles_no_arrow() {
let ns = parse_names("0/2 filled");
assert!(ns.is_empty());
}
#[test]
fn stable_id_is_deterministic() {
let a = stable_id("2026-04-20T00:00:00Z", "fill: Welder x2 in Toledo, OH");
let b = stable_id("2026-04-20T00:00:00Z", "fill: Welder x2 in Toledo, OH");
assert_eq!(a, b);
assert!(a.starts_with("pb-"));
}
#[test]
fn boost_caps_per_worker() {
// Even with 100 similar playbooks all endorsing the same name, the
// boost never exceeds MAX_BOOST_PER_WORKER.
let pm = PlaybookMemory::new(Arc::new(object_store::memory::InMemory::new()));
let entries: Vec<PlaybookEntry> = (0..100)
.map(|i| PlaybookEntry {
playbook_id: format!("pb-{i}"),
operation: "fill: Welder x1 in Toledo, OH".into(),
approach: "transfer".into(),
context: "".into(),
timestamp: "2026-04-20".into(),
endorsed_names: vec!["Deborah Powell".into()],
city: Some("Toledo".into()),
state: Some("OH".into()),
embedding: Some(vec![1.0, 0.0, 0.0]),
})
.collect();
tokio::runtime::Runtime::new().unwrap().block_on(async {
pm.set_entries(entries).await.unwrap();
let boosts = pm.compute_boost_for(&[1.0, 0.0, 0.0], 100, 0.5).await;
let key = ("Toledo".into(), "OH".into(), "Deborah Powell".into());
let entry = boosts.get(&key).expect("boost entry present");
assert!(entry.boost <= MAX_BOOST_PER_WORKER + 1e-6,
"boost {} exceeded cap {}", entry.boost, MAX_BOOST_PER_WORKER);
});
}
}
crates/vectord/src/service.rs
+500 -19
View File
@@ -12,7 +12,7 @@ use std::sync::Arc;
use aibridge::client::{AiClient, EmbedRequest, GenerateRequest};
use catalogd::registry::Registry as CatalogRegistry;
use storaged::registry::BucketRegistry;
use crate::{agent, autotune, chunker, embedding_cache, harness, hnsw, index_registry, jobs, lance_backend, promotion, rag, refresh, search, store, supervisor, trial};
use crate::{agent, autotune, chunker, embedding_cache, harness, hnsw, index_registry, jobs, lance_backend, playbook_memory, promotion, rag, refresh, search, store, supervisor, trial};
#[derive(Clone)]
pub struct VectorState {
@@ -23,6 +23,9 @@ pub struct VectorState {
pub hnsw_store: hnsw::HnswStore,
pub embedding_cache: embedding_cache::EmbeddingCache,
pub trial_journal: trial::TrialJournal,
/// Federation-aware harness store — resolves eval artifacts to each
/// index's recorded bucket, falling back to primary for legacy evals.
pub harness_store: harness::HarnessStore,
/// Catalog registry — needed by the Phase C refresh path to mark/clear
/// staleness and look up dataset manifests.
pub catalog: CatalogRegistry,
@@ -46,6 +49,10 @@ pub struct VectorState {
/// ADR-019 hybrid: handles to Lance datasets keyed by index name.
/// Lazy-created on first /vectors/lance/* call.
pub lance: lance_backend::LanceRegistry,
/// Phase 19 — meta-index feedback. Embeds past successful_playbooks
/// and, when `use_playbook_memory` is set on /vectors/hybrid, boosts
/// workers that were actually filled in semantically-similar past ops.
pub playbook_memory: playbook_memory::PlaybookMemory,
}
/// What the active-profile singleton records. Narrow — we don't need the
@@ -63,6 +70,7 @@ pub fn router(state: VectorState) -> Router {
.route("/index", post(create_index))
.route("/indexes", get(list_indexes))
.route("/indexes/{name}", get(get_index_meta))
.route("/indexes/{name}/bucket", axum::routing::patch(migrate_index_bucket))
.route("/jobs", get(list_jobs))
.route("/jobs/{id}", get(get_job))
.route("/search", post(search_index))
@@ -110,6 +118,12 @@ pub fn router(state: VectorState) -> Router {
.route("/lance/stats/{index_name}", get(lance_stats))
.route("/lance/scalar-index/{index_name}/{column}", post(lance_build_scalar_index))
.route("/lance/recall/{index_name}", post(lance_recall_harness))
// Phase 19: playbook memory — the meta-index feedback loop
.route("/playbook_memory/rebuild", post(rebuild_playbook_memory))
.route("/playbook_memory/stats", get(playbook_memory_stats))
.route("/playbook_memory/seed", post(seed_playbook_memory))
.route("/playbook_memory/persist_sql", post(persist_playbook_memory_sql))
.route("/playbook_memory/patterns", post(discover_playbook_patterns))
.with_state(state)
}
@@ -259,6 +273,174 @@ async fn get_index_meta(
}
}
#[derive(Deserialize)]
struct MigrateBucketRequest {
dest_bucket: String,
/// If true, delete artifacts from the source bucket after the pointer
/// flip. Default false — keeping source copies means a failed migration
/// is recoverable by editing IndexMeta.bucket back, and a successful
/// migration leaves inspectable forensics until an operator sweeps.
#[serde(default)]
delete_source: bool,
}
#[derive(Serialize)]
struct MigrateBucketReport {
index_name: String,
source_bucket: String,
dest_bucket: String,
/// Artifact keys that were copied (or attempted). Order follows copy order.
copied: Vec<String>,
/// Artifact prefixes that had nothing to copy (optional files missing,
/// trial journal empty, etc).
skipped: Vec<String>,
/// Subset of `copied` that was subsequently deleted from the source.
deleted_source: Vec<String>,
duration_secs: f32,
}
/// Move an index's artifacts from its current bucket to `dest_bucket`.
/// Parquet-backed indexes only — Lance migration needs URI rewriting that
/// isn't in scope for this endpoint. Copies the vector data, trial journal,
/// promotion file, and auto-generated harness; updates `IndexMeta.bucket`
/// last so a mid-flight failure leaves the index still usable at its
/// original location. Evicts the `EmbeddingCache` entry so the next load
/// re-reads from the new bucket.
async fn migrate_index_bucket(
State(state): State<VectorState>,
Path(name): Path<String>,
Json(req): Json<MigrateBucketRequest>,
) -> Result<Json<MigrateBucketReport>, (StatusCode, String)> {
let t0 = std::time::Instant::now();
let mut meta = state
.index_registry
.get(&name)
.await
.ok_or_else(|| (StatusCode::NOT_FOUND, format!("index '{name}' not found")))?;
if meta.vector_backend == shared::types::VectorBackend::Lance {
return Err((
StatusCode::BAD_REQUEST,
"Lance-backed indexes cannot be migrated via this endpoint — \
Lance URIs are bucket-specific; a separate migrate_lance tool \
is needed".into(),
));
}
if !state.bucket_registry.contains(&req.dest_bucket) {
return Err((
StatusCode::BAD_REQUEST,
format!("dest bucket '{}' not registered", req.dest_bucket),
));
}
let source_bucket = meta.bucket.clone();
if source_bucket == req.dest_bucket {
return Err((
StatusCode::BAD_REQUEST,
format!("source and dest are both '{source_bucket}' — nothing to migrate"),
));
}
let src = state
.bucket_registry
.get(&source_bucket)
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e))?;
let dst = state
.bucket_registry
.get(&req.dest_bucket)
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e))?;
let mut copied: Vec<String> = Vec::new();
let mut skipped: Vec<String> = Vec::new();
// 1. Vector data (single parquet file for this backend).
copy_key(&src, &dst, &meta.storage_key)
.await
.map_err(|e| {
(StatusCode::INTERNAL_SERVER_ERROR,
format!("copy {}: {e}", meta.storage_key))
})?;
copied.push(meta.storage_key.clone());
// 2. Trial journal batches — per-index directory of JSONL files.
let trial_prefix = format!("_hnsw_trials/{name}/");
let trial_keys = storaged::ops::list(&src, Some(&trial_prefix))
.await
.unwrap_or_default();
if trial_keys.is_empty() {
skipped.push(trial_prefix);
}
for k in &trial_keys {
copy_key(&src, &dst, k)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("copy {k}: {e}")))?;
copied.push(k.clone());
}
// 3. Promotion file (optional — absent for never-promoted indexes).
let promo_key = format!("_hnsw_promotions/{name}.json");
match copy_key(&src, &dst, &promo_key).await {
Ok(()) => copied.push(promo_key),
Err(_) => skipped.push(promo_key),
}
// 4. Auto-generated harness (optional — absent if agent never ran).
let harness_key = format!("_hnsw_evals/{name}_auto.json");
match copy_key(&src, &dst, &harness_key).await {
Ok(()) => copied.push(harness_key),
Err(_) => skipped.push(harness_key),
}
// 5. Pointer flip — IndexMeta.bucket now points at destination. This
// is the commit point; earlier failures leave copies in dest but the
// index still usable at source.
meta.bucket = req.dest_bucket.clone();
state
.index_registry
.register(meta)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("update meta: {e}")))?;
// 6. Cache eviction — next load reads the new bucket's parquet.
state.embedding_cache.evict(&name).await;
// 7. Optional source cleanup.
let mut deleted_source: Vec<String> = Vec::new();
if req.delete_source {
for k in &copied {
if storaged::ops::delete(&src, k).await.is_ok() {
deleted_source.push(k.clone());
}
}
}
Ok(Json(MigrateBucketReport {
index_name: name,
source_bucket,
dest_bucket: req.dest_bucket,
copied,
skipped,
deleted_source,
duration_secs: t0.elapsed().as_secs_f32(),
}))
}
/// Stream a single object from one bucket to another. Uses the existing
/// `storaged::ops` get + put primitives — no native copy in object_store
/// across heterogeneous backends (local ↔ S3), so an in-memory hop is
/// unavoidable. Bounded by individual object size, which for our parquet
/// + jsonl artifacts tops out around a few hundred MB.
async fn copy_key(
src: &Arc<dyn ObjectStore>,
dst: &Arc<dyn ObjectStore>,
key: &str,
) -> Result<(), String> {
let data = storaged::ops::get(src, key).await?;
storaged::ops::put(dst, key, data).await
}
// --- unused legacy function below, kept for reference ---
#[allow(dead_code)]
@@ -420,6 +602,15 @@ struct HybridRequest {
/// If false, just return the ranked matches (faster, no Ollama gen).
#[serde(default = "default_true")]
generate: bool,
/// Phase 19: consult `playbook_memory` and boost workers that past
/// similar playbooks successfully filled. Off by default so current
/// callers keep deterministic ranking; opt-in unlocks the feedback.
#[serde(default)]
use_playbook_memory: bool,
/// Number of past playbooks to consider when `use_playbook_memory`
/// is on. Ignored otherwise. Defaults to 5.
#[serde(default)]
playbook_memory_k: Option<usize>,
}
fn default_true() -> bool { true }
@@ -442,8 +633,18 @@ struct HybridSource {
chunk_text: String,
score: f32,
sql_verified: bool,
/// Phase 19: how much the playbook_memory boost lifted this hit's
/// score. 0.0 when `use_playbook_memory=false` or no past playbook
/// endorsed this worker.
#[serde(default, skip_serializing_if = "is_zero")]
playbook_boost: f32,
/// playbook_ids whose endorsement contributed to `playbook_boost`.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
playbook_citations: Vec<String>,
}
fn is_zero(x: &f32) -> bool { x.abs() < 1e-6 }
async fn hybrid_search(
State(state): State<VectorState>,
Json(req): Json<HybridRequest>,
@@ -556,6 +757,11 @@ async fn hybrid_search(
.and_then(|m| m.id_prefix.clone());
let sql_count = valid_ids.as_ref().map(|s| s.len()).unwrap_or(0);
// Phase 19: when playbook_memory is consulted, pull a wider candidate
// pool so endorsed workers outside the vanilla top-K can still be
// boosted into visibility. 5× is a conservative multiplier — plenty
// for a +0.25 boost to flip rankings without dragging the cost up.
let fetch_k = if req.use_playbook_memory { req.top_k * 5 } else { req.top_k };
let filtered: Vec<search::SearchResult> = if let Some(ref ids) = valid_ids {
all_results.into_iter()
.filter(|r| {
@@ -572,20 +778,54 @@ async fn hybrid_search(
};
ids.contains(raw_id)
})
.take(req.top_k)
.take(fetch_k)
.collect()
} else {
all_results.into_iter().take(req.top_k).collect()
all_results.into_iter().take(fetch_k).collect()
};
// Step 4: Build sources with SQL-verified flag.
let sources: Vec<HybridSource> = filtered.iter().map(|r| HybridSource {
let mut sources: Vec<HybridSource> = filtered.iter().map(|r| HybridSource {
doc_id: r.doc_id.clone(),
chunk_text: r.chunk_text.clone(),
score: r.score,
sql_verified: valid_ids.is_some(),
playbook_boost: 0.0,
playbook_citations: Vec::new(),
}).collect();
// Step 4b (Phase 19): if use_playbook_memory, look up semantically
// similar past playbooks and boost workers they endorsed. Name-match
// is on the tuple (city, state, name) extracted from chunk_text —
// hybrid_search's SQL filter already narrowed to one city+state, so
// this just needs to check the name against each playbook's endorsed
// set. Additive boost on the existing vector score, then re-sort.
if req.use_playbook_memory {
let boost_k = req.playbook_memory_k.unwrap_or(playbook_memory::DEFAULT_TOP_K_PLAYBOOKS);
// We embedded the question as `qv` above — reuse it for the
// playbook similarity lookup so we don't double-pay Ollama.
let boosts = state.playbook_memory.compute_boost_for(&qv, boost_k, 0.5).await;
for src in sources.iter_mut() {
// Parse "{Name} — {Role} in {City}, {State}. …" chunk. Being
// defensive: chunks from other datasets may not follow this
// exact shape, so absent fields just skip the boost.
if let Some((name, city, state)) = parse_worker_chunk(&src.chunk_text) {
let key = (city, state, name);
if let Some(entry) = boosts.get(&key) {
src.score += entry.boost;
src.playbook_boost = entry.boost;
src.playbook_citations = entry.citations.clone();
}
}
}
// Re-rank: boosted scores can flip ordering.
sources.sort_by(|a, b| b.score.partial_cmp(&a.score).unwrap_or(std::cmp::Ordering::Equal));
// Finally trim to the caller's requested top_k — we pulled fetch_k
// (5× wider) above specifically so the boost could reach workers
// that would otherwise have been trimmed pre-boost.
sources.truncate(req.top_k);
}
// Step 5: Generate answer if requested.
let answer = if req.generate && !sources.is_empty() {
let context: String = sources.iter().enumerate().map(|(i, s)| {
@@ -734,7 +974,7 @@ async fn run_trial(
State(state): State<VectorState>,
Json(req): Json<TrialRequest>,
) -> Result<Json<trial::Trial>, (StatusCode, String)> {
let mut harness_set = harness::EvalSet::load(&state.store, &req.harness)
let mut harness_set = state.harness_store.load_for_index(&req.index_name, &req.harness)
.await
.map_err(|e| (StatusCode::NOT_FOUND, format!("harness not found: {e}")))?;
@@ -764,8 +1004,8 @@ async fn run_trial(
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("ground truth: {e}")))?;
tracing::info!("trial: ground truth built in {:.1}s", t0.elapsed().as_secs_f32());
harness_set
.save(&state.store)
state.harness_store
.save(&harness_set)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("save harness: {e}")))?;
}
@@ -890,17 +1130,14 @@ async fn best_trial(
// --- Harness management ---
async fn list_evals(State(state): State<VectorState>) -> impl IntoResponse {
match harness::EvalSet::list(&state.store).await {
Ok(names) => Ok(Json(names)),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
Json(state.harness_store.list_all().await)
}
async fn get_eval(
State(state): State<VectorState>,
Path(name): Path<String>,
) -> impl IntoResponse {
match harness::EvalSet::load(&state.store, &name).await {
match state.harness_store.get_any(&name).await {
Ok(e) => Ok(Json(e)),
Err(err) => Err((StatusCode::NOT_FOUND, err)),
}
@@ -916,7 +1153,7 @@ async fn put_eval(
.queries
.iter()
.all(|q| q.ground_truth.is_some());
match harness_set.save(&state.store).await {
match state.harness_store.save(&harness_set).await {
Ok(()) => Ok(Json(harness_set)),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
@@ -957,8 +1194,8 @@ async fn autogen_eval(
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("ground truth: {e}")))?;
harness_set
.save(&state.store)
state.harness_store
.save(&harness_set)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("save: {e}")))?;
@@ -1407,7 +1644,12 @@ async fn profile_scoped_search(
let lance_store = state.lance.store_for(&req.index_name).await
.map_err(|e| (StatusCode::BAD_REQUEST, e))?;
let t0 = std::time::Instant::now();
match lance_store.search(&query_vec, top_k).await {
match lance_store.search(
&query_vec,
top_k,
Some(LANCE_DEFAULT_NPROBES),
Some(LANCE_DEFAULT_REFINE_FACTOR),
).await {
Ok(hits) => Ok(Json(serde_json::json!({
"profile": profile.id,
"source": index_meta.source,
@@ -1516,6 +1758,7 @@ async fn run_autotune_endpoint(
&state.index_registry,
&state.trial_journal,
&state.promotion_registry,
&state.harness_store,
&state.job_tracker,
).await {
Ok(result) => Ok(Json(result)),
@@ -1636,8 +1879,25 @@ struct LanceSearchRequest {
query: String,
#[serde(default = "default_top_k")]
top_k: usize,
/// IVF partitions to probe. `None` uses Lance's built-in default of
/// 1, which caps recall well below the index's real capability.
/// Recommended: 5–10% of num_partitions (≈20 for a 316-partition
/// index). Omitting it here picks the server-side default.
#[serde(default)]
nprobes: Option<usize>,
/// Refine factor — re-rank `top_k * factor` PQ-approximate candidates
/// with exact distances before returning `top_k`. Recovers recall
/// lost to product quantization.
#[serde(default)]
refine_factor: Option<u32>,
}
/// Server-side defaults when the caller doesn't pin nprobes / refine
/// themselves. Tuned for the ~100K × 768d reference workload; see
/// docs/ADR-019-vector-storage.md for the recall / latency trade-off.
const LANCE_DEFAULT_NPROBES: usize = 20;
const LANCE_DEFAULT_REFINE_FACTOR: u32 = 5;
fn default_top_k() -> usize { 5 }
/// Vector search against a Lance dataset. Embeds the query text via the
@@ -1660,7 +1920,9 @@ async fn lance_search(
.map_err(|e| (StatusCode::BAD_REQUEST, e))?;
let t0 = std::time::Instant::now();
let hits = lance_store.search(&qv, req.top_k).await
let nprobes = req.nprobes.or(Some(LANCE_DEFAULT_NPROBES));
let refine = req.refine_factor.or(Some(LANCE_DEFAULT_REFINE_FACTOR));
let hits = lance_store.search(&qv, req.top_k, nprobes, refine).await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e))?;
Ok(Json(serde_json::json!({
@@ -1761,6 +2023,12 @@ struct LanceRecallRequest {
harness: String,
#[serde(default = "default_top_k")]
top_k: usize,
/// Override server defaults so operators can sweep nprobes /
/// refine_factor to chart the recall-vs-latency curve.
#[serde(default)]
nprobes: Option<usize>,
#[serde(default)]
refine_factor: Option<u32>,
}
#[derive(serde::Serialize)]
@@ -1784,6 +2052,214 @@ struct LanceRecallQuery {
hits_returned: usize,
}
// --- Phase 19: playbook memory endpoints ---
/// Extract (name, city, state) from a chunk formatted like
/// "{Name} — {Role} in {City}, {State}. Skills: …".
/// Returns None if the chunk doesn't match the shape; callers simply
/// skip the boost for that hit.
fn parse_worker_chunk(chunk: &str) -> Option<(String, String, String)> {
// "Name — Role in City, ST. …" → split on "—" then " in " then ","
let (name_part, rest) = chunk.split_once('—')?;
let rest = rest.trim();
let (_role, loc_part) = rest.split_once(" in ")?;
let loc_part = loc_part.trim();
let (city, state_plus) = loc_part.split_once(',')?;
let state: String = state_plus.trim()
.chars()
.take_while(|c| c.is_ascii_alphabetic())
.collect();
let name = name_part.trim().to_string();
let city = city.trim().to_string();
if name.is_empty() || city.is_empty() || state.is_empty() {
return None;
}
Some((name, city, state))
}
#[derive(Deserialize)]
struct SeedPlaybookRequest {
/// One playbook with {operation, approach, context, endorsed_names}.
/// City + state are parsed from the operation text.
operation: String,
#[serde(default)]
approach: String,
#[serde(default)]
context: String,
endorsed_names: Vec<String>,
/// Append to the existing memory rather than replacing. Default true —
/// seeding is a bootstrap/demo tool, not a rebuild substitute.
#[serde(default = "default_true")]
append: bool,
}
/// Bootstrap / test-only: inject a playbook entry directly into
/// `playbook_memory` without going through `successful_playbooks`. Useful
/// when the source dataset has stale or phantom entries (as the initial
/// staffing seed did — names that don't correspond to real workers), and
/// you want to demonstrate the feedback loop with a known-good fixture.
///
/// Production path is always `/rebuild` — this endpoint is for operators
/// who need to prime the memory before real playbooks accumulate.
async fn seed_playbook_memory(
State(state): State<VectorState>,
Json(req): Json<SeedPlaybookRequest>,
) -> impl IntoResponse {
// Embed the entry through the same text shape `rebuild` uses so
// similarity math is comparable across seed + real entries.
let tmp_entry = playbook_memory::PlaybookEntry {
playbook_id: String::new(),
operation: req.operation.clone(),
approach: req.approach.clone(),
context: req.context.clone(),
timestamp: chrono::Utc::now().to_rfc3339(),
endorsed_names: req.endorsed_names.clone(),
city: None, state: None, embedding: None,
};
let text = format!(
"{} | {} | {} | fills: {}",
tmp_entry.operation, tmp_entry.approach, tmp_entry.context,
tmp_entry.endorsed_names.join(", "),
);
let resp = match state.ai_client.embed(EmbedRequest { texts: vec![text], model: None }).await {
Ok(r) => r,
Err(e) => return Err((StatusCode::BAD_GATEWAY, format!("embed seed: {e}"))),
};
if resp.embeddings.is_empty() {
return Err((StatusCode::BAD_GATEWAY, "embed returned nothing".into()));
}
let emb: Vec<f32> = resp.embeddings[0].iter().map(|&x| x as f32).collect();
// Parse city/state from the operation ("fill: Role xN in City, ST").
// Parser lives in playbook_memory::rebuild — expose via a tiny helper
// or inline the same logic here; duplicated briefly since this seed
// path is stable but infrequently called.
let (city, state_) = {
let after_in = req.operation.split(" in ").nth(1).unwrap_or("");
let mut parts = after_in.splitn(2, ',');
let city = parts.next().map(|s| s.trim().to_string()).filter(|s| !s.is_empty());
let state = parts.next().map(|s| s.trim().chars().take_while(|c| c.is_ascii_alphabetic()).collect::<String>()).filter(|s| !s.is_empty());
(city, state)
};
if city.is_none() || state_.is_none() {
return Err((StatusCode::BAD_REQUEST,
"operation must match 'fill: Role xN in City, ST' shape".into()));
}
// Stable id: hash of timestamp + operation. Callers get the id back
// so they can reference it in citations.
let ts = chrono::Utc::now().to_rfc3339();
use sha2::{Digest, Sha256};
let mut h = Sha256::new();
h.update(ts.as_bytes());
h.update(b"|");
h.update(req.operation.as_bytes());
let bytes = h.finalize();
let pid = format!("pb-seed-{}", bytes.iter().take(8).map(|b| format!("{b:02x}")).collect::<String>());
let new_entry = playbook_memory::PlaybookEntry {
playbook_id: pid.clone(),
operation: req.operation,
approach: req.approach,
context: req.context,
timestamp: ts,
endorsed_names: req.endorsed_names,
city, state: state_,
embedding: Some(emb),
};
let mut current = state.playbook_memory.snapshot().await;
if req.append {
current.push(new_entry);
} else {
current = vec![new_entry];
}
if let Err(e) = state.playbook_memory.set_entries(current).await {
return Err((StatusCode::INTERNAL_SERVER_ERROR, format!("persist: {e}")));
}
Ok(Json(serde_json::json!({ "playbook_id": pid, "entries_after": state.playbook_memory.entry_count().await })))
}
async fn rebuild_playbook_memory(
State(state): State<VectorState>,
) -> impl IntoResponse {
match playbook_memory::rebuild(
&state.playbook_memory,
&state.ai_client,
&state.catalog,
&state.bucket_registry,
).await {
Ok(report) => Ok(Json(report)),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
}
// Path 2 foundation — dump in-memory playbook_memory state to a fresh
// `successful_playbooks_live` dataset. Cheap to call (writes one parquet,
// updates one manifest), so /log can call it after every seed to keep the
// SQL-queryable surface honest without the destructive REPLACE bug that
// /ingest/file has.
async fn persist_playbook_memory_sql(
State(state): State<VectorState>,
) -> impl IntoResponse {
match playbook_memory::persist_to_sql(&state.playbook_memory, &state.catalog).await {
Ok(report) => Ok(Json(report)),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
}
#[derive(Deserialize)]
struct PatternsRequest {
query: String,
#[serde(default = "default_pattern_k")]
top_k_playbooks: usize,
/// Minimum frequency (0.0-1.0) for a trait to make the report.
/// Default 0.4 — at least 40% of examined workers must share it.
#[serde(default = "default_pattern_min_freq")]
min_trait_frequency: f32,
}
fn default_pattern_k() -> usize { 10 }
fn default_pattern_min_freq() -> f32 { 0.4 }
// Path 2 — meta-index discovery surface. "What did past similar fills
// have in common that I didn't ask about?" — surfaces signals like
// recurring certifications, skill clusters, archetype tendencies.
async fn discover_playbook_patterns(
State(state): State<VectorState>,
Json(req): Json<PatternsRequest>,
) -> impl IntoResponse {
match playbook_memory::discover_patterns(
&state.playbook_memory,
&state.ai_client,
&state.catalog,
&state.bucket_registry,
&req.query,
req.top_k_playbooks,
req.min_trait_frequency,
).await {
Ok(report) => Ok(Json(report)),
Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
}
}
async fn playbook_memory_stats(
State(state): State<VectorState>,
) -> impl IntoResponse {
let entries = state.playbook_memory.snapshot().await;
Json(serde_json::json!({
"entries": entries.len(),
"total_names_endorsed": entries.iter().map(|e| e.endorsed_names.len()).sum::<usize>(),
"entries_with_embeddings": entries.iter().filter(|e| e.embedding.is_some()).count(),
"sample": entries.iter().take(3).map(|e| serde_json::json!({
"id": e.playbook_id,
"operation": e.operation,
"city": e.city,
"state": e.state,
"endorsed": e.endorsed_names,
})).collect::<Vec<_>>(),
}))
}
async fn lance_recall_harness(
State(state): State<VectorState>,
Path(index_name): Path<String>,
@@ -1791,7 +2267,7 @@ async fn lance_recall_harness(
) -> impl IntoResponse {
let t0 = std::time::Instant::now();
let harness_set = harness::EvalSet::load(&state.store, &req.harness).await
let harness_set = state.harness_store.load_for_index(&index_name, &req.harness).await
.map_err(|e| (StatusCode::NOT_FOUND, format!("harness: {e}")))?;
if !harness_set.ground_truth_built {
return Err((StatusCode::BAD_REQUEST,
@@ -1817,7 +2293,12 @@ async fn lance_recall_harness(
};
let qt0 = std::time::Instant::now();
let hits = lance_store.search(qv, k).await
let hits = lance_store.search(
qv,
k,
Some(req.nprobes.unwrap_or(LANCE_DEFAULT_NPROBES)),
Some(req.refine_factor.unwrap_or(LANCE_DEFAULT_REFINE_FACTOR)),
).await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("search: {e}")))?;
let lat_us = qt0.elapsed().as_micros() as f32;