luft_storage/
reader.rs

1//! UI-ready query API.
2//!
3//! All functions read from the global SQLite DB and return structs that map
4//! directly to UI components (run list, conversation view, run tree).
5
6use crate::error::StorageResult;
7use crate::DbPool;
8use luft_core::contract::ids::{AgentId, RunId};
9use sqlx::Row;
10
11/// Run summary for list views.
12#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
13pub struct RunSummary {
14    pub run_id: RunId,
15    pub task: String,
16    pub status: String,
17    pub started_ts: String,
18    pub finished_ts: Option<String>,
19    pub elapsed_ms: i64,
20    pub input_tokens: i64,
21    pub output_tokens: i64,
22}
23
24/// Per-agent overview within a run.
25#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
26pub struct AgentOverview {
27    pub agent_id: AgentId,
28    pub phase_id: Option<i64>,
29    pub model: Option<String>,
30    pub status: String,
31    pub prompt_preview: Option<String>,
32    pub input_tokens: i64,
33    pub output_tokens: i64,
34    pub elapsed_ms: i64,
35    pub started_ts: String,
36    pub done_ts: Option<String>,
37    pub retry_count: i64,
38}
39
40/// One row of an agent's conversation stream.
41#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
42pub struct TurnRow {
43    pub seq: i64,
44    pub ts: String,
45    pub kind: String,
46    pub role: Option<String>,
47    pub text: Option<String>,
48    pub tool_call_id: Option<String>,
49    pub name: Option<String>,
50    pub input: Option<String>,
51    pub output: Option<String>,
52    pub tool_status: Option<String>,
53    pub file_path: Option<String>,
54    pub file_op: Option<String>,
55    pub diff: Option<String>,
56}
57
58/// Aggregated overview of a run.
59#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
60pub struct RunOverview {
61    pub run: RunSummary,
62    pub agents: Vec<AgentOverview>,
63    pub turn_counts: Vec<TurnKindCount>,
64    pub total_messages: i64,
65    pub total_tool_calls: i64,
66    pub total_file_edits: i64,
67}
68
69#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
70pub struct TurnKindCount {
71    pub kind: String,
72    pub count: i64,
73}
74
75/// Span row used to build the run orchestration tree.
76#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
77pub struct SpanRow {
78    pub span_id: i64,
79    pub kind: String,
80    pub phase_id: Option<i64>,
81    pub parent_span_id: Option<i64>,
82    pub label: Option<String>,
83    pub path: Option<String>,
84    pub items: Option<i64>,
85    pub max_rounds: Option<i64>,
86    pub rounds: Option<i64>,
87    pub converged: Option<i64>,
88    pub ok: Option<i64>,
89    pub failed: Option<i64>,
90    pub result: Option<String>,
91    pub error: Option<String>,
92    pub started_ts: Option<String>,
93    pub done_ts: Option<String>,
94    pub elapsed_ms: i64,
95}
96
97// ---------------------------------------------------------------------------
98// Queries
99// ---------------------------------------------------------------------------
100
101/// List recent runs, newest first. Paginated.
102pub async fn list_runs(pool: &DbPool, limit: i64, offset: i64) -> StorageResult<Vec<RunSummary>> {
103    let rows = sqlx::query(
104        r#"SELECT run_id, task, status, started_ts, finished_ts, elapsed_ms,
105                  input_tokens, output_tokens
106           FROM runs
107           ORDER BY started_ts DESC
108           LIMIT ? OFFSET ?"#,
109    )
110    .bind(limit)
111    .bind(offset)
112    .fetch_all(pool)
113    .await?;
114
115    rows.into_iter()
116        .map(|r| {
117            Ok(RunSummary {
118                run_id: r.try_get("run_id")?,
119                task: r.try_get("task")?,
120                status: r.try_get("status")?,
121                started_ts: r.try_get("started_ts")?,
122                finished_ts: r.try_get("finished_ts")?,
123                elapsed_ms: r.try_get("elapsed_ms")?,
124                input_tokens: r.try_get("input_tokens")?,
125                output_tokens: r.try_get("output_tokens")?,
126            })
127        })
128        .collect()
129}
130
131/// List turns for an agent, in seq order. Paginated.
132pub async fn get_agent_turns(
133    pool: &DbPool,
134    run_id: RunId,
135    agent_id: AgentId,
136    limit: i64,
137    offset: i64,
138) -> StorageResult<Vec<TurnRow>> {
139    let rows = sqlx::query(
140        r#"SELECT seq, ts, kind, role, text, tool_call_id, name, input, output,
141                  tool_status, file_path, file_op, diff
142           FROM turns
143           WHERE run_id = ? AND agent_id = ?
144           ORDER BY seq
145           LIMIT ? OFFSET ?"#,
146    )
147    .bind(run_id)
148    .bind(agent_id)
149    .bind(limit)
150    .bind(offset)
151    .fetch_all(pool)
152    .await?;
153
154    rows.into_iter()
155        .map(|r| {
156            Ok(TurnRow {
157                seq: r.try_get("seq")?,
158                ts: r.try_get("ts")?,
159                kind: r.try_get("kind")?,
160                role: r.try_get("role")?,
161                text: r.try_get("text")?,
162                tool_call_id: r.try_get("tool_call_id")?,
163                name: r.try_get("name")?,
164                input: r.try_get("input")?,
165                output: r.try_get("output")?,
166                tool_status: r.try_get("tool_status")?,
167                file_path: r.try_get("file_path")?,
168                file_op: r.try_get("file_op")?,
169                diff: r.try_get("diff")?,
170            })
171        })
172        .collect()
173}
174
175/// Single-agent overview (status, tokens, model).
176pub async fn get_agent_overview(
177    pool: &DbPool,
178    run_id: RunId,
179    agent_id: AgentId,
180) -> StorageResult<AgentOverview> {
181    let r = sqlx::query(
182        r#"SELECT agent_id, phase_id, model, status, prompt_preview,
183                  input_tokens, output_tokens, elapsed_ms, started_ts, done_ts,
184                  retry_count
185           FROM agents WHERE run_id = ? AND agent_id = ?"#,
186    )
187    .bind(run_id)
188    .bind(agent_id)
189    .fetch_one(pool)
190    .await?;
191
192    Ok(AgentOverview {
193        agent_id: r.try_get("agent_id")?,
194        phase_id: r.try_get("phase_id")?,
195        model: r.try_get("model")?,
196        status: r.try_get("status")?,
197        prompt_preview: r.try_get("prompt_preview")?,
198        input_tokens: r.try_get("input_tokens")?,
199        output_tokens: r.try_get("output_tokens")?,
200        elapsed_ms: r.try_get("elapsed_ms")?,
201        started_ts: r.try_get("started_ts")?,
202        done_ts: r.try_get("done_ts")?,
203        retry_count: r.try_get("retry_count")?,
204    })
205}
206
207/// All agents for a run.
208pub async fn get_run_agents(pool: &DbPool, run_id: RunId) -> StorageResult<Vec<AgentOverview>> {
209    let rows = sqlx::query(
210        r#"SELECT agent_id, phase_id, model, status, prompt_preview,
211                  input_tokens, output_tokens, elapsed_ms, started_ts, done_ts,
212                  retry_count
213           FROM agents
214           WHERE run_id = ?
215           ORDER BY started_ts"#,
216    )
217    .bind(run_id)
218    .fetch_all(pool)
219    .await?;
220
221    rows.into_iter()
222        .map(|r| {
223            Ok(AgentOverview {
224                agent_id: r.try_get("agent_id")?,
225                phase_id: r.try_get("phase_id")?,
226                model: r.try_get("model")?,
227                status: r.try_get("status")?,
228                prompt_preview: r.try_get("prompt_preview")?,
229                input_tokens: r.try_get("input_tokens")?,
230                output_tokens: r.try_get("output_tokens")?,
231                elapsed_ms: r.try_get("elapsed_ms")?,
232                started_ts: r.try_get("started_ts")?,
233                done_ts: r.try_get("done_ts")?,
234                retry_count: r.try_get("retry_count")?,
235            })
236        })
237        .collect()
238}
239
240/// Aggregated run overview with turn counts.
241pub async fn get_run_overview(pool: &DbPool, run_id: RunId) -> StorageResult<RunOverview> {
242    let run_row = sqlx::query(
243        r#"SELECT run_id, task, status, started_ts, finished_ts, elapsed_ms,
244                  input_tokens, output_tokens
245           FROM runs WHERE run_id = ?"#,
246    )
247    .bind(run_id)
248    .fetch_one(pool)
249    .await?;
250
251    let run = RunSummary {
252        run_id: run_row.try_get("run_id")?,
253        task: run_row.try_get("task")?,
254        status: run_row.try_get("status")?,
255        started_ts: run_row.try_get("started_ts")?,
256        finished_ts: run_row.try_get("finished_ts")?,
257        elapsed_ms: run_row.try_get("elapsed_ms")?,
258        input_tokens: run_row.try_get("input_tokens")?,
259        output_tokens: run_row.try_get("output_tokens")?,
260    };
261
262    let agents = get_run_agents(pool, run_id).await?;
263
264    let count_rows =
265        sqlx::query("SELECT kind, COUNT(*) AS c FROM turns WHERE run_id = ? GROUP BY kind")
266            .bind(run_id)
267            .fetch_all(pool)
268            .await?;
269
270    let mut turn_counts = Vec::new();
271    let mut total_messages = 0i64;
272    let mut total_tool_calls = 0i64;
273    let mut total_file_edits = 0i64;
274    for r in count_rows {
275        let kind: String = r.try_get("kind")?;
276        let count: i64 = r.try_get("c")?;
277        match kind.as_str() {
278            "message" => total_messages = count,
279            "tool_call" | "tool_result" => total_tool_calls += count,
280            "file_edit" => total_file_edits = count,
281            _ => {}
282        }
283        turn_counts.push(TurnKindCount { kind, count });
284    }
285
286    Ok(RunOverview {
287        run,
288        agents,
289        turn_counts,
290        total_messages,
291        total_tool_calls,
292        total_file_edits,
293    })
294}
295
296/// Orchestration spans for a run (caller assembles into a tree).
297pub async fn get_run_spans(pool: &DbPool, run_id: RunId) -> StorageResult<Vec<SpanRow>> {
298    let rows = sqlx::query(
299        r#"SELECT span_id, kind, phase_id, parent_span_id, label, path,
300                  items, max_rounds, rounds, converged, ok, failed,
301                  result, error, started_ts, done_ts, elapsed_ms
302           FROM spans WHERE run_id = ? ORDER BY span_id"#,
303    )
304    .bind(run_id)
305    .fetch_all(pool)
306    .await?;
307
308    rows.into_iter()
309        .map(|r| {
310            Ok(SpanRow {
311                span_id: r.try_get("span_id")?,
312                kind: r.try_get("kind")?,
313                phase_id: r.try_get("phase_id")?,
314                parent_span_id: r.try_get("parent_span_id")?,
315                label: r.try_get("label")?,
316                path: r.try_get("path")?,
317                items: r.try_get("items")?,
318                max_rounds: r.try_get("max_rounds")?,
319                rounds: r.try_get("rounds")?,
320                converged: r.try_get("converged")?,
321                ok: r.try_get("ok")?,
322                failed: r.try_get("failed")?,
323                result: r.try_get("result")?,
324                error: r.try_get("error")?,
325                started_ts: r.try_get("started_ts")?,
326                done_ts: r.try_get("done_ts")?,
327                elapsed_ms: r.try_get("elapsed_ms")?,
328            })
329        })
330        .collect()
331}
332
333
334/// Search turns by free-text query (basic LIKE-based, sufficient for now;
335/// will be upgraded to FTS5 in P4).
336pub async fn search_turns(
337    pool: &DbPool,
338    run_id: RunId,
339    query: &str,
340    limit: i64,
341) -> StorageResult<Vec<TurnRow>> {
342    let like = format!("%{query}%");
343    let rows = sqlx::query(
344        r#"SELECT seq, ts, kind, role, text, tool_call_id, name, input, output,
345                  tool_status, file_path, file_op, diff
346           FROM turns
347           WHERE run_id = ? AND text LIKE ?
348           ORDER BY seq
349           LIMIT ?"#,
350    )
351    .bind(run_id)
352    .bind(like)
353    .bind(limit)
354    .fetch_all(pool)
355    .await?;
356
357    rows.into_iter()
358        .map(|r| {
359            Ok(TurnRow {
360                seq: r.try_get("seq")?,
361                ts: r.try_get("ts")?,
362                kind: r.try_get("kind")?,
363                role: r.try_get("role")?,
364                text: r.try_get("text")?,
365                tool_call_id: r.try_get("tool_call_id")?,
366                name: r.try_get("name")?,
367                input: r.try_get("input")?,
368                output: r.try_get("output")?,
369                tool_status: r.try_get("tool_status")?,
370                file_path: r.try_get("file_path")?,
371                file_op: r.try_get("file_op")?,
372                diff: r.try_get("diff")?,
373            })
374        })
375        .collect()
376}
377
378// ---------------------------------------------------------------------------
379// Tests
380// ---------------------------------------------------------------------------
381
382#[cfg(test)]
383mod tests {
384    use super::*;
385    use crate::db::open_db;
386    use crate::EventWriter;
387    use chrono::Utc;
388    use luft_core::contract::backend::AgentStatus;
389    use luft_core::contract::event::{AgentEvent, ProgressDelta, RunStatus};
390    use luft_core::contract::ids::TokenUsage;
391    use std::path::PathBuf;
392    use tempfile::tempdir;
393
394    async fn setup() -> (tempfile::TempDir, DbPool, EventWriter) {
395        let dir = tempdir().unwrap();
396        let pool = open_db(&dir.path().join("test.db")).await.unwrap();
397        let writer = EventWriter::new(pool.clone());
398        (dir, pool, writer)
399    }
400
401    async fn seed_run_with_agent(
402        w: &EventWriter,
403        task: &str,
404        agent_count: usize,
405        msg_per_agent: usize,
406    ) -> (RunId, Vec<AgentId>) {
407        let run_id = uuid::Uuid::now_v7();
408        w.write_event(&AgentEvent::RunStarted {
409            run_id,
410            task: task.into(),
411            ts: Utc::now(),
412        })
413        .await
414        .unwrap();
415
416        let mut agents = Vec::new();
417        for i in 0..agent_count {
418            let agent_id = uuid::Uuid::now_v7();
419            agents.push(agent_id);
420            w.write_event(&AgentEvent::AgentStarted {
421                run_id,
422                phase_id: 0,
423                agent_id,
424                prompt_preview: format!("task {i}"),
425                model: Some("claude-sonnet-4".into()),
426                description: None,
427                role: None,
428                name: None,
429                agent_seq: 0,
430                ts: Default::default(),
431            })
432            .await
433            .unwrap();
434            for j in 0..msg_per_agent {
435                w.write_event(&AgentEvent::AgentProgress {
436                    run_id,
437                    agent_id,
438                    delta: ProgressDelta::Message {
439                        text: format!("agent {i} msg {j}"),
440                    },
441                })
442                .await
443                .unwrap();
444            }
445            w.write_event(&AgentEvent::AgentDone {
446                run_id,
447                agent_id,
448                status: AgentStatus::Ok,
449                tokens: TokenUsage {
450                    input: 10,
451                    output: 5,
452                    cache_read: 0,
453                    cache_write: 0,
454                },
455                elapsed_ms: 100,
456                name: None,
457                agent_seq: 0,
458                output: serde_json::Value::Null,
459                findings: Vec::new(),
460                prompt: String::new(),
461                retry_count: 0,
462                ts: Default::default(),
463            })
464            .await
465            .unwrap();
466        }
467
468        w.write_event(&AgentEvent::RunDone {
469            run_id,
470            status: RunStatus::Completed,
471            total_tokens: TokenUsage {
472                input: 10,
473                output: 5,
474                cache_read: 0,
475                cache_write: 0,
476            },
477            report: serde_json::json!({}),
478            ts: chrono::Utc::now(),
479        })
480        .await
481        .unwrap();
482
483        (run_id, agents)
484    }
485
486    #[tokio::test]
487    async fn list_runs_returns_seeded_runs() {
488        let (_dir, pool, writer) = setup().await;
489        for i in 0..3 {
490            seed_run_with_agent(&writer, &format!("task {i}"), 1, 1).await;
491        }
492
493        let runs = list_runs(&pool, 10, 0).await.unwrap();
494        assert_eq!(runs.len(), 3);
495        assert!(runs.iter().all(|r| r.status == "completed"));
496    }
497
498    #[tokio::test]
499    async fn list_runs_pagination() {
500        let (_dir, pool, writer) = setup().await;
501        for i in 0..5 {
502            seed_run_with_agent(&writer, &format!("t{i}"), 1, 0).await;
503        }
504
505        let page1 = list_runs(&pool, 2, 0).await.unwrap();
506        let page2 = list_runs(&pool, 2, 2).await.unwrap();
507        let page3 = list_runs(&pool, 2, 4).await.unwrap();
508        assert_eq!(page1.len(), 2);
509        assert_eq!(page2.len(), 2);
510        assert_eq!(page3.len(), 1);
511
512        // Pages must not overlap.
513        let ids: std::collections::HashSet<_> = page1
514            .iter()
515            .chain(&page2)
516            .chain(&page3)
517            .map(|r| r.run_id)
518            .collect();
519        assert_eq!(ids.len(), 5);
520    }
521
522    #[tokio::test]
523    async fn get_agent_turns_preserves_order() {
524        let (_dir, pool, writer) = setup().await;
525        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 4).await;
526        let agent_id = agents[0];
527
528        let turns = get_agent_turns(&pool, run_id, agent_id, 100, 0)
529            .await
530            .unwrap();
531        assert_eq!(turns.len(), 4);
532        let seqs: Vec<i64> = turns.iter().map(|t| t.seq).collect();
533        assert!(seqs.windows(2).all(|w| w[0] < w[1]));
534        for (i, t) in turns.iter().enumerate() {
535            assert_eq!(t.kind, "message");
536            assert_eq!(t.text.as_deref(), Some(format!("agent 0 msg {i}").as_str()));
537        }
538    }
539
540    #[tokio::test]
541    async fn get_agent_overview_returns_tokens() {
542        let (_dir, pool, writer) = setup().await;
543        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 0).await;
544        let agent_id = agents[0];
545
546        let overview = get_agent_overview(&pool, run_id, agent_id).await.unwrap();
547        assert_eq!(overview.status, "ok");
548        assert_eq!(overview.input_tokens, 10);
549        assert_eq!(overview.output_tokens, 5);
550        assert_eq!(overview.model.as_deref(), Some("claude-sonnet-4"));
551    }
552
553    #[tokio::test]
554    async fn get_run_overview_aggregates_turns() {
555        let (_dir, pool, writer) = setup().await;
556        let (run_id, _) = seed_run_with_agent(&writer, "t", 2, 3).await;
557
558        let overview = get_run_overview(&pool, run_id).await.unwrap();
559        assert_eq!(overview.run.task, "t");
560        assert_eq!(overview.agents.len(), 2);
561        assert_eq!(overview.total_messages, 6); // 2 agents × 3 messages
562        assert_eq!(overview.turn_counts.len(), 1);
563        assert_eq!(overview.turn_counts[0].kind, "message");
564        assert_eq!(overview.turn_counts[0].count, 6);
565    }
566
567    #[tokio::test]
568    async fn get_run_spans_returns_empty_when_no_spans() {
569        let (_dir, pool, writer) = setup().await;
570        let (run_id, _) = seed_run_with_agent(&writer, "t", 1, 1).await;
571
572        let spans = get_run_spans(&pool, run_id).await.unwrap();
573        assert!(spans.is_empty());
574    }
575
576    #[tokio::test]
577    async fn search_turns_finds_matching_text() {
578        let (_dir, pool, writer) = setup().await;
579        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 3).await;
580        let agent_id = agents[0];
581
582        // Add a distinctive message.
583        writer
584            .write_event(&AgentEvent::AgentProgress {
585                run_id,
586                agent_id,
587                delta: ProgressDelta::Message {
588                    text: "the quick brown fox jumps".into(),
589                },
590            })
591            .await
592            .unwrap();
593
594        let results = search_turns(&pool, run_id, "fox", 10).await.unwrap();
595        assert_eq!(results.len(), 1);
596        assert!(results[0].text.as_ref().unwrap().contains("fox"));
597    }
598
599    #[tokio::test]
600    async fn file_edit_turns_appear_in_conversation() {
601        let (_dir, pool, writer) = setup().await;
602        let run_id = uuid::Uuid::now_v7();
603        let agent_id = uuid::Uuid::now_v7();
604
605        writer
606            .write_event(&AgentEvent::AgentStarted {
607                run_id,
608                phase_id: 0,
609                agent_id,
610                prompt_preview: "".into(),
611                model: None,
612                description: None,
613                role: None,
614                name: None,
615                agent_seq: 0,
616                ts: Default::default(),
617            })
618            .await
619            .unwrap();
620
621        writer
622            .write_event(&AgentEvent::AgentProgress {
623                run_id,
624                agent_id,
625                delta: ProgressDelta::FileEdit {
626                    path: PathBuf::from("src/lib.rs"),
627                },
628            })
629            .await
630            .unwrap();
631
632        let turns = get_agent_turns(&pool, run_id, agent_id, 100, 0)
633            .await
634            .unwrap();
635        assert_eq!(turns.len(), 1);
636        assert_eq!(turns[0].kind, "file_edit");
637        assert_eq!(turns[0].file_path.as_deref(), Some("src/lib.rs"));
638    }
639}
luft_storage/reader.rs

luft_storage/
reader.rs
