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                ts: Default::default(),
435            })
436            .await
437            .unwrap();
438            for j in 0..msg_per_agent {
439                w.write_event(&AgentEvent::AgentProgress {
440                    run_id,
441                    agent_id,
442                    delta: ProgressDelta::Message {
443                        text: format!("agent {i} msg {j}"),
444                    },
445                })
446                .await
447                .unwrap();
448            }
449            w.write_event(&AgentEvent::AgentDone {
450                run_id,
451                agent_id,
452                status: AgentStatus::Ok,
453                tokens: TokenUsage {
454                    input: 10,
455                    output: 5,
456                    cache_read: 0,
457                    cache_write: 0,
458                },
459                elapsed_ms: 100,
460                name: None,
461                agent_seq: 0,
462                output: serde_json::Value::Null,
463                findings: Vec::new(),
464                prompt: String::new(),
465                retry_count: 0,
466                ts: Default::default(),
467            })
468            .await
469            .unwrap();
470        }
471
472        w.write_event(&AgentEvent::RunDone {
473            run_id,
474            status: RunStatus::Completed,
475            total_tokens: TokenUsage {
476                input: 10,
477                output: 5,
478                cache_read: 0,
479                cache_write: 0,
480            },
481            report: serde_json::json!({}),
482            ts: chrono::Utc::now(),
483        })
484        .await
485        .unwrap();
486
487        (run_id, agents)
488    }
489
490    #[tokio::test]
491    async fn list_runs_returns_seeded_runs() {
492        let (_dir, pool, writer) = setup().await;
493        for i in 0..3 {
494            seed_run_with_agent(&writer, &format!("task {i}"), 1, 1).await;
495        }
496
497        let runs = list_runs(&pool, 10, 0).await.unwrap();
498        assert_eq!(runs.len(), 3);
499        assert!(runs.iter().all(|r| r.status == "completed"));
500    }
501
502    #[tokio::test]
503    async fn list_runs_pagination() {
504        let (_dir, pool, writer) = setup().await;
505        for i in 0..5 {
506            seed_run_with_agent(&writer, &format!("t{i}"), 1, 0).await;
507        }
508
509        let page1 = list_runs(&pool, 2, 0).await.unwrap();
510        let page2 = list_runs(&pool, 2, 2).await.unwrap();
511        let page3 = list_runs(&pool, 2, 4).await.unwrap();
512        assert_eq!(page1.len(), 2);
513        assert_eq!(page2.len(), 2);
514        assert_eq!(page3.len(), 1);
515
516        // Pages must not overlap.
517        let ids: std::collections::HashSet<_> = page1
518            .iter()
519            .chain(&page2)
520            .chain(&page3)
521            .map(|r| r.run_id)
522            .collect();
523        assert_eq!(ids.len(), 5);
524    }
525
526    #[tokio::test]
527    async fn get_agent_turns_preserves_order() {
528        let (_dir, pool, writer) = setup().await;
529        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 4).await;
530        let agent_id = agents[0];
531
532        let turns = get_agent_turns(&pool, run_id, agent_id, 100, 0)
533            .await
534            .unwrap();
535        assert_eq!(turns.len(), 4);
536        let seqs: Vec<i64> = turns.iter().map(|t| t.seq).collect();
537        assert!(seqs.windows(2).all(|w| w[0] < w[1]));
538        for (i, t) in turns.iter().enumerate() {
539            assert_eq!(t.kind, "message");
540            assert_eq!(t.text.as_deref(), Some(format!("agent 0 msg {i}").as_str()));
541        }
542    }
543
544    #[tokio::test]
545    async fn get_agent_overview_returns_tokens() {
546        let (_dir, pool, writer) = setup().await;
547        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 0).await;
548        let agent_id = agents[0];
549
550        let overview = get_agent_overview(&pool, run_id, agent_id).await.unwrap();
551        assert_eq!(overview.status, "ok");
552        assert_eq!(overview.input_tokens, 10);
553        assert_eq!(overview.output_tokens, 5);
554        assert_eq!(overview.model.as_deref(), Some("claude-sonnet-4"));
555    }
556
557    #[tokio::test]
558    async fn get_run_overview_aggregates_turns() {
559        let (_dir, pool, writer) = setup().await;
560        let (run_id, _) = seed_run_with_agent(&writer, "t", 2, 3).await;
561
562        let overview = get_run_overview(&pool, run_id).await.unwrap();
563        assert_eq!(overview.run.task, "t");
564        assert_eq!(overview.agents.len(), 2);
565        assert_eq!(overview.total_messages, 6); // 2 agents × 3 messages
566        assert_eq!(overview.turn_counts.len(), 1);
567        assert_eq!(overview.turn_counts[0].kind, "message");
568        assert_eq!(overview.turn_counts[0].count, 6);
569    }
570
571    #[tokio::test]
572    async fn get_run_spans_returns_empty_when_no_spans() {
573        let (_dir, pool, writer) = setup().await;
574        let (run_id, _) = seed_run_with_agent(&writer, "t", 1, 1).await;
575
576        let spans = get_run_spans(&pool, run_id).await.unwrap();
577        assert!(spans.is_empty());
578    }
579
580    #[tokio::test]
581    async fn search_turns_finds_matching_text() {
582        let (_dir, pool, writer) = setup().await;
583        let (run_id, agents) = seed_run_with_agent(&writer, "t", 1, 3).await;
584        let agent_id = agents[0];
585
586        // Add a distinctive message.
587        writer
588            .write_event(&AgentEvent::AgentProgress {
589                run_id,
590                agent_id,
591                delta: ProgressDelta::Message {
592                    text: "the quick brown fox jumps".into(),
593                },
594            })
595            .await
596            .unwrap();
597
598        let results = search_turns(&pool, run_id, "fox", 10).await.unwrap();
599        assert_eq!(results.len(), 1);
600        assert!(results[0].text.as_ref().unwrap().contains("fox"));
601    }
602
603    #[tokio::test]
604    async fn file_edit_turns_appear_in_conversation() {
605        let (_dir, pool, writer) = setup().await;
606        let run_id = uuid::Uuid::now_v7();
607        let agent_id = uuid::Uuid::now_v7();
608
609        writer
610            .write_event(&AgentEvent::AgentStarted {
611                run_id,
612                phase_id: 0,
613                agent_id,
614                prompt_preview: "".into(),
615                model: None,
616                description: None,
617                role: None,
618                name: None,
619                agent_seq: 0,
620                ts: Default::default(),
621            })
622            .await
623            .unwrap();
624
625        writer
626            .write_event(&AgentEvent::AgentProgress {
627                run_id,
628                agent_id,
629                delta: ProgressDelta::FileEdit {
630                    path: PathBuf::from("src/lib.rs"),
631                },
632            })
633            .await
634            .unwrap();
635
636        let turns = get_agent_turns(&pool, run_id, agent_id, 100, 0)
637            .await
638            .unwrap();
639        assert_eq!(turns.len(), 1);
640        assert_eq!(turns[0].kind, "file_edit");
641        assert_eq!(turns[0].file_path.as_deref(), Some("src/lib.rs"));
642    }
643}
luft_storage/reader.rs

luft_storage/
reader.rs
