agentmux_srv/backend/
tool_calls.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Which agent tool call each running Bash command is: `agentmux-bashwrap`,
5//! running a call as Claude Code's shell prefix, reports its own pid with the
6//! call's id and description (`prefix.rs`, `{op:"call"}` on the wrapper's
7//! publish route). Tower labels that process, the root of the call's process
8//! tree, with the description ("started by").
9//! docs/specs/SPEC_TOWER_AGENT_CENTRIC_VIEWS_2026_10_08.md §5.3.
10
11use std::collections::HashMap;
12use std::sync::{Mutex, OnceLock};
13
14/// A call's report is kept this long: a background `task dev` runs for hours.
15const KEEP_MS: u64 = 12 * 60 * 60 * 1000;
16/// At most this many reports, the newest kept.
17const MAX_CALLS: usize = 4096;
18/// A process is the reporting wrapper only if it started at most this long
19/// before the report (the report is sent as the wrapper starts its command):
20/// a later process reusing the pid never inherits the label.
21const START_SLACK_MS: u64 = 60_000;
22/// And at most this long after it: a start time the OS gives in whole
23/// seconds (Linux's boot time) can land just after the report.
24const CLOCK_SLACK_MS: u64 = 2_000;
25
26/// One tool call a wrapper reported.
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct Call {
29    pub block_id: String,
30    pub tool_id: String,
31    pub description: String,
32    /// When it was reported, unix ms.
33    pub at_ms: u64,
34}
35
36fn calls() -> &'static Mutex<HashMap<u32, Call>> {
37    static CALLS: OnceLock<Mutex<HashMap<u32, Call>>> = OnceLock::new();
38    CALLS.get_or_init(|| Mutex::new(HashMap::new()))
39}
40
41/// Note a wrapper's report: `pid` runs `call`.
42pub fn record(pid: u32, call: Call) {
43    let mut map = calls().lock().unwrap_or_else(|e| e.into_inner());
44    let cutoff = call.at_ms.saturating_sub(KEEP_MS);
45    map.retain(|_, c| c.at_ms >= cutoff);
46    if map.len() >= MAX_CALLS {
47        if let Some(oldest) = map.iter().min_by_key(|(_, c)| c.at_ms).map(|(p, _)| *p) {
48            map.remove(&oldest);
49        }
50    }
51    map.insert(pid, call);
52}
53
54/// The call the process with `pid`, started at `started_at_ms`, runs, if it
55/// is the wrapper that reported one. A process whose start is unknown is
56/// trusted on its pid alone.
57pub fn lookup(pid: u32, started_at_ms: Option<u64>) -> Option<Call> {
58    let map = calls().lock().unwrap_or_else(|e| e.into_inner());
59    let call = map.get(&pid)?;
60    match started_at_ms {
61        Some(started) if started > call.at_ms + CLOCK_SLACK_MS || call.at_ms.saturating_sub(started) > START_SLACK_MS => None,
62        _ => Some(call.clone()),
63    }
64}
65
66/// A `{op:"call"}` report from the wrapper's publish, if `data` is one:
67/// `(pid, call)`, with the block from the publish's `block:<id>` scope.
68pub fn parse_report(scopes: &[String], data: &serde_json::Value) -> Option<(u32, Call)> {
69    if data.get("op").and_then(|v| v.as_str()) != Some("call") {
70        return None;
71    }
72    let pid = u32::try_from(data.get("pid")?.as_u64()?).ok()?;
73    let tool_id = data.get("tool_id")?.as_str()?.to_string();
74    let description = data.get("description")?.as_str()?.trim().to_string();
75    if tool_id.is_empty() || description.is_empty() {
76        return None;
77    }
78    let block_id = scopes.iter().find_map(|s| s.strip_prefix("block:")).unwrap_or("").to_string();
79    let at_ms = data
80        .get("timestamp")
81        .and_then(|v| v.as_u64())
82        .unwrap_or_else(|| agentmux_common::time::now_ms() as u64);
83    Some((pid, Call { block_id, tool_id, description, at_ms }))
84}
85
86#[cfg(test)]
87mod tests {
88    use super::*;
89    use serde_json::json;
90
91    #[test]
92    fn parses_a_wrappers_report_and_nothing_else() {
93        let scopes = vec!["block:b1".to_string()];
94        let (pid, call) = parse_report(
95            &scopes,
96            &json!({"op": "call", "tool_id": "toolu_1", "pid": 4242, "description": " Run the srv tests ", "timestamp": 1000}),
97        )
98        .unwrap();
99        assert_eq!(pid, 4242);
100        assert_eq!(
101            call,
102            Call { block_id: "b1".into(), tool_id: "toolu_1".into(), description: "Run the srv tests".into(), at_ms: 1000 }
103        );
104        for other in [
105            json!({"op": "chunk", "tool_id": "t", "content": "x"}),
106            json!({"op": "pid", "tool_id": "t", "pid": 1}),
107            json!({"op": "call", "tool_id": "", "pid": 1, "description": "d"}),
108            json!({"op": "call", "tool_id": "t", "pid": 1, "description": "  "}),
109            json!({"op": "call", "tool_id": "t", "description": "d"}),
110        ] {
111            assert!(parse_report(&scopes, &other).is_none(), "{other}");
112        }
113    }
114
115    #[test]
116    fn a_process_is_the_reporting_wrapper_only_if_it_started_before_the_report() {
117        let call = |at_ms| Call { block_id: "b".into(), tool_id: "t".into(), description: "Build".into(), at_ms };
118        record(910_001, call(500_000));
119        assert_eq!(lookup(910_001, Some(499_000)).unwrap().description, "Build");
120        assert!(lookup(910_001, None).is_some(), "an unknown start is trusted on the pid");
121        assert!(lookup(910_001, Some(501_000)).is_some(), "a start a second late is the clock's");
122        assert!(lookup(910_001, Some(500_000 + CLOCK_SLACK_MS + 1)).is_none(), "started after the report: the pid was reused");
123        assert!(lookup(910_001, Some(500_000 - START_SLACK_MS - 1)).is_none(), "far older than the report");
124        assert!(lookup(910_002, Some(499_000)).is_none());
125    }
126}
agentmux_srv/backend/tool_calls.rs

agentmux_srv/backend/
tool_calls.rs
