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/// Alias for backwards-compat with the design doc.
334pub async fn get_run_tree(pool: &DbPool, run_id: RunId) -> StorageResult<Vec<SpanRow>> {
335    get_run_spans(pool, run_id).await
336}
337
338/// Search turns by free-text query (basic LIKE-based, sufficient for now;
339/// will be upgraded to FTS5 in P4).
340pub async fn search_turns(
341    pool: &DbPool,
342    run_id: RunId,
343    query: &str,
344    limit: i64,
345) -> StorageResult<Vec<TurnRow>> {
346    let like = format!("%{query}%");
347    let rows = sqlx::query(
348        r#"SELECT seq, ts, kind, role, text, tool_call_id, name, input, output,
349                  tool_status, file_path, file_op, diff
350           FROM turns
351           WHERE run_id = ? AND text LIKE ?
352           ORDER BY seq
353           LIMIT ?"#,
354    )
355    .bind(run_id)
356    .bind(like)
357    .bind(limit)
358    .fetch_all(pool)
359    .await?;
360
361    rows.into_iter()
362        .map(|r| {
363            Ok(TurnRow {
364                seq: r.try_get("seq")?,
365                ts: r.try_get("ts")?,
366                kind: r.try_get("kind")?,
367                role: r.try_get("role")?,
368                text: r.try_get("text")?,
369                tool_call_id: r.try_get("tool_call_id")?,
370                name: r.try_get("name")?,
371                input: r.try_get("input")?,
372                output: r.try_get("output")?,
373                tool_status: r.try_get("tool_status")?,
374                file_path: r.try_get("file_path")?,
375                file_op: r.try_get("file_op")?,
376                diff: r.try_get("diff")?,
377            })
378        })
379        .collect()
380}
381
382// ---------------------------------------------------------------------------
383// Tests
384// ---------------------------------------------------------------------------
385
386#[cfg(test)]
387mod tests {
388    use super::*;
389    use crate::db::open_db;
390    use crate::EventWriter;
391    use chrono::Utc;
392    use luft_core::contract::backend::AgentStatus;
393    use luft_core::contract::event::{AgentEvent, ProgressDelta, RunStatus};
394    use luft_core::contract::ids::TokenUsage;
395    use std::path::PathBuf;
396    use tempfile::tempdir;
397
398    async fn setup() -> (tempfile::TempDir, DbPool, EventWriter) {
399        let dir = tempdir().unwrap();
400        let pool = open_db(&dir.path().join("test.db")).await.unwrap();
401        let writer = EventWriter::new(pool.clone());
402        (dir, pool, writer)
403    }
404
405    async fn seed_run_with_agent(
406        w: &EventWriter,
407        task: &str,
408        agent_count: usize,
409        msg_per_agent: usize,
410    ) -> (RunId, Vec<AgentId>) {
411        let run_id = uuid::Uuid::now_v7();
412        w.write_event(&AgentEvent::RunStarted {
413            run_id,
414            task: task.into(),
415            ts: Utc::now(),
416        })
417        .await
418        .unwrap();
419
420        let mut agents = Vec::new();
421        for i in 0..agent_count {
422            let agent_id = uuid::Uuid::now_v7();
423            agents.push(agent_id);
424            w.write_event(&AgentEvent::AgentStarted {
425                run_id,
426                phase_id: 0,
427                agent_id,
428                prompt_preview: format!("task {i}"),
429                model: Some("claude-sonnet-4".into()),
430                description: None,
431                role: None,
432                name: None,
433                agent_seq: 0,
434            })
435            .await
436            .unwrap();
437            for j in 0..msg_per_agent {
438                w.write_event(&AgentEvent::AgentProgress {
439                    run_id,
440                    agent_id,
441                    delta: ProgressDelta::Message {
442                        text: format!("agent {i} msg {j}"),
443                    },
444                })
445                .await
446                .unwrap();
447            }
448            w.write_event(&AgentEvent::AgentDone {
449                run_id,
450                agent_id,
451                status: AgentStatus::Ok,
452                tokens: TokenUsage {
453                    input: 10,
454                    output: 5,
455                    cache_read: 0,
456                    cache_write: 0,
457                },
458                elapsed_ms: 100,
459                name: None,
460                agent_seq: 0,
461                output: serde_json::Value::Null,
462                findings: Vec::new(),
463                prompt: String::new(),
464                retry_count: 0,
465            })
466            .await
467            .unwrap();
468        }
469
470        w.write_event(&AgentEvent::RunDone {
471            run_id,
472            status: RunStatus::Completed,
473            total_tokens: TokenUsage {
474                input: 10,
475                output: 5,
476                cache_read: 0,
477                cache_write: 0,
478            },
479            report: serde_json::json!({}),
480            ts: chrono::Utc::now(),
481        })
482        .await
483        .unwrap();
484
485        (run_id, agents)
486    }
487
488    #[tokio::test]
489    async fn list_runs_returns_seeded_runs() {
490        let (_dir, pool, writer) = setup().await;
491        for i in 0..3 {
492            seed_run_with_agent(&writer, &format!("task {i}"), 1, 1).await;
493        }
494
495        let runs = list_runs(&pool, 10, 0).await.unwrap();
496        assert_eq!(runs.len(), 3);
497        assert!(runs.iter().all(|r| r.status == "completed"));
498    }
499
500    #[tokio::test]
501    async fn list_runs_pagination() {
502        let (_dir, pool, writer) = setup().await;
503        for i in 0..5 {
504            seed_run_with_agent(&writer, &format!("t{i}"), 1, 0).await;
505        }
506
507        let page1 = list_runs(&pool, 2, 0).await.unwrap();
508        let page2 = list_runs(&pool, 2, 2).await.unwrap();
509        let page3 = list_runs(&pool, 2, 4).await.unwrap();
510        assert_eq!(page1.len(), 2);
511        assert_eq!(page2.len(), 2);
512        assert_eq!(page3.len(), 1);
513
514        // Pages must not overlap.
515        let ids: std::collections::HashSet<_> = page1
516            .iter()
517            .chain(&page2)
518            .chain(&page3)
519            .map(|r| r.run_id)
520            .collect();
521        assert_eq!(ids.len(), 5);
522    }
523
524    #[tokio::test]
525    async fn get_agent_turns_preserves_order() {
526        let (_dir, pool, writer) = setup().await;
527        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 4).await;
528        let agent_id = agents[0];
529
530        let turns = get_agent_turns(&pool, run_id, agent_id, 100, 0)
531            .await
532            .unwrap();
533        assert_eq!(turns.len(), 4);
534        let seqs: Vec<i64> = turns.iter().map(|t| t.seq).collect();
535        assert!(seqs.windows(2).all(|w| w[0] < w[1]));
536        for (i, t) in turns.iter().enumerate() {
537            assert_eq!(t.kind, "message");
538            assert_eq!(t.text.as_deref(), Some(format!("agent 0 msg {i}").as_str()));
539        }
540    }
541
542    #[tokio::test]
543    async fn get_agent_overview_returns_tokens() {
544        let (_dir, pool, writer) = setup().await;
545        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 0).await;
546        let agent_id = agents[0];
547
548        let overview = get_agent_overview(&pool, run_id, agent_id).await.unwrap();
549        assert_eq!(overview.status, "ok");
550        assert_eq!(overview.input_tokens, 10);
551        assert_eq!(overview.output_tokens, 5);
552        assert_eq!(overview.model.as_deref(), Some("claude-sonnet-4"));
553    }
554
555    #[tokio::test]
556    async fn get_run_overview_aggregates_turns() {
557        let (_dir, pool, writer) = setup().await;
558        let (run_id, _) = seed_run_with_agent(&writer, "t", 2, 3).await;
559
560        let overview = get_run_overview(&pool, run_id).await.unwrap();
561        assert_eq!(overview.run.task, "t");
562        assert_eq!(overview.agents.len(), 2);
563        assert_eq!(overview.total_messages, 6); // 2 agents × 3 messages
564        assert_eq!(overview.turn_counts.len(), 1);
565        assert_eq!(overview.turn_counts[0].kind, "message");
566        assert_eq!(overview.turn_counts[0].count, 6);
567    }
568
569    #[tokio::test]
570    async fn get_run_spans_returns_empty_when_no_spans() {
571        let (_dir, pool, writer) = setup().await;
572        let (run_id, _) = seed_run_with_agent(&writer, "t", 1, 1).await;
573
574        let spans = get_run_spans(&pool, run_id).await.unwrap();
575        assert!(spans.is_empty());
576    }
577
578    #[tokio::test]
579    async fn search_turns_finds_matching_text() {
580        let (_dir, pool, writer) = setup().await;
581        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 3).await;
582        let agent_id = agents[0];
583
584        // Add a distinctive message.
585        writer
586            .write_event(&AgentEvent::AgentProgress {
587                run_id,
588                agent_id,
589                delta: ProgressDelta::Message {
590                    text: "the quick brown fox jumps".into(),
591                },
592            })
593            .await
594            .unwrap();
595
596        let results = search_turns(&pool, run_id, "fox", 10).await.unwrap();
597        assert_eq!(results.len(), 1);
598        assert!(results[0].text.as_ref().unwrap().contains("fox"));
599    }
600
601    #[tokio::test]
602    async fn file_edit_turns_appear_in_conversation() {
603        let (_dir, pool, writer) = setup().await;
604        let run_id = uuid::Uuid::now_v7();
605        let agent_id = uuid::Uuid::now_v7();
606
607        writer
608            .write_event(&AgentEvent::AgentStarted {
609                run_id,
610                phase_id: 0,
611                agent_id,
612                prompt_preview: "".into(),
613                model: None,
614                description: None,
615                role: None,
616                name: None,
617                agent_seq: 0,
618            })
619            .await
620            .unwrap();
621
622        writer
623            .write_event(&AgentEvent::AgentProgress {
624                run_id,
625                agent_id,
626                delta: ProgressDelta::FileEdit {
627                    path: PathBuf::from("src/lib.rs"),
628                },
629            })
630            .await
631            .unwrap();
632
633        let turns = get_agent_turns(&pool, run_id, agent_id, 100, 0)
634            .await
635            .unwrap();
636        assert_eq!(turns.len(), 1);
637        assert_eq!(turns[0].kind, "file_edit");
638        assert_eq!(turns[0].file_path.as_deref(), Some("src/lib.rs"));
639    }
640}
luft_storage/reader.rs

luft_storage/
reader.rs
