agentd/runtime/
children.rs

1// SPDX-License-Identifier: Apache-2.0
2//! The runtime's **flat child tree** (RFC 0026 §2, D3): every turn worker and
3//! subagent is a direct child of the supervisor, spawned through the 1.x
4//! machinery (`supervisor::spawn` + the reaper + PDEATHSIG + process groups),
5//! tracked here with its purpose, liveness and cancellation, and torn down
6//! by the kill ladder on drain.
7
8use crate::subagent::protocol::{AgentMsg, ControlMsg, SpawnPayload};
9use crate::supervisor::kill::{Ladder, LadderAction, kill_group, term_group};
10use crate::supervisor::liveness::{Health, Liveness, LivenessConfig};
11use crate::supervisor::reap::Reaped;
12use crate::supervisor::spawn::{Subagent, spawn};
13use crate::supervisor::tree::NodeId;
14use serde_json::{Value, json};
15use std::collections::HashMap;
16use std::path::PathBuf;
17use std::sync::mpsc::Sender;
18use std::time::{Duration, Instant};
19
20/// Why a child exists.
21#[derive(Debug, Clone, PartialEq)]
22pub enum ChildKind {
23    /// A root/conversation turn for context `ctx`, triggered by inbox event `event`.
24    RootTurn {
25        ctx: String,
26        event: Option<String>,
27        reservation: Option<u64>,
28    },
29    /// A workflow step turn (`agent` / `think`).
30    StepTurn {
31        run: String,
32        step: String,
33        reservation: Option<u64>,
34    },
35    /// A structured think serving an internal request (compaction, `think` tool,
36    /// preflight). `reply_to` = the requesting child + request id when a tool
37    /// call is waiting on it.
38    Think {
39        purpose: String,
40        ctx: Option<String>,
41        reply_to: Option<(NodeId, u64)>,
42        extra: Value,
43        reservation: Option<u64>,
44    },
45    /// A subagent (RFC 0009 payload) with a registry handle.
46    Subagent { handle: String },
47}
48
49/// One live child.
50pub struct Child {
51    pub sub: Subagent,
52    pub kind: ChildKind,
53    pub started: Instant,
54    pub liveness: Liveness,
55    pub cancelled: bool,
56    pub tokens: u64,
57    /// Whether this child's unit has been settled — a terminal frame
58    /// (`TurnDone` / `Failed`) was folded back into the durable state. The
59    /// reap path needs the answer *after* the record has left the table (see
60    /// [`Children::is_settled`]): the child's mere presence cannot give it,
61    /// because a settled worker also stays in the table until it is reaped.
62    pub settled: bool,
63}
64
65/// The children registry.
66pub struct Children {
67    exe: PathBuf,
68    events: Sender<(NodeId, AgentMsg)>,
69    reap_tx: Sender<Reaped>,
70    map: HashMap<NodeId, Child>,
71    pid_to_node: HashMap<i32, NodeId>,
72    next: u64,
73    liveness_cfg: LivenessConfig,
74    ladder: Option<Ladder>,
75    last_ping: Instant,
76    ping_seq: u64,
77    /// The child reaped most recently, kept past its removal from `map` as
78    /// `(node, kind, settled)`: the reap path asks "did this worker report a
79    /// terminal frame?" — and, when it did not, needs the kind to route the
80    /// failure and release the reservation the kind carries. One slot is
81    /// enough because the reactor drains reaps one at a time and finishes with
82    /// a node before taking the next.
83    last_reaped: Option<(NodeId, ChildKind, bool)>,
84}
85
86impl Children {
87    pub fn new(
88        exe: PathBuf,
89        events: Sender<(NodeId, AgentMsg)>,
90        reap_tx: Sender<Reaped>,
91    ) -> Children {
92        Children {
93            exe,
94            events,
95            reap_tx,
96            map: HashMap::new(),
97            pid_to_node: HashMap::new(),
98            next: 1,
99            liveness_cfg: LivenessConfig::from_env(),
100            ladder: None,
101            last_ping: Instant::now(),
102            ping_seq: 0,
103            last_reaped: None,
104        }
105    }
106
107    pub fn len(&self) -> usize {
108        self.map.len()
109    }
110    pub fn is_empty(&self) -> bool {
111        self.map.is_empty()
112    }
113    pub fn get(&self, node: NodeId) -> Option<&Child> {
114        self.map.get(&node)
115    }
116    pub fn get_mut(&mut self, node: NodeId) -> Option<&mut Child> {
117        self.map.get_mut(&node)
118    }
119    pub fn iter(&self) -> impl Iterator<Item = (&NodeId, &Child)> {
120        self.map.iter()
121    }
122    pub fn count_kind(&self, f: impl Fn(&ChildKind) -> bool) -> usize {
123        self.map.values().filter(|c| f(&c.kind)).count()
124    }
125
126    /// Spawn a child (tracked with the reaper). The payload's `telemetry`
127    /// must already carry the correlation ids.
128    pub fn spawn(
129        &mut self,
130        payload: &SpawnPayload,
131        kind: ChildKind,
132        deadline: Duration,
133    ) -> std::io::Result<NodeId> {
134        let node = NodeId(self.next);
135        self.next += 1;
136        let exe = self.exe.clone();
137        let events = self.events.clone();
138        let sub = crate::supervisor::reaper::spawn_tracked(&self.reap_tx, || {
139            spawn(&exe, payload, node, events)
140        })?;
141        let now = Instant::now();
142        let child = Child {
143            liveness: Liveness::new(
144                now,
145                now + deadline + Duration::from_secs(60),
146                self.liveness_cfg,
147            ),
148            sub,
149            kind,
150            started: now,
151            cancelled: false,
152            tokens: 0,
153            settled: false,
154        };
155        self.pid_to_node.insert(child.sub.pid(), node);
156        self.map.insert(node, child);
157        crate::obs::metrics::record_subagent_spawned();
158        Ok(node)
159    }
160
161    /// A frame arrived from `node`: refresh liveness (returns whether known).
162    pub fn on_frame(&mut self, node: NodeId, msg: &AgentMsg) -> bool {
163        let Some(c) = self.map.get_mut(&node) else {
164            return false;
165        };
166        let now = Instant::now();
167        match msg {
168            AgentMsg::Pong { .. } => c.liveness.on_pong(now),
169            AgentMsg::Usage(u) => {
170                c.tokens += u.total();
171                c.liveness.on_event(now);
172            }
173            _ => c.liveness.on_event(now),
174        }
175        true
176    }
177
178    /// A terminal frame was folded back in for `node`: its unit is settled, so
179    /// the reap path must not fail it a second time. Marks the just-reaped
180    /// record too, so a failure routed *from* the reap path is not re-entered.
181    pub fn mark_settled(&mut self, node: NodeId) {
182        if let Some(c) = self.map.get_mut(&node) {
183            c.settled = true;
184        }
185        if let Some((n, _, settled)) = self.last_reaped.as_mut()
186            && *n == node
187        {
188            *settled = true;
189        }
190    }
191
192    /// Whether `node`'s unit has been settled by a terminal frame — answerable
193    /// after the child is gone, which is the only time the question is asked.
194    /// A node we never knew counts as settled: there is nothing left to fail.
195    pub fn is_settled(&self, node: NodeId) -> bool {
196        if let Some(c) = self.map.get(&node) {
197            return c.settled;
198        }
199        match &self.last_reaped {
200            Some((n, _, settled)) if *n == node => *settled,
201            _ => true,
202        }
203    }
204
205    /// The kind of the most recently reaped child, so its failure can still be
206    /// routed (and its reservation released) once `on_reaped` has removed it.
207    pub fn reaped_kind(&self, node: NodeId) -> Option<ChildKind> {
208        match &self.last_reaped {
209            Some((n, kind, _)) if *n == node => Some(kind.clone()),
210            _ => None,
211        }
212    }
213
214    /// A child was reaped: forget it and return its record.
215    pub fn on_reaped(&mut self, r: &Reaped) -> Option<(NodeId, Child)> {
216        let node = self.pid_to_node.remove(&r.pid)?;
217        let mut c = self.map.remove(&node)?;
218        c.sub.mark_reaped();
219        c.liveness.on_eof();
220        self.last_reaped = Some((node, c.kind.clone(), c.settled));
221        crate::obs::metrics::record_subagent_exited(match r.outcome {
222            crate::supervisor::reap::WaitOutcome::Exited(0) => "completed",
223            crate::supervisor::reap::WaitOutcome::Exited(_) => "crashed",
224            crate::supervisor::reap::WaitOutcome::Signaled(_) => "cancelled",
225        });
226        Some((node, c))
227    }
228
229    /// Send a control frame to a child.
230    pub fn send(&mut self, node: NodeId, msg: &ControlMsg) -> bool {
231        self.map
232            .get_mut(&node)
233            .is_some_and(|c| c.sub.send(msg).is_ok())
234    }
235
236    /// Cancel a child gracefully (the kill ladder escalates on drain).
237    pub fn cancel(&mut self, node: NodeId, reason: &str) -> bool {
238        let Some(c) = self.map.get_mut(&node) else {
239            return false;
240        };
241        c.cancelled = true;
242        c.sub
243            .send(&ControlMsg::Cancel {
244                reason: reason.to_string(),
245            })
246            .is_ok()
247    }
248
249    /// Kill a child now (its whole process group).
250    pub fn kill(&mut self, node: NodeId) {
251        if let Some(c) = self.map.get_mut(&node) {
252            c.sub.kill();
253        }
254    }
255
256    /// Periodic maintenance: pings + liveness. Returns the nodes that must be
257    /// torn down (stuck / past deadline).
258    pub fn tick(&mut self) -> Vec<(NodeId, Health)> {
259        let now = Instant::now();
260        if now.duration_since(self.last_ping) >= self.liveness_cfg.ping_interval {
261            self.last_ping = now;
262            self.ping_seq += 1;
263            let seq = self.ping_seq;
264            for c in self.map.values_mut() {
265                let _ = c.sub.send(&ControlMsg::Ping { seq });
266            }
267        }
268        self.map
269            .iter()
270            .filter_map(|(n, c)| {
271                let h = c.liveness.classify(now);
272                h.needs_teardown().then_some((*n, h))
273            })
274            .collect()
275    }
276
277    /// Begin the drain: cancel every child; the ladder escalates.
278    pub fn begin_drain(&mut self, reason: &str) {
279        for c in self.map.values_mut() {
280            c.cancelled = true;
281            let _ = c.sub.send(&ControlMsg::Cancel {
282                reason: reason.to_string(),
283            });
284        }
285        if self.ladder.is_none() {
286            self.ladder = Some(Ladder::with_defaults(Instant::now()));
287        }
288    }
289
290    /// Drive the ladder: `true` when every child is gone.
291    pub fn drive_drain(&mut self, force: bool) -> bool {
292        let all_exited = self.map.is_empty();
293        let Some(ladder) = self.ladder.as_mut() else {
294            return all_exited;
295        };
296        match ladder.poll(Instant::now(), all_exited, force) {
297            LadderAction::Wait => false,
298            LadderAction::Term => {
299                for c in self.map.values() {
300                    term_group(c.sub.pgid());
301                }
302                false
303            }
304            LadderAction::Kill => {
305                for c in self.map.values() {
306                    kill_group(c.sub.pgid());
307                }
308                false
309            }
310            LadderAction::Done => true,
311        }
312    }
313
314    /// Forget every remaining child (after a forced kill at abandon).
315    pub fn abandon(&mut self) {
316        for (_, mut c) in self.map.drain() {
317            c.sub.kill();
318        }
319        self.pid_to_node.clear();
320    }
321
322    /// A status view.
323    pub fn status(&self) -> Value {
324        json!(
325            self.map
326                .iter()
327                .map(|(n, c)| json!({"node": n.0, "pid": c.sub.pid(), "kind": kind_label(&c.kind), "age_ms": c.started.elapsed().as_millis() as u64, "tokens": c.tokens, "cancelled": c.cancelled}))
328                .collect::<Vec<_>>()
329        )
330    }
331}
332
333pub fn kind_label(k: &ChildKind) -> String {
334    match k {
335        ChildKind::RootTurn { ctx, .. } => format!("turn:{ctx}"),
336        ChildKind::StepTurn { run, step, .. } => format!("step:{run}/{step}"),
337        ChildKind::Think { purpose, .. } => format!("think:{purpose}"),
338        ChildKind::Subagent { handle } => format!("subagent:{handle}"),
339    }
340}
agentd/runtime/children.rs

agentd/runtime/
children.rs
