Skip to main content

kaizen/shell/
cli.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2//! CLI command implementations.
3
4use crate::collect::tail::claude::scan_claude_session_dir;
5use crate::collect::tail::codex::scan_codex_session_dir;
6use crate::collect::tail::copilot_cli::scan_copilot_cli_workspace;
7use crate::collect::tail::copilot_vscode::scan_copilot_vscode_workspace;
8use crate::collect::tail::cursor::scan_session_dir_all;
9use crate::collect::tail::goose::scan_goose_workspace;
10use crate::collect::tail::openclaw::scan_openclaw_workspace;
11use crate::collect::tail::opencode::scan_opencode_workspace;
12use crate::core::config;
13use crate::core::event::{Event, SessionRecord};
14use crate::metrics::report;
15use crate::shell::fmt::fmt_ts;
16use crate::shell::scope;
17use crate::store::{SYNC_STATE_LAST_AGENT_SCAN_MS, SYNC_STATE_LAST_AUTO_PRUNE_MS, Store};
18use anyhow::Result;
19use serde::Serialize;
20use std::collections::HashMap;
21use std::io::IsTerminal;
22use std::path::{Path, PathBuf};
23
24pub use crate::shell::init::cmd_init;
25pub use crate::shell::insights::cmd_insights;
26
27#[derive(Serialize)]
28struct SessionsListJson {
29    workspace: String,
30    #[serde(skip_serializing_if = "Vec::is_empty")]
31    workspaces: Vec<String>,
32    count: usize,
33    sessions: Vec<SessionRecord>,
34}
35
36#[derive(Serialize)]
37struct SummaryJsonOut {
38    workspace: String,
39    #[serde(skip_serializing_if = "Vec::is_empty")]
40    workspaces: Vec<String>,
41    #[serde(flatten)]
42    stats: crate::store::SummaryStats,
43    cost_usd: f64,
44    #[serde(skip_serializing_if = "Option::is_none")]
45    cost_note: Option<String>,
46    #[serde(skip_serializing_if = "Option::is_none")]
47    hotspot: Option<crate::metrics::types::RankedFile>,
48    #[serde(skip_serializing_if = "Option::is_none")]
49    slowest_tool: Option<crate::metrics::types::RankedTool>,
50}
51
52/// Summary/MCP: sessions exist but rollup has no stored micro-USD — show honest footnote, not invented spend.
53pub(crate) fn summary_needs_cost_rollup_note(session_count: u64, total_cost_usd_e6: i64) -> bool {
54    session_count > 0 && total_cost_usd_e6 == 0
55}
56
57pub(crate) fn cost_rollup_zero_note_paragraph() -> &'static str {
58    "Cost rollup shows $0.00 because stored events have no cost_usd_e6 — common when Cursor agent-transcript lines omit usage/tokens. \
59If you expect non-zero spend, ingest Claude/Codex transcripts with usage, hooks with total_cost_usd, or Kaizen proxy Cost events; run `kaizen summary --refresh` after ingest changes. \
60See docs/usage.md#cost-shows-zero."
61}
62
63pub(crate) fn cost_rollup_zero_doctor_hint() -> &'static str {
64    "Cost rollup $0.00 with sessions but no cost_usd_e6 — often Cursor transcripts without usage; see docs/usage.md#cost-shows-zero"
65}
66
67struct ScanSpinner(Option<indicatif::ProgressBar>);
68
69impl ScanSpinner {
70    fn start(msg: &'static str) -> Self {
71        if !std::io::stdout().is_terminal() {
72            return Self(None);
73        }
74        let p = indicatif::ProgressBar::new_spinner();
75        p.set_message(msg.to_string());
76        p.enable_steady_tick(std::time::Duration::from_millis(120));
77        Self(Some(p))
78    }
79}
80
81impl Drop for ScanSpinner {
82    fn drop(&mut self) {
83        if let Some(p) = self.0.take() {
84            p.finish_and_clear();
85        }
86    }
87}
88
89fn now_ms_u64() -> u64 {
90    std::time::SystemTime::now()
91        .duration_since(std::time::UNIX_EPOCH)
92        .unwrap_or_default()
93        .as_millis() as u64
94}
95
96/// Minimum interval between automatic local DB prunes after a successful rescan (24h).
97const AUTO_PRUNE_INTERVAL_MS: u64 = 86_400_000;
98
99pub(crate) fn maybe_auto_prune_after_scan(store: &Store, cfg: &config::Config) -> Result<()> {
100    if cfg.retention.hot_days == 0 {
101        return Ok(());
102    }
103    let now = now_ms_u64();
104    if let Some(last) = store.sync_state_get_u64(SYNC_STATE_LAST_AUTO_PRUNE_MS)?
105        && now.saturating_sub(last) < AUTO_PRUNE_INTERVAL_MS
106    {
107        return Ok(());
108    }
109    let cutoff = now.saturating_sub((cfg.retention.hot_days as u64).saturating_mul(86_400_000));
110    store.prune_sessions_started_before(cutoff as i64)?;
111    store.sync_state_set_u64(SYNC_STATE_LAST_AUTO_PRUNE_MS, now)?;
112    Ok(())
113}
114
115/// Full transcript rescan unless throttled by `[scan].min_rescan_seconds` or `refresh` is true.
116pub(crate) fn maybe_scan_all_agents(
117    ws: &Path,
118    cfg: &config::Config,
119    ws_str: &str,
120    store: &Store,
121    refresh: bool,
122) -> Result<()> {
123    let interval_ms = cfg.scan.min_rescan_seconds.saturating_mul(1000);
124    let now = now_ms_u64();
125    if !refresh
126        && interval_ms > 0
127        && let Some(last) = store.sync_state_get_u64(SYNC_STATE_LAST_AGENT_SCAN_MS)?
128        && now.saturating_sub(last) < interval_ms
129    {
130        return Ok(());
131    }
132    scan_all_agents(ws, cfg, ws_str, store)?;
133    store.sync_state_set_u64(SYNC_STATE_LAST_AGENT_SCAN_MS, now_ms_u64())?;
134    Ok(())
135}
136
137pub(crate) fn maybe_refresh_store(workspace: &Path, store: &Store, refresh: bool) -> Result<()> {
138    if !refresh {
139        return Ok(());
140    }
141    let cfg = config::load(workspace)?;
142    let ws_str = workspace.to_string_lossy().to_string();
143    maybe_scan_all_agents(workspace, &cfg, &ws_str, store, true)
144}
145
146fn combine_counts(rows: Vec<Vec<(String, u64)>>) -> Vec<(String, u64)> {
147    let mut counts = HashMap::new();
148    for set in rows {
149        for (key, value) in set {
150            *counts.entry(key).or_insert(0_u64) += value;
151        }
152    }
153    let mut out = counts.into_iter().collect::<Vec<_>>();
154    out.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
155    out
156}
157
158fn workspace_names(roots: &[PathBuf]) -> Vec<String> {
159    roots
160        .iter()
161        .map(|path| path.to_string_lossy().to_string())
162        .collect()
163}
164
165fn open_workspace_store(workspace: &Path) -> Result<Store> {
166    Store::open(&crate::core::workspace::db_path(workspace))
167}
168
169pub(crate) fn open_workspace_read_store(workspace: &Path, refresh: bool) -> Result<Store> {
170    let db_path = crate::core::workspace::db_path(workspace);
171    if refresh || !db_path.exists() {
172        Store::open(&db_path)
173    } else {
174        Store::open_query(&db_path)
175    }
176}
177
178/// `kaizen sessions list` — same output as CLI stdout.
179pub fn sessions_list_text(
180    workspace: Option<&Path>,
181    json_out: bool,
182    refresh: bool,
183    all_workspaces: bool,
184    limit: Option<usize>,
185) -> Result<String> {
186    let roots = scope::resolve(workspace, all_workspaces)?;
187    let mut sessions = Vec::new();
188    if crate::daemon::enabled() && !refresh {
189        for workspace in &roots {
190            let ws_str = workspace.to_string_lossy().to_string();
191            let response =
192                crate::daemon::request_blocking(crate::ipc::DaemonRequest::ListSessions {
193                    workspace: ws_str,
194                    offset: 0,
195                    limit: i64::MAX as usize,
196                    filter: crate::store::SessionFilter::default(),
197                })?;
198            match response {
199                crate::ipc::DaemonResponse::Sessions(page) => sessions.extend(page.rows),
200                crate::ipc::DaemonResponse::Error { message, .. } => anyhow::bail!(message),
201                _ => anyhow::bail!("unexpected daemon sessions response"),
202            }
203        }
204    } else {
205        for workspace in &roots {
206            let store = open_workspace_read_store(workspace, refresh)?;
207            maybe_refresh_store(workspace, &store, refresh)?;
208            let ws_str = workspace.to_string_lossy().to_string();
209            sessions.extend(store.list_sessions(&ws_str)?);
210        }
211    }
212    sessions.sort_by(|a, b| {
213        b.started_at_ms
214            .cmp(&a.started_at_ms)
215            .then_with(|| a.id.cmp(&b.id))
216    });
217    let output_limit = limit.unwrap_or(100);
218    if output_limit > 0 {
219        let n = output_limit;
220        sessions.truncate(n);
221    }
222    let scope_label = scope::label(&roots);
223    let workspaces = if roots.len() > 1 {
224        workspace_names(&roots)
225    } else {
226        Vec::new()
227    };
228    if json_out {
229        return Ok(format!(
230            "{}\n",
231            serde_json::to_string_pretty(&SessionsListJson {
232                workspace: scope_label,
233                workspaces,
234                count: sessions.len(),
235                sessions,
236            })?
237        ));
238    }
239    use std::fmt::Write;
240    let mut out = String::new();
241    if roots.len() > 1 {
242        writeln!(&mut out, "Scope: {scope_label}").unwrap();
243        writeln!(&mut out).unwrap();
244    }
245    writeln!(
246        &mut out,
247        "{:<40} {:<10} {:<10} STARTED",
248        "ID", "AGENT", "STATUS"
249    )
250    .unwrap();
251    writeln!(&mut out, "{}", "-".repeat(80)).unwrap();
252    for s in &sessions {
253        writeln!(
254            &mut out,
255            "{:<40} {:<10} {:<10} {}",
256            s.id,
257            s.agent,
258            format!("{:?}", s.status),
259            fmt_ts(s.started_at_ms),
260        )
261        .unwrap();
262    }
263    if sessions.is_empty() {
264        writeln!(&mut out, "(no sessions)").unwrap();
265        sessions_empty_state_hints(&mut out);
266    }
267    Ok(out)
268}
269
270fn sessions_empty_state_hints(out: &mut String) {
271    use std::fmt::Write;
272    let _ = writeln!(out);
273    let _ = writeln!(out, "No sessions found for this workspace. Try:");
274    let _ = writeln!(out, "  · `kaizen doctor` — verify config and hooks");
275    let _ = writeln!(out, "  · a short agent session in this repo, then re-run");
276    let _ = writeln!(
277        out,
278        "  · docs: https://github.com/marquesds/kaizen/blob/main/docs/config.md (sources)"
279    );
280}
281
282/// `kaizen sessions list` — scan all agent transcripts, upsert sessions, print table.
283pub fn cmd_sessions_list(
284    workspace: Option<&Path>,
285    json_out: bool,
286    refresh: bool,
287    all_workspaces: bool,
288    limit: Option<usize>,
289) -> Result<()> {
290    print!(
291        "{}",
292        sessions_list_text(workspace, json_out, refresh, all_workspaces, limit)?
293    );
294    Ok(())
295}
296
297/// `kaizen sessions show` — same output as CLI stdout.
298pub fn session_show_text(id: &str, workspace: Option<&Path>) -> Result<String> {
299    let ws = workspace_path(workspace)?;
300    let store = open_workspace_store(&ws)?;
301    use std::fmt::Write;
302    let mut out = String::new();
303    match store.get_session(id)? {
304        Some(s) => {
305            writeln!(&mut out, "id:           {}", s.id).unwrap();
306            writeln!(&mut out, "agent:        {}", s.agent).unwrap();
307            writeln!(
308                &mut out,
309                "model:        {}",
310                s.model.as_deref().unwrap_or("-")
311            )
312            .unwrap();
313            writeln!(&mut out, "workspace:    {}", s.workspace).unwrap();
314            writeln!(&mut out, "started_at:   {}", fmt_ts(s.started_at_ms)).unwrap();
315            writeln!(
316                &mut out,
317                "ended_at:     {}",
318                s.ended_at_ms.map(fmt_ts).unwrap_or_else(|| "-".to_string())
319            )
320            .unwrap();
321            writeln!(&mut out, "status:       {:?}", s.status).unwrap();
322            writeln!(&mut out, "trace_path:   {}", s.trace_path).unwrap();
323            if let Some(fp) = &s.prompt_fingerprint {
324                writeln!(&mut out, "prompt_fp:    {fp}").unwrap();
325                if let Ok(Some(snap)) = store.get_prompt_snapshot(fp) {
326                    for f in snap.files() {
327                        writeln!(&mut out, "  - {}", f.path).unwrap();
328                    }
329                }
330            }
331        }
332        None => anyhow::bail!("session not found: {id} — try `kaizen sessions list`"),
333    }
334    let evals = store.list_evals_for_session(id).unwrap_or_default();
335    if !evals.is_empty() {
336        writeln!(&mut out, "evals:").unwrap();
337        for e in &evals {
338            writeln!(
339                &mut out,
340                "  {} score={:.2} flagged={} {}",
341                e.rubric_id, e.score, e.flagged, e.rationale
342            )
343            .unwrap();
344        }
345    }
346    let fb = store
347        .feedback_for_sessions(&[id.to_string()])
348        .unwrap_or_default();
349    if let Some(r) = fb.get(id) {
350        let score = r
351            .score
352            .as_ref()
353            .map(|s| s.0.to_string())
354            .unwrap_or_else(|| "-".into());
355        let label = r
356            .label
357            .as_ref()
358            .map(|l| l.to_string())
359            .unwrap_or_else(|| "-".into());
360        writeln!(&mut out, "feedback:     score={score} label={label}").unwrap();
361        if let Some(n) = &r.note {
362            writeln!(&mut out, "  note: {n}").unwrap();
363        }
364    }
365    Ok(out)
366}
367
368/// `kaizen sessions show <id>` — print full session fields.
369pub fn cmd_session_show(id: &str, workspace: Option<&Path>) -> Result<()> {
370    print!("{}", session_show_text(id, workspace)?);
371    Ok(())
372}
373
374pub fn sessions_tree_text(id: &str, max_depth: u32, workspace: Option<&Path>) -> Result<String> {
375    let ws = workspace_path(workspace)?;
376    let store = open_workspace_store(&ws)?;
377    let nodes = store.session_span_tree(id)?;
378    if nodes.is_empty() {
379        if store.get_session(id)?.is_none() {
380            anyhow::bail!("session not found: {id}");
381        }
382        return Ok(format!("(no tool spans for session {id})\n"));
383    }
384    let total_cost: i64 = nodes.iter().map(|n| n.subtree_cost_usd_e6).sum();
385    let mut out = String::new();
386    for node in &nodes {
387        render_node(&mut out, node, 0, max_depth, total_cost);
388    }
389    Ok(out)
390}
391
392fn render_node(
393    out: &mut String,
394    node: &crate::store::span_tree::SpanNode,
395    depth: u32,
396    max_depth: u32,
397    session_total: i64,
398) {
399    use std::fmt::Write;
400    if depth > max_depth {
401        return;
402    }
403    let indent = "│  ".repeat(depth as usize);
404    let prefix = if depth == 0 { "┌─ " } else { "├─ " };
405    let cost_str = match node.span.subtree_cost_usd_e6 {
406        Some(c) => {
407            let pct = if session_total > 0 {
408                c * 100 / session_total
409            } else {
410                0
411            };
412            let flag = if pct > 40 { " ⚡" } else { "" };
413            format!(" ${:.4}{}", c as f64 / 1_000_000.0, flag)
414        }
415        None => String::new(),
416    };
417    writeln!(
418        out,
419        "{}{}{} [{}]{}",
420        indent, prefix, node.span.tool, node.span.status, cost_str
421    )
422    .unwrap();
423    for child in &node.children {
424        render_node(out, child, depth + 1, max_depth, session_total);
425    }
426}
427
428/// `kaizen sessions tree <id>` — produce text output (ASCII or JSON).
429pub fn cmd_sessions_tree_text(
430    id: &str,
431    depth: u32,
432    json: bool,
433    workspace: Option<&Path>,
434) -> Result<String> {
435    if json {
436        let ws = workspace_path(workspace)?;
437        let store = open_workspace_read_store(&ws, false)?;
438        let nodes = store.session_span_tree(id)?;
439        Ok(serde_json::to_string_pretty(&nodes)?)
440    } else {
441        sessions_tree_text(id, depth, workspace)
442    }
443}
444
445/// `kaizen sessions tree <id>` — print ASCII span tree.
446pub fn cmd_sessions_tree(id: &str, depth: u32, json: bool, workspace: Option<&Path>) -> Result<()> {
447    print!("{}", cmd_sessions_tree_text(id, depth, json, workspace)?);
448    Ok(())
449}
450
451/// `kaizen summary` — same output as CLI stdout.
452pub fn summary_text(
453    workspace: Option<&Path>,
454    json_out: bool,
455    refresh: bool,
456    all_workspaces: bool,
457    source: crate::core::data_source::DataSource,
458) -> Result<String> {
459    let roots = scope::resolve(workspace, all_workspaces)?;
460    let mut total_cost_usd_e6 = 0_i64;
461    let mut session_count = 0_u64;
462    let mut by_agent = Vec::new();
463    let mut by_model = Vec::new();
464    let mut top_tools = Vec::new();
465    let mut hottest = Vec::new();
466    let mut slowest = Vec::new();
467
468    for workspace in &roots {
469        let cfg = config::load(workspace)?;
470        let store = open_workspace_read_store(
471            workspace,
472            refresh || source != crate::core::data_source::DataSource::Local,
473        )?;
474        crate::shell::remote_pull::maybe_telemetry_pull(workspace, &store, &cfg, source, refresh)?;
475        maybe_refresh_store(workspace, &store, refresh)?;
476        let ws_str = workspace.to_string_lossy().to_string();
477        let read_store = open_workspace_read_store(workspace, false)?;
478        let query = crate::store::query::QueryStore::open(&workspace.join(".kaizen"))?;
479        let mut stats = query.summary_stats(&read_store, &ws_str)?;
480        if source != crate::core::data_source::DataSource::Local
481            && let Ok(Some(agg)) =
482                crate::shell::remote_observe::try_remote_event_agg(&read_store, &cfg, workspace)
483        {
484            stats = crate::shell::remote_observe::merge_summary_stats(stats, &agg, source);
485        }
486        total_cost_usd_e6 += stats.total_cost_usd_e6;
487        session_count += stats.session_count;
488        by_agent.push(stats.by_agent);
489        by_model.push(stats.by_model);
490        top_tools.push(stats.top_tools);
491        if let Ok(metrics) = report::build_report(&read_store, &ws_str, 7) {
492            if let Some(file) = metrics.hottest_files.first().cloned() {
493                hottest.push(if roots.len() == 1 {
494                    file
495                } else {
496                    crate::metrics::types::RankedFile {
497                        path: scope::decorate_path(workspace, &file.path),
498                        ..file
499                    }
500                });
501            }
502            if let Some(tool) = metrics.slowest_tools.first().cloned() {
503                slowest.push(tool);
504            }
505        }
506    }
507
508    let stats = crate::store::SummaryStats {
509        session_count,
510        total_cost_usd_e6,
511        by_agent: combine_counts(by_agent),
512        by_model: combine_counts(by_model),
513        top_tools: combine_counts(top_tools),
514    };
515    let cost_dollars = stats.total_cost_usd_e6 as f64 / 1_000_000.0;
516    let hotspot = hottest
517        .into_iter()
518        .max_by(|a, b| a.value.cmp(&b.value).then_with(|| b.path.cmp(&a.path)));
519    let slowest_tool = slowest.into_iter().max_by(|a, b| {
520        a.p95_ms
521            .unwrap_or(0)
522            .cmp(&b.p95_ms.unwrap_or(0))
523            .then_with(|| b.tool.cmp(&a.tool))
524    });
525    let scope_label = scope::label(&roots);
526    let workspaces = if roots.len() > 1 {
527        workspace_names(&roots)
528    } else {
529        Vec::new()
530    };
531    let cost_note = summary_needs_cost_rollup_note(stats.session_count, stats.total_cost_usd_e6)
532        .then_some(cost_rollup_zero_note_paragraph().to_string());
533    if json_out {
534        return Ok(format!(
535            "{}\n",
536            serde_json::to_string_pretty(&SummaryJsonOut {
537                workspace: scope_label,
538                workspaces,
539                cost_usd: cost_dollars,
540                stats,
541                cost_note,
542                hotspot,
543                slowest_tool,
544            })?
545        ));
546    }
547    use std::fmt::Write;
548    let mut out = String::new();
549    if roots.len() > 1 {
550        writeln!(&mut out, "Scope: {}", scope::label(&roots)).unwrap();
551    }
552    writeln!(
553        &mut out,
554        "Sessions: {}   Cost: ${:.2}",
555        stats.session_count, cost_dollars
556    )
557    .unwrap();
558
559    if !stats.by_agent.is_empty() {
560        let parts: Vec<String> = stats
561            .by_agent
562            .iter()
563            .map(|(a, n)| format!("{a} {n}"))
564            .collect();
565        writeln!(&mut out, "By agent:  {}", parts.join(" · ")).unwrap();
566    }
567    if !stats.by_model.is_empty() {
568        let parts: Vec<String> = stats
569            .by_model
570            .iter()
571            .map(|(m, n)| format!("{m} {n}"))
572            .collect();
573        writeln!(&mut out, "By model:  {}", parts.join(" · ")).unwrap();
574    }
575    if !stats.top_tools.is_empty() {
576        let parts: Vec<String> = stats
577            .top_tools
578            .iter()
579            .take(5)
580            .map(|(t, n)| format!("{t} {n}"))
581            .collect();
582        writeln!(&mut out, "Top tools: {}", parts.join(" · ")).unwrap();
583    }
584    if let Some(file) = hotspot {
585        writeln!(&mut out, "Hotspot:   {} ({})", file.path, file.value).unwrap();
586    }
587    if let Some(tool) = slowest_tool {
588        let p95 = tool
589            .p95_ms
590            .map(|v| format!("{v}ms"))
591            .unwrap_or_else(|| "-".into());
592        writeln!(&mut out, "Slowest:   {} p95 {}", tool.tool, p95).unwrap();
593    }
594    if cost_note.is_some() {
595        writeln!(&mut out).unwrap();
596        writeln!(&mut out, "Note: {}", cost_rollup_zero_note_paragraph()).unwrap();
597    }
598    Ok(out)
599}
600
601/// `kaizen summary` — aggregate session + cost stats across all agents.
602pub fn cmd_summary(
603    workspace: Option<&Path>,
604    json_out: bool,
605    refresh: bool,
606    all_workspaces: bool,
607    source: crate::core::data_source::DataSource,
608) -> Result<()> {
609    print!(
610        "{}",
611        summary_text(workspace, json_out, refresh, all_workspaces, source,)?
612    );
613    Ok(())
614}
615
616pub(crate) fn scan_all_agents(
617    ws: &Path,
618    cfg: &config::Config,
619    ws_str: &str,
620    store: &Store,
621) -> Result<()> {
622    let _spin = ScanSpinner::start("Scanning agent sessions…");
623    let slug = workspace_slug(ws_str);
624    let sync_ctx = crate::sync::ingest_ctx(cfg, ws.to_path_buf());
625
626    for root in &cfg.scan.roots {
627        let expanded = expand_home(root);
628        let cursor_dir = PathBuf::from(&expanded)
629            .join(&slug)
630            .join("agent-transcripts");
631        scan_agent_dirs(
632            &cursor_dir,
633            store,
634            |p| {
635                scan_session_dir_all(p).map(|sessions| {
636                    sessions
637                        .into_iter()
638                        .map(|(mut r, evs)| {
639                            r.workspace = ws_str.to_string();
640                            (r, evs)
641                        })
642                        .collect()
643                })
644            },
645            sync_ctx.as_ref(),
646        )?;
647    }
648
649    let home = std::env::var("HOME").unwrap_or_default();
650
651    let claude_dir = PathBuf::from(&home)
652        .join(".claude/projects")
653        .join(&slug)
654        .join("sessions");
655    scan_agent_dirs(
656        &claude_dir,
657        store,
658        |p| {
659            scan_claude_session_dir(p).map(|(mut r, evs)| {
660                r.workspace = ws_str.to_string();
661                vec![(r, evs)]
662            })
663        },
664        sync_ctx.as_ref(),
665    )?;
666
667    let codex_dir = PathBuf::from(&home).join(".codex/sessions").join(&slug);
668    scan_agent_dirs(
669        &codex_dir,
670        store,
671        |p| {
672            scan_codex_session_dir(p).map(|(mut r, evs)| {
673                r.workspace = ws_str.to_string();
674                vec![(r, evs)]
675            })
676        },
677        sync_ctx.as_ref(),
678    )?;
679
680    let tail = &cfg.sources.tail;
681    let home_pb = PathBuf::from(&home);
682    if tail.goose {
683        let sessions = scan_goose_workspace(&home_pb, ws)?;
684        persist_session_batch(store, sessions, sync_ctx.as_ref())?;
685    }
686    if tail.openclaw {
687        let sessions = scan_openclaw_workspace(ws)?;
688        persist_session_batch(store, sessions, sync_ctx.as_ref())?;
689    }
690    if tail.opencode {
691        let sessions = scan_opencode_workspace(ws)?;
692        persist_session_batch(store, sessions, sync_ctx.as_ref())?;
693    }
694    if tail.copilot_cli {
695        let sessions = scan_copilot_cli_workspace(ws)?;
696        persist_session_batch(store, sessions, sync_ctx.as_ref())?;
697    }
698    if tail.copilot_vscode {
699        let sessions = scan_copilot_vscode_workspace(ws)?;
700        persist_session_batch(store, sessions, sync_ctx.as_ref())?;
701    }
702
703    maybe_auto_prune_after_scan(store, cfg)?;
704    Ok(())
705}
706
707fn persist_session_batch(
708    store: &Store,
709    sessions: Vec<(SessionRecord, Vec<Event>)>,
710    sync_ctx: Option<&crate::sync::SyncIngestContext>,
711) -> Result<()> {
712    for (mut record, events) in sessions {
713        if record.start_commit.is_none() && !record.workspace.is_empty() {
714            let binding = crate::core::repo::binding_for_session(
715                Path::new(&record.workspace),
716                record.started_at_ms,
717                record.ended_at_ms,
718            );
719            record.start_commit = binding.start_commit;
720            record.end_commit = binding.end_commit;
721            record.branch = binding.branch;
722            record.dirty_start = binding.dirty_start;
723            record.dirty_end = binding.dirty_end;
724            record.repo_binding_source = binding.source;
725        }
726        store.upsert_session(&record)?;
727        let flush_ms = record.ended_at_ms.unwrap_or(record.started_at_ms);
728        for ev in events {
729            store.append_event_with_sync(&ev, sync_ctx)?;
730        }
731        if record.status == crate::core::event::SessionStatus::Done {
732            store.flush_projector_session(&record.id, flush_ms)?;
733        }
734    }
735    Ok(())
736}
737
738pub(crate) fn scan_agent_dirs<F>(
739    dir: &Path,
740    store: &Store,
741    scanner: F,
742    sync_ctx: Option<&crate::sync::SyncIngestContext>,
743) -> Result<()>
744where
745    F: Fn(&Path) -> Result<Vec<(SessionRecord, Vec<Event>)>>,
746{
747    if !dir.exists() {
748        return Ok(());
749    }
750    for entry in std::fs::read_dir(dir)?.filter_map(|e| e.ok()) {
751        if !entry.file_type().map(|t| t.is_dir()).unwrap_or(false) {
752            continue;
753        }
754        match scanner(&entry.path()) {
755            Ok(sessions) => {
756                for (mut record, events) in sessions {
757                    if record.start_commit.is_none() && !record.workspace.is_empty() {
758                        let binding = crate::core::repo::binding_for_session(
759                            Path::new(&record.workspace),
760                            record.started_at_ms,
761                            record.ended_at_ms,
762                        );
763                        record.start_commit = binding.start_commit;
764                        record.end_commit = binding.end_commit;
765                        record.branch = binding.branch;
766                        record.dirty_start = binding.dirty_start;
767                        record.dirty_end = binding.dirty_end;
768                        record.repo_binding_source = binding.source;
769                    }
770                    store.upsert_session(&record)?;
771                    let flush_ms = record.ended_at_ms.unwrap_or(record.started_at_ms);
772                    for ev in events {
773                        store.append_event_with_sync(&ev, sync_ctx)?;
774                    }
775                    if record.status == crate::core::event::SessionStatus::Done {
776                        store.flush_projector_session(&record.id, flush_ms)?;
777                    }
778                }
779            }
780            Err(e) => tracing::warn!("scan {:?}: {e}", entry.path()),
781        }
782    }
783    Ok(())
784}
785
786pub(crate) fn workspace_path(workspace: Option<&Path>) -> Result<PathBuf> {
787    crate::core::workspace::resolve(workspace)
788}
789
790/// Convert workspace path to cursor project slug.
791/// `/Users/lucas/Projects/kaizen` → `Users-lucas-Projects-kaizen`
792pub(crate) fn workspace_slug(ws: &str) -> String {
793    ws.trim_start_matches('/').replace('/', "-")
794}
795
796pub(crate) fn expand_home(path: &str) -> String {
797    if let (Some(rest), Ok(home)) = (path.strip_prefix("~/"), std::env::var("HOME")) {
798        return format!("{home}/{rest}");
799    }
800    path.to_string()
801}
802
803#[cfg(test)]
804mod cost_rollup_note_tests {
805    use super::*;
806
807    #[test]
808    fn needs_note_only_when_sessions_and_zero_cost() {
809        assert!(summary_needs_cost_rollup_note(1, 0));
810        assert!(!summary_needs_cost_rollup_note(0, 0));
811        assert!(!summary_needs_cost_rollup_note(1, 1));
812    }
813
814    #[test]
815    fn paragraph_names_gap_and_doc_anchor() {
816        let s = cost_rollup_zero_note_paragraph();
817        assert!(s.contains("cost_usd_e6"));
818        assert!(s.contains("usage"));
819        assert!(s.contains("docs/usage.md#cost-shows-zero"));
820    }
821}