clustervision 0.18.0

clustervision CLI — search, read, and port AI coding-agent sessions across harnesses.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

//! `cv scry` — live activity across harnesses (`tail -f` for sessions).

use crate::util::{home_rel, parse_harness, poll_interval, short_id, Wait};
use anyhow::Result;
use cv_core::ir::{truncate, SessionRef};
use cv_core::watch::{EventKind, Filter, SessionEvent, Watcher};

/// `cv scry`'s options, named once at the command boundary.
pub(crate) struct ScryArgs {
    pub harness: Option<String>,
    pub cwd: Option<String>,
    pub interval: f64,
    pub existing: bool,
    pub follow: bool,
    pub for_: Option<String>,
}

pub(crate) fn cmd_scry(args: ScryArgs) -> Result<()> {
    let ScryArgs {
        harness,
        cwd,
        interval,
        existing,
        follow,
        for_,
    } = args;
    let interval = poll_interval(interval)?;
    let filter = Filter {
        harness: parse_harness(&harness)?,
        cwd_contains: cwd,
    };
    let wait = Wait::resolve(follow, for_.as_deref())?;
    wait.announce("scry", "live agent activity");
    if matches!(wait, Wait::Once) && !existing {
        // The current state: the sessions a follower would baseline, in the order it polls them.
        for r in Watcher::new(filter, true).discover() {
            println!(
                "{}",
                header("   ·", &r, r.message_count.map_or("–".into(), |n| n.to_string()))
            );
        }
        return Ok(());
    }
    // The baseline read of a large fleet can outlive a short `--for`: it stops at the deadline.
    let mut watcher = Watcher::with_discovery_until(filter, existing, cv_core::discover_all, || wait.expired());
    if !existing && wait.expired() {
        eprintln!(
            "cv scry: --for elapsed during the baseline read, {} sessions in",
            watcher.following()
        );
        return Ok(());
    }
    if !matches!(wait, Wait::Once) {
        eprintln!("cv scry: following {} sessions", watcher.following());
    }
    loop {
        let mut events = Vec::new();
        watcher.poll(cv_core::watch::Poll::default(), |ev| {
            let taken = ev.new_messages.len();
            events.push(ev);
            taken
        });
        for ev in events {
            print_event(&ev);
        }
        if !wait.pause(interval) {
            return Ok(());
        }
    }
}

fn header(tag: &str, r: &SessionRef, messages: String) -> String {
    let where_ = r.cwd.as_deref().map(home_rel).unwrap_or_else(|| "?".into());
    format!(
        "{tag}  {:8} {:8}  {}  ({messages} msg)",
        r.harness.as_str(),
        short_id(&r.id),
        where_
    )
}

fn print_event(ev: &SessionEvent) {
    let tag = match ev.kind {
        EventKind::New => "✷ new",
        EventKind::Updated => "   +",
    };
    println!("{}", header(tag, &ev.reference, ev.new_messages.len().to_string()));
    for m in &ev.new_messages {
        if let Some(t) = m.text() {
            println!("      {} {}", cv_core::render::role_label(m.role), truncate(&t, 100));
        }
    }
}