Skip to main content

scv_tools/delegate/
records.rs

1//! What SCV started, so it can list, stop, and clean up delegated agents.
2//!
3//! Every delegated process is tagged through its environment
4//! (`SCV_PARENT=<instance>/<session>/<handle>`, chained through nested SCVs,
5//! and `SCV_DELEGATION_DEPTH`) and recorded in
6//! `$SCV_HOME/state/delegations/<handle>.json` while it runs. A record whose
7//! owning SCV process died is an orphan: the daemon's reconciliation kills its
8//! process group and anything still carrying its tag, then removes it.
9//!
10//! This is cooperative bookkeeping. Delegated agents run as the user, so one
11//! that deliberately clears its environment or leaves its process group can
12//! escape it; the tags and records exist to clean up accidental leaks.
13
14use std::{
15    collections::{HashMap, HashSet},
16    ffi::OsString,
17    path::{Path, PathBuf},
18    sync::{
19        Arc, Mutex,
20        atomic::{AtomicBool, Ordering},
21    },
22    time::{Duration, SystemTime, UNIX_EPOCH},
23};
24
25use scv_client::Layout;
26use serde::{Deserialize, Serialize};
27use sha2::{Digest, Sha256};
28
29use crate::{process::ProcessGroup, sync::lock};
30
31/// Environment variable carrying the delegation chain.
32pub const PARENT_VARIABLE: &str = "SCV_PARENT";
33/// Environment variable carrying how deeply this process is delegated.
34pub use scv_client::DELEGATION_DEPTH_VARIABLE as DEPTH_VARIABLE;
35/// Grace between TERM and KILL when stopping delegated processes.
36pub(crate) const STOP_GRACE: Duration = Duration::from_secs(2);
37/// Largest record file read.
38const MAX_RECORD_BYTES: u64 = 64 * 1024;
39/// A zombie child younger than this may still be awaited by its spawner.
40const ZOMBIE_MIN_AGE: Duration = Duration::from_secs(10);
41
42/// Delegation depth of the current process: 0 unless an SCV started it.
43///
44/// Always 0 in this crate's unit tests: callers without a delegation context
45/// fall back to it, and the test runner may itself be a delegated agent. The
46/// black-box tests cover the inherited read.
47pub fn current_depth() -> u32 {
48    #[cfg(test)]
49    {
50        0
51    }
52    #[cfg(not(test))]
53    {
54        scv_client::inherited_delegation_depth().unwrap_or(0)
55    }
56}
57
58/// A process, identified by PID plus start time so a reused PID never matches.
59#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
60pub struct ProcessIdentity {
61    pub pid: u32,
62    pub(crate) start_time: u64,
63}
64
65impl ProcessIdentity {
66    pub fn current() -> Option<Self> {
67        Self::of(std::process::id())
68    }
69
70    pub fn of(pid: u32) -> Option<Self> {
71        process_start_time(pid).map(|start_time| Self { pid, start_time })
72    }
73
74    /// Whether this exact process still runs. An exited process that its
75    /// parent has not yet collected (a zombie) does not count.
76    pub fn is_alive(&self) -> bool {
77        #[cfg(target_os = "linux")]
78        {
79            linux::stat(self.pid)
80                .is_some_and(|info| info.start_time == self.start_time && info.state != 'Z')
81        }
82        #[cfg(not(target_os = "linux"))]
83        {
84            Self::of(self.pid) == Some(*self) && !process_is_zombie(self.pid)
85        }
86    }
87}
88
89/// One delegated run, as recorded on disk.
90#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
91pub struct DelegationRecord {
92    pub handle: String,
93    pub agent: String,
94    pub instance: String,
95    pub session: String,
96    pub owner: ProcessIdentity,
97    /// The agent process, which leads its own process group.
98    pub process: ProcessIdentity,
99    pub pgid: u32,
100    pub cwd: PathBuf,
101    pub started_unix: u64,
102    /// Depth of the delegated process (the owner's depth plus one).
103    pub depth: u32,
104    /// The conversation this run is a turn of, and which turn.
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub conversation: Option<String>,
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub turn: Option<u32>,
109    /// A live child that keeps its process between turns (a nested SCV or an
110    /// ACP agent) and has no turn running: when its last turn ended. Absent
111    /// while a turn runs, and always for a per-turn run.
112    #[serde(default, skip_serializing_if = "Option::is_none")]
113    pub idle_since_unix: Option<u64>,
114    /// A nested SCV's own background jobs that still run or whose results
115    /// its model has not yet taken in: finished and not yet reported, or in
116    /// the turn it started to report them, until that turn ends. Absent when
117    /// there are none, for every other agent, and from older releases.
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub background_jobs: Option<u32>,
120}
121
122/// A record plus what SCV currently observes about it.
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct DelegationEntry {
125    pub record: DelegationRecord,
126    /// The owning SCV process is gone; reconciliation will clean it up.
127    pub orphaned: bool,
128    /// Live processes in its group plus tagged processes outside it.
129    pub processes: usize,
130}
131
132impl DelegationEntry {
133    /// Whether the run is still at work: it has live processes and is not a
134    /// live child waiting between turns of its conversation with no
135    /// background jobs of its own.
136    pub fn working(&self) -> bool {
137        self.processes > 0
138            && (self.record.idle_since_unix.is_none()
139                || self.record.background_jobs.is_some_and(|jobs| jobs > 0))
140    }
141}
142
143/// A delegation named in an `SCV_PARENT` chain, and the session that
144/// started it.
145#[derive(Debug, Clone, PartialEq, Eq)]
146pub struct ChainRun {
147    /// The delegation's handle, such as `codex-3f9a2c`.
148    pub handle: String,
149    /// The SCV session that started it.
150    pub session: String,
151}
152
153/// What a reconciliation pass did.
154#[derive(Debug, Clone, Default, PartialEq, Eq)]
155pub struct ReconcileReport {
156    /// Orphaned delegations whose processes were stopped.
157    pub reaped: Vec<String>,
158    /// Orphaned records whose processes had already exited.
159    pub removed: usize,
160    /// Conversation markers left by SCV processes that no longer run.
161    pub stale_markers: usize,
162}
163
164#[derive(Debug, Default)]
165struct Inner {
166    active: HashMap<String, Arc<AtomicBool>>,
167    reaped: u64,
168}
169
170/// The delegations one SCV process started, backed by the instance's records.
171#[derive(Debug)]
172pub struct DelegationRegistry {
173    record_dir: PathBuf,
174    conversation_dir: PathBuf,
175    instance: String,
176    owner: Option<ProcessIdentity>,
177    depth: u32,
178    chain: Option<String>,
179    inner: Mutex<Inner>,
180}
181
182/// A delegation about to start: its handle and the environment tagging it.
183pub(crate) struct PendingDelegation {
184    pub(crate) handle: String,
185    pub(crate) environment: Vec<(OsString, OsString)>,
186    agent: String,
187    session: String,
188    cwd: PathBuf,
189    conversation: Option<(String, u32)>,
190    /// Depth of the delegated process.
191    depth: u32,
192}
193
194impl DelegationRegistry {
195    /// The registry for the SCV instance at `layout`: records in
196    /// [`Layout::delegations`], conversation markers in
197    /// [`Layout::conversations`], and an instance ID hashed from its home.
198    pub fn new(layout: &Layout) -> Self {
199        let digest = Sha256::digest(layout.home().as_os_str().as_encoded_bytes());
200        let instance = digest[..4]
201            .iter()
202            .map(|byte| format!("{byte:02x}"))
203            .collect();
204        Self {
205            record_dir: layout.delegations(),
206            conversation_dir: layout.conversations(),
207            instance,
208            owner: ProcessIdentity::current(),
209            depth: current_depth(),
210            chain: std::env::var(PARENT_VARIABLE)
211                .ok()
212                .filter(|value| !value.trim().is_empty()),
213            inner: Mutex::new(Inner::default()),
214        }
215    }
216
217    /// A unit-test fixture whose caller is independent of the test runner.
218    /// [`Self::new`] reads the inherited chain in every build.
219    #[cfg(test)]
220    pub(crate) fn for_test(layout: &Layout) -> Self {
221        Self {
222            depth: 0,
223            chain: None,
224            ..Self::new(layout)
225        }
226    }
227
228    /// This process's own delegation depth.
229    pub(crate) fn depth(&self) -> u32 {
230        self.depth
231    }
232
233    pub fn record_dir(&self) -> &Path {
234        &self.record_dir
235    }
236
237    /// Short identifier of the SCV instance, shared by all its processes.
238    pub fn instance(&self) -> &str {
239        &self.instance
240    }
241
242    /// Delegations this process stopped as orphans since it started.
243    pub fn reaped_total(&self) -> u64 {
244        lock(&self.inner).reaped
245    }
246
247    /// Where live conversations leave markers for `scv agents gc`.
248    pub fn conversation_dir(&self) -> &Path {
249        &self.conversation_dir
250    }
251
252    #[cfg(test)]
253    pub(crate) fn begin(
254        &self,
255        agent: &str,
256        session: &str,
257        cwd: &Path,
258        conversation: Option<(&str, u32)>,
259    ) -> PendingDelegation {
260        self.begin_at(self.depth, agent, session, cwd, conversation)
261    }
262
263    /// Start recording a delegation whose owner is at `owner_depth`: the
264    /// process's own depth, or more when its client is itself delegated.
265    pub(crate) fn begin_at(
266        &self,
267        owner_depth: u32,
268        agent: &str,
269        session: &str,
270        cwd: &Path,
271        conversation: Option<(&str, u32)>,
272    ) -> PendingDelegation {
273        let suffix = uuid::Uuid::new_v4().simple().to_string();
274        let handle = format!("{agent}-{}", &suffix[..6]);
275        let entry = format!("{}/{session}/{handle}", self.instance);
276        let chain = match &self.chain {
277            Some(chain) => format!("{chain};{entry}"),
278            None => entry,
279        };
280        PendingDelegation {
281            environment: vec![
282                (PARENT_VARIABLE.into(), chain.into()),
283                (
284                    DEPTH_VARIABLE.into(),
285                    owner_depth.saturating_add(1).to_string().into(),
286                ),
287            ],
288            handle,
289            agent: agent.to_owned(),
290            session: session.to_owned(),
291            cwd: cwd.to_owned(),
292            conversation: conversation.map(|(handle, turn)| (handle.to_owned(), turn)),
293            depth: owner_depth.saturating_add(1),
294        }
295    }
296
297    /// Record a spawned delegation. The returned guard removes the record and
298    /// stops leftovers when the run ends, even if the run is abandoned.
299    pub(crate) fn register(
300        self: &Arc<Self>,
301        pending: PendingDelegation,
302        pid: u32,
303    ) -> std::io::Result<DelegationGuard> {
304        let killed = Arc::new(AtomicBool::new(false));
305        let record = DelegationRecord {
306            handle: pending.handle.clone(),
307            agent: pending.agent,
308            instance: self.instance.clone(),
309            session: pending.session,
310            owner: self.owner.unwrap_or(ProcessIdentity {
311                pid: std::process::id(),
312                start_time: 0,
313            }),
314            process: ProcessIdentity::of(pid).unwrap_or(ProcessIdentity { pid, start_time: 0 }),
315            pgid: pid,
316            cwd: pending.cwd,
317            started_unix: unix_now(),
318            depth: pending.depth,
319            conversation: pending
320                .conversation
321                .as_ref()
322                .map(|(handle, _)| handle.clone()),
323            turn: pending.conversation.as_ref().map(|(_, turn)| *turn),
324            idle_since_unix: None,
325            background_jobs: None,
326        };
327        lock(&self.inner)
328            .active
329            .insert(record.handle.clone(), Arc::clone(&killed));
330        if let Err(error) = write_record(&self.record_dir, &record) {
331            lock(&self.inner).active.remove(&record.handle);
332            return Err(error);
333        }
334        Ok(DelegationGuard {
335            registry: Arc::clone(self),
336            handle: record.handle,
337            pgid: pid,
338            killed,
339            finished: false,
340        })
341    }
342
343    /// Delegations of this instance that are still running. With
344    /// `include_orphans`, also records whose owner died and await cleanup.
345    pub fn list(&self, include_orphans: bool) -> Vec<DelegationEntry> {
346        let table = ProcessTable::snapshot();
347        let mut entries: Vec<_> = self
348            .records()
349            .into_iter()
350            .filter_map(|record| {
351                let orphaned = !self.owner_alive(&record);
352                if orphaned && !include_orphans {
353                    return None;
354                }
355                let processes = table.members(&record).len();
356                Some(DelegationEntry {
357                    record,
358                    orphaned,
359                    processes,
360                })
361            })
362            .collect();
363        entries.sort_by(|a, b| {
364            a.record
365                .started_unix
366                .cmp(&b.record.started_unix)
367                .then_with(|| a.record.handle.cmp(&b.record.handle))
368        });
369        entries
370    }
371
372    /// The running delegation this process started that an `SCV_PARENT`
373    /// `chain` names: the one a caller runs inside, whatever nested SCVs lie
374    /// between. Entries of other instances, runs other processes own, and
375    /// malformed entries are skipped.
376    pub fn own_run(&self, chain: &str) -> Option<ChainRun> {
377        let own = std::process::id();
378        let entries = self.list(true);
379        chain.split(';').find_map(|entry| {
380            let mut parts = entry.splitn(3, '/');
381            let (instance, session, handle) = (parts.next()?, parts.next()?, parts.next()?);
382            if instance != self.instance {
383                return None;
384            }
385            entries
386                .iter()
387                .find(|running| running.record.handle == handle && running.record.owner.pid == own)
388                .map(|_| ChainRun {
389                    handle: handle.to_owned(),
390                    session: session.to_owned(),
391                })
392        })
393    }
394
395    /// Stop one delegation of this instance, whichever process owns it.
396    pub async fn kill(&self, handle: &str) -> Result<(), String> {
397        let record = self
398            .records()
399            .into_iter()
400            .find(|record| record.handle == handle)
401            .ok_or_else(|| format!("no running delegation {handle:?}"))?;
402        let local = lock(&self.inner).active.get(handle).cloned();
403        if let Some(killed) = &local {
404            killed.store(true, Ordering::Release);
405        }
406        stop_delegation(&record).await;
407        if local.is_none() && !self.owner_alive(&record) {
408            remove_record(&self.record_dir, handle);
409            lock(&self.inner).reaped += 1;
410        }
411        Ok(())
412    }
413
414    /// Stop and remove every orphaned delegation of this instance.
415    pub async fn reconcile(&self) -> ReconcileReport {
416        let mut report = ReconcileReport::default();
417        for record in self.records() {
418            if self.owner_alive(&record) {
419                continue;
420            }
421            if stop_delegation(&record).await {
422                report.reaped.push(record.handle.clone());
423            } else {
424                report.removed += 1;
425            }
426            remove_record(&self.record_dir, &record.handle);
427        }
428        lock(&self.inner).reaped += report.reaped.len() as u64;
429        report.stale_markers =
430            crate::delegate::conversation::remove_stale_markers(self.conversation_dir());
431        report
432    }
433
434    /// Whether the process that owns `record` still runs it. A record this
435    /// process owns counts only while its run is active here.
436    fn owner_alive(&self, record: &DelegationRecord) -> bool {
437        if Some(record.owner) == self.owner {
438            return lock(&self.inner).active.contains_key(&record.handle);
439        }
440        record.owner.is_alive()
441    }
442
443    fn records(&self) -> Vec<DelegationRecord> {
444        let Ok(entries) = std::fs::read_dir(&self.record_dir) else {
445            return Vec::new();
446        };
447        entries
448            .filter_map(Result::ok)
449            .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "json"))
450            .filter_map(|entry| read_record(&entry.path()))
451            .filter(|record| record.instance == self.instance)
452            .collect()
453    }
454
455    fn finish_local(&self, handle: &str) {
456        lock(&self.inner).active.remove(handle);
457        remove_record(&self.record_dir, handle);
458    }
459}
460
461/// Keeps a delegation recorded while it runs.
462pub(crate) struct DelegationGuard {
463    registry: Arc<DelegationRegistry>,
464    handle: String,
465    pgid: u32,
466    killed: Arc<AtomicBool>,
467    finished: bool,
468}
469
470impl DelegationGuard {
471    #[cfg(test)]
472    pub(crate) fn handle(&self) -> &str {
473        &self.handle
474    }
475
476    /// Record that the run moved on to `turn` of its conversation and is at
477    /// work, for a live child that serves every turn.
478    pub(crate) fn set_turn(&self, turn: u32) {
479        self.update(|record| {
480            record.turn = Some(turn);
481            record.idle_since_unix = None;
482        });
483    }
484
485    /// Record that a live child ended its turn and waits for the next one.
486    pub(crate) fn set_idle(&self) {
487        let now = unix_now();
488        self.update(|record| record.idle_since_unix = Some(now));
489    }
490
491    /// Record how many of a live child's own background jobs still run or
492    /// wait to be reported to it.
493    pub(crate) fn set_background_jobs(&self, jobs: usize) {
494        let jobs = (jobs > 0).then(|| u32::try_from(jobs).unwrap_or(u32::MAX));
495        self.update(|record| record.background_jobs = jobs);
496    }
497
498    /// Rewrite the record. Bookkeeping only: a failure never fails the turn,
499    /// and a record already removed stays removed.
500    fn update(&self, change: impl FnOnce(&mut DelegationRecord)) {
501        let dir = &self.registry.record_dir;
502        if let Some(mut record) = read_record(&dir.join(format!("{}.json", self.handle))) {
503            change(&mut record);
504            if let Err(error) = write_record(dir, &record) {
505                tracing::debug!(handle = %record.handle, %error, "could not update a delegation record");
506            }
507        }
508    }
509
510    /// Whether `scv agents kill` stopped this run.
511    pub(crate) fn was_killed(&self) -> bool {
512        self.killed.load(Ordering::Acquire)
513    }
514
515    /// The run ended: stop anything still tagged with it, then forget it.
516    pub(crate) async fn finish(mut self) {
517        self.finished = true;
518        stop_tagged(&self.handle).await;
519        self.registry.finish_local(&self.handle);
520    }
521}
522
523impl Drop for DelegationGuard {
524    fn drop(&mut self) {
525        if self.finished {
526            return;
527        }
528        // The run was abandoned mid-flight: kill its group now and sweep
529        // tagged leftovers in the background.
530        if let Some(group) = ProcessGroup::new(self.pgid) {
531            group.signal(libc::SIGKILL);
532        }
533        self.registry.finish_local(&self.handle);
534        let handle = self.handle.clone();
535        if let Ok(runtime) = tokio::runtime::Handle::try_current() {
536            runtime.spawn(async move { stop_tagged(&handle).await });
537        } else {
538            for identity in tagged_processes(&handle) {
539                signal(identity.pid, libc::SIGKILL);
540            }
541        }
542    }
543}
544
545/// Stop a delegation's process group and tagged processes: TERM, then KILL
546/// after a short grace. Returns whether anything was still running.
547async fn stop_delegation(record: &DelegationRecord) -> bool {
548    let mut stopped = false;
549    // The group ID is the leader's PID, which the kernel does not reuse while
550    // the group has members. A live leader with a different start time means
551    // the PID was reused, so the group is not ours.
552    let leader = ProcessIdentity::of(record.process.pid);
553    let group_is_ours = record.pgid == record.process.pid
554        && match leader {
555            Some(leader) => leader == record.process,
556            None => group_exists(record.pgid),
557        };
558    if group_is_ours && group_exists(record.pgid) {
559        stopped = true;
560        let group = ProcessGroup::new(record.pgid);
561        if let Some(group) = group {
562            group.signal(libc::SIGTERM);
563        }
564        let deadline = tokio::time::Instant::now() + STOP_GRACE;
565        while group_exists(record.pgid) && tokio::time::Instant::now() < deadline {
566            tokio::time::sleep(Duration::from_millis(50)).await;
567        }
568        if let Some(group) = group {
569            group.signal(libc::SIGKILL);
570        }
571    }
572    stopped | stop_tagged(&record.handle).await
573}
574
575/// TERM, then KILL, every process tagged with `handle`. Returns whether any was found.
576async fn stop_tagged(handle: &str) -> bool {
577    let tagged = tagged_processes(handle);
578    if tagged.is_empty() {
579        return false;
580    }
581    for identity in &tagged {
582        signal(identity.pid, libc::SIGTERM);
583    }
584    let deadline = tokio::time::Instant::now() + STOP_GRACE;
585    while tagged.iter().any(ProcessIdentity::is_alive) && tokio::time::Instant::now() < deadline {
586        tokio::time::sleep(Duration::from_millis(50)).await;
587    }
588    for identity in tagged.iter().filter(|identity| identity.is_alive()) {
589        signal(identity.pid, libc::SIGKILL);
590    }
591    true
592}
593
594/// Processes whose `SCV_PARENT` chain names `handle`.
595fn tagged_processes(handle: &str) -> Vec<ProcessIdentity> {
596    let own = std::process::id();
597    ProcessTable::snapshot()
598        .tagged
599        .into_iter()
600        .filter(|(identity, chain)| identity.pid != own && chain_names(chain, handle))
601        .map(|(identity, _)| identity)
602        .collect()
603}
604
605fn chain_names(chain: &str, handle: &str) -> bool {
606    chain
607        .split(';')
608        .any(|entry| entry.rsplit('/').next() == Some(handle))
609}
610
611fn signal(pid: u32, signal: i32) {
612    if let Ok(pid) = i32::try_from(pid)
613        && pid > 0
614    {
615        // SAFETY: kill(2) takes plain integers and touches no memory of
616        // ours; a positive PID addresses exactly one process.
617        unsafe {
618            libc::kill(pid, signal);
619        }
620    }
621}
622
623/// Whether the process group still has a running member (zombies excluded on Linux).
624pub(crate) fn group_exists(pgid: u32) -> bool {
625    let Some(group) = ProcessGroup::new(pgid) else {
626        return false;
627    };
628    let signalable = group.is_signalable();
629    #[cfg(target_os = "linux")]
630    {
631        signalable
632            && linux::all_stats()
633                .iter()
634                .any(|info| info.pgid == pgid && info.state != 'Z')
635    }
636    #[cfg(not(target_os = "linux"))]
637    {
638        signalable
639    }
640}
641
642fn unix_now() -> u64 {
643    SystemTime::now()
644        .duration_since(UNIX_EPOCH)
645        .map_or(0, |elapsed| elapsed.as_secs())
646}
647
648fn write_record(dir: &Path, record: &DelegationRecord) -> std::io::Result<()> {
649    write_private_json(dir, &format!("{}.json", record.handle), record)
650}
651
652/// Atomically write `value` as `dir/name` with mode 0600, creating `dir` and
653/// keeping it and its parent (`run/`) private.
654pub(crate) fn write_private_json(
655    dir: &Path,
656    name: &str,
657    value: &impl Serialize,
658) -> std::io::Result<()> {
659    use std::os::unix::fs::PermissionsExt as _;
660    std::fs::create_dir_all(dir)?;
661    if let Some(run) = dir.parent() {
662        std::fs::set_permissions(run, std::fs::Permissions::from_mode(0o700))?;
663    }
664    std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))?;
665    let bytes = serde_json::to_vec_pretty(value).map_err(std::io::Error::other)?;
666    scv_client::fs::replace_private(&dir.join(name), &bytes)
667}
668
669fn read_record(path: &Path) -> Option<DelegationRecord> {
670    let bytes = match std::fs::File::open(path).and_then(|file| {
671        let mut bytes = Vec::new();
672        std::io::Read::read_to_end(&mut std::io::Read::take(file, MAX_RECORD_BYTES), &mut bytes)
673            .map(|_| bytes)
674    }) {
675        Ok(bytes) => bytes,
676        Err(error) => {
677            // A record removed between listing and reading is normal.
678            if error.kind() != std::io::ErrorKind::NotFound {
679                tracing::debug!(path = %path.display(), %error, "unreadable delegation record");
680            }
681            return None;
682        }
683    };
684    let record: DelegationRecord = match serde_json::from_slice(&bytes) {
685        Ok(record) => record,
686        Err(error) => {
687            tracing::debug!(path = %path.display(), %error, "malformed delegation record");
688            return None;
689        }
690    };
691    // Only a record named after its own handle is trusted.
692    if path.file_stem().and_then(|stem| stem.to_str()) == Some(record.handle.as_str()) {
693        Some(record)
694    } else {
695        tracing::debug!(path = %path.display(), "delegation record named for another handle");
696        None
697    }
698}
699
700fn remove_record(dir: &Path, handle: &str) {
701    let path = dir.join(format!("{handle}.json"));
702    if let Err(error) = std::fs::remove_file(&path)
703        && error.kind() != std::io::ErrorKind::NotFound
704    {
705        tracing::debug!(path = %path.display(), %error, "could not remove a delegation record");
706    }
707}
708
709/// Make this process the reaper of orphaned descendants (Linux), so processes
710/// a delegated agent leaves behind stay in SCV's process tree.
711pub fn become_child_subreaper() -> bool {
712    #[cfg(target_os = "linux")]
713    {
714        // SAFETY: PR_SET_CHILD_SUBREAPER takes integer arguments only and
715        // changes only this process's own reaping attribute.
716        unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) == 0 }
717    }
718    #[cfg(not(target_os = "linux"))]
719    {
720        false
721    }
722}
723
724static SPAWNED: Mutex<Option<HashSet<u32>>> = Mutex::new(None);
725
726/// Note a child this process spawned and will wait for itself.
727pub(crate) fn track_spawned(pid: u32) {
728    lock(&SPAWNED).get_or_insert_with(HashSet::new).insert(pid);
729}
730
731pub(crate) fn untrack_spawned(pid: u32) {
732    if let Some(spawned) = lock(&SPAWNED).as_mut() {
733        spawned.remove(&pid);
734    }
735}
736
737/// Collect exited orphans reparented to this subreaper. Children SCV spawned
738/// itself are left to their own waiters.
739pub fn reap_orphaned_zombies() -> usize {
740    #[cfg(target_os = "linux")]
741    {
742        let own = std::process::id();
743        let spawned = lock(&SPAWNED).clone().unwrap_or_default();
744        let uptime = linux::uptime_ticks();
745        let mut reaped = 0;
746        for info in linux::all_stats() {
747            if info.ppid != own || info.state != 'Z' || spawned.contains(&info.pid) {
748                continue;
749            }
750            let old_enough = uptime.is_some_and(|now| {
751                now.saturating_sub(info.start_time)
752                    >= ZOMBIE_MIN_AGE.as_secs() * linux::clock_ticks()
753            });
754            if !old_enough {
755                continue;
756            }
757            let mut status = 0;
758            // SAFETY: `status` is a live local that waitpid(2) writes one
759            // int into; WNOHANG keeps the call from blocking.
760            if unsafe { libc::waitpid(info.pid as i32, &raw mut status, libc::WNOHANG) }
761                == info.pid as i32
762            {
763                reaped += 1;
764            }
765        }
766        reaped
767    }
768    #[cfg(not(target_os = "linux"))]
769    {
770        0
771    }
772}
773
774/// Processes of interest at one moment: group membership and tags.
775struct ProcessTable {
776    groups: Vec<(ProcessIdentity, u32)>,
777    tagged: Vec<(ProcessIdentity, String)>,
778}
779
780impl ProcessTable {
781    fn members(&self, record: &DelegationRecord) -> HashSet<u32> {
782        let mut members: HashSet<u32> = self
783            .groups
784            .iter()
785            .filter(|(_, pgid)| *pgid == record.pgid)
786            .map(|(identity, _)| identity.pid)
787            .collect();
788        members.extend(
789            self.tagged
790                .iter()
791                .filter(|(_, chain)| chain_names(chain, &record.handle))
792                .map(|(identity, _)| identity.pid),
793        );
794        members
795    }
796
797    #[cfg(target_os = "linux")]
798    fn snapshot() -> Self {
799        let mut groups = Vec::new();
800        let mut tagged = Vec::new();
801        for info in linux::all_stats() {
802            if info.state == 'Z' {
803                continue;
804            }
805            let identity = ProcessIdentity {
806                pid: info.pid,
807                start_time: info.start_time,
808            };
809            groups.push((identity, info.pgid));
810            if let Some(chain) = linux::parent_chain(info.pid) {
811                tagged.push((identity, chain));
812            }
813        }
814        Self { groups, tagged }
815    }
816
817    #[cfg(not(target_os = "linux"))]
818    fn snapshot() -> Self {
819        let mut groups = Vec::new();
820        let mut tagged = Vec::new();
821        // `ps -E` appends each process's environment to its command line.
822        let Ok(output) = std::process::Command::new("ps")
823            .args(["-E", "-ww", "-axo", "pid=,pgid=,command="])
824            .output()
825        else {
826            return Self { groups, tagged };
827        };
828        for line in String::from_utf8_lossy(&output.stdout).lines() {
829            let mut fields = line.split_whitespace();
830            let (Some(pid), Some(pgid)) = (
831                fields.next().and_then(|value| value.parse::<u32>().ok()),
832                fields.next().and_then(|value| value.parse::<u32>().ok()),
833            ) else {
834                continue;
835            };
836            let Some(identity) = ProcessIdentity::of(pid) else {
837                continue;
838            };
839            groups.push((identity, pgid));
840            if let Some(chain) = fields.find_map(|field| {
841                field
842                    .strip_prefix(PARENT_VARIABLE)
843                    .and_then(|rest| rest.strip_prefix('='))
844            }) {
845                tagged.push((identity, chain.to_owned()));
846            }
847        }
848        Self { groups, tagged }
849    }
850}
851
852#[cfg(target_os = "linux")]
853fn process_start_time(pid: u32) -> Option<u64> {
854    linux::stat(pid).map(|info| info.start_time)
855}
856
857#[cfg(target_os = "macos")]
858fn process_start_time(pid: u32) -> Option<u64> {
859    // SAFETY: proc_bsdinfo is a plain C struct of integers and byte arrays,
860    // for which all-zero bytes are a valid value.
861    let mut info: libc::proc_bsdinfo = unsafe { std::mem::zeroed() };
862    let size = std::mem::size_of::<libc::proc_bsdinfo>() as i32;
863    // SAFETY: the buffer is `info` itself and `size` is its exact size, so
864    // proc_pidinfo(2) writes at most that many bytes into it.
865    let written = unsafe {
866        libc::proc_pidinfo(
867            pid as i32,
868            libc::PROC_PIDTBSDINFO,
869            0,
870            (&mut info as *mut libc::proc_bsdinfo).cast(),
871            size,
872        )
873    };
874    (written == size).then(|| info.pbi_start_tvsec * 1_000_000 + info.pbi_start_tvusec)
875}
876
877#[cfg(not(target_os = "linux"))]
878fn process_is_zombie(pid: u32) -> bool {
879    std::process::Command::new("ps")
880        .args(["-o", "stat=", "-p", &pid.to_string()])
881        .output()
882        .ok()
883        .filter(|output| output.status.success())
884        .is_some_and(|output| {
885            String::from_utf8_lossy(&output.stdout)
886                .trim_start()
887                .starts_with('Z')
888        })
889}
890
891#[cfg(not(any(target_os = "linux", target_os = "macos")))]
892fn process_start_time(pid: u32) -> Option<u64> {
893    // SAFETY: signal 0 only checks that the process exists; kill(2) takes
894    // plain integers and touches no memory of ours.
895    let alive = unsafe { libc::kill(pid as i32, 0) } == 0;
896    alive.then_some(0)
897}
898
899#[cfg(target_os = "linux")]
900mod linux {
901    pub(super) struct Stat {
902        pub(crate) pid: u32,
903        pub(crate) ppid: u32,
904        pub(crate) pgid: u32,
905        pub(crate) state: char,
906        pub(crate) start_time: u64,
907    }
908
909    pub(super) fn stat(pid: u32) -> Option<Stat> {
910        let text = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
911        // The command name is parenthesized and may contain spaces or ')'.
912        let rest = &text[text.rfind(')')? + 2..];
913        let fields: Vec<&str> = rest.split_whitespace().collect();
914        // After the name: state(3) ppid(4) pgrp(5) ... starttime(22).
915        Some(Stat {
916            pid,
917            state: fields.first()?.chars().next()?,
918            ppid: fields.get(1)?.parse().ok()?,
919            pgid: fields.get(2)?.parse().ok()?,
920            start_time: fields.get(19)?.parse().ok()?,
921        })
922    }
923
924    pub(super) fn all_stats() -> Vec<Stat> {
925        let Ok(entries) = std::fs::read_dir("/proc") else {
926            return Vec::new();
927        };
928        entries
929            .filter_map(Result::ok)
930            .filter_map(|entry| entry.file_name().to_str()?.parse::<u32>().ok())
931            .filter_map(stat)
932            .collect()
933    }
934
935    /// `SCV_PARENT` from a process's environment, when readable.
936    pub(super) fn parent_chain(pid: u32) -> Option<String> {
937        let environ = std::fs::read(format!("/proc/{pid}/environ")).ok()?;
938        let prefix = format!("{}=", super::PARENT_VARIABLE);
939        environ.split(|byte| *byte == 0).find_map(|entry| {
940            entry
941                .strip_prefix(prefix.as_bytes())
942                .map(|value| String::from_utf8_lossy(value).into_owned())
943        })
944    }
945
946    pub(super) fn clock_ticks() -> u64 {
947        // SAFETY: sysconf(3) only reads a system constant.
948        let ticks = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
949        u64::try_from(ticks)
950            .ok()
951            .filter(|ticks| *ticks > 0)
952            .unwrap_or(100)
953    }
954
955    pub(super) fn uptime_ticks() -> Option<u64> {
956        let text = std::fs::read_to_string("/proc/uptime").ok()?;
957        let seconds: f64 = text.split_whitespace().next()?.parse().ok()?;
958        Some((seconds * clock_ticks() as f64) as u64)
959    }
960}
961
962#[cfg(test)]
963mod tests;