import Foundation

#if canImport(Darwin)
import Darwin
#elseif canImport(Musl)
import Musl
#elseif canImport(Glibc)
import Glibc
#endif

// MARK: - ProcessHost contract (ADAPTERS §0.1)

public enum StdinMode: Sendable {
    case pipe
    case closed
}

public struct ProcessSpec: Sendable {
    public let executable: String
    public let args: [String]
    public let cwd: String
    public let env: [String: String]
    public let stdinMode: StdinMode
    /// Keys to strip from the *inherited* app environment before the child launches. A launched
    /// child normally inherits Nucleic's own process environment; this lets a caller guarantee a
    /// credential the app happened to inherit (e.g. an ambient `ANTHROPIC_API_KEY`) never reaches a
    /// subprocess. `env` still wins over the inherited value; removal applies after that merge.
    public let removeFromEnvironment: Set<String>

    public init(
        executable: String, args: [String], cwd: String,
        env: [String: String] = [:], stdinMode: StdinMode = .pipe,
        removeFromEnvironment: Set<String> = []
    ) {
        self.executable = executable
        self.args = args
        self.cwd = cwd
        self.env = env
        self.stdinMode = stdinMode
        self.removeFromEnvironment = removeFromEnvironment
    }
}

public protocol ProcessHandle: Sendable {
    /// NDJSON-friendly: one line per element, newline stripped, partial trailing
    /// bytes held across pipe reads.
    var stdoutLines: AsyncThrowingStream<Data, Error> { get }
    var stderrLines: AsyncThrowingStream<Data, Error> { get }
    var processID: Int32 { get }
    func writeLine(_ data: Data) throws
    func closeStdin()
    func sendSignal(_ sig: Int32)
    func wait() async -> Int32
    /// Finish the stdout/stderr line streams immediately, independent of the process's exit, so a
    /// consumer blocked on `for await stdoutLines` is released. Only meaningful when the streams can
    /// outlive a killed process (the containerized handle, whose stream closure rides a guest RPC
    /// that can stall); the host pipe handle EOFs reliably on exit, so its default is a no-op.
    func forceCloseStreams()

    /// Monotonic uptime-nanoseconds (`DispatchTime`) of the most recent stdout/stderr chunk this
    /// process produced — or of launch, before any output. Drives ``ProcessStallMonitor``: a handle
    /// whose output has gone silent is a hang candidate. Default: "just now", so a handle that
    /// doesn't track activity never looks stalled.
    var lastActivityNanos: UInt64 { get }

    /// Whether ``processID`` is a real host pid whose CPU time can be sampled from this machine (a
    /// host ``ChildProcess``). False for guest/VM handles, whose `processID` is a guest pid or a
    /// socket fd — for those ``ProcessStallMonitor`` falls back to the output-idle signal alone.
    var supportsHostCPUSampling: Bool { get }
}

extension ProcessHandle {
    /// Default: nothing to force. The host `ChildProcess` reads OS pipes that EOF the moment the
    /// process dies, so the line streams always finish on their own. ``ContainerizedProcessHandle``
    /// overrides this because its streams close only when a guest wait RPC resolves.
    public func forceCloseStreams() {}

    /// Default: treat the handle as perpetually active, so a handle that opts out of activity
    /// tracking is never mistaken for a hang. The concrete host/container/VM handles override this.
    public var lastActivityNanos: UInt64 { DispatchTime.now().uptimeNanoseconds }

    /// Default: not host-CPU-samplable. Only the host ``ChildProcess`` overrides this to `true`.
    public var supportsHostCPUSampling: Bool { false }
}

/// A thread-safe "last time output was seen" marker for one process, sampled by
/// ``ProcessStallMonitor`` to tell a busy-but-quiet command from a wedged one. Backed by
/// `DispatchTime` uptime nanos (monotonic — immune to wall-clock jumps). Bumped from the
/// stdout/stderr chunk sinks of each ``ProcessHandle`` implementation.
final class ActivityClock: @unchecked Sendable {
    private let lock = NSLock()
    private var nanos: UInt64 = DispatchTime.now().uptimeNanoseconds
    func bump() {
        let now = DispatchTime.now().uptimeNanoseconds
        lock.lock(); nanos = now; lock.unlock()
    }
    var lastNanos: UInt64 { lock.lock(); defer { lock.unlock() }; return nanos }
}

/// One-shot claim guard: the first caller to `claim()` wins, the rest get `false`. Lets two racing
/// tasks (exit vs. timeout) resume a single `CheckedContinuation` exactly once.
private final class ResumeOnceFlag: @unchecked Sendable {
    private let lock = NSLock()
    private var claimed = false
    func claim() -> Bool {
        lock.lock(); defer { lock.unlock() }
        if claimed { return false }
        claimed = true
        return true
    }
}

extension ProcessHandle {
    /// Await the process's exit, but give up after `nanos`. Returns `true` iff it exited in time.
    /// On timeout the background wait is left running (harmless — it resolves once the process
    /// finally dies, e.g. after a follow-up SIGKILL), so this never blocks longer than `nanos`.
    /// Used to bound every Stop/shutdown so a wedged agent can't hang the actor on `wait()`.
    func waitForExit(within nanos: UInt64) async -> Bool {
        let once = ResumeOnceFlag()
        return await withCheckedContinuation { (cont: CheckedContinuation<Bool, Never>) in
            Task { _ = await self.wait(); if once.claim() { cont.resume(returning: true) } }
            Task {
                try? await Task.sleep(nanoseconds: nanos)
                if once.claim() { cont.resume(returning: false) }
            }
        }
    }

    /// Stop the process for good and return once it has actually exited: SIGTERM, then — if it
    /// hasn't gone within `graceNanos` — SIGKILL. For a containerized handle the signals reach the
    /// agent's whole process group. Bounding the wait is what lets a stopped/ended session settle
    /// instead of spinning forever on an agent that ignores the polite signal.
    func terminate(graceNanos: UInt64 = 3 * 1_000_000_000) async {
        sendSignal(SIGTERM)
        if await waitForExit(within: graceNanos) { return }
        sendSignal(SIGKILL)
        // SIGKILL is uncatchable, so the process exits and `wait()` resolves; still bound it so a
        // broken wait can never hang teardown.
        if await waitForExit(within: graceNanos) { return }
        // Killed, but `wait()` still hasn't resolved — the guest wait/stdio RPC is wedged. Finish the
        // line streams ourselves so a consumer blocked on `stdoutLines` is released and teardown ends.
        forceCloseStreams()
    }

    /// Like ``terminate(graceNanos:)`` but reaps the process's ENTIRE descendant tree, not just the
    /// direct child. This is what a host `host_exec` build/test command needs: the login shell we
    /// spawn (`/bin/zsh -lc …`) forks a deep tree — `make → swift → swiftpm-testing-helper →
    /// *.xctest → fake-* / sh`. Signaling only the shell's pid (all ``sendSignal`` can do) orphans
    /// that tree; the orphans keep holding the shared trunk's SwiftPM `.build` lock and wedge every
    /// later build/test until they're killed by hand — the exact "host processes need to be
    /// terminated to continue" failure this exists to prevent.
    ///
    /// `Foundation.Process` puts the child in the app's OWN process group, so a `killpg` would kill
    /// the app too; there's no supported way to make the child a group leader. A snapshot tree-walk is
    /// the portable equivalent: capture the descendants BEFORE any signal (once the shell exits its
    /// children reparent to launchd/init and the ancestry link is lost, but pids stay valid across
    /// reparenting), then escalate SIGTERM → SIGKILL over the captured set. Accurate for the idle
    /// (hung/aborted) trees this cleans up — nothing is forking new grandchildren at that point. On a
    /// platform where the process table can't be read, it degrades to ``terminate(graceNanos:)``.
    func terminateTree(graceNanos: UInt64 = 3 * 1_000_000_000) async {
        let root = processID
        guard root > 0 else { await terminate(graceNanos: graceNanos); return }
        // Snapshot first — root + every transitive descendant — while the links still exist.
        let tree = [root] + hostDescendantPIDs(of: root)
        for pid in tree where pid > 0 { kill(pid, SIGTERM) }
        if await waitForExit(within: graceNanos) {
            // Root is gone; sweep any descendant that outlived it (a build child mid-shutdown).
            for pid in tree where pid > 0 && pid != root { kill(pid, SIGKILL) }
            return
        }
        // Root ignored SIGTERM — SIGKILL the whole captured set (uncatchable).
        for pid in tree where pid > 0 { kill(pid, SIGKILL) }
        if await waitForExit(within: graceNanos) {
            for pid in tree where pid > 0 && pid != root { kill(pid, SIGKILL) }
            return
        }
        // Even the kill didn't settle `wait()` — release any consumer blocked on the line streams.
        forceCloseStreams()
    }

    /// Best-effort escalation for an interactive Stop: SIGINT now (let the agent end its turn
    /// cleanly / emit an interrupted result), then SIGKILL the process group if it's still alive
    /// after `graceNanos`. Detached so the caller's `interrupt()` stays non-blocking; the run loop
    /// settles when the process finally exits. A responsive agent exits on SIGINT and the escalation
    /// no-ops; a wedged one is force-killed so Stop always takes effect.
    func interruptThenKill(graceNanos: UInt64 = 5 * 1_000_000_000) {
        sendSignal(SIGINT)
        Task { [self] in
            if await waitForExit(within: graceNanos) { return }
            sendSignal(SIGKILL)
            // If even the kill doesn't settle `wait()` (wedged guest RPC), release any consumer
            // blocked on the line streams so Stop always takes effect.
            if await waitForExit(within: graceNanos) { return }
            forceCloseStreams()
        }
    }

    /// Safety net behind a *cooperative* cancel (e.g. an `turn/interrupt` / `session/cancel` RPC the
    /// agent might ignore): force-SIGKILL the process group if it's still alive after `graceNanos`,
    /// without sending an upfront signal that could disrupt a clean cancel in flight. Detached and a
    /// no-op if the process exits on its own — so a well-behaved agent stops gracefully and a wedged
    /// one is still guaranteed to stop.
    func killGroupAfter(graceNanos: UInt64) {
        Task { [self] in
            if await waitForExit(within: graceNanos) { return }
            sendSignal(SIGKILL)
            if await waitForExit(within: graceNanos) { return }
            forceCloseStreams()
        }
    }
}

// MARK: - Executable resolution (shared by every host spawn site)

/// What to hand `Process` in order to run `executable` with `args`, resolving a bare name the way
/// the platform can. `nil` means the name is not on `path` — Windows only.
///
/// This exists because there are three host spawn sites (``ChildProcess``, `GitRunner.run`,
/// `GitRunner.runRedirected`) and they must not each invent their own answer. They did, briefly:
/// item 9 taught `ChildProcess` to search PATH on Windows and left `GitRunner`'s two copies
/// pointing at `/usr/bin/env`, so every `git` call on Windows failed with "The file doesn't
/// exist" — 13 Carbon tests, one missed call site.
///
/// - POSIX keeps delegating to `/usr/bin/env`, which does the PATH search itself (and reports its
///   own exit 127 when it fails), so this never returns nil there.
/// - Windows has no `env`, and `Process` resolves a relative `executableURL` against the *current
///   directory* — a bare `git` would try to spawn `<cwd>\git`. So the search happens here, over
///   PATH × PATHEXT.
func resolvedSpawnTarget(
    executable: String, args: [String], path: String?
) -> (executableURL: URL, arguments: [String])? {
    #if os(Windows)
    // Either separator counts as "already a path": a Windows absolute path (`C:\…\git.exe`)
    // contains no forward slash, and a caller may well hand us either spelling.
    if executable.contains("/") || executable.contains("\\") {
        return (URL(fileURLWithPath: executable), args)
    }
    guard let resolved = windowsResolveExecutable(executable, path: path) else { return nil }
    return (URL(fileURLWithPath: resolved), args)
    #else
    _ = path
    if executable.contains("/") {
        return (URL(fileURLWithPath: executable), args)
    }
    return (URL(fileURLWithPath: "/usr/bin/env"), [executable] + args)
    #endif
}

/// Spawns and monitors child processes with line-buffered stdio.
public actor ProcessHost {
    public init() {}

    public func launch(_ spec: ProcessSpec) throws -> any ProcessHandle {
        try ChildProcess(spec: spec)
    }
}

// MARK: - Implementation

/// All mutable state is confined: Process control calls (signal/terminate) are
/// thread-safe, stdin writes are serialized behind a lock, and the line streams
/// are produced by dedicated reader tasks.
final class ChildProcess: ProcessHandle, @unchecked Sendable {
    private let process: Process
    private let stdinHandle: FileHandle?
    private let stdinLock = NSLock()
    private var stdinClosed = false
    #if os(Windows)
    /// The child's Job Object — assigned right after launch, released when this handle dies
    /// (which kills any survivor). Nil only if the job couldn't be created, in which case
    /// termination degrades to the single-pid path POSIX uses.
    private var job: WindowsJobObject?
    #endif

    let stdoutLines: AsyncThrowingStream<Data, Error>
    let stderrLines: AsyncThrowingStream<Data, Error>
    private let exitTask: Task<Int32, Never>
    /// Bumped on every stdout/stderr chunk (both pipes share it) so ``ProcessStallMonitor`` can tell
    /// how long this command has been silent.
    private let activity = ActivityClock()

    var processID: Int32 { process.processIdentifier }
    var lastActivityNanos: UInt64 { activity.lastNanos }
    /// A host `ChildProcess`'s `processID` is a real host pid, so its (and its descendants') CPU time
    /// can be sampled — the second stall signal alongside output-idle.
    var supportsHostCPUSampling: Bool { true }

    init(spec: ProcessSpec) throws {
        let process = Process()
        // Resolve against the SAME widened PATH the child will receive (below) — otherwise a `git`
        // that only the augmented PATH can see would be declared missing here.
        let searchPath = LoginShellPATH.augment(
            ProcessInfo.processInfo.environment.merging(spec.env) { _, new in new })["PATH"]
        guard let target = resolvedSpawnTarget(
            executable: spec.executable, args: spec.args, path: searchPath)
        else {
            throw BackendError.spawnFailed("\(spec.executable): not found on PATH")
        }
        process.executableURL = target.executableURL
        process.arguments = target.arguments
        process.currentDirectoryURL = URL(fileURLWithPath: spec.cwd)
        // Widen PATH to the user's login-shell PATH (+ well-known install dirs) before spawning.
        // A GUI-launched app inherits launchd's minimal PATH, which omits where `claude`/`codex`/
        // `grok` install — so resolving a bare name via `/usr/bin/env` would fail with
        // `env: <name>: No such file or directory` (exit 127). See ``LoginShellPATH``.
        var childEnv = ProcessInfo.processInfo.environment.merging(spec.env) { _, new in new }
        // Underlay the user's login-shell env delta (Homebrew/version-manager/toolchain exports)
        // beneath the composed env — the non-PATH half of what `zsh -lc` used to re-source per
        // command. Additive only (explicit spec/app values and the Managed Git pins always win),
        // gated by `nucleic.host.loginEnvSnapshot`. See ``LoginShellEnv`` (docs/NASH.md §7.7.2).
        // The spec's removals run AFTER the underlay so the snapshot can never re-introduce a key
        // the spawn site asked to drop (its own credential filter already ran at snapshot time).
        childEnv = LoginShellEnv.inject(into: childEnv)
        for key in spec.removeFromEnvironment { childEnv.removeValue(forKey: key) }
        process.environment = LoginShellPATH.augment(childEnv)

        let stdoutPipe = Pipe()
        let stderrPipe = Pipe()
        process.standardOutput = stdoutPipe
        process.standardError = stderrPipe

        switch spec.stdinMode {
        case .pipe:
            let stdinPipe = Pipe()
            process.standardInput = stdinPipe
            stdinHandle = stdinPipe.fileHandleForWriting
        case .closed:
            process.standardInput = FileHandle.nullDevice
            stdinHandle = nil
        }

        let exitStream = AsyncStream<Int32> { continuation in
            process.terminationHandler = { p in
                continuation.yield(p.terminationStatus)
                continuation.finish()
            }
        }
        exitTask = Task {
            for await code in exitStream { return code }
            return -1
        }

        stdoutLines = Self.lineStream(from: stdoutPipe.fileHandleForReading, activity: activity)
        stderrLines = Self.lineStream(from: stderrPipe.fileHandleForReading, activity: activity)

        do {
            try process.run()
        } catch {
            throw BackendError.spawnFailed("\(spec.executable): \(error.localizedDescription)")
        }
        self.process = process
        #if os(Windows)
        // Bind the child (and everything it goes on to spawn) to a Job Object, so SIGKILL can
        // reap the whole tree and so nothing survives this handle being released — the Windows
        // stand-in for the POSIX process-group semantics the shared code assumes. See
        // `Windows/ProcessTree.swift`.
        job = WindowsJobObject(attaching: process.processIdentifier)
        #endif
    }

    private static func lineStream(
        from handle: FileHandle, activity: ActivityClock
    ) -> AsyncThrowingStream<Data, Error> {
        AsyncThrowingStream { continuation in
            // Chunked reads via `readabilityHandler` rather than `FileHandle.bytes`:
            // the byte-by-byte `AsyncBytes` sequence terminates early / drops data on
            // large bursty CLI output, truncating a big tool-result line and losing
            // everything the agent streamed after it (the final answer + result event).
            let splitter = LineSplitter()
            handle.readabilityHandler = { fh in
                let chunk = fh.availableData
                if chunk.isEmpty {  // EOF
                    fh.readabilityHandler = nil
                    if let rest = splitter.drain() { continuation.yield(rest) }
                    continuation.finish()
                    return
                }
                // Any output — even a chunk with no complete line yet — is a sign of life for the
                // stall monitor.
                activity.bump()
                for line in splitter.push(chunk) { continuation.yield(line) }
            }
            continuation.onTermination = { _ in handle.readabilityHandler = nil }
        }
    }

    func writeLine(_ data: Data) throws {
        stdinLock.lock()
        defer { stdinLock.unlock() }
        guard let stdinHandle, !stdinClosed else { throw BackendError.notRunning }
        var framed = data
        framed.append(0x0A)
        try stdinHandle.write(contentsOf: framed)
    }

    func closeStdin() {
        stdinLock.lock()
        defer { stdinLock.unlock() }
        guard !stdinClosed else { return }
        stdinClosed = true
        try? stdinHandle?.close()
    }

    func sendSignal(_ sig: Int32) {
        let pid = process.processIdentifier
        guard pid > 0 else { return }
        #if os(Windows)
        // A tree-wide kill is the whole point of the job; fall through to the single-pid
        // `TerminateProcess` only when there is no job.
        if sig == SIGKILL, let job {
            job.terminate()
            return
        }
        #endif
        kill(pid, sig)
    }

    func wait() async -> Int32 {
        await exitTask.value
    }
}

/// Accumulates raw stdout/stderr bytes and splits them into newline-delimited lines,
/// holding any partial trailing line across chunk boundaries. `readabilityHandler`
/// fires serially, but it's locked anyway so a final drain can't race the last chunk.
///
/// Module-internal (not file-private) so the containerization adapter
/// (`ContainerizedProcessHandle`) can reuse the exact same NDJSON line framing the host
/// `ChildProcess` uses — guest stdio arrives as raw byte chunks over vsock and must be split
/// identically for the backend decode loop to behave the same in-container as on the host.
final class LineSplitter: @unchecked Sendable {
    private let lock = NSLock()
    private var buffer = Data()
    /// Number of bytes at the front of `buffer` already inspected for a newline. A native
    /// stream-json event can be several megabytes when a Write tool carries a large file, and
    /// pipe reads deliver that one line in many small chunks. Restarting the search at byte zero
    /// for every chunk makes framing quadratic in the tool input's length and can wedge the decode
    /// loop even though no bytes were lost.
    private var scannedCount = 0

    /// Append a chunk; return every complete line it produced (newline stripped, CR-trimmed).
    func push(_ chunk: Data) -> [Data] {
        lock.lock(); defer { lock.unlock() }
        buffer.append(chunk)
        var lines: [Data] = []
        var start = buffer.startIndex
        // Everything before `scannedCount` was proven newline-free by an earlier push. Begin at
        // the first newly appended byte; after finding a newline, scan the remainder normally.
        var searchStart = buffer.index(
            buffer.startIndex, offsetBy: min(scannedCount, buffer.count))
        while searchStart < buffer.endIndex,
              let newline = buffer[searchStart...].firstIndex(of: 0x0A) {
            var line = Data(buffer[start..<newline])
            if line.last == 0x0D { line.removeLast() }
            lines.append(line)
            start = buffer.index(after: newline)
            searchStart = start
        }
        // Rebase the unconsumed remainder so indices stay zero-based.
        buffer = start == buffer.startIndex ? buffer : Data(buffer[start...])
        // The entire remainder has now been searched. The next push only needs to inspect bytes
        // appended after it, making total scan work linear in the stream size.
        scannedCount = buffer.count
        return lines
    }

    /// Any trailing bytes with no terminating newline (flushed at EOF).
    func drain() -> Data? {
        lock.lock(); defer { lock.unlock() }
        guard !buffer.isEmpty else { return nil }
        defer {
            buffer.removeAll()
            scannedCount = 0
        }
        return buffer
    }
}

// MARK: - Process-tree enumeration (host)

/// The transitive descendant pids of `root`, from a single snapshot of the host process table.
/// Returns just the descendants (NOT `root`). Empty when `root` has no children or the table can't
/// be read — callers degrade to signaling `root` alone. Best-effort by design: a hung build tree is
/// idle, so one snapshot captures it faithfully; a live-forking tree could miss a just-spawned
/// grandchild. Used by ``ProcessHandle/terminateTree(graceNanos:)``.
func hostDescendantPIDs(of root: pid_t) -> [pid_t] {
    let table = hostProcessTable()
    guard !table.isEmpty else { return [] }
    var childrenByParent: [pid_t: [pid_t]] = [:]
    for entry in table { childrenByParent[entry.ppid, default: []].append(entry.pid) }
    var result: [pid_t] = []
    var visited: Set<pid_t> = [root]  // guards against pid-reuse cycles in the snapshot
    var queue = childrenByParent[root] ?? []
    var i = 0
    while i < queue.count {
        let pid = queue[i]; i += 1
        guard visited.insert(pid).inserted else { continue }
        result.append(pid)
        if let kids = childrenByParent[pid] { queue.append(contentsOf: kids) }
    }
    return result
}

#if canImport(Darwin)
/// (pid, ppid) for every process, via `sysctl(KERN_PROC_ALL)`. Over-allocates a little slack so the
/// table growing between the size probe and the fetch doesn't ENOMEM; on any error returns [].
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] {
    var mib: [Int32] = [CTL_KERN, KERN_PROC, KERN_PROC_ALL, 0]
    var size = 0
    if sysctl(&mib, u_int(mib.count), nil, &size, nil, 0) != 0 || size == 0 { return [] }
    let stride = MemoryLayout<kinfo_proc>.stride
    let capacity = size / stride + 32  // slack for processes spawned since the probe
    var procs = [kinfo_proc](repeating: kinfo_proc(), count: capacity)
    var byteLen = capacity * stride
    let rc = procs.withUnsafeMutableBytes { buf in
        sysctl(&mib, u_int(mib.count), buf.baseAddress, &byteLen, nil, 0)
    }
    if rc != 0 { return [] }
    let count = byteLen / stride
    return (0..<count).map { i in
        (pid: procs[i].kp_proc.p_pid, ppid: procs[i].kp_eproc.e_ppid)
    }
}
#elseif os(Linux)
/// (pid, ppid) for every process, from `/proc/<pid>/stat`. The `comm` field (field 2) can contain
/// spaces and parentheses, so ppid is read as the second whitespace token AFTER the final ')'.
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] {
    guard let entries = try? FileManager.default.contentsOfDirectory(atPath: "/proc") else { return [] }
    var table: [(pid: pid_t, ppid: pid_t)] = []
    for entry in entries {
        guard let pid = pid_t(entry),
            let stat = try? String(contentsOfFile: "/proc/\(entry)/stat", encoding: .utf8),
            let close = stat.lastIndex(of: ")")
        else { continue }
        let fields = stat[stat.index(after: close)...]
            .split(separator: " ", omittingEmptySubsequences: true)
        // fields[0] = state, fields[1] = ppid
        guard fields.count >= 2, let ppid = pid_t(fields[1]) else { continue }
        table.append((pid: pid, ppid: ppid))
    }
    return table
}
#elseif os(Windows)
/// (pid, ppid) for every process, from a Toolhelp snapshot — see `Windows/ProcessTree.swift`.
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] { windowsProcessTable() }
#else
private func hostProcessTable() -> [(pid: pid_t, ppid: pid_t)] { [] }
#endif

// MARK: - Process-tree CPU sampling (host)

/// Cumulative CPU time (user + system) in nanoseconds for `root` plus every transitive descendant,
/// from a single process-table snapshot. `0` when the tree can't be read or `root` is invalid.
/// ``ProcessStallMonitor`` diffs consecutive samples to get a "% of one core" over the interval —
/// the signal that tells a wedged (idle-CPU) command apart from one that's quietly working hard.
/// Reuses ``hostDescendantPIDs(of:)`` so it covers the whole `zsh → make → swift → xctest` fork
/// tree, not just the shell.
func hostProcessTreeCPUNanos(of root: pid_t) -> UInt64 {
    guard root > 0 else { return 0 }
    var total = hostProcessCPUNanos(root)
    for pid in hostDescendantPIDs(of: root) where pid > 0 { total &+= hostProcessCPUNanos(pid) }
    return total
}

#if canImport(Darwin)
/// One process's cumulative user+system CPU nanoseconds via `proc_pid_rusage`. `0` on any error
/// (dead pid, permission) — a dead child just contributes nothing to the tree total.
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 {
    var info = rusage_info_v4()
    let rc = withUnsafeMutablePointer(to: &info) { ptr in
        ptr.withMemoryRebound(to: rusage_info_t?.self, capacity: 1) { reb in
            proc_pid_rusage(pid, RUSAGE_INFO_V4, reb)
        }
    }
    guard rc == 0 else { return 0 }
    // `ri_user_time` / `ri_system_time` are already in nanoseconds.
    return info.ri_user_time &+ info.ri_system_time
}
#elseif os(Linux)
/// One process's cumulative user+system CPU nanoseconds from `/proc/<pid>/stat` (utime+stime, in
/// clock ticks → nanos via `_SC_CLK_TCK`). `0` on any error.
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 {
    guard let stat = try? String(contentsOfFile: "/proc/\(pid)/stat", encoding: .utf8),
        let close = stat.lastIndex(of: ")")
    else { return 0 }
    // After the final ')', the first token is field 3 (state); utime is field 14 → index 11,
    // stime is field 15 → index 12 (mirrors the ppid-at-index-1 arithmetic in `hostProcessTable`).
    let fields = stat[stat.index(after: close)...]
        .split(separator: " ", omittingEmptySubsequences: true)
    guard fields.count > 12, let utime = UInt64(fields[11]), let stime = UInt64(fields[12])
    else { return 0 }
    let hz = UInt64(max(sysconf(Int32(_SC_CLK_TCK)), 1))
    return (utime &+ stime) &* 1_000_000_000 / hz
}
#elseif os(Windows)
/// One process's cumulative user+kernel CPU nanoseconds via `GetProcessTimes`.
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 { windowsProcessCPUNanos(pid) }
#else
private func hostProcessCPUNanos(_ pid: pid_t) -> UInt64 { 0 }
#endif

// MARK: - Process-tree disk-I/O sampling (host)

/// Cumulative disk-I/O bytes (read + written) for `root` plus every transitive descendant, from a
/// single process-table snapshot. `0` when the tree can't be read or `root` is invalid.
/// ``ProcessStallMonitor`` diffs consecutive samples to tell a wedged command apart from one that is
/// quietly *moving data* at near-zero CPU — a `cctl images push` streaming blobs to a registry, a
/// slow `cp` — the exact case CPU sampling alone can't exonerate. Reuses ``hostDescendantPIDs(of:)``
/// so it covers the whole fork tree, matching the CPU sampler.
func hostProcessTreeIOBytes(of root: pid_t) -> UInt64 {
    guard root > 0 else { return 0 }
    var total = hostProcessIOBytes(root)
    for pid in hostDescendantPIDs(of: root) where pid > 0 { total &+= hostProcessIOBytes(pid) }
    return total
}

#if canImport(Darwin)
/// One process's cumulative disk bytes (read + written) via `proc_pid_rusage` — the same call the CPU
/// sampler already makes. `0` on any error (dead pid, permission).
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 {
    var info = rusage_info_v4()
    let rc = withUnsafeMutablePointer(to: &info) { ptr in
        ptr.withMemoryRebound(to: rusage_info_t?.self, capacity: 1) { reb in
            proc_pid_rusage(pid, RUSAGE_INFO_V4, reb)
        }
    }
    guard rc == 0 else { return 0 }
    return info.ri_diskio_bytesread &+ info.ri_diskio_byteswritten
}
#elseif os(Linux)
/// One process's cumulative I/O bytes from `/proc/<pid>/io` (`rchar` + `wchar` — the bytes the
/// process caused to be read/written, the closest analogue to Darwin's disk counter). `0` on any
/// error (the file is readable only by the process owner).
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 {
    guard let io = try? String(contentsOfFile: "/proc/\(pid)/io", encoding: .utf8) else { return 0 }
    var total: UInt64 = 0
    for line in io.split(separator: "\n") where line.hasPrefix("rchar:") || line.hasPrefix("wchar:") {
        total &+= line.split(separator: " ").last.flatMap { UInt64($0) } ?? 0
    }
    return total
}
#elseif os(Windows)
/// One process's cumulative I/O bytes via `GetProcessIoCounters`.
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 { windowsProcessIOBytes(pid) }
#else
private func hostProcessIOBytes(_ pid: pid_t) -> UInt64 { 0 }
#endif
