use crate::paths::AgentsHome;
use serde::Serialize;
use serde_json::Value;
use std::collections::HashSet;
use std::path::{Path, PathBuf};
#[derive(Debug, Serialize, PartialEq)]
pub struct DigestSummary {
pub session: String,
pub since: String,
pub commits: usize,
pub pr_number: Option<u64>,
pub pr_url: Option<String>,
pub pr_state: Option<String>,
pub ci: Option<String>,
pub reviewed: bool,
pub blocked_episodes: usize,
pub last_block_reason: Option<String>,
pub resolved: bool,
pub cost_usd: f64,
pub state: String,
pub skipped_lines: usize,
pub lines: Vec<String>,
}
fn field<'a>(v: &'a Value, key: &str) -> Option<&'a Value> {
v.get("data")
.and_then(|d| d.get(key))
.or_else(|| v.get(key))
}
fn event_kind(v: &Value) -> Option<&str> {
v.get("type")
.and_then(|t| t.as_str())
.or_else(|| v.get("kind").and_then(|k| k.as_str()))
}
fn event_ts(v: &Value) -> &str {
v.get("ts").and_then(|t| t.as_str()).unwrap_or("")
}
#[derive(Clone, Copy)]
enum Since<'a> {
Str(&'a str),
Epoch(u64),
}
impl Since<'_> {
fn includes(&self, ts: &str) -> bool {
match self {
Since::Str(s) => ts >= *s,
Since::Epoch(e) => to_epoch_lenient(ts).is_none_or(|secs| secs >= *e),
}
}
fn label(&self) -> String {
match self {
Since::Str(s) => s.to_string(),
Since::Epoch(e) => format!("epoch:{e}"),
}
}
}
fn to_epoch_lenient(ts: &str) -> Option<u64> {
crate::state::rfc3339_like_to_secs(ts).or_else(|| {
let secs = ts.get(..19)?;
crate::state::rfc3339_like_to_secs(&format!("{secs}Z"))
})
}
fn event_session_in(v: &Value, set: &HashSet<&str>) -> bool {
field(v, "session_id")
.and_then(|s| s.as_str())
.is_some_and(|s| set.contains(s))
}
pub fn fold(events_raw: &str, ledger_raw: &str, selector: &str, since: &str) -> DigestSummary {
fold_since(events_raw, ledger_raw, selector, Since::Str(since))
}
fn fold_since(
events_raw: &str,
ledger_raw: &str,
selector: &str,
since: Since<'_>,
) -> DigestSummary {
let ledger = resolve_from_ledger(ledger_raw, selector, since);
let mut session_ids: HashSet<&str> = ledger.session_ids.iter().map(String::as_str).collect();
session_ids.insert(selector);
let mut skipped_lines = 0usize;
let mut commit_shas: Vec<String> = Vec::new(); let mut pr_state: Option<String> = None;
let mut ci: Option<String> = None;
let mut reviewed = false;
let mut blocked_episodes = 0usize;
let mut last_block_reason: Option<String> = None;
let mut resolved = false;
let mut terminated = false;
let mut last_decision_blocked = false;
let mut seen_lines: HashSet<&str> = HashSet::new();
for line in events_raw.lines() {
if line.trim().is_empty() {
continue;
}
let Ok(v) = serde_json::from_str::<Value>(line) else {
skipped_lines += 1;
continue;
};
if !event_session_in(&v, &session_ids) {
continue;
}
if !since.includes(event_ts(&v)) {
continue;
}
if !seen_lines.insert(line) {
continue; }
match event_kind(&v) {
Some("loop_check") => {
if let Some(fp) = field(&v, "fingerprint").and_then(|f| f.as_str()) {
let head = fp.split('|').next().unwrap_or("");
if !head.is_empty() && head != "none" {
if commit_shas.last().map(String::as_str) != Some(head) {
commit_shas.push(head.to_string());
}
}
}
if let Some(s) = field(&v, "pr_state").and_then(|s| s.as_str()) {
if s != "none" {
pr_state = Some(s.to_string());
}
}
if let Some(c) = field(&v, "ci").and_then(|c| c.as_str()) {
if c != "none" {
ci = Some(c.to_string());
}
}
if let Some(r) = field(&v, "reviewed").and_then(|r| r.as_bool()) {
reviewed = r;
}
let decision = field(&v, "decision").and_then(|d| d.as_str());
if decision == Some("block") {
if !last_decision_blocked {
blocked_episodes += 1;
resolved = false;
}
last_decision_blocked = true;
let ci_blocking = field(&v, "ci")
.and_then(|c| c.as_str())
.filter(|c| c.starts_with("FAILURE") || *c == "PENDING");
last_block_reason = ci_blocking
.or_else(|| field(&v, "intent").and_then(|i| i.as_str()))
.filter(|r| *r != "none")
.map(str::to_string);
} else if decision == Some("allow") {
if last_decision_blocked {
resolved = true;
}
last_decision_blocked = false;
}
}
Some("termination") | Some("session_finalized") | Some("node_closed") => {
terminated = true;
}
_ => {}
}
}
let commits = commit_shas.len().saturating_sub(1);
let state = if terminated {
"done"
} else if last_decision_blocked {
"blocked"
} else {
"working"
}
.to_string();
let mut summary = DigestSummary {
session: selector.to_string(),
since: since.label(),
commits,
pr_number: ledger.pr_number,
pr_url: ledger.pr_url,
pr_state,
ci,
reviewed,
blocked_episodes,
last_block_reason,
resolved,
cost_usd: ledger.cost_usd,
state,
skipped_lines,
lines: Vec::new(),
};
summary.lines = render_lines(&summary);
summary
}
#[derive(Default)]
struct LedgerFold {
session_ids: HashSet<String>,
pr_number: Option<u64>,
pr_url: Option<String>,
cost_usd: f64,
}
fn basename(path: &str) -> &str {
path.rsplit('/').next().unwrap_or(path)
}
fn ledger_entry_matches(entry: &Value, selector: &str) -> bool {
if entry.get("session_id").and_then(|s| s.as_str()) == Some(selector) {
return true;
}
if entry
.get("sessions")
.and_then(|s| s.as_array())
.is_some_and(|arr| arr.iter().any(|s| s.as_str() == Some(selector)))
{
return true;
}
for key in ["graph_node_id", "title"] {
if entry.get(key).and_then(|v| v.as_str()) == Some(selector) {
return true;
}
}
for key in ["worktree", "root_path"] {
if entry
.get(key)
.and_then(|v| v.as_str())
.is_some_and(|p| basename(p) == selector)
{
return true;
}
}
false
}
fn resolve_from_ledger(ledger_raw: &str, selector: &str, since: Since<'_>) -> LedgerFold {
let mut out = LedgerFold::default();
let Ok(root) = serde_json::from_str::<Value>(ledger_raw) else {
return out;
};
let Some(entries) = root
.get("entries")
.and_then(|e| e.as_array())
.or_else(|| root.as_array())
else {
return out;
};
for entry in entries {
if !ledger_entry_matches(entry, selector) {
continue;
}
if let Some(sid) = entry.get("session_id").and_then(|s| s.as_str()) {
out.session_ids.insert(sid.to_string());
}
if let Some(arr) = entry.get("sessions").and_then(|s| s.as_array()) {
for s in arr.iter().filter_map(|s| s.as_str()) {
out.session_ids.insert(s.to_string());
}
}
let ts = ["timestamp", "completed", "ts"]
.iter()
.find_map(|k| entry.get(*k).and_then(|v| v.as_str()))
.unwrap_or("");
if !since.includes(ts) {
continue;
}
if let Some(c) = entry.get("cost_usd").and_then(|c| c.as_f64()) {
out.cost_usd += c;
}
if let Some(n) = entry.get("pr_number").and_then(|n| n.as_u64()) {
out.pr_number = Some(n);
}
if let Some(u) = entry.get("pr_url").and_then(|u| u.as_str()) {
out.pr_url = Some(u.to_string());
}
}
out
}
fn is_empty_digest(s: &DigestSummary) -> bool {
s.commits == 0
&& s.pr_number.is_none()
&& s.blocked_episodes == 0
&& s.cost_usd == 0.0
&& s.state != "done"
&& s.state != "blocked"
}
fn render_lines(s: &DigestSummary) -> Vec<String> {
if is_empty_digest(s) {
return Vec::new();
}
let mut lines = Vec::new();
if s.blocked_episodes > 0 {
let reason = s.last_block_reason.as_deref().unwrap_or("gate");
let resolution = if s.resolved { "resolved" } else { "UNRESOLVED" };
let plural = if s.blocked_episodes == 1 { "" } else { "s" };
lines.push(format!(
"! {} block{plural} (last: {reason}) - {resolution}",
s.blocked_episodes
));
}
match (s.pr_number, s.pr_state.as_deref()) {
(Some(n), state) => {
let state = state.unwrap_or("open");
let ci = s.ci.as_deref().unwrap_or("?");
let review = if s.reviewed { "reviewed" } else { "unreviewed" };
lines.push(format!("PR #{n} {state} - CI {ci} - {review}"));
}
(None, _) => lines.push("no PR yet".to_string()),
}
let commit_word = if s.commits == 1 { "commit" } else { "commits" };
lines.push(format!(
"{} {commit_word} - ${:.2} - state: {}",
s.commits, s.cost_usd, s.state
));
lines
}
fn default_sources(home: &AgentsHome) -> (Vec<PathBuf>, PathBuf) {
let fno_dir = home
.root()
.parent()
.map(Path::to_path_buf)
.unwrap_or_else(|| PathBuf::from(".fno"));
let global_events = fno_dir.join("events.jsonl");
let project_events = PathBuf::from(".fno").join("events.jsonl");
let ledger = fno_dir.join("ledger.json");
(vec![project_events, global_events], ledger)
}
struct DigestArgs {
session: String,
since: String,
since_epoch: Option<u64>,
json: bool,
events_override: Vec<PathBuf>,
ledger_override: Option<PathBuf>,
}
fn parse_args(rest: &[String]) -> Result<DigestArgs, String> {
let mut session: Option<String> = None;
let mut since = String::new();
let mut since_epoch: Option<u64> = None;
let mut json = false;
let mut events_override: Vec<PathBuf> = Vec::new();
let mut ledger_override: Option<PathBuf> = None;
let mut it = expand_eq(rest).into_iter();
while let Some(a) = it.next() {
match a.as_str() {
"--session" => session = it.next(),
"--since" => since = it.next().ok_or("--since needs a value")?,
"--since-epoch" => {
since_epoch = Some(
it.next()
.and_then(|v| v.parse::<u64>().ok())
.ok_or("--since-epoch needs a non-negative integer")?,
)
}
"--json" | "-J" => json = true,
"--events" => {
events_override.push(PathBuf::from(it.next().ok_or("--events needs a path")?))
}
"--ledger" => {
ledger_override = Some(PathBuf::from(it.next().ok_or("--ledger needs a path")?))
}
other => return Err(format!("unknown digest flag: {other}")),
}
}
let session = match session {
Some(s) if !s.is_empty() => s,
_ => return Err("digest needs --session".into()),
};
Ok(DigestArgs {
session,
since,
since_epoch,
json,
events_override,
ledger_override,
})
}
fn expand_eq(rest: &[String]) -> Vec<String> {
let mut out = Vec::with_capacity(rest.len());
for a in rest {
if let Some(eq) = a.find('=') {
if a.starts_with("--") && eq > 2 {
out.push(a[..eq].to_string());
out.push(a[eq + 1..].to_string());
continue;
}
}
out.push(a.clone());
}
out
}
pub async fn run_digest(rest: &[String], home: &AgentsHome) -> i32 {
let args = match parse_args(rest) {
Ok(a) => a,
Err(msg) => {
eprintln!("fno-agents: {msg}");
return 2;
}
};
let (default_events, default_ledger) = default_sources(home);
let event_paths = if args.events_override.is_empty() {
default_events
} else {
args.events_override
};
let ledger_path = args.ledger_override.unwrap_or(default_ledger);
let mut events_raw = String::new();
for p in &event_paths {
if let Ok(content) = std::fs::read_to_string(p) {
events_raw.push_str(&content);
if !content.ends_with('\n') {
events_raw.push('\n');
}
}
}
let ledger_raw = std::fs::read_to_string(&ledger_path).unwrap_or_default();
let since = match args.since_epoch {
Some(e) => Since::Epoch(e),
None => Since::Str(&args.since),
};
let summary = fold_since(&events_raw, &ledger_raw, &args.session, since);
if args.json {
println!(
"{}",
serde_json::to_string(&summary).expect("serializing an owned value never fails")
);
} else {
for line in &summary.lines {
println!("{line}");
}
}
0
}
#[cfg(test)]
mod tests {
use super::*;
fn loop_check(
ts: &str,
session: &str,
head: &str,
pr: &str,
ci: &str,
reviewed: bool,
decision: &str,
) -> String {
format!(
r#"{{"ts":"{ts}","type":"loop_check","source":"hook","data":{{"session_id":"{session}","fingerprint":"{head}|{pr}|{ci}|none","pr_state":"{pr}","ci":"{ci}","reviewed":{reviewed},"decision":"{decision}","intent":"promise"}}}}"#
)
}
#[test]
fn happy_names_pr_blocked_and_cost() {
let events = [
loop_check(
"2026-07-03T01:00:00Z",
"sess-A",
"base",
"none",
"PENDING",
false,
"block",
),
loop_check(
"2026-07-03T02:00:00Z",
"sess-A",
"c1",
"OPEN",
"PENDING",
false,
"block",
),
loop_check(
"2026-07-03T03:00:00Z",
"sess-A",
"c2",
"OPEN",
"SUCCESS",
true,
"allow",
),
]
.join("\n");
let ledger = r#"{"entries":[{"session_id":"sess-A","cost_usd":1.5,"pr_number":42,"pr_url":"https://x/pull/42","completed":"2026-07-03T03:00:00Z"}]}"#;
let d = fold(&events, ledger, "sess-A", "2026-07-03T00:00:00Z");
assert_eq!(d.pr_number, Some(42), "PR named");
assert_eq!(d.commits, 2, "base->c1->c2 = 2 commits");
assert_eq!(d.blocked_episodes, 1);
assert!(d.resolved, "a later allow resolved the block");
assert_eq!(d.cost_usd, 1.5);
assert!(d.reviewed);
assert_eq!(d.skipped_lines, 0);
assert!(
d.lines.iter().any(|l| l.contains("#42")),
"lines name the PR: {:?}",
d.lines
);
assert!(
d.lines.iter().any(|l| l.contains("block")),
"lines name the block"
);
}
#[test]
fn corrupt_line_is_skipped_and_counted() {
let events = [
loop_check(
"2026-07-03T02:00:00Z",
"sess-A",
"c1",
"OPEN",
"SUCCESS",
true,
"allow",
),
"{ this is not valid json".to_string(),
"".to_string(),
]
.join("\n");
let d = fold(&events, "", "sess-A", "");
assert_eq!(d.skipped_lines, 1, "one bad line, blank line not counted");
assert_eq!(d.pr_state.as_deref(), Some("OPEN"));
}
#[test]
fn since_in_future_is_empty() {
let events = loop_check(
"2026-07-03T02:00:00Z",
"sess-A",
"c1",
"OPEN",
"SUCCESS",
true,
"allow",
);
let d = fold(&events, "", "sess-A", "2099-01-01T00:00:00Z");
assert_eq!(d.commits, 0);
assert_eq!(d.blocked_episodes, 0);
assert!(d.pr_state.is_none());
assert_eq!(d.cost_usd, 0.0);
assert!(
d.lines.is_empty(),
"empty digest renders nothing: {:?}",
d.lines
);
}
#[test]
fn other_session_ignored() {
let events = [
loop_check(
"2026-07-03T02:00:00Z",
"sess-A",
"c1",
"OPEN",
"SUCCESS",
true,
"allow",
),
loop_check(
"2026-07-03T02:00:00Z",
"sess-B",
"z9",
"MERGED",
"SUCCESS",
true,
"allow",
),
]
.join("\n");
let d = fold(&events, "", "sess-A", "");
assert_eq!(
d.pr_state.as_deref(),
Some("OPEN"),
"sess-B's MERGED must not leak in"
);
}
#[test]
fn cost_matches_sessions_array_membership() {
let ledger = r#"[{"sessions":["uuid-1","sess-A"],"cost_usd":23.45,"pr_number":7,"pr_url":"https://x/pull/7","completed":"2026-07-03T03:00:00Z"}]"#;
let l = resolve_from_ledger(ledger, "sess-A", Since::Str(""));
assert_eq!(l.pr_number, Some(7));
assert_eq!(l.pr_url.as_deref(), Some("https://x/pull/7"));
assert_eq!(l.cost_usd, 23.45);
}
#[test]
fn ledger_cost_respects_since() {
let ledger = r#"[{"session_id":"sess-A","cost_usd":9.0,"timestamp":"2026-07-01T00:00:00Z"},{"session_id":"sess-A","cost_usd":1.0,"timestamp":"2026-07-03T00:00:00Z"}]"#;
let l = resolve_from_ledger(ledger, "sess-A", Since::Str("2026-07-02T00:00:00Z"));
assert_eq!(
l.cost_usd, 1.0,
"the pre-since entry is excluded from the delta"
);
}
#[test]
fn node_id_selector_resolves_via_ledger_to_events() {
let ledger = r#"[{"graph_node_id":"x-4e2d","worktree":"/w/footnote/x-4e2d","session_id":"20260703T-abc","cost_usd":2.0,"pr_number":99,"pr_url":"https://x/pull/99","completed":"2026-07-03T03:00:00Z"}]"#;
let events = loop_check(
"2026-07-03T02:00:00Z",
"20260703T-abc",
"c1",
"OPEN",
"SUCCESS",
true,
"block",
);
let d = fold(&events, ledger, "x-4e2d", "");
assert_eq!(
d.pr_number,
Some(99),
"PR resolved from the ledger by node id"
);
assert_eq!(d.cost_usd, 2.0);
assert_eq!(
d.pr_state.as_deref(),
Some("OPEN"),
"events found via the resolved session id"
);
assert_eq!(d.blocked_episodes, 1);
}
#[test]
fn mirrored_events_are_folded_once() {
let one = loop_check(
"2026-07-03T01:00:00Z",
"s",
"a",
"OPEN",
"FAILURE:x",
false,
"block",
);
let events = [one.clone(), one].join("\n");
let d = fold(&events, "", "s", "");
assert_eq!(d.blocked_episodes, 1, "the duplicate line is folded once");
}
#[test]
fn repeated_block_rows_are_one_episode() {
let events = [
loop_check(
"2026-07-03T01:00:00Z",
"s",
"a",
"OPEN",
"FAILURE:x",
false,
"block",
),
loop_check(
"2026-07-03T01:05:00Z",
"s",
"a",
"OPEN",
"FAILURE:x",
false,
"block",
),
loop_check(
"2026-07-03T01:10:00Z",
"s",
"a",
"OPEN",
"FAILURE:x",
false,
"block",
),
]
.join("\n");
let d = fold(&events, "", "s", "");
assert_eq!(
d.blocked_episodes, 1,
"three block rows, one unresolved spell"
);
}
#[test]
fn ledger_epoch_filter_parses_non_z_microsecond_ts() {
let ledger =
r#"[{"session_id":"s","cost_usd":5.0,"completed":"2026-07-01T00:00:00.123456"}]"#;
let after = crate::state::rfc3339_like_to_secs("2026-07-02T00:00:00Z").unwrap();
let l = resolve_from_ledger(ledger, "s", Since::Epoch(after));
assert_eq!(
l.cost_usd, 0.0,
"the pre-window microsecond-ts row is excluded"
);
}
#[test]
fn reblock_after_resolve_reads_unresolved() {
let events = [
loop_check(
"2026-07-03T01:00:00Z",
"s",
"a",
"OPEN",
"FAILURE:x",
false,
"block",
),
loop_check(
"2026-07-03T02:00:00Z",
"s",
"b",
"OPEN",
"SUCCESS",
true,
"allow",
),
loop_check(
"2026-07-03T03:00:00Z",
"s",
"c",
"OPEN",
"FAILURE:y",
false,
"block",
),
]
.join("\n");
let d = fold(&events, "", "s", "");
assert_eq!(d.blocked_episodes, 2);
assert!(!d.resolved, "the current (last) block is unresolved");
assert_eq!(d.state, "blocked");
assert!(d.lines.iter().any(|l| l.contains("UNRESOLVED")));
}
#[test]
fn since_epoch_scopes_to_the_absence_window() {
let events = [
loop_check(
"2026-07-03T01:00:00Z",
"sess-A",
"old",
"OPEN",
"PENDING",
false,
"block",
),
loop_check(
"2026-07-03T05:00:00Z",
"sess-A",
"new",
"OPEN",
"SUCCESS",
true,
"allow",
),
]
.join("\n");
let detach = crate::state::rfc3339_like_to_secs("2026-07-03T03:00:00Z").unwrap();
let d = fold_since(&events, "", "sess-A", Since::Epoch(detach));
assert_eq!(
d.blocked_episodes, 0,
"the pre-detach block is out of window"
);
assert!(d.reviewed, "the in-window allow still counts");
assert!(d.since.starts_with("epoch:"));
}
#[test]
fn worktree_basename_selector_matches() {
let entry: Value =
serde_json::from_str(r#"{"worktree":"/Users/x/conductor/workspaces/footnote/x-4e2d"}"#)
.unwrap();
assert!(ledger_entry_matches(&entry, "x-4e2d"));
assert!(!ledger_entry_matches(&entry, "footnote"));
}
}