keel_cli/
flows.rs

1//! `keel flows` and `keel trace <flow>` — the Tier 2 durable-flow inspectors
2//! (dx-spec §6; architecture spec §4.3–4.4).
3//!
4//! Both read only `.keel/journal.db` (the frozen `contracts/journal.sql`
5//! schema), so they are pure queries any SQLite tool could reproduce:
6//!
7//! - `keel flows` lists each flow — id, entrypoint, status, steps done/total,
8//!   age — with `--dead` narrowing to the poison flows a resume gave up on.
9//! - `keel trace <flow>` walks one flow's steps in order with their outcome,
10//!   attempts, and duration.
11//!
12//! Internal control rows never surface: the reserved seq-0 attempt counter and
13//! any `flow:*` replay-branch markers (`kind = 'marker'`) are excluded from both
14//! the step counts and the trace, so the user sees only real work.
15//!
16//! Determinism (dx-spec §5): the `--json` twin carries only values read from the
17//! DB (ids, statuses, `created_at`/`updated_at` in ms) — never a wall-clock
18//! "age", which lives in the human view alone. `age` is computed against an
19//! injected `now`, so the human output is reproducible under test.
20
21use std::path::Path;
22
23use rusqlite::{Connection, OpenFlags};
24use serde::Serialize;
25
26use crate::render::to_json;
27use crate::{EXIT_FAILURE, Rendered, evidence};
28
29/// One flow row for `keel flows`.
30#[derive(Debug, Serialize)]
31struct FlowRow {
32    created_at: i64,
33    entrypoint: String,
34    flow_id: String,
35    status: String,
36    steps_done: i64,
37    steps_total: i64,
38    updated_at: i64,
39}
40
41/// The `keel flows` report — one struct so the human table and `--json` cannot
42/// drift.
43#[derive(Debug, Serialize)]
44struct FlowsReport {
45    count: usize,
46    dead_only: bool,
47    flows: Vec<FlowRow>,
48    journal_present: bool,
49}
50
51/// `keel flows [--dead]` for `project`, dating ages against `now_ms`.
52pub fn flows(project: &Path, dead_only: bool, now_ms: i64) -> Rendered {
53    // Honor the policy's `journal` key (file: locations), like the engine does.
54    let path = evidence::resolved_journal(project).path;
55    if !path.exists() {
56        return Rendered::ok(
57            "keel \u{25b8} no flows yet.\n  Run a flow with `keel run <script>` (a `[flows]` entrypoint) to record one."
58                .to_owned(),
59            to_json(&FlowsReport {
60                count: 0,
61                dead_only,
62                flows: Vec::new(),
63                journal_present: false,
64            }),
65        );
66    }
67    let rows = match read_flows(&path, dead_only) {
68        Ok(r) => r,
69        Err(e) => return soft_error(&e),
70    };
71    let report = FlowsReport {
72        count: rows.len(),
73        dead_only,
74        flows: rows,
75        journal_present: true,
76    };
77    let human = flows_human(&report, now_ms);
78    Rendered::ok(human, to_json(&report))
79}
80
81/// Read the flows table (and per-flow step counts), newest-updated first.
82fn read_flows(path: &Path, dead_only: bool) -> Result<Vec<FlowRow>, String> {
83    let conn = open_ro(path)?;
84    let sql = if dead_only {
85        "SELECT flow_id, entrypoint, status, created_at, updated_at FROM flows \
86         WHERE status = 'dead' ORDER BY updated_at DESC, flow_id"
87    } else {
88        "SELECT flow_id, entrypoint, status, created_at, updated_at FROM flows \
89         ORDER BY updated_at DESC, flow_id"
90    };
91    let mut stmt = conn.prepare(sql).map_err(|e| q(&e))?;
92    let raw = stmt
93        .query_map([], |row| {
94            Ok((
95                row.get::<_, String>(0)?,
96                row.get::<_, String>(1)?,
97                row.get::<_, String>(2)?,
98                row.get::<_, i64>(3)?,
99                row.get::<_, i64>(4)?,
100            ))
101        })
102        .map_err(|e| q(&e))?
103        .collect::<rusqlite::Result<Vec<_>>>()
104        .map_err(|e| q(&e))?;
105
106    let mut out = Vec::with_capacity(raw.len());
107    for (flow_id, entrypoint, status, created_at, updated_at) in raw {
108        let (steps_total, steps_done) = step_counts(&conn, &flow_id)?;
109        out.push(FlowRow {
110            created_at,
111            entrypoint,
112            flow_id,
113            status,
114            steps_done,
115            steps_total,
116            updated_at,
117        });
118    }
119    Ok(out)
120}
121
122/// `(total, done)` real steps for a flow — excluding internal `marker` rows (the
123/// seq-0 attempt counter and replay-branch markers). `done` counts terminal
124/// (`ok`/`error`) outcomes; a still-`running` step is counted only in `total`.
125fn step_counts(conn: &Connection, flow_id: &str) -> Result<(i64, i64), String> {
126    conn.query_row(
127        "SELECT COUNT(*), COALESCE(SUM(CASE WHEN outcome IN ('ok','error') THEN 1 ELSE 0 END), 0) \
128         FROM steps WHERE flow_id = ?1 AND kind != 'marker'",
129        [flow_id],
130        |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
131    )
132    .map_err(|e| q(&e))
133}
134
135/// The human flows table, derived entirely from [`FlowsReport`].
136fn flows_human(report: &FlowsReport, now_ms: i64) -> String {
137    if report.flows.is_empty() {
138        let what = if report.dead_only {
139            "no dead flows \u{2014} nothing has exhausted its resume cap."
140        } else {
141            "no flows recorded yet."
142        };
143        return format!("keel \u{25b8} {what}");
144    }
145    let scope = if report.dead_only { " (dead)" } else { "" };
146    let mut lines = vec![format!(
147        "keel \u{25b8} flows{scope}: {} total\n",
148        report.count
149    )];
150    for f in &report.flows {
151        lines.push(format!(
152            "  {}  {}  {}  steps {}/{}  {}\n",
153            f.flow_id,
154            f.entrypoint,
155            f.status,
156            f.steps_done,
157            f.steps_total,
158            fmt_age(now_ms, f.updated_at),
159        ));
160    }
161    lines.concat()
162}
163
164/// One step line for `keel trace`.
165#[derive(Debug, Serialize)]
166struct TraceStep {
167    attempt: i64,
168    duration_ms: Option<i64>,
169    ended_at: Option<i64>,
170    kind: String,
171    outcome: String,
172    seq: i64,
173    started_at: i64,
174    step_key: String,
175}
176
177/// The `keel trace <flow>` report.
178#[derive(Debug, Serialize)]
179struct TraceReport {
180    created_at: i64,
181    entrypoint: String,
182    flow_id: String,
183    status: String,
184    steps: Vec<TraceStep>,
185    updated_at: i64,
186}
187
188/// `keel trace <flow>` for `project`. `flow` is an exact `flow_id`, or a
189/// substring of an id/entrypoint that resolves to exactly one flow.
190pub fn trace(project: &Path, flow: &str) -> Rendered {
191    let path = evidence::resolved_journal(project).path;
192    if !path.exists() {
193        return soft_error("no journal yet (.keel/journal.db). Run a flow first with `keel run`.");
194    }
195    let conn = match open_ro(&path) {
196        Ok(c) => c,
197        Err(e) => return soft_error(&e),
198    };
199    let resolved = match resolve_flow(&conn, flow) {
200        Ok(r) => r,
201        Err(e) => return soft_error(&e),
202    };
203    let steps = match read_steps(&conn, &resolved.flow_id) {
204        Ok(s) => s,
205        Err(e) => return soft_error(&e),
206    };
207    let report = TraceReport {
208        created_at: resolved.created_at,
209        entrypoint: resolved.entrypoint,
210        flow_id: resolved.flow_id,
211        status: resolved.status,
212        steps,
213        updated_at: resolved.updated_at,
214    };
215    let human = trace_human(&report);
216    Rendered::ok(human, to_json(&report))
217}
218
219/// A resolved flow header (before its steps are read). Shared with
220/// [`replay`](crate::replay), which resolves flows the same way.
221pub(crate) struct ResolvedFlow {
222    pub(crate) flow_id: String,
223    pub(crate) entrypoint: String,
224    pub(crate) status: String,
225    pub(crate) created_at: i64,
226    pub(crate) updated_at: i64,
227}
228
229/// Resolve `flow` to exactly one flow: an exact `flow_id`, else a unique
230/// substring match on id or entrypoint. Ambiguity and absence are precise
231/// errors, never a silent pick.
232pub(crate) fn resolve_flow(conn: &Connection, flow: &str) -> Result<ResolvedFlow, String> {
233    let like = format!("%{flow}%");
234    let mut stmt = conn
235        .prepare(
236            "SELECT flow_id, entrypoint, status, created_at, updated_at FROM flows \
237             WHERE flow_id = ?1 OR flow_id LIKE ?2 OR entrypoint LIKE ?2 \
238             ORDER BY (flow_id = ?1) DESC, updated_at DESC, flow_id",
239        )
240        .map_err(|e| q(&e))?;
241    let matches = stmt
242        .query_map([flow, &like], |row| {
243            Ok(ResolvedFlow {
244                flow_id: row.get(0)?,
245                entrypoint: row.get(1)?,
246                status: row.get(2)?,
247                created_at: row.get(3)?,
248                updated_at: row.get(4)?,
249            })
250        })
251        .map_err(|e| q(&e))?
252        .collect::<rusqlite::Result<Vec<_>>>()
253        .map_err(|e| q(&e))?;
254
255    // An exact id match wins outright even if the substring also hits others.
256    if let Some(first) = matches.first()
257        && first.flow_id == flow
258    {
259        return Ok(matches.into_iter().next().expect("first exists"));
260    }
261    match matches.len() {
262        0 => Err(format!(
263            "no flow matches {flow:?} in the journal. Run `keel flows` to list recorded flows."
264        )),
265        1 => Ok(matches.into_iter().next().expect("one match")),
266        n => {
267            let ids: Vec<&str> = matches.iter().take(5).map(|m| m.flow_id.as_str()).collect();
268            Err(format!(
269                "{flow:?} matches {n} flows ({}); use a full flow_id.",
270                ids.join(", ")
271            ))
272        }
273    }
274}
275
276/// Read a flow's real steps (excluding `marker` rows) in seq order.
277fn read_steps(conn: &Connection, flow_id: &str) -> Result<Vec<TraceStep>, String> {
278    let mut stmt = conn
279        .prepare(
280            "SELECT seq, step_key, kind, attempt, outcome, started_at, ended_at FROM steps \
281             WHERE flow_id = ?1 AND kind != 'marker' ORDER BY seq",
282        )
283        .map_err(|e| q(&e))?;
284    let steps = stmt
285        .query_map([flow_id], |row| {
286            let seq: i64 = row.get(0)?;
287            let step_key: String = row.get(1)?;
288            let kind: String = row.get(2)?;
289            let attempt: i64 = row.get(3)?;
290            let outcome: String = row.get(4)?;
291            let started_at: i64 = row.get(5)?;
292            let ended_at: Option<i64> = row.get(6)?;
293            Ok(TraceStep {
294                attempt,
295                duration_ms: ended_at.map(|e| e - started_at),
296                ended_at,
297                kind,
298                outcome,
299                seq,
300                started_at,
301                step_key,
302            })
303        })
304        .map_err(|e| q(&e))?
305        .collect::<rusqlite::Result<Vec<_>>>()
306        .map_err(|e| q(&e))?;
307    Ok(steps)
308}
309
310/// The human trace, derived entirely from [`TraceReport`].
311fn trace_human(report: &TraceReport) -> String {
312    let mut lines = vec![
313        format!(
314            "keel \u{25b8} trace {}\n  entrypoint: {}\n  status:     {}\n",
315            report.flow_id, report.entrypoint, report.status
316        ),
317        format!("  steps:      {}\n", report.steps.len()),
318    ];
319    for s in &report.steps {
320        let dur = match s.duration_ms {
321            Some(ms) => format!("{ms}ms"),
322            None => "\u{2014}".to_owned(), // still running
323        };
324        let attempts = if s.attempt > 1 {
325            format!(" ({} attempts)", s.attempt)
326        } else {
327            String::new()
328        };
329        lines.push(format!(
330            "    {:>3}. {:<7} {:<28} {:<8} {}{}\n",
331            s.seq, s.kind, s.step_key, s.outcome, dur, attempts,
332        ));
333    }
334    lines.concat()
335}
336
337/// A coarse human age ("5s", "3m", "2h", "1d") from `then_ms` to `now_ms`.
338fn fmt_age(now_ms: i64, then_ms: i64) -> String {
339    let secs = (now_ms - then_ms).max(0) / 1000;
340    if secs < 60 {
341        format!("{secs}s ago")
342    } else if secs < 3600 {
343        format!("{}m ago", secs / 60)
344    } else if secs < 86_400 {
345        format!("{}h ago", secs / 3600)
346    } else {
347        format!("{}d ago", secs / 86_400)
348    }
349}
350
351pub(crate) fn open_ro(path: &Path) -> Result<Connection, String> {
352    Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)
353        .map_err(|e| format!("could not open {}: {e}", path.display()))
354}
355
356pub(crate) fn q(e: &rusqlite::Error) -> String {
357    format!("journal query failed: {e}")
358}
359
360/// A read failure (exit 1, on stderr) — mirrors `keel status`. Shared with
361/// [`crate::replay`] (the same journal reads) and `keel flows suggest` (same
362/// failure mode there: a corrupt `.keel/discovery.db`).
363pub(crate) fn soft_error(message: &str) -> Rendered {
364    #[derive(Serialize)]
365    struct ErrReport<'a> {
366        error: &'a str,
367    }
368    Rendered {
369        human: format!("keel \u{25b8} {message}"),
370        json: to_json(&ErrReport { error: message }),
371        exit: EXIT_FAILURE,
372        to_stderr: true,
373    }
374}
375
376#[cfg(test)]
377mod tests {
378    use super::*;
379    use std::path::PathBuf;
380
381    const T0: i64 = 1_783_728_000_000;
382
383    /// Build a project dir with `.keel/journal.db` from a golden fixture
384    /// (`conformance/fixtures/journal/<fixture>`), applied over the frozen
385    /// schema — self-contained, no reliance on the checked-in `.gen/` build.
386    fn project_with_fixture(fixture: &str) -> (tempfile::TempDir, PathBuf) {
387        let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
388        let schema = std::fs::read_to_string(root.join("contracts/journal.sql")).unwrap();
389        let sql = std::fs::read_to_string(root.join("conformance/fixtures/journal").join(fixture))
390            .unwrap();
391        let dir = tempfile::TempDir::new().unwrap();
392        let keel = dir.path().join(".keel");
393        std::fs::create_dir_all(&keel).unwrap();
394        let db = keel.join("journal.db");
395        let conn = Connection::open(&db).unwrap();
396        conn.execute_batch(&schema).unwrap();
397        conn.execute_batch(&sql).unwrap();
398        let project = dir.path().to_path_buf();
399        (dir, project)
400    }
401
402    #[test]
403    fn flows_lists_completed_with_step_counts() {
404        let (_d, project) = project_with_fixture("completed-flow.sql");
405        let r = flows(&project, false, T0 + 10_000);
406        assert_eq!(r.exit, crate::EXIT_OK);
407        assert_eq!(r.json["count"], 1);
408        let f = &r.json["flows"][0];
409        assert_eq!(f["entrypoint"], "py:pipeline.ingest:main");
410        assert_eq!(f["status"], "completed");
411        assert_eq!(f["steps_done"], 5);
412        assert_eq!(f["steps_total"], 5);
413        // Human view is deterministic under a fixed `now`.
414        assert!(r.human.contains("steps 5/5"));
415        assert!(r.human.contains("ago"));
416    }
417
418    #[test]
419    fn flows_interrupted_shows_incomplete_step_count() {
420        let (_d, project) = project_with_fixture("interrupted-flow.sql");
421        let r = flows(&project, false, T0 + 60_000);
422        let f = &r.json["flows"][0];
423        assert_eq!(f["status"], "running");
424        // 4 recorded steps; step 4 is still `running` → 3 done of 4.
425        assert_eq!(f["steps_total"], 4);
426        assert_eq!(f["steps_done"], 3);
427    }
428
429    #[test]
430    fn flows_dead_filter_selects_only_dead() {
431        let (_d, project) = project_with_fixture("dead-flow.sql");
432        let all = flows(&project, false, T0);
433        assert_eq!(all.json["count"], 1);
434        let dead = flows(&project, true, T0);
435        assert_eq!(dead.json["count"], 1);
436        assert_eq!(dead.json["dead_only"], true);
437        assert_eq!(dead.json["flows"][0]["status"], "dead");
438        // A completed fixture has no dead flows.
439        let (_d2, p2) = project_with_fixture("completed-flow.sql");
440        assert_eq!(flows(&p2, true, T0).json["count"], 0);
441    }
442
443    #[test]
444    fn flows_absent_journal_nudges() {
445        let dir = tempfile::TempDir::new().unwrap();
446        let r = flows(dir.path(), false, T0);
447        assert_eq!(r.exit, crate::EXIT_OK);
448        assert_eq!(r.json["journal_present"], false);
449        assert_eq!(r.json["count"], 0);
450    }
451
452    #[test]
453    fn trace_walks_steps_with_outcomes_and_durations() {
454        let (_d, project) = project_with_fixture("completed-flow.sql");
455        let r = trace(&project, "py:pipeline.ingest:main");
456        assert_eq!(r.exit, crate::EXIT_OK);
457        assert_eq!(r.json["status"], "completed");
458        let steps = r.json["steps"].as_array().unwrap();
459        assert_eq!(steps.len(), 5);
460        // seq 1: the fetch effect, 240ms.
461        assert_eq!(steps[0]["seq"], 1);
462        assert_eq!(steps[0]["step_key"], "api.source.internal#q1");
463        assert_eq!(steps[0]["outcome"], "ok");
464        assert_eq!(steps[0]["duration_ms"], 240);
465        // seq 2: the virtualized time read.
466        assert_eq!(steps[1]["kind"], "time");
467        assert_eq!(steps[1]["step_key"], "py:time.time#-");
468        // seq 3: enrich succeeded on its 2nd attempt.
469        assert_eq!(steps[2]["attempt"], 2);
470        assert!(r.human.contains("2 attempts"));
471    }
472
473    #[test]
474    fn trace_running_step_has_no_duration() {
475        let (_d, project) = project_with_fixture("interrupted-flow.sql");
476        let r = trace(&project, "01JZWY0A0000000000000002");
477        let steps = r.json["steps"].as_array().unwrap();
478        assert_eq!(steps.len(), 4);
479        assert_eq!(steps[3]["outcome"], "running");
480        assert!(steps[3]["duration_ms"].is_null());
481        assert!(steps[3]["ended_at"].is_null());
482    }
483
484    #[test]
485    fn trace_unknown_flow_is_a_soft_error() {
486        let (_d, project) = project_with_fixture("completed-flow.sql");
487        let r = trace(&project, "does-not-exist");
488        assert_eq!(r.exit, EXIT_FAILURE);
489        assert!(r.to_stderr);
490        assert!(r.human.contains("no flow matches"));
491    }
492}
keel_cli/flows.rs

keel_cli/
flows.rs
