lean_ctx/core/agents/
friction.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Friction signals of agent coordination, counted locally for the retro
3//! loop: `<data>/agents/friction.jsonl`, one `{"at", "signal", "agent"?,
4//! "root"?}` line per event (fixed keys and ids, never message text),
5//! summed by `lean-ctx agent friction [--since <dur>]` and per hour, worker
6//! and vault by the Hive connector's friction counter for the vault bus
7//! (V16.1).
8
9use std::collections::BTreeMap;
10use std::io::Write as _;
11use std::path::Path;
12
13use chrono::{DateTime, Utc};
14
15/// A task claim Hive refused because the agent was not on the bus.
16pub const CLAIM_REFUSED_PRESENCE: &str = "claim_refused_presence";
17/// A task claim Hive refused because the agent was on the bus in another role.
18pub const CLAIM_REFUSED_ROLE: &str = "claim_refused_role";
19/// A task claim Hive refused for a role or consent reason it did not name.
20pub const CLAIM_REFUSED_PERMIT: &str = "claim_refused_permit";
21/// A claim this machine held ended before its agent finished (stale or closed).
22pub const CLAIM_EXPIRED: &str = "claim_expired";
23/// An agent's registration had expired and a heartbeat renewed it.
24pub const REGISTRATION_RENEWED: &str = "registration_renewed";
25/// `ctx_agent wait` / `lean-ctx agent wait` ended without a message.
26pub const WAIT_TIMEOUT: &str = "wait_timeout";
27/// The bound checkout had unmerged paths while an agent worked in it.
28pub const MERGE_CONFLICT: &str = "merge_conflict";
29
30const FILE: &str = "friction.jsonl";
31/// Past this size the log keeps its newest half, so it stays small.
32const MAX_BYTES: u64 = 256 * 1024;
33
34#[derive(serde::Serialize, serde::Deserialize)]
35struct Line {
36    at: String,
37    signal: String,
38    /// The agent the signal belongs to, when the call site knows it.
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    agent: Option<String>,
41    /// The checkout it happened in, when known (else the agent's).
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    root: Option<String>,
44}
45
46/// One recorded signal.
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct Event {
49    pub at: DateTime<Utc>,
50    pub signal: String,
51    pub agent: Option<String>,
52    pub root: Option<String>,
53}
54
55/// Count one `signal` now for `agent` in checkout `root` (either unknown).
56/// Best effort: a failure is logged, never raised.
57pub fn record_for(signal: &str, agent: Option<&str>, root: Option<&str>) {
58    let Ok(dir) = super::persistence::agents_dir() else {
59        return;
60    };
61    if let Err(error) = record_in(&dir, signal, agent, root, Utc::now()) {
62        tracing::debug!("lean-ctx: friction log: {error}");
63    }
64}
65
66fn record_in(
67    dir: &Path,
68    signal: &str,
69    agent: Option<&str>,
70    root: Option<&str>,
71    at: DateTime<Utc>,
72) -> std::io::Result<()> {
73    std::fs::create_dir_all(dir)?;
74    let path = dir.join(FILE);
75    if std::fs::metadata(&path).is_ok_and(|m| m.len() > MAX_BYTES) {
76        let text = std::fs::read_to_string(&path).unwrap_or_default();
77        let lines: Vec<&str> = text.lines().collect();
78        let kept = lines[lines.len() / 2..].join("\n");
79        crate::core::atomic_fs::try_atomic_write(&path, format!("{kept}\n").as_bytes(), None)
80            .map_err(std::io::Error::other)?;
81    }
82    let line = serde_json::to_string(&Line {
83        at: at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
84        signal: signal.to_string(),
85        agent: agent.map(str::to_string),
86        root: root.map(str::to_string),
87    })
88    .map_err(std::io::Error::other)?;
89    let mut options = std::fs::OpenOptions::new();
90    options.create(true).append(true);
91    #[cfg(unix)]
92    std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
93    let mut file = options.open(&path)?;
94    writeln!(file, "{line}")
95}
96
97/// The signals recorded in `[from, to)`, oldest first.
98pub fn events(from: DateTime<Utc>, to: DateTime<Utc>) -> Vec<Event> {
99    super::persistence::agents_dir()
100        .map(|dir| events_in(&dir, from, to))
101        .unwrap_or_default()
102}
103
104pub(crate) fn events_in(dir: &Path, from: DateTime<Utc>, to: DateTime<Utc>) -> Vec<Event> {
105    let Ok(text) = std::fs::read_to_string(dir.join(FILE)) else {
106        return Vec::new();
107    };
108    text.lines()
109        .filter_map(|line| serde_json::from_str::<Line>(line).ok())
110        .filter_map(|line| {
111            let at = DateTime::parse_from_rfc3339(&line.at)
112                .ok()?
113                .with_timezone(&Utc);
114            (from <= at && at < to).then_some(Event {
115                at,
116                signal: line.signal,
117                agent: line.agent,
118                root: line.root,
119            })
120        })
121        .collect()
122}
123
124/// Signal → count of the events at or after `since` (all when `None`).
125pub fn summary(since: Option<DateTime<Utc>>) -> BTreeMap<String, u64> {
126    super::persistence::agents_dir()
127        .map(|dir| summary_in(&dir, since))
128        .unwrap_or_default()
129}
130
131fn summary_in(dir: &Path, since: Option<DateTime<Utc>>) -> BTreeMap<String, u64> {
132    let mut counts = BTreeMap::new();
133    let Ok(text) = std::fs::read_to_string(dir.join(FILE)) else {
134        return counts;
135    };
136    for line in text.lines() {
137        let Ok(line) = serde_json::from_str::<Line>(line) else {
138            continue;
139        };
140        let Ok(at) = DateTime::parse_from_rfc3339(&line.at) else {
141            continue;
142        };
143        if since.is_none_or(|since| at >= since) {
144            *counts.entry(line.signal).or_insert(0) += 1;
145        }
146    }
147    counts
148}
149
150/// The summary as text: one `signal count` line each, sorted by signal.
151pub fn render(counts: &BTreeMap<String, u64>) -> String {
152    if counts.is_empty() {
153        return "No friction recorded.".to_string();
154    }
155    let width = counts.keys().map(String::len).max().unwrap_or(0);
156    counts
157        .iter()
158        .map(|(signal, n)| format!("{signal:<width$}  {n}"))
159        .collect::<Vec<_>>()
160        .join("\n")
161}
162
163/// `30m`, `12h`, `7d` or plain seconds.
164pub fn parse_duration(text: &str) -> Option<chrono::Duration> {
165    let text = text.trim();
166    let (number, unit) = text
167        .find(|c: char| !c.is_ascii_digit())
168        .map_or((text, ""), |at| text.split_at(at));
169    let n: i64 = number.parse().ok()?;
170    match unit {
171        "" | "s" => Some(chrono::Duration::seconds(n)),
172        "m" => Some(chrono::Duration::minutes(n)),
173        "h" => Some(chrono::Duration::hours(n)),
174        "d" => Some(chrono::Duration::days(n)),
175        _ => None,
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use super::*;
182
183    #[test]
184    fn signals_are_counted_by_key_and_window() {
185        let dir = tempfile::tempdir().unwrap();
186        let at = |s: &str| DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc);
187        record_in(
188            dir.path(),
189            WAIT_TIMEOUT,
190            None,
191            None,
192            at("2026-10-24T08:00:00Z"),
193        )
194        .unwrap();
195        record_in(
196            dir.path(),
197            WAIT_TIMEOUT,
198            None,
199            None,
200            at("2026-10-24T10:00:00Z"),
201        )
202        .unwrap();
203        record_in(
204            dir.path(),
205            CLAIM_REFUSED_PRESENCE,
206            None,
207            None,
208            at("2026-10-24T10:30:00Z"),
209        )
210        .unwrap();
211        std::fs::OpenOptions::new()
212            .append(true)
213            .open(dir.path().join(FILE))
214            .unwrap()
215            .write_all(b"not json\n")
216            .unwrap();
217        let all = summary_in(dir.path(), None);
218        assert_eq!(all[WAIT_TIMEOUT], 2);
219        assert_eq!(all[CLAIM_REFUSED_PRESENCE], 1);
220        let recent = summary_in(dir.path(), Some(at("2026-10-24T09:00:00Z")));
221        assert_eq!(recent[WAIT_TIMEOUT], 1);
222        assert_eq!(
223            render(&recent),
224            "claim_refused_presence  1\nwait_timeout            1"
225        );
226        assert_eq!(render(&BTreeMap::new()), "No friction recorded.");
227        let text = std::fs::read_to_string(dir.path().join(FILE)).unwrap();
228        assert!(
229            text.lines()
230                .next()
231                .unwrap()
232                .contains(r#""signal":"wait_timeout""#)
233        );
234        #[cfg(unix)]
235        {
236            use std::os::unix::fs::PermissionsExt as _;
237            let mode = std::fs::metadata(dir.path().join(FILE))
238                .unwrap()
239                .permissions()
240                .mode();
241            assert_eq!(mode & 0o777, 0o600);
242        }
243    }
244
245    #[test]
246    fn the_log_stays_bounded() {
247        let dir = tempfile::tempdir().unwrap();
248        let now = Utc::now();
249        for _ in 0..6000 {
250            record_in(dir.path(), CLAIM_EXPIRED, None, None, now).unwrap();
251        }
252        assert!(std::fs::metadata(dir.path().join(FILE)).unwrap().len() <= MAX_BYTES + 128);
253        assert!(summary_in(dir.path(), None)[CLAIM_EXPIRED] > 1000);
254    }
255
256    /// V16.1: a signal carries its agent and checkout when the call site
257    /// knows them; a window query returns `[from, to)` only.
258    #[test]
259    fn events_carry_agent_and_checkout_and_respect_the_window() {
260        let dir = tempfile::tempdir().unwrap();
261        let at = |s: &str| DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc);
262        record_in(
263            dir.path(),
264            WAIT_TIMEOUT,
265            Some("mcp-1"),
266            Some("/w/a"),
267            at("2026-10-24T08:10:00Z"),
268        )
269        .unwrap();
270        record_in(
271            dir.path(),
272            CLAIM_EXPIRED,
273            None,
274            None,
275            at("2026-10-24T09:00:00Z"),
276        )
277        .unwrap();
278        let hour = events_in(
279            dir.path(),
280            at("2026-10-24T08:00:00Z"),
281            at("2026-10-24T09:00:00Z"),
282        );
283        assert_eq!(
284            hour,
285            [Event {
286                at: at("2026-10-24T08:10:00Z"),
287                signal: WAIT_TIMEOUT.into(),
288                agent: Some("mcp-1".into()),
289                root: Some("/w/a".into()),
290            }]
291        );
292        let text = std::fs::read_to_string(dir.path().join(FILE)).unwrap();
293        assert!(
294            text.lines()
295                .nth(1)
296                .unwrap()
297                .ends_with(r#""signal":"claim_expired"}"#),
298            "unknown agent and checkout are left out: {text}"
299        );
300    }
301
302    #[test]
303    fn durations_parse() {
304        assert_eq!(parse_duration("90"), Some(chrono::Duration::seconds(90)));
305        assert_eq!(parse_duration("30m"), Some(chrono::Duration::minutes(30)));
306        assert_eq!(parse_duration("12h"), Some(chrono::Duration::hours(12)));
307        assert_eq!(parse_duration("7d"), Some(chrono::Duration::days(7)));
308        assert_eq!(parse_duration("7w"), None);
309        assert_eq!(parse_duration("h"), None);
310    }
311}
lean_ctx/core/agents/friction.rs

lean_ctx/core/agents/
friction.rs
