agentd/runtime/
children.rs

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

agentd/runtime/
children.rs
