innate_core/storage/
metrics.rs

1use super::*;
2use serde_json::json;
3
4/// One operation_runs row to persist (P3a, schema 4.20). Holds only an aggregatable
5/// summary — never prompt/response (raw LLM detail stays in `llm_trace.log`).
6pub struct OperationRun {
7    pub id: String,
8    pub trace_id: Option<String>,
9    pub op: String,
10    pub source: Option<String>,
11    pub agent: Option<String>,
12    pub status: String, // ok / error / timeout
13    pub error_kind: Option<String>,
14    pub started_at: String,
15    pub duration_ms: i64,
16    pub counts_json: Option<String>,
17    pub params_json: Option<String>,
18}
19
20/// Raw row used for windowed aggregation in inspect().
21pub struct OpRunRow {
22    pub op: String,
23    pub status: String,
24    pub error_kind: Option<String>,
25    pub duration_ms: i64,
26    pub source: Option<String>,
27    pub agent: Option<String>,
28    pub context: Option<String>,
29}
30
31impl Storage {
32    pub fn insert_operation_run(&self, run: &OperationRun) -> Result<()> {
33        self.conn.execute(
34            "INSERT OR IGNORE INTO operation_runs
35             (id, trace_id, op, source, agent, status, error_kind,
36              started_at, duration_ms, counts_json, params_json)
37             VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",
38            params![
39                run.id,
40                run.trace_id,
41                run.op,
42                run.source,
43                run.agent,
44                run.status,
45                run.error_kind,
46                run.started_at,
47                run.duration_ms,
48                run.counts_json,
49                run.params_json,
50            ],
51        )?;
52        Ok(())
53    }
54
55    /// Raw rows since a cutoff, for aggregation (incl. source/agent/context dimensions).
56    /// `context` is **trace-derived** — operation_runs has no context column, so it is
57    /// pulled via a correlated lookup on `episodic_log.context_key` and is therefore only
58    /// populated for ops that carry a `trace_id` (e.g. record); trace-less ops → None.
59    pub fn operation_runs_since(&self, since_ts: &str) -> Result<Vec<OpRunRow>> {
60        let mut stmt = self.conn.prepare(
61            "SELECT op, status, error_kind, duration_ms, source, agent,
62                    (SELECT el.context_key FROM episodic_log el
63                     WHERE el.trace_id = operation_runs.trace_id LIMIT 1) AS context
64             FROM operation_runs WHERE started_at >= ?1",
65        )?;
66        let rows = stmt.query_map(params![since_ts], |r| {
67            Ok(OpRunRow {
68                op: r.get(0)?,
69                status: r.get(1)?,
70                error_kind: r.get::<_, Option<String>>(2)?,
71                duration_ms: r.get(3)?,
72                source: r.get::<_, Option<String>>(4)?,
73                agent: r.get::<_, Option<String>>(5)?,
74                context: r.get::<_, Option<String>>(6)?,
75            })
76        })?;
77        Ok(rows.filter_map(|r| r.ok()).collect())
78    }
79
80    pub fn count_operation_runs(&self) -> Result<i64> {
81        Ok(self
82            .conn
83            .query_row("SELECT COUNT(*) FROM operation_runs", [], |r| r.get(0))?)
84    }
85
86    /// Retention: drop run rows older than `before_ts` (called by curate).
87    pub fn purge_operation_runs(&self, before_ts: &str) -> Result<usize> {
88        Ok(self.conn.execute(
89            "DELETE FROM operation_runs WHERE started_at < ?1",
90            params![before_ts],
91        )?)
92    }
93
94    pub fn insert_metric_snapshot(&self, ts: &str, kpis_json: &str) -> Result<()> {
95        self.conn.execute(
96            "INSERT OR REPLACE INTO metric_snapshots(ts, kpis) VALUES (?1, ?2)",
97            params![ts, kpis_json],
98        )?;
99        Ok(())
100    }
101
102    /// Most recent snapshot `(ts, kpis_json)`, if any.
103    pub fn latest_snapshot(&self) -> Result<Option<(String, String)>> {
104        Ok(self
105            .conn
106            .query_row(
107                "SELECT ts, kpis FROM metric_snapshots ORDER BY ts DESC LIMIT 1",
108                [],
109                |r| Ok((r.get(0)?, r.get(1)?)),
110            )
111            .optional()?)
112    }
113
114    /// Recent snapshots newest-first as `{ts, kpis:{…}}` rows (for the Web trend view).
115    pub fn recent_snapshots(&self, limit: usize) -> Result<Vec<serde_json::Value>> {
116        let mut stmt = self
117            .conn
118            .prepare("SELECT ts, kpis FROM metric_snapshots ORDER BY ts DESC LIMIT ?1")?;
119        let rows = stmt.query_map(params![limit as i64], |r| {
120            Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
121        })?;
122        Ok(rows
123            .filter_map(|r| r.ok())
124            .map(|(ts, kpis)| {
125                let parsed: serde_json::Value =
126                    serde_json::from_str(&kpis).unwrap_or(serde_json::Value::Null);
127                json!({ "ts": ts, "kpis": parsed })
128            })
129            .collect())
130    }
131
132    /// Nearest snapshot at or before `ts` — the trend baseline for week-over-week deltas.
133    pub fn snapshot_at_or_before(&self, ts: &str) -> Result<Option<(String, String)>> {
134        Ok(self
135            .conn
136            .query_row(
137                "SELECT ts, kpis FROM metric_snapshots WHERE ts <= ?1 ORDER BY ts DESC LIMIT 1",
138                params![ts],
139                |r| Ok((r.get(0)?, r.get(1)?)),
140            )
141            .optional()?)
142    }
143}
144
145/// Map an `InnateError` to a bounded, aggregatable `error_kind` (design doc §5.3.1).
146/// The vocabulary is closed so the "error_kind top list" groups cleanly; new failure
147/// types must extend this table rather than inventing strings at the call site.
148pub fn classify_error(e: &crate::errors::InnateError) -> &'static str {
149    use crate::errors::InnateError as E;
150    match e {
151        E::EmbeddingUnavailable(m) => {
152            if has_arrearage(m) {
153                "embedding_arrearage"
154            } else {
155                "embedding_unavailable"
156            }
157        }
158        E::Db(_) => {
159            let s = e.to_string().to_lowercase();
160            if s.contains("locked") || s.contains("busy") {
161                "db_locked"
162            } else {
163                "db_error"
164            }
165        }
166        E::Json(_) => "json_parse",
167        E::ChunkNotFound(_) => "chunk_not_found",
168        E::InvalidState(_) => "invalid_state",
169        E::Io(_) => "io_error",
170        E::Other(m) => classify_message(m),
171    }
172}
173
174/// Classify a free-text error message (daemon shell-out / HTTP wrappers / `Other`).
175pub fn classify_message(m: &str) -> &'static str {
176    let lm = m.to_lowercase();
177    if has_arrearage(m) {
178        "embedding_arrearage"
179    } else if lm.contains("no such file") || lm.contains("text file busy") {
180        "spawn_failed"
181    } else if lm.contains("timeout") || lm.contains("deadline") {
182        "llm_timeout"
183    } else if lm.contains("status: 4") || lm.contains("status: 5") || lm.contains("http error") {
184        "llm_http_error"
185    } else if lm.contains("locked") || lm.contains("busy") {
186        "db_locked"
187    } else {
188        "other"
189    }
190}
191
192fn has_arrearage(m: &str) -> bool {
193    let lm = m.to_lowercase();
194    lm.contains("arrearage") || m.contains("欠费")
195}
196
197/// Aggregate windowed operation rows into a JSON summary: count + p50/p95 latency +
198/// ok/error/timeout split **broken down by op / source / agent / context** (design doc
199/// §5.3 "按 source/agent/context 分解性能"), plus a global `error_kind` top list. Pure
200/// function over already-fetched rows so it is unit-testable without a db. `by_context`
201/// is trace-derived (operation_runs has no context column) so it only covers ops that
202/// carry a trace_id — group_perf skips None keys, so trace-less ops just don't appear.
203pub fn aggregate_ops(rows: &[OpRunRow]) -> serde_json::Value {
204    let mut err_kind: std::collections::HashMap<&str, i64> = std::collections::HashMap::new();
205    for r in rows {
206        if r.status != "ok" {
207            if let Some(k) = r.error_kind.as_deref() {
208                *err_kind.entry(k).or_insert(0) += 1;
209            }
210        }
211    }
212    let mut top: Vec<(&str, i64)> = err_kind.into_iter().collect();
213    top.sort_by_key(|entry| std::cmp::Reverse(entry.1));
214    let error_kind_top: Vec<serde_json::Value> = top
215        .into_iter()
216        .take(10)
217        .map(|(k, n)| json!({"error_kind": k, "count": n}))
218        .collect();
219    json!({
220        "by_op": group_perf(rows, |r| Some(r.op.as_str())),
221        "by_source": group_perf(rows, |r| r.source.as_deref()),
222        "by_agent": group_perf(rows, |r| r.agent.as_deref()),
223        "by_context": group_perf(rows, |r| r.context.as_deref()),
224        "error_kind_top": error_kind_top,
225    })
226}
227
228/// Group rows by a key extractor and emit per-group count + status split + p50/p95.
229/// Rows whose key is `None` (e.g. an unattributed source/agent) are skipped, so the
230/// breakdown only reports dimensions that were actually recorded.
231fn group_perf<'a>(
232    rows: &'a [OpRunRow],
233    key: impl Fn(&'a OpRunRow) -> Option<&'a str>,
234) -> serde_json::Value {
235    use std::collections::HashMap;
236    let mut g: HashMap<&str, (Vec<i64>, i64, i64, i64)> = HashMap::new();
237    for r in rows {
238        let Some(k) = key(r) else { continue };
239        let e = g.entry(k).or_default();
240        e.0.push(r.duration_ms);
241        match r.status.as_str() {
242            "ok" => e.1 += 1,
243            "timeout" => e.3 += 1,
244            _ => e.2 += 1,
245        }
246    }
247    let mut out = serde_json::Map::new();
248    for (k, (mut durs, ok, err, timeout)) in g {
249        durs.sort_unstable();
250        let total = ok + err + timeout;
251        out.insert(
252            k.to_string(),
253            json!({
254                "count": total,
255                "ok": ok,
256                "error": err,
257                "timeout": timeout,
258                "success_rate": if total > 0 { (ok as f64 / total as f64 * 1000.0).round() / 1000.0 } else { 0.0 },
259                "p50_ms": percentile(&durs, 50),
260                "p95_ms": percentile(&durs, 95),
261            }),
262        );
263    }
264    serde_json::Value::Object(out)
265}
266
267/// Nearest-rank percentile over a pre-sorted ascending slice. Empty → 0.
268fn percentile(sorted: &[i64], p: usize) -> i64 {
269    if sorted.is_empty() {
270        return 0;
271    }
272    let rank = (p * sorted.len()).div_ceil(100); // ceil(p% * n), nearest-rank
273    let idx = rank.saturating_sub(1).min(sorted.len() - 1);
274    sorted[idx]
275}
innate_core/storage/metrics.rs

innate_core/storage/
metrics.rs
