agentmux_srv/backend/
identity_spawn.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Which spawned agent processes carry an identity token — identity M4a
5//! (`docs/specs/SPEC_AGENT_IDENTITY_CARRIED_NOT_DERIVED_2026_09_23.md`
6//! §6.5.6).
7//!
8//! Two measurements, because request-side counters cannot gate anything: an
9//! unattributed request from the UI and one from a tokenless agent look the
10//! same (§6.5.1).
11//!
12//! - **`spawn.no_token.<path>`** counts process spawns, per controller, of
13//!   an agent that has a row (a UID) but got no `AGENTMUX_AGENT_TOKEN` — the
14//!   gap M4b closes, gated on these reaching zero. "Has a row" is decided by
15//!   the store, not by the environment: the gap paths (`agent.send`, App
16//!   Server) are exactly the ones whose environment carries no
17//!   `AGENTMUX_AGENT_UID` although their block has a row (adversarial review
18//!   of #3571). **`spawn.no_row.<path>`**
19//!   counts spawns with no row at all (quick-launch panes, template-based
20//!   continuations before M4b binds them), kept apart so the gate can be met.
21//!   Recorded where a process is actually started, never where an
22//!   environment is merely built: a turn delivered to a running process
23//!   rebuilds the environment without spawning anything (review on #3571).
24//! - **`live.tokenless_or_unknown`** counts live registered agents (with a
25//!   row, so they *should* carry a token) whose process is not known to carry
26//!   one — including a process spawned before this srv started, which never
27//!   passes a spawn site again. M4d and M5 wait for this to drain to zero,
28//!   not for a counter that merely stops moving.
29
30use std::collections::HashMap;
31use std::sync::{Arc, Mutex, OnceLock};
32
33use crate::backend::reactive::AgentRegistration;
34use crate::backend::storage::store::Store;
35
36static CARRIED_BY_BLOCK: OnceLock<Mutex<HashMap<String, bool>>> = OnceLock::new();
37
38fn carried_by_block() -> &'static Mutex<HashMap<String, bool>> {
39    CARRIED_BY_BLOCK.get_or_init(Default::default)
40}
41
42/// Record that this srv spawned a process on `block_id`, and whether it
43/// carried a token. `no_token_site` names the launch path's counter; `None`
44/// for a path that is not an agent launch by itself (a terminal pane, where
45/// an agent CLI is only one possibility — the live gauge covers that case).
46pub(crate) fn record_spawn(
47    block_id: &str,
48    carried_token: bool,
49    no_token_site: Option<&'static str>,
50) {
51    if block_id.is_empty() {
52        return;
53    }
54    carried_by_block()
55        .lock()
56        .unwrap_or_else(|e| e.into_inner())
57        .insert(block_id.to_string(), carried_token);
58    if !carried_token {
59        if let Some(site) = no_token_site {
60            crate::backend::agent_resolve::record_uid_fallback(site);
61        }
62    }
63}
64
65/// Drop `block_id`'s record when its processes are released
66/// (`blockcontroller::release_block_processes`), so the map is bounded by
67/// open blocks, not by every block ever opened.
68pub(crate) fn forget_block(block_id: &str) {
69    carried_by_block()
70        .lock()
71        .unwrap_or_else(|e| e.into_inner())
72        .remove(block_id);
73}
74
75/// The controller a process was spawned by — the `<path>` in the counters.
76#[derive(Debug, Clone, Copy)]
77pub(crate) enum SpawnPath {
78    Persistent,
79    Subprocess,
80    Acp,
81    AppServer,
82    Container,
83}
84
85/// This channel's object store, for deciding whether a spawned block has a
86/// row. Attached once at boot (`bootstrap/stores.rs`), beside the token index, so
87/// `record_process_spawn`'s callers need not pass a store.
88static ROW_STORE: OnceLock<Arc<Store>> = OnceLock::new();
89
90pub(crate) fn attach_row_store(store: Arc<Store>) {
91    let _ = ROW_STORE.set(store);
92}
93
94/// Whether `block_id` shows an agent that has a row — the same block → row
95/// resolution `build_persistent_spawn_env` uses, which also finds the row a
96/// template-based continuation folded into. A synchronous store read; spawn
97/// paths call it only when the environment carries no UID.
98pub(crate) fn block_has_row(block_id: &str) -> bool {
99    ROW_STORE
100        .get()
101        .is_some_and(|store| store_block_has_row(store, block_id))
102}
103
104/// [`block_has_row`] against a given store. A store fault answers "has a
105/// row" (counted): a gate must never read clear because a read failed.
106pub(crate) fn store_block_has_row(store: &Store, block_id: &str) -> bool {
107    match store.instance_get_active_for_block(block_id) {
108        Ok(Some(instance)) => !instance.id.trim().is_empty(),
109        Ok(None) => false,
110        Err(e) => {
111            crate::backend::agent_resolve::record_uid_fallback("identity.row_lookup_error");
112            tracing::warn!(block_id, error = %e, "identity: block row lookup failed — counted as row-backed");
113            true
114        }
115    }
116}
117
118/// Record a process spawn on `block_id` from the environment it was
119/// actually given. Call at the point the process is started.
120pub(crate) fn record_process_spawn(block_id: &str, path: SpawnPath, env: &HashMap<String, String>) {
121    record_process_spawn_with(block_id, path, env, block_has_row);
122}
123
124fn record_process_spawn_with(
125    block_id: &str,
126    path: SpawnPath,
127    env: &HashMap<String, String>,
128    has_row: impl FnOnce(&str) -> bool,
129) {
130    let has = |k: &str| env.get(k).is_some_and(|v| !v.trim().is_empty());
131    let carried = has("AGENTMUX_AGENT_TOKEN");
132    // A token implies a row (tokens are minted for rows); otherwise a carried
133    // UID does, and failing both, the store decides.
134    let row = carried || has("AGENTMUX_AGENT_UID") || has_row(block_id);
135    let site = match (row, path) {
136        (true, SpawnPath::Persistent) => "spawn.no_token.persistent",
137        (true, SpawnPath::Subprocess) => "spawn.no_token.subprocess",
138        (true, SpawnPath::Acp) => "spawn.no_token.acp",
139        (true, SpawnPath::AppServer) => "spawn.no_token.app_server",
140        (true, SpawnPath::Container) => "spawn.no_token.container",
141        (false, SpawnPath::Persistent) => "spawn.no_row.persistent",
142        (false, SpawnPath::Subprocess) => "spawn.no_row.subprocess",
143        (false, SpawnPath::Acp) => "spawn.no_row.acp",
144        (false, SpawnPath::AppServer) => "spawn.no_row.app_server",
145        (false, SpawnPath::Container) => "spawn.no_row.container",
146    };
147    record_spawn(block_id, carried, Some(site));
148}
149
150/// Whether the process this srv last spawned on `block_id` carried a token;
151/// `None` if this srv has not spawned one there.
152pub(crate) fn carried_token(block_id: &str) -> Option<bool> {
153    carried_by_block()
154        .lock()
155        .unwrap_or_else(|e| e.into_inner())
156        .get(block_id)
157        .copied()
158}
159
160/// The two live gauges over `registrations`:
161/// `(tokenless_or_unknown, unidentified)`. An agent with a row whose process
162/// is not known to carry a token counts toward the first; an agent with no
163/// row (a quick-launch pane — outside the identity system, §6.5.6) toward
164/// the second. A registration without a UID is not taken as row-less on its
165/// word: a block driven only through `agent.send` registers without one
166/// although it has a row, so `has_row` asks the store. `is_live` filters out
167/// registrations whose block no longer runs anything here, so a stale
168/// registration cannot hold the gauge up.
169pub(crate) fn live_gauges(
170    registrations: &[AgentRegistration],
171    is_live: impl Fn(&str) -> bool,
172    has_row: impl Fn(&str) -> bool,
173) -> (usize, usize) {
174    let mut tokenless_or_unknown = 0;
175    let mut unidentified = 0;
176    for reg in registrations {
177        if !is_live(&reg.block_id) {
178            continue;
179        }
180        if carried_token(&reg.block_id) == Some(true) {
181            continue;
182        }
183        if reg.uid.is_some() || has_row(&reg.block_id) {
184            tokenless_or_unknown += 1;
185        } else {
186            unidentified += 1;
187        }
188    }
189    (tokenless_or_unknown, unidentified)
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195
196    fn reg(block: &str, uid: Option<&str>) -> AgentRegistration {
197        let h = crate::backend::reactive::handler::ReactiveHandler::new();
198        h.register_agent_full("AgentY", block, None, 0, None, uid, "test.m4a")
199            .unwrap();
200        h.list_agents().into_iter().next().unwrap()
201    }
202
203    /// A row-backed agent counts until this srv has seen it spawned with a
204    /// token; one spawned before this srv started stays "unknown" — the case
205    /// spawn counters alone would miss. A row-less pane is counted apart.
206    #[test]
207    fn live_gauges_count_row_backed_agents_not_known_to_carry_a_token() {
208        let (with, without, unknown, rowless) = (
209            "m4a-blk-with",
210            "m4a-blk-without",
211            "m4a-blk-unknown",
212            "m4a-blk-rowless",
213        );
214        record_spawn(with, true, Some("test.m4a.no_token"));
215        record_spawn(without, false, Some("test.m4a.no_token"));
216        let regs = vec![
217            reg(with, Some("uid-1")),
218            reg(without, Some("uid-2")),
219            reg(unknown, Some("uid-3")),
220            reg(rowless, None),
221        ];
222        let no_row = |_: &str| false;
223        assert_eq!(live_gauges(&regs, |_| true, no_row), (2, 1));
224        // A registration whose block runs nothing here does not count.
225        assert_eq!(live_gauges(&regs, |b| b != unknown, no_row), (1, 1));
226        // A registration without a UID whose block has a row (an
227        // `agent.send`-only block) is tokenless, not unidentified.
228        assert_eq!(live_gauges(&regs, |_| true, |b| b == rowless), (3, 0));
229        assert_eq!(carried_token(with), Some(true));
230        assert_eq!(carried_token(unknown), None);
231    }
232
233    /// Classified from the environment the process actually got: a UID
234    /// without a token is the gap; no UID at all is a row-less spawn.
235    #[test]
236    fn a_spawn_is_classified_from_its_actual_environment() {
237        let env = |pairs: &[(&str, &str)]| -> HashMap<String, String> {
238            pairs
239                .iter()
240                .map(|(k, v)| (k.to_string(), v.to_string()))
241                .collect()
242        };
243        record_process_spawn(
244            "m4a-blk-env-tok",
245            SpawnPath::Persistent,
246            &env(&[("AGENTMUX_AGENT_UID", "u"), ("AGENTMUX_AGENT_TOKEN", "t")]),
247        );
248        assert_eq!(carried_token("m4a-blk-env-tok"), Some(true));
249        let count = |site: &str| {
250            crate::backend::agent_resolve::uid_fallback_counts()
251                .into_iter()
252                .find(|(s, _)| *s == site)
253                .map_or(0, |(_, n)| n)
254        };
255        let (gap, rowless) = (count("spawn.no_token.acp"), count("spawn.no_row.acp"));
256        record_process_spawn(
257            "m4a-blk-env-gap",
258            SpawnPath::Acp,
259            &env(&[("AGENTMUX_AGENT_UID", "u")]),
260        );
261        record_process_spawn_with("m4a-blk-env-rowless", SpawnPath::Acp, &env(&[]), |_| false);
262        assert!(count("spawn.no_token.acp") > gap);
263        assert!(count("spawn.no_row.acp") > rowless);
264        assert_eq!(carried_token("m4a-blk-env-gap"), Some(false));
265    }
266
267    /// The gap paths (`agent.send`, App Server) spawn with no
268    /// `AGENTMUX_AGENT_UID` although their block has a row: the store
269    /// decides, so they count as the gap, not as row-less (adversarial
270    /// review of #3571 — otherwise the M4b gate passes trivially).
271    #[test]
272    fn a_row_backed_spawn_without_a_carried_uid_is_the_gap() {
273        let count = |site: &str| {
274            crate::backend::agent_resolve::uid_fallback_counts()
275                .into_iter()
276                .find(|(s, _)| *s == site)
277                .map_or(0, |(_, n)| n)
278        };
279        let (gap, rowless) = (
280            count("spawn.no_token.app_server"),
281            count("spawn.no_row.app_server"),
282        );
283        let mut asked = None;
284        record_process_spawn_with(
285            "m4a-blk-appserver",
286            SpawnPath::AppServer,
287            &HashMap::new(),
288            |b| {
289                asked = Some(b.to_string());
290                true
291            },
292        );
293        assert_eq!(asked.as_deref(), Some("m4a-blk-appserver"));
294        assert_eq!(count("spawn.no_token.app_server"), gap + 1);
295        assert_eq!(count("spawn.no_row.app_server"), rowless);
296
297        // A token alone implies a row; the store is not asked.
298        let before = count("spawn.no_row.container");
299        record_process_spawn_with(
300            "m4a-blk-container",
301            SpawnPath::Container,
302            &[("AGENTMUX_AGENT_TOKEN".to_string(), "t".to_string())]
303                .into_iter()
304                .collect(),
305            |_| panic!("not asked"),
306        );
307        assert_eq!(carried_token("m4a-blk-container"), Some(true));
308        assert_eq!(count("spawn.no_row.container"), before);
309    }
310
311    /// Releasing a block's processes drops its record (ReAgent P2 on #3571:
312    /// the map must not grow with every pane ever opened).
313    #[test]
314    fn releasing_a_block_forgets_its_record() {
315        record_spawn("m4a-blk-closed", true, None);
316        crate::backend::blockcontroller::release_block_processes("m4a-blk-closed");
317        assert_eq!(carried_token("m4a-blk-closed"), None);
318    }
319}
agentmux_srv/backend/identity_spawn.rs

agentmux_srv/backend/
identity_spawn.rs
