zeph_memory/store/
experiments.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4use super::SqliteStore;
5use crate::error::MemoryError;
6use zeph_common::SessionId;
7#[allow(unused_imports)]
8use zeph_db::sql;
9
10#[derive(Debug, Clone)]
11pub struct ExperimentResultRow {
12    pub id: i64,
13    pub session_id: SessionId,
14    pub parameter: String,
15    pub value_json: String,
16    pub baseline_score: f64,
17    pub candidate_score: f64,
18    pub delta: f64,
19    pub latency_ms: i64,
20    pub tokens_used: i64,
21    pub accepted: bool,
22    pub source: String,
23    pub created_at: String,
24}
25
26#[derive(Debug, Clone)]
27pub struct NewExperimentResult<'a> {
28    pub session_id: &'a str,
29    pub parameter: &'a str,
30    pub value_json: &'a str,
31    pub baseline_score: f64,
32    pub candidate_score: f64,
33    pub delta: f64,
34    pub latency_ms: i64,
35    pub tokens_used: i64,
36    pub accepted: bool,
37    pub source: &'a str,
38}
39
40#[derive(Debug, Clone)]
41pub struct SessionSummaryRow {
42    pub session_id: SessionId,
43    pub total: i64,
44    pub accepted_count: i64,
45    pub best_delta: f64,
46    pub total_tokens: i64,
47}
48
49/// Validate that `s` looks like `YYYY-MM-DD HH:MM:SS` or `YYYY-MM-DDTHH:MM:SS`.
50fn validate_timestamp(s: &str) -> Result<(), MemoryError> {
51    let bytes = s.as_bytes();
52    // Minimum length: "2000-01-01 00:00:00" = 19 chars
53    if bytes.len() < 19 {
54        return Err(MemoryError::InvalidInput(format!(
55            "invalid timestamp format (too short): {s:?}"
56        )));
57    }
58    let sep = bytes[10];
59    if sep != b' ' && sep != b'T' {
60        return Err(MemoryError::InvalidInput(format!(
61            "invalid timestamp format (expected space or T at position 10): {s:?}"
62        )));
63    }
64    // Check digit positions: YYYY-MM-DD HH:MM:SS
65    let digits_at = [0, 1, 2, 3, 5, 6, 8, 9, 11, 12, 14, 15, 17, 18];
66    let dashes_at = [4, 7];
67    let colons_at = [13, 16];
68    for i in digits_at {
69        if !bytes[i].is_ascii_digit() {
70            return Err(MemoryError::InvalidInput(format!(
71                "invalid timestamp format (expected digit at {i}): {s:?}"
72            )));
73        }
74    }
75    for i in dashes_at {
76        if bytes[i] != b'-' {
77            return Err(MemoryError::InvalidInput(format!(
78                "invalid timestamp format (expected '-' at {i}): {s:?}"
79            )));
80        }
81    }
82    for i in colons_at {
83        if bytes[i] != b':' {
84            return Err(MemoryError::InvalidInput(format!(
85                "invalid timestamp format (expected ':' at {i}): {s:?}"
86            )));
87        }
88    }
89    Ok(())
90}
91
92type ResultTuple = (
93    i64,
94    String,
95    String,
96    String,
97    f64,
98    f64,
99    f64,
100    i64,
101    i64,
102    bool,
103    String,
104    String,
105);
106
107fn row_from_tuple(t: ResultTuple) -> ExperimentResultRow {
108    ExperimentResultRow {
109        id: t.0,
110        session_id: SessionId::new(t.1),
111        parameter: t.2,
112        value_json: t.3,
113        baseline_score: t.4,
114        candidate_score: t.5,
115        delta: t.6,
116        latency_ms: t.7,
117        tokens_used: t.8,
118        accepted: t.9,
119        source: t.10,
120        created_at: t.11,
121    }
122}
123
124impl SqliteStore {
125    /// Insert an experiment result and return the new row ID.
126    ///
127    /// # Errors
128    ///
129    /// Returns an error if the insert fails.
130    pub async fn insert_experiment_result(
131        &self,
132        result: &NewExperimentResult<'_>,
133    ) -> Result<i64, MemoryError> {
134        let row: (i64,) = zeph_db::query_as(sql!(
135            "INSERT INTO experiment_results \
136             (session_id, parameter, value_json, baseline_score, candidate_score, \
137              delta, latency_ms, tokens_used, accepted, source) \
138             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id"
139        ))
140        .bind(result.session_id)
141        .bind(result.parameter)
142        .bind(result.value_json)
143        .bind(result.baseline_score)
144        .bind(result.candidate_score)
145        .bind(result.delta)
146        .bind(result.latency_ms)
147        .bind(result.tokens_used)
148        .bind(result.accepted)
149        .bind(result.source)
150        .fetch_one(&self.pool)
151        .await?;
152        Ok(row.0)
153    }
154
155    /// List experiment results, optionally filtered by `session_id`, newest first.
156    ///
157    /// # Errors
158    ///
159    /// Returns an error if the query fails.
160    pub async fn list_experiment_results(
161        &self,
162        session_id: Option<&str>,
163        limit: u32,
164    ) -> Result<Vec<ExperimentResultRow>, MemoryError> {
165        // `created_at` is `TIMESTAMPTZ` on Postgres (`TEXT` on SQLite); project through
166        // `Dialect::select_as_text` so it decodes into the `String` tuple field below.
167        // `latency_ms`/`tokens_used` are `INTEGER` (`INT4`) on Postgres but the tuple decodes
168        // them as `i64`; `CAST(... AS BIGINT)` widens them (same un-gated idiom already used
169        // for `weight`/`confidence_fast`/`confidence_slow` in `graph/store/mod.rs`).
170        let created_at_sel =
171            <zeph_db::ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
172        let rows: Vec<ResultTuple> = if let Some(sid) = session_id {
173            let raw = format!(
174                "SELECT id, session_id, parameter, value_json, baseline_score, candidate_score, \
175                 delta, CAST(latency_ms AS BIGINT) AS latency_ms, \
176                 CAST(tokens_used AS BIGINT) AS tokens_used, accepted, source, {created_at_sel} \
177                 FROM experiment_results WHERE session_id = ? ORDER BY id DESC LIMIT ?"
178            );
179            let query_sql = zeph_db::rewrite_placeholders(&raw);
180            zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
181                .bind(sid)
182                .bind(i64::from(limit))
183                .fetch_all(&self.pool)
184                .await?
185        } else {
186            let raw = format!(
187                "SELECT id, session_id, parameter, value_json, baseline_score, candidate_score, \
188                 delta, CAST(latency_ms AS BIGINT) AS latency_ms, \
189                 CAST(tokens_used AS BIGINT) AS tokens_used, accepted, source, {created_at_sel} \
190                 FROM experiment_results ORDER BY id DESC LIMIT ?"
191            );
192            let query_sql = zeph_db::rewrite_placeholders(&raw);
193            zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
194                .bind(i64::from(limit))
195                .fetch_all(&self.pool)
196                .await?
197        };
198        Ok(rows.into_iter().map(row_from_tuple).collect())
199    }
200
201    /// Get the best accepted result (highest delta), optionally filtered by parameter.
202    ///
203    /// # Errors
204    ///
205    /// Returns an error if the query fails.
206    pub async fn best_experiment_result(
207        &self,
208        parameter: Option<&str>,
209    ) -> Result<Option<ExperimentResultRow>, MemoryError> {
210        // `created_at` is `TIMESTAMPTZ` on Postgres — see `list_experiment_results`.
211        // `latency_ms`/`tokens_used` need the same `CAST(... AS BIGINT)` widening — see
212        // `list_experiment_results`.
213        let created_at_sel =
214            <zeph_db::ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
215        let row: Option<ResultTuple> = if let Some(param) = parameter {
216            let raw = format!(
217                "SELECT id, session_id, parameter, value_json, baseline_score, candidate_score, \
218                 delta, CAST(latency_ms AS BIGINT) AS latency_ms, \
219                 CAST(tokens_used AS BIGINT) AS tokens_used, accepted, source, {created_at_sel} \
220                 FROM experiment_results \
221                 WHERE accepted = TRUE AND parameter = ? ORDER BY delta DESC LIMIT 1"
222            );
223            let query_sql = zeph_db::rewrite_placeholders(&raw);
224            zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
225                .bind(param)
226                .fetch_optional(&self.pool)
227                .await?
228        } else {
229            let raw = format!(
230                "SELECT id, session_id, parameter, value_json, baseline_score, candidate_score, \
231                 delta, CAST(latency_ms AS BIGINT) AS latency_ms, \
232                 CAST(tokens_used AS BIGINT) AS tokens_used, accepted, source, {created_at_sel} \
233                 FROM experiment_results \
234                 WHERE accepted = TRUE ORDER BY delta DESC LIMIT 1"
235            );
236            let query_sql = zeph_db::rewrite_placeholders(&raw);
237            zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
238                .fetch_optional(&self.pool)
239                .await?
240        };
241        Ok(row.map(row_from_tuple))
242    }
243
244    /// Get all results since a given ISO-8601 timestamp (`YYYY-MM-DD HH:MM:SS` or `YYYY-MM-DDTHH:MM:SS`).
245    ///
246    /// # Errors
247    ///
248    /// Returns `MemoryError::Other` if `since` does not match the expected timestamp format.
249    /// Returns an error if the query fails.
250    pub async fn experiment_results_since(
251        &self,
252        since: &str,
253    ) -> Result<Vec<ExperimentResultRow>, MemoryError> {
254        validate_timestamp(since)?;
255        // `created_at` is `TIMESTAMPTZ` on Postgres (`TEXT` on SQLite): the output projects
256        // through `Dialect::select_as_text` to decode into the `String` tuple field below, and
257        // the `since` bind parameter needs an explicit `TIMESTAMPTZ_CAST` — Postgres has no
258        // implicit text->timestamptz cast for a bound (non-literal) parameter — mirroring
259        // `agent_sessions.rs::upsert_agent_session`'s fix for the same class of mismatch.
260        // `latency_ms`/`tokens_used` need the same `CAST(... AS BIGINT)` widening — see
261        // `list_experiment_results`.
262        let created_at_sel =
263            <zeph_db::ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
264        let ts_cast = <zeph_db::ActiveDialect as zeph_db::dialect::Dialect>::TIMESTAMPTZ_CAST;
265        let raw = format!(
266            "SELECT id, session_id, parameter, value_json, baseline_score, candidate_score, \
267             delta, CAST(latency_ms AS BIGINT) AS latency_ms, \
268             CAST(tokens_used AS BIGINT) AS tokens_used, accepted, source, {created_at_sel} \
269             FROM experiment_results WHERE created_at >= ?{ts_cast} ORDER BY id DESC LIMIT 10000"
270        );
271        let query_sql = zeph_db::rewrite_placeholders(&raw);
272        let rows: Vec<ResultTuple> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
273            .bind(since)
274            .fetch_all(&self.pool)
275            .await?;
276        Ok(rows.into_iter().map(row_from_tuple).collect())
277    }
278
279    /// Get a summary for a specific session.
280    ///
281    /// # Errors
282    ///
283    /// Returns an error if the query fails.
284    pub async fn experiment_session_summary(
285        &self,
286        session_id: &str,
287    ) -> Result<Option<SessionSummaryRow>, MemoryError> {
288        let row: Option<(String, i64, i64, Option<f64>, i64)> = zeph_db::query_as(sql!(
289            "SELECT session_id, COUNT(*) as total, \
290             SUM(CASE WHEN accepted = TRUE THEN 1 ELSE 0 END) as accepted_count, \
291             MAX(CASE WHEN accepted = TRUE THEN delta ELSE NULL END) as best_delta, \
292             SUM(tokens_used) as total_tokens \
293             FROM experiment_results WHERE session_id = ? GROUP BY session_id"
294        ))
295        .bind(session_id)
296        .fetch_optional(&self.pool)
297        .await?;
298        Ok(row.map(
299            |(sid, total, accepted_count, best_delta, total_tokens)| SessionSummaryRow {
300                session_id: SessionId::new(sid),
301                total,
302                accepted_count,
303                best_delta: best_delta.unwrap_or(0.0),
304                total_tokens,
305            },
306        ))
307    }
308}
309
310#[cfg(test)]
311mod tests {
312    use super::*;
313
314    async fn test_store() -> SqliteStore {
315        SqliteStore::new(":memory:").await.unwrap()
316    }
317
318    fn make_result<'a>(
319        session_id: &'a str,
320        parameter: &'a str,
321        accepted: bool,
322        delta: f64,
323    ) -> NewExperimentResult<'a> {
324        NewExperimentResult {
325            session_id,
326            parameter,
327            value_json: r#"{"type":"Float","value":0.7}"#,
328            baseline_score: 7.0,
329            candidate_score: 7.0 + delta,
330            delta,
331            latency_ms: 500,
332            tokens_used: 100,
333            accepted,
334            source: "manual",
335        }
336    }
337
338    #[tokio::test]
339    async fn insert_and_list_results() {
340        let store = test_store().await;
341        let r = make_result("session-1", "temperature", true, 1.0);
342        let id = store.insert_experiment_result(&r).await.unwrap();
343        assert!(id > 0);
344
345        let rows = store
346            .list_experiment_results(Some("session-1"), 10)
347            .await
348            .unwrap();
349        assert_eq!(rows.len(), 1);
350        assert_eq!(rows[0].session_id, "session-1");
351        assert_eq!(rows[0].parameter, "temperature");
352        assert!(rows[0].accepted);
353        assert!((rows[0].delta - 1.0).abs() < f64::EPSILON);
354    }
355
356    #[tokio::test]
357    async fn list_results_no_filter_returns_all() {
358        let store = test_store().await;
359        store
360            .insert_experiment_result(&make_result("s1", "temperature", true, 1.0))
361            .await
362            .unwrap();
363        store
364            .insert_experiment_result(&make_result("s2", "top_p", false, -0.2))
365            .await
366            .unwrap();
367
368        let rows = store.list_experiment_results(None, 10).await.unwrap();
369        assert_eq!(rows.len(), 2);
370        // newest first
371        assert_eq!(rows[0].session_id, "s2");
372    }
373
374    #[tokio::test]
375    async fn best_result_returns_accepted_highest_delta() {
376        let store = test_store().await;
377        store
378            .insert_experiment_result(&make_result("s1", "temperature", false, 2.0))
379            .await
380            .unwrap();
381        store
382            .insert_experiment_result(&make_result("s1", "temperature", true, 0.5))
383            .await
384            .unwrap();
385        store
386            .insert_experiment_result(&make_result("s1", "temperature", true, 1.5))
387            .await
388            .unwrap();
389
390        let best = store.best_experiment_result(None).await.unwrap().unwrap();
391        assert!(best.accepted);
392        assert!((best.delta - 1.5).abs() < f64::EPSILON);
393    }
394
395    #[tokio::test]
396    async fn best_result_filtered_by_parameter() {
397        let store = test_store().await;
398        store
399            .insert_experiment_result(&make_result("s1", "temperature", true, 2.0))
400            .await
401            .unwrap();
402        store
403            .insert_experiment_result(&make_result("s1", "top_p", true, 1.0))
404            .await
405            .unwrap();
406
407        let best = store
408            .best_experiment_result(Some("top_p"))
409            .await
410            .unwrap()
411            .unwrap();
412        assert_eq!(best.parameter, "top_p");
413    }
414
415    #[tokio::test]
416    async fn best_result_no_accepted_returns_none() {
417        let store = test_store().await;
418        store
419            .insert_experiment_result(&make_result("s1", "temperature", false, 2.0))
420            .await
421            .unwrap();
422        let best = store.best_experiment_result(None).await.unwrap();
423        assert!(best.is_none());
424    }
425
426    #[tokio::test]
427    async fn session_summary_aggregation() {
428        let store = test_store().await;
429        store
430            .insert_experiment_result(&make_result("sess", "temperature", true, 1.0))
431            .await
432            .unwrap();
433        store
434            .insert_experiment_result(&make_result("sess", "top_p", false, -0.2))
435            .await
436            .unwrap();
437        store
438            .insert_experiment_result(&make_result("sess", "top_k", true, 0.8))
439            .await
440            .unwrap();
441
442        let summary = store
443            .experiment_session_summary("sess")
444            .await
445            .unwrap()
446            .unwrap();
447        assert_eq!(summary.session_id, "sess");
448        assert_eq!(summary.total, 3);
449        assert_eq!(summary.accepted_count, 2);
450        assert!((summary.best_delta - 1.0).abs() < f64::EPSILON);
451        assert_eq!(summary.total_tokens, 300);
452    }
453
454    #[tokio::test]
455    async fn session_summary_unknown_session_returns_none() {
456        let store = test_store().await;
457        let summary = store
458            .experiment_session_summary("nonexistent")
459            .await
460            .unwrap();
461        assert!(summary.is_none());
462    }
463
464    #[tokio::test]
465    async fn results_since_time_filtering() {
466        let store = test_store().await;
467        // Insert a result, then query with a future timestamp — expect nothing
468        store
469            .insert_experiment_result(&make_result("s1", "temperature", true, 1.0))
470            .await
471            .unwrap();
472
473        let rows = store
474            .experiment_results_since("2099-01-01 00:00:00")
475            .await
476            .unwrap();
477        assert!(rows.is_empty());
478
479        // Query with a past timestamp — expect the result
480        let rows = store
481            .experiment_results_since("2000-01-01 00:00:00")
482            .await
483            .unwrap();
484        assert_eq!(rows.len(), 1);
485    }
486
487    #[tokio::test]
488    async fn results_since_rejects_invalid_timestamp() {
489        let store = test_store().await;
490        let bad = [
491            "",
492            "not-a-date",
493            "0000-00-00",
494            "2024-01-01",
495            "2024/01/01 00:00:00",
496        ];
497        for ts in bad {
498            let err = store.experiment_results_since(ts).await;
499            assert!(err.is_err(), "expected error for timestamp: {ts:?}");
500        }
501        // ISO-8601 with T separator should work
502        let store2 = test_store().await;
503        let rows = store2
504            .experiment_results_since("2000-01-01T00:00:00")
505            .await
506            .unwrap();
507        assert!(rows.is_empty());
508    }
509
510    #[tokio::test]
511    async fn list_results_respects_limit() {
512        let store = test_store().await;
513        for i in 0..5 {
514            store
515                .insert_experiment_result(&make_result(
516                    "s1",
517                    "temperature",
518                    i % 2 == 0,
519                    f64::from(i),
520                ))
521                .await
522                .unwrap();
523        }
524        let rows = store.list_experiment_results(None, 3).await.unwrap();
525        assert_eq!(rows.len(), 3);
526    }
527}
zeph_memory/store/experiments.rs

zeph_memory/store/
experiments.rs
