use super::*;
impl KnowledgeBase {
pub fn inspect(&self) -> Result<Value> {
let total: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE origin!='spark'",
)?;
let active: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE state='active' AND origin!='spark'",
)?;
let pending: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE state='pending' AND origin!='spark'",
)?;
let archived: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE state='archived' AND origin!='spark'",
)?;
let sparks: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE origin='spark' AND state!='archived'",
)?;
let open_logs: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM episodic_log WHERE distill_state='open'",
)?;
let new_logs: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM episodic_log WHERE distill_state='new'",
)?;
let embed_rebuild: i64 = count_query(&self.storage,
"SELECT COUNT(*) FROM chunks WHERE embed_version=0 OR embed_version < (SELECT COALESCE(CAST(value AS INTEGER),1) FROM meta WHERE key='embed_version')")?;
let schema_version = self.storage.get_meta_or("schema_version", "?");
let lib_id = self.storage.get_meta_or("lib_id", "?");
let last_agg = self.storage.get_meta_or("last_agg_ts", "never");
let metric_window_start = days_ago(&utc_now_iso(), 30);
let trace_metrics = self.storage.query_chunks_params(
"SELECT COUNT(*) AS total,
SUM(CASE WHEN task_state='completed' THEN 1 ELSE 0 END) AS completed,
SUM(CASE WHEN task_state='timed_out' THEN 1 ELSE 0 END) AS timed_out,
SUM(CASE WHEN task_state='completed' AND usage_state!='unknown'
THEN 1 ELSE 0 END) AS usage_known,
SUM(CASE WHEN task_state='completed' AND usage_state='known_some'
THEN 1 ELSE 0 END) AS usage_some,
SUM(CASE WHEN task_state='completed'
AND outcome IN ('ok','fail')
THEN 1 ELSE 0 END) AS outcome_known,
SUM(CASE WHEN outcome='ok' THEN 1 ELSE 0 END) AS succeeded
FROM episodic_log WHERE ts >= ?",
rusqlite::params![metric_window_start],
)?;
let trace_row = trace_metrics.first();
let trace_total = trace_row
.and_then(|row| row.get("total"))
.and_then(Value::as_i64)
.unwrap_or(0);
let trace_completed = trace_row
.and_then(|row| row.get("completed"))
.and_then(Value::as_i64)
.unwrap_or(0);
let trace_timed_out = trace_row
.and_then(|row| row.get("timed_out"))
.and_then(Value::as_i64)
.unwrap_or(0);
let usage_known = trace_row
.and_then(|row| row.get("usage_known"))
.and_then(Value::as_i64)
.unwrap_or(0);
let usage_some = trace_row
.and_then(|row| row.get("usage_some"))
.and_then(Value::as_i64)
.unwrap_or(0);
let succeeded = trace_row
.and_then(|row| row.get("succeeded"))
.and_then(Value::as_i64)
.unwrap_or(0);
let outcome_known = trace_row
.and_then(|row| row.get("outcome_known"))
.and_then(Value::as_i64)
.unwrap_or(0);
let usage_rows = self.storage.query_chunks_params(
"SELECT recall_snapshot, used_ids FROM episodic_log
WHERE task_state='completed'
AND usage_state!='unknown' AND used_complete=1
AND recall_snapshot IS NOT NULL AND used_ids IS NOT NULL
AND ts >= ?",
rusqlite::params![metric_window_start],
)?;
let mut selected_total = 0_i64;
let mut selected_used = 0_i64;
for row in usage_rows {
let selected: HashSet<String> = row
.get("recall_snapshot")
.and_then(Value::as_str)
.and_then(|raw| serde_json::from_str::<Value>(raw).ok())
.and_then(|snapshot| snapshot.get("selected").cloned())
.and_then(|value| serde_json::from_value::<Vec<String>>(value).ok())
.unwrap_or_default()
.into_iter()
.collect();
let used: HashSet<String> = row
.get("used_ids")
.and_then(Value::as_str)
.and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
.unwrap_or_default()
.into_iter()
.collect();
selected_total += selected.len() as i64;
selected_used += selected.intersection(&used).count() as i64;
}
let feedback_count = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM feedback_events WHERE ts >= ?",
rusqlite::params![metric_window_start],
)?;
let feedback_traces = count_query_params(
&self.storage,
"SELECT COUNT(DISTINCT f.trace_id)
FROM feedback_events f
JOIN episodic_log e ON e.trace_id=f.trace_id
WHERE f.ts >= ? AND e.ts >= ? AND e.task_state='completed'",
rusqlite::params![metric_window_start, metric_window_start],
)?;
let pending_evolve = count_query(
&self.storage,
"SELECT COUNT(*) FROM evolve_requests WHERE state IN ('pending','running')",
)?;
let governance_pending = count_query(
&self.storage,
"SELECT COUNT(*) FROM governance_proposals WHERE state='pending'",
)?;
let failed_evolve = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM evolve_requests
WHERE last_failed_at >= ?",
rusqlite::params![metric_window_start],
)?;
let failed_distill = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM episodic_log
WHERE distill_last_failed_at >= ?",
rusqlite::params![metric_window_start],
)?;
let confidence_buckets = self.storage.query_chunks(&format!(
"SELECT
SUM(CASE WHEN confidence < 0.25 THEN 1 ELSE 0 END) AS low,
SUM(CASE WHEN confidence >= 0.25 AND confidence < {0} THEN 1 ELSE 0 END) AS medium,
SUM(CASE WHEN confidence >= {0} THEN 1 ELSE 0 END) AS high
FROM chunks WHERE origin!='spark' AND state!='archived'",
self.promote_confidence_min
))?;
let confidence_row = confidence_buckets.first();
let pending_oldest_ts = self.storage.query_chunks(
"SELECT MIN(created_at) AS oldest FROM chunks WHERE state='pending' AND origin!='spark'",
)?.into_iter().next()
.and_then(|r| r.get("oldest").cloned())
.filter(|v| !v.is_null());
let zombie_cutoff = days_ago(&utc_now_iso(), 14);
let zombie: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM chunks
WHERE origin!='spark' AND state='active'
AND confidence >= 0.4 AND confidence <= 0.6
AND last_used_at IS NOT NULL
AND created_at < ?",
rusqlite::params![zombie_cutoff],
)?;
let debt_numerator = pending + zombie;
let debt_denominator = active.max(1);
let debt_ratio = debt_numerator as f64 / debt_denominator as f64;
let screening_cutoff = minutes_ago(&utc_now_iso(), self.screening_timeout_minutes);
let stale_screening: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM episodic_log
WHERE distill_state='screening' AND distill_locked_at < ?",
rusqlite::params![screening_cutoff],
)?;
let distill_period_start = self.distill_token_period_start(&utc_now_iso())?;
let distill_cost = self.storage.query_chunks_params(
"SELECT COALESCE(SUM(prompt_tokens),0) AS pt,
COALESCE(SUM(completion_tokens),0) AS ct
FROM distill_token_usage
WHERE accounted_at >= ?",
rusqlite::params![distill_period_start],
)?;
let prompt_tokens = distill_cost
.first()
.and_then(|r| r.get("pt"))
.and_then(Value::as_i64)
.unwrap_or(0);
let completion_tokens = distill_cost
.first()
.and_then(|r| r.get("ct"))
.and_then(Value::as_i64)
.unwrap_or(0);
let spark_threshold: i64 = self
.storage
.get_meta("curate.soft_mature_threshold")
.ok()
.flatten()
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(5);
let recurring_sparks = self.storage.query_chunks_params(
"SELECT ut.chunk_id, COUNT(*) AS cnt,
c.content, c.trigger_desc, c.maturity
FROM usage_trace ut
JOIN chunks c ON c.id = ut.chunk_id
WHERE ut.event='retrieved'
AND c.origin='spark'
GROUP BY ut.chunk_id HAVING cnt >= ?",
rusqlite::params![spark_threshold],
)?;
let recurring_spark_ids: Vec<Value> = recurring_sparks
.iter()
.map(|r| {
json!({
"id": r.get("chunk_id").and_then(Value::as_str).unwrap_or(""),
"retrieved_count": r.get("cnt").and_then(Value::as_i64).unwrap_or(0),
"maturity": r.get("maturity").and_then(Value::as_str).unwrap_or(""),
"content_preview": r.get("content").and_then(Value::as_str).unwrap_or("")
.chars().take(80).collect::<String>(),
})
})
.collect();
let mut suggestions: Vec<Value> = Vec::new();
if embed_rebuild > 0 {
suggestions.push(json!({"action": "innate evolve --rebuild-embeddings", "reason": format!("{embed_rebuild} chunk(s) missing embeddings")}));
}
if new_logs > 0 {
suggestions.push(json!({"action": "innate evolve --trigger manual", "reason": format!("{new_logs} episodic log(s) ready to distill")}));
}
if pending > 0 {
suggestions.push(json!({"action": "innate approve <id> # or innate archive <id>", "reason": format!("{pending} pending chunk(s) awaiting review")}));
}
if !recurring_spark_ids.is_empty() {
suggestions.push(json!({"action": "innate promote-spark <id> --to note", "reason": format!("{} spark(s) recalled ≥{spark_threshold}× — consider promoting", recurring_spark_ids.len())}));
}
if stale_screening > 0 {
suggestions.push(json!({"action": "innate evolve --trigger manual", "reason": format!("{stale_screening} episodic log(s) stuck in screening")}));
}
if governance_pending > 0 {
suggestions.push(json!({
"action": "review governance_proposals",
"reason": format!("{governance_pending} chunk(s) have repeated negative feedback")
}));
}
let mcpcli_null: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM episodic_log
WHERE event_source IN ('mcp','cli') AND agent IS NULL AND ts >= ?1",
rusqlite::params![days_ago(&utc_now_iso(), 7)],
)?;
if mcpcli_null > 0 {
suggestions.push(json!({
"action": "set INNATE_AGENT in the agent's MCP env (innate install)",
"reason": format!("{mcpcli_null} mcp/cli call(s) in 7d have NULL agent — agent attribution misconfigured")
}));
}
let intuition = self.intuition_calibration(&metric_window_start)?;
let appraisals = intuition
.get("appraisals")
.and_then(Value::as_i64)
.unwrap_or(0);
let mono_gap = intuition
.get("monotonicity_gap")
.and_then(Value::as_f64)
.unwrap_or(0.0);
let false_alarm = intuition
.get("false_alarm_rate")
.and_then(Value::as_f64)
.unwrap_or(0.0);
if appraisals >= 20 && mono_gap <= 0.0 {
suggestions.push(json!({
"action": "tune recall.w_* / situation.coarse_keys",
"reason": "appraise strength may be noise — strong tier does not beat weak on task_ok"
}));
}
if appraisals >= 20 && false_alarm >= 0.5 {
suggestions.push(json!({
"action": "review caution chunks / raise appraise.tier_strong",
"reason": format!("intuition false-alarm rate {false_alarm} — strong cautions often end ok")
}));
}
let usage_trace_total = count_query(&self.storage, "SELECT COUNT(*) FROM usage_trace")?;
let episodic_log_total = count_query(&self.storage, "SELECT COUNT(*) FROM episodic_log")?;
let page_count = count_query(&self.storage, "PRAGMA page_count")?;
let page_size = count_query(&self.storage, "PRAGMA page_size")?;
let db_size_bytes = page_count * page_size;
let now_obs = utc_now_iso();
let observability = self.observability_block(&now_obs)?;
let operational = self.operational_block(&now_obs)?;
let trends = self.trends_block(&now_obs)?;
let mut out = json!({
"schema_version": schema_version,
"lib_id": lib_id,
"last_agg_ts": last_agg,
"chunks": {
"total": total, "active": active, "pending": pending, "archived": archived,
"pending_oldest_ts": pending_oldest_ts,
},
"storage": {
"usage_trace_rows": usage_trace_total,
"episodic_log_rows": episodic_log_total,
"db_size_bytes": db_size_bytes,
"db_size_mb": (db_size_bytes as f64 / 1_048_576.0 * 100.0).round() / 100.0,
},
"sparks": sparks,
"episodic_log": {"open": open_logs, "new": new_logs},
"embed_rebuild_queue": embed_rebuild,
"knowledge_debt_ratio": (debt_ratio * 100.0).round() / 100.0,
"stale_screening_count": stale_screening,
"feedback_loop": {
"trace_completion_rate": ratio(trace_completed, trace_total),
"usage_annotation_rate": ratio(usage_known, trace_completed),
"trace_use_rate": ratio(usage_some, usage_known),
"selected_to_used_rate": ratio(selected_used, selected_total),
"task_success_rate": ratio(succeeded, outcome_known),
"feedback_coverage": ratio(feedback_traces, trace_completed),
"feedback_events": feedback_count,
"timed_out_traces": trace_timed_out,
"pending_evolve_requests": pending_evolve,
"failed_evolve_requests_30d": failed_evolve,
"failed_distill_logs_30d": failed_distill,
"pending_governance_proposals": governance_pending,
"window_days": 30,
"confidence_distribution": {
"low": confidence_row.and_then(|row| row.get("low")).and_then(Value::as_i64).unwrap_or(0),
"medium": confidence_row.and_then(|row| row.get("medium")).and_then(Value::as_i64).unwrap_or(0),
"high": confidence_row.and_then(|row| row.get("high")).and_then(Value::as_i64).unwrap_or(0),
}
},
"intuition_calibration": intuition,
"distill_cost_estimate": {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens},
"recurring_sparks": recurring_sparks.len(),
"recurring_spark_ids": recurring_spark_ids,
"params": {
"recall.w_content": self.w_content,
"recall.w_trigger": self.w_trigger,
"recall.w_context": self.w_context,
"recall.w_activation": self.w_activation,
"recall.w_spread": self.w_spread,
"recall.top_k_candidates": self.top_k_candidates,
"curate.low_conf_threshold": self.low_conf_threshold,
"curate.low_conf_idle_days": self.low_conf_idle_days,
"curate.repeat_select_min": self.repeat_select_min,
"curate.never_used_age_days": self.never_used_age_days,
"curate.promote_used_success_min": self.promote_used_success_min,
"curate.promote_confidence_min": self.promote_confidence_min,
"curate.screening_timeout_minutes": self.screening_timeout_minutes,
"curate.open_ttl_days": self.open_ttl_days,
"curate.log_compact_days": self.log_compact_days,
"evolve.schedule_interval_hours": self.evolve_schedule_interval_hours,
},
"suggestions": suggestions
});
if let Some(obj) = out.as_object_mut() {
obj.insert("observability".to_string(), observability);
obj.insert("operational".to_string(), operational);
if let Some(trends) = trends {
obj.insert("trends".to_string(), trends);
}
}
Ok(out)
}
fn observability_block(&self, now: &str) -> Result<Value> {
let windows = json!({
"1d": self.window_rates(&days_ago(now, 1))?,
"7d": self.window_rates(&days_ago(now, 7))?,
"30d": self.window_rates(&days_ago(now, 30))?,
});
let mut by_source = serde_json::Map::new();
let rows = self.storage.query_chunks_params(
"SELECT COALESCE(event_source,'unknown') AS src, COUNT(*) AS total,
SUM(CASE WHEN agent IS NOT NULL THEN 1 ELSE 0 END) AS with_agent
FROM episodic_log WHERE ts >= ?1 GROUP BY event_source",
rusqlite::params![days_ago(now, 30)],
)?;
for r in &rows {
let src = r
.get("src")
.and_then(Value::as_str)
.unwrap_or("unknown")
.to_string();
let total = r.get("total").and_then(Value::as_i64).unwrap_or(0);
let with_agent = r.get("with_agent").and_then(Value::as_i64).unwrap_or(0);
by_source.insert(
src,
json!({
"total": total,
"with_agent": with_agent,
"agent_coverage": ratio(with_agent, total),
}),
);
}
let cut7 = days_ago(now, 7);
let by_dimension = json!({
"event_source": { "agent_coverage": by_source,
"rates": self.dimension_rates("event_source", &cut7, 0)? },
"agent": self.dimension_rates("agent", &cut7, 0)?,
"context_key": self.dimension_rates("context_key", &cut7, 10)?,
});
let zombie: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM chunks
WHERE state!='archived' AND origin!='spark'
AND selected_count >= ?1 AND used_count = 0",
rusqlite::params![self.repeat_select_min],
)?;
let avg_pack = self.storage.query_chunks_params(
"SELECT
AVG(r) AS avg_retrieved,
AVG(s) AS avg_selected
FROM (
SELECT trace_id,
SUM(CASE WHEN event='retrieved' THEN 1 ELSE 0 END) AS r,
SUM(CASE WHEN event='selected' THEN 1 ELSE 0 END) AS s
FROM usage_trace WHERE ts >= ?1 GROUP BY trace_id)",
rusqlite::params![cut7],
)?;
let avg_get = |k: &str| {
avg_pack
.first()
.and_then(|r| r.get(k))
.and_then(Value::as_f64)
.map(|x| (x * 100.0).round() / 100.0)
.unwrap_or(0.0)
};
let selected_unused_top = self.storage.query_chunks(
"SELECT id, selected_count FROM chunks
WHERE state!='archived' AND origin!='spark' AND used_count=0 AND selected_count>0
ORDER BY selected_count DESC LIMIT 5",
)?;
let sel_any: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE origin!='spark' AND selected_count>0",
)?;
let sel_unused: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE origin!='spark' AND selected_count>0 AND used_count=0",
)?;
let mrr_rows = self.storage.query_chunks_params(
"SELECT AVG(1.0/ut.rank) AS mrr FROM usage_trace ut
JOIN chunks c ON c.id = ut.chunk_id
WHERE ut.event='selected' AND ut.rank IS NOT NULL AND ut.rank>0
AND c.used_count>0 AND ut.ts >= ?1",
rusqlite::params![cut7],
)?;
let used_rank_mrr = mrr_rows
.first()
.and_then(|r| r.get("mrr"))
.and_then(Value::as_f64)
.map(|x| (x * 1000.0).round() / 1000.0)
.unwrap_or(0.0);
let hook_total: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM episodic_log WHERE event_source='hook' AND ts >= ?1",
rusqlite::params![cut7],
)?;
let hook_silent: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM episodic_log
WHERE event_source='hook' AND usage_state='known_none' AND ts >= ?1",
rusqlite::params![cut7],
)?;
let rank_hist = self.storage.query_chunks_params(
"SELECT
SUM(CASE WHEN rank=1 THEN 1 ELSE 0 END) AS r1,
SUM(CASE WHEN rank BETWEEN 2 AND 3 THEN 1 ELSE 0 END) AS r2_3,
SUM(CASE WHEN rank BETWEEN 4 AND 10 THEN 1 ELSE 0 END) AS r4_10,
SUM(CASE WHEN rank > 10 THEN 1 ELSE 0 END) AS r11plus
FROM usage_trace WHERE event='selected' AND rank IS NOT NULL AND ts >= ?1",
rusqlite::params![cut7],
)?;
let rh = |k: &str| {
rank_hist
.first()
.and_then(|r| r.get(k))
.and_then(Value::as_i64)
.unwrap_or(0)
};
let high_rank_unused = self.storage.query_chunks_params(
"SELECT ut.chunk_id AS id, MIN(ut.rank) AS best_rank
FROM usage_trace ut JOIN chunks c ON c.id = ut.chunk_id
WHERE ut.event='selected' AND ut.rank<=3 AND c.used_count=0 AND ut.ts >= ?1
GROUP BY ut.chunk_id ORDER BY best_rank LIMIT 5",
rusqlite::params![cut7],
)?;
let low_rank_used = self.storage.query_chunks_params(
"SELECT ut.chunk_id AS id, MIN(ut.rank) AS best_rank
FROM usage_trace ut JOIN chunks c ON c.id = ut.chunk_id
WHERE ut.event='selected' AND ut.rank>10 AND c.used_count>0 AND ut.ts >= ?1
GROUP BY ut.chunk_id ORDER BY best_rank DESC LIMIT 5",
rusqlite::params![cut7],
)?;
let promotions: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM chunks
WHERE state='active' AND origin!='spark' AND state_updated_at >= ?1
AND state_reason IN ('repeated_success','approved','restore')",
rusqlite::params![cut7],
)?;
let evictions: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM chunks
WHERE state='archived' AND origin!='spark' AND state_updated_at >= ?1",
rusqlite::params![cut7],
)?;
let pending_oldest = self.storage.query_chunks(
"SELECT MIN(created_at) AS oldest FROM chunks WHERE state='pending' AND origin!='spark'",
)?;
let pending_oldest_ts = pending_oldest
.first()
.and_then(|r| r.get("oldest"))
.and_then(Value::as_str)
.map(|s| s.to_string());
let gov_backlog = self.storage.query_chunks(
"SELECT MIN(created_at) AS oldest FROM governance_proposals WHERE state='pending'",
)?;
let gov_backlog_oldest_ts = gov_backlog
.first()
.and_then(|r| r.get("oldest"))
.and_then(Value::as_str)
.map(|s| s.to_string());
Ok(json!({
"windows": windows,
"by_dimension": by_dimension,
"recall_pack": {
"zombie_chunks": zombie,
"avg_retrieved": avg_get("avg_retrieved"),
"avg_selected": avg_get("avg_selected"),
"selected_unused_rate": ratio(sel_unused, sel_any),
"selected_unused_top": selected_unused_top,
"used_rank_mrr": used_rank_mrr,
"hook_silence_rate": ratio(hook_silent, hook_total),
"selected_rank_distribution": {
"1": rh("r1"), "2-3": rh("r2_3"), "4-10": rh("r4_10"), "11+": rh("r11plus")
},
"high_rank_unused": high_rank_unused,
"low_rank_used": low_rank_used,
},
"lifecycle": {
"pending_oldest_ts": pending_oldest_ts,
"governance_backlog_oldest_ts": gov_backlog_oldest_ts,
"state_transition_approx": {
"promotions_7d": promotions,
"evictions_7d": evictions,
"note": "approx via state_updated_at/state_reason; not a strict rate"
}
}
}))
}
fn dimension_rates(&self, group_col: &str, cutoff: &str, top_n: i64) -> Result<Value> {
let limit = if top_n > 0 {
format!(" ORDER BY recalls DESC LIMIT {top_n}")
} else {
String::new()
};
let sql = format!(
"SELECT COALESCE({group_col},'(null)') AS g,
COUNT(*) AS recalls,
SUM(CASE WHEN usage_state='known_none' THEN 1 ELSE 0 END) AS empty,
SUM(CASE WHEN task_state='completed' THEN 1 ELSE 0 END) AS completed,
SUM(CASE WHEN outcome='ok' THEN 1 ELSE 0 END) AS ok,
SUM(CASE WHEN outcome IN ('ok','fail') THEN 1 ELSE 0 END) AS outcome_known,
SUM(CASE WHEN task_state='completed' AND usage_state!='unknown' THEN 1 ELSE 0 END) AS annotated
FROM episodic_log WHERE ts >= ?1 GROUP BY {group_col}{limit}"
);
let rows = self.storage.query_chunks_params(&sql, rusqlite::params![cutoff])?;
let su_sql = format!(
"SELECT COALESCE(el.{group_col},'(null)') AS g,
SUM(CASE WHEN ut.event='selected' THEN 1 ELSE 0 END) AS sel,
SUM(CASE WHEN ut.event='used' THEN 1 ELSE 0 END) AS used
FROM usage_trace ut JOIN episodic_log el ON el.trace_id = ut.trace_id
WHERE ut.ts >= ?1 GROUP BY el.{group_col}"
);
let su_rows = self.storage.query_chunks_params(&su_sql, rusqlite::params![cutoff])?;
let mut su_map: std::collections::HashMap<String, (i64, i64)> = std::collections::HashMap::new();
for r in &su_rows {
let g = r.get("g").and_then(Value::as_str).unwrap_or("(null)").to_string();
su_map.insert(
g,
(
r.get("sel").and_then(Value::as_i64).unwrap_or(0),
r.get("used").and_then(Value::as_i64).unwrap_or(0),
),
);
}
let fb_sql = format!(
"SELECT COALESCE(el.{group_col},'(null)') AS g,
COUNT(DISTINCT fe.trace_id) AS fb
FROM feedback_events fe JOIN episodic_log el ON el.trace_id = fe.trace_id
WHERE fe.ts >= ?1 GROUP BY el.{group_col}"
);
let fb_rows = self.storage.query_chunks_params(&fb_sql, rusqlite::params![cutoff])?;
let mut fb_map: std::collections::HashMap<String, i64> = std::collections::HashMap::new();
for r in &fb_rows {
let g = r.get("g").and_then(Value::as_str).unwrap_or("(null)").to_string();
fb_map.insert(g, r.get("fb").and_then(Value::as_i64).unwrap_or(0));
}
let mut out = serde_json::Map::new();
for r in &rows {
let g = r.get("g").and_then(Value::as_str).unwrap_or("(null)").to_string();
let i = |k: &str| r.get(k).and_then(Value::as_i64).unwrap_or(0);
let recalls = i("recalls");
let completed = i("completed");
let (sel, used) = su_map.get(&g).copied().unwrap_or((0, 0));
let fb = fb_map.get(&g).copied().unwrap_or(0);
out.insert(
g,
json!({
"recalls": recalls,
"empty_recall_rate": ratio(i("empty"), recalls),
"completed_rate": ratio(completed, recalls),
"task_success_rate": ratio(i("ok"), i("outcome_known")),
"usage_annotation_rate": ratio(i("annotated"), completed),
"selected_to_used_rate": ratio(used, sel),
"feedback_coverage": ratio(fb, completed),
}),
);
}
Ok(Value::Object(out))
}
fn window_rates(&self, cutoff: &str) -> Result<Value> {
let rows = self.storage.query_chunks_params(
"SELECT COUNT(*) AS recalls,
SUM(CASE WHEN usage_state='known_none' THEN 1 ELSE 0 END) AS empty,
SUM(CASE WHEN task_state='completed' THEN 1 ELSE 0 END) AS completed,
SUM(CASE WHEN task_state='timed_out' THEN 1 ELSE 0 END) AS timed_out,
SUM(CASE WHEN outcome='ok' THEN 1 ELSE 0 END) AS ok,
SUM(CASE WHEN outcome IN ('ok','fail') THEN 1 ELSE 0 END) AS outcome_known,
SUM(CASE WHEN task_state='completed' AND usage_state!='unknown' THEN 1 ELSE 0 END) AS annotated
FROM episodic_log WHERE ts >= ?1",
rusqlite::params![cutoff],
)?;
let g = |k: &str| {
rows.first()
.and_then(|r| r.get(k))
.and_then(Value::as_i64)
.unwrap_or(0)
};
let su = self.storage.query_chunks_params(
"SELECT SUM(CASE WHEN event='selected' THEN 1 ELSE 0 END) AS sel,
SUM(CASE WHEN event='used' THEN 1 ELSE 0 END) AS used
FROM usage_trace WHERE ts >= ?1",
rusqlite::params![cutoff],
)?;
let sug = |k: &str| {
su.first()
.and_then(|r| r.get(k))
.and_then(Value::as_i64)
.unwrap_or(0)
};
let fb_traces: i64 = count_query_params(
&self.storage,
"SELECT COUNT(DISTINCT trace_id) FROM feedback_events WHERE ts >= ?1",
rusqlite::params![cutoff],
)?;
let recalls = g("recalls");
let completed = g("completed");
Ok(json!({
"recalls": recalls,
"empty_recall_rate": ratio(g("empty"), recalls),
"completed_rate": ratio(completed, recalls),
"timeout_rate": ratio(g("timed_out"), recalls),
"task_success_rate": ratio(g("ok"), g("outcome_known")),
"usage_annotation_rate": ratio(g("annotated"), completed),
"selected_to_used_rate": ratio(sug("used"), sug("sel")),
"feedback_coverage": ratio(fb_traces, completed),
}))
}
fn operational_block(&self, now: &str) -> Result<Value> {
let daemon = crate::daemon::health(
&crate::paths::daemon_state_path(),
&crate::paths::daemon_pid_path(),
now,
);
let mut block = serde_json::Map::new();
block.insert("daemon".to_string(), daemon);
if self.storage.count_operation_runs().unwrap_or(0) > 0 {
let rows = self.storage.operation_runs_since(&days_ago(now, 7))?;
block.insert(
"ops".to_string(),
crate::storage::metrics::aggregate_ops(&rows),
);
}
Ok(Value::Object(block))
}
fn trends_block(&self, now: &str) -> Result<Option<Value>> {
let Some((cur_ts, cur_json)) = self.storage.latest_snapshot()? else {
return Ok(None);
};
let cur: Value = serde_json::from_str(&cur_json).unwrap_or_else(|_| json!({}));
let mut out = json!({ "current_ts": cur_ts, "current": cur.clone() });
if let Some((base_ts, base_json)) =
self.storage.snapshot_at_or_before(&days_ago(now, 7))?
{
let base: Value = serde_json::from_str(&base_json).unwrap_or_else(|_| json!({}));
let mut delta = serde_json::Map::new();
if let (Some(c), Some(b)) = (cur.as_object(), base.as_object()) {
for (k, cv) in c {
if let (Some(cf), Some(bf)) = (cv.as_f64(), b.get(k).and_then(Value::as_f64)) {
delta.insert(k.clone(), json!(((cf - bf) * 1000.0).round() / 1000.0));
}
}
}
out["baseline_ts"] = json!(base_ts);
out["delta_vs_7d"] = Value::Object(delta);
}
Ok(Some(out))
}
pub fn recall_weights(&self) -> Value {
json!({
"w_content": self.w_content,
"w_trigger": self.w_trigger,
"w_lexical": self.w_lexical,
"w_context": self.w_context,
"w_activation": self.w_activation,
"w_spread": self.w_spread,
})
}
pub fn write_metric_snapshot(&self) -> Result<Value> {
let now = utc_now_iso();
let kpis = self.collect_kpis(&now)?;
self.storage.insert_metric_snapshot(&now, &kpis.to_string())?;
Ok(kpis)
}
pub(crate) fn collect_kpis(&self, now: &str) -> Result<Value> {
let active: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE state='active' AND origin!='spark'",
)?;
let pending: i64 = count_query(
&self.storage,
"SELECT COUNT(*) FROM chunks WHERE state='pending' AND origin!='spark'",
)?;
let zombie: i64 = count_query_params(
&self.storage,
"SELECT COUNT(*) FROM chunks
WHERE state!='archived' AND origin!='spark'
AND selected_count >= ?1 AND used_count = 0",
rusqlite::params![self.repeat_select_min],
)?;
let debt_ratio = if active > 0 {
(pending as f64 / active as f64 * 100.0).round() / 100.0
} else {
pending as f64
};
let w7 = self.window_rates(&days_ago(now, 7))?;
let daemon = crate::daemon::health(
&crate::paths::daemon_state_path(),
&crate::paths::daemon_pid_path(),
now,
);
Ok(json!({
"active": active,
"pending": pending,
"knowledge_debt_ratio": debt_ratio,
"zombie_chunks": zombie,
"task_success_rate_7d": w7.get("task_success_rate").and_then(Value::as_f64).unwrap_or(0.0),
"empty_recall_rate_7d": w7.get("empty_recall_rate").and_then(Value::as_f64).unwrap_or(0.0),
"daemon_errors_24h": daemon.get("errors_24h").and_then(Value::as_i64).unwrap_or(0),
}))
}
fn intuition_calibration(&self, window_start: &str) -> Result<Value> {
let rows = self.storage.query_chunks_params(
"SELECT recall_snapshot, outcome FROM episodic_log
WHERE ts >= ? AND recall_snapshot LIKE '%\"appraise\"%'",
rusqlite::params![window_start],
)?;
let mut buckets: std::collections::BTreeMap<String, [f64; 4]> =
std::collections::BTreeMap::new();
for tier in ["weak", "medium", "strong"] {
buckets.insert(tier.to_string(), [0.0; 4]);
}
let mut total = 0_i64;
let mut silent = 0_i64;
let mut caution_strong = 0_i64;
let mut caution_strong_false = 0_i64;
for row in &rows {
let snapshot = row
.get("recall_snapshot")
.and_then(Value::as_str)
.and_then(|raw| serde_json::from_str::<Value>(raw).ok());
let Some(appraise) = snapshot.as_ref().and_then(|s| s.get("appraise")) else {
continue;
};
let tier = appraise
.get("tier")
.and_then(Value::as_str)
.unwrap_or("weak");
let valence = appraise
.get("valence")
.and_then(Value::as_str)
.unwrap_or("neutral");
let strength = appraise
.get("strength")
.and_then(Value::as_f64)
.unwrap_or(0.0);
let outcome = row.get("outcome").and_then(Value::as_str);
total += 1;
if tier == "weak" || valence == "neutral" {
silent += 1;
}
let has_outcome = matches!(outcome, Some("ok") | Some("fail"));
let is_ok = outcome == Some("ok");
if let Some(b) = buckets.get_mut(tier) {
b[0] += 1.0;
if has_outcome {
b[1] += 1.0;
b[3] += strength;
if is_ok {
b[2] += 1.0;
}
}
}
if valence == "caution" && tier == "strong" && has_outcome {
caution_strong += 1;
if is_ok {
caution_strong_false += 1;
}
}
}
let hit_rate = |b: &[f64; 4]| if b[1] > 0.0 { b[2] / b[1] } else { 0.0 };
let weak = buckets.get("weak").copied().unwrap_or([0.0; 4]);
let strong = buckets.get("strong").copied().unwrap_or([0.0; 4]);
let monotonicity_gap = hit_rate(&strong) - hit_rate(&weak);
let outcome_total: f64 = buckets.values().map(|b| b[1]).sum();
let ece = if outcome_total > 0.0 {
buckets
.values()
.filter(|b| b[1] > 0.0)
.map(|b| {
let avg_strength = b[3] / b[1];
(b[1] / outcome_total) * (avg_strength - hit_rate(b)).abs()
})
.sum::<f64>()
} else {
0.0
};
let bucket_detail: Vec<Value> = ["weak", "medium", "strong"]
.iter()
.map(|tier| {
let b = buckets.get(*tier).copied().unwrap_or([0.0; 4]);
json!({
"tier": tier,
"n": b[0] as i64,
"n_with_outcome": b[1] as i64,
"avg_strength": if b[1] > 0.0 { (b[3] / b[1] * 1000.0).round() / 1000.0 } else { 0.0 },
"actual_hit_rate": (hit_rate(&b) * 1000.0).round() / 1000.0,
})
})
.collect();
let (vl_total, vl_abstained, vl_observed) =
self.storage.verdict_log_overview().unwrap_or((0, 0, 0));
let samples = self.storage.verdict_calibration_samples().unwrap_or_default();
let bins = self.calibration_bins.max(2);
let mut vhit = vec![0.0_f64; bins as usize];
let mut vtot = vec![0.0_f64; bins as usize];
for (_strength, conf, h) in &samples {
let b = ((conf * bins as f64).floor() as i64).clamp(0, bins - 1) as usize;
vtot[b] += 1.0;
vhit[b] += *h;
}
let n_obs: f64 = vtot.iter().sum();
let verdict_ece = if n_obs > 0.0 {
(0..bins as usize)
.filter(|&b| vtot[b] > 0.0)
.map(|b| {
let claimed = (b as f64 + 0.5) / bins as f64;
let actual = vhit[b] / vtot[b];
(vtot[b] / n_obs) * (claimed - actual).abs()
})
.sum::<f64>()
} else {
0.0
};
Ok(json!({
"appraisals": total,
"monotonicity_gap": (monotonicity_gap * 1000.0).round() / 1000.0,
"ece": (ece * 1000.0).round() / 1000.0,
"false_alarm_rate": ratio(caution_strong_false, caution_strong),
"silence_rate": ratio(silent, total),
"buckets": bucket_detail,
"verdict_log": {
"total": vl_total,
"abstained": vl_abstained,
"abstain_rate": ratio(vl_abstained, vl_total),
"observed": vl_observed,
"ece": (verdict_ece * 1000.0).round() / 1000.0,
},
}))
}
pub fn rebuild_embeddings(&self) -> Result<usize> {
Ok(self.rebuild_embeddings_capped(None)?.0)
}
pub fn rebuild_embeddings_capped(&self, max: Option<usize>) -> Result<(usize, usize)> {
let meta_version = self
.storage
.get_meta("embed_version")?
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(1);
let mut stale = self.storage.query_chunks_params(
"SELECT id, content, trigger_desc, state_reason FROM chunks
WHERE embed_version = 0 OR embed_version < ?",
rusqlite::params![meta_version],
)?;
let total_stale = stale.len();
let remaining = match max {
Some(n) if n < total_stale => {
stale.truncate(n);
total_stale - n
}
_ => 0,
};
self.storage.invalidate_vector_caches();
let mut count = 0;
for row in &stale {
let id = match row.get("id").and_then(Value::as_str) {
Some(v) => v,
None => continue,
};
let content = row.get("content").and_then(Value::as_str).unwrap_or("");
let trigger = row
.get("trigger_desc")
.and_then(Value::as_str)
.unwrap_or(content);
let state_reason = row
.get("state_reason")
.and_then(Value::as_str)
.unwrap_or("");
let (cvec_res, tvec_res) = self.embed_pair(content, trigger, "rebuild");
let cvec = match cvec_res {
Ok(v) => v,
Err(_) => continue,
};
let tvec = match tvec_res {
Ok(v) => v,
Err(_) => continue,
};
self.storage.begin_immediate()?;
let r = (|| -> Result<()> {
self.store_vec_content(id, &cvec)?;
self.store_vec_trigger(id, &tvec)?;
let new_reason = if state_reason.starts_with("embedding_pending:target=") {
let target_state = state_reason.trim_start_matches("embedding_pending:target=");
let now = utc_now_iso();
self.storage.update_chunk_state(
id,
target_state,
Some("embedding_rebuilt"),
&now,
)?;
"embedding_rebuilt".to_string()
} else {
"embedding_rebuilt".to_string()
};
let now = utc_now_iso();
self.storage.conn_execute(
"UPDATE chunks SET embed_version=?, state_reason=?, updated_at=? WHERE id=?",
rusqlite::params![meta_version, new_reason, now, id],
)?;
self.storage.commit()
})();
if r.is_err() {
let _ = self.storage.rollback();
} else {
count += 1;
}
}
Ok((count, remaining))
}
pub fn inspect_id(&self, id: &str) -> Result<Value> {
if let Some(chunk) = self.storage.get_chunk(id)? {
let traces = self.storage.query_chunks_params(
"SELECT * FROM usage_trace WHERE chunk_id=? ORDER BY ts DESC LIMIT 20",
rusqlite::params![id],
)?;
let derived = self.storage.query_chunks_params(
"SELECT id, state, confidence FROM chunks WHERE distilled_from IN (
SELECT id FROM episodic_log WHERE trace_id IN (
SELECT trace_id FROM usage_trace WHERE chunk_id=?
)
) LIMIT 10",
rusqlite::params![id],
)?;
return Ok(json!({
"kind": "chunk",
"chunk": chunk,
"recent_traces": traces,
"derived_chunks": derived,
}));
}
if let Some(log) = self.storage.get_episodic_log(id)? {
let traces = self.storage.query_chunks_params(
"SELECT * FROM usage_trace WHERE trace_id=? ORDER BY ts ASC",
rusqlite::params![id],
)?;
return Ok(json!({
"kind": "trace",
"episodic_log": log,
"usage_traces": traces,
}));
}
Err(InnateError::ChunkNotFound(id.to_string()))
}
pub(super) fn sanitize_content(&self, content: &str) -> (String, SanitizeAction) {
self.sanitizer.sanitize(content)
}
}