#!/usr/bin/bash
# SPDX-License-Identifier: AGPL-3.0-only
# /usr/local/libexec/ai-tools/ai-tools-stop
# Stops running agent sessions and everything they spawned. This is the stop rung of the incident ladder: the one
# control that acts on a session ALREADY RUNNING, where every other operator control (unclaim, disable a provider,
# revoke an operator) only changes what the NEXT launch gets.
#
# THE PROPERTY THIS FILE EXISTS TO HOLD: a stop that is asked for and reported as done HAS HAPPENED. The design
# that follows from it -- why sessions are found by CGROUP rather than by process tree, why it does not take a target,
# where containment ends, and the residual failure modes -- is documented once, in stop.rule.md. This header states only
# what a reader of THIS FILE needs; each function carries its own local mechanism.
#
# ── Two inverted conventions, stated here so they are not "fixed" back ───────────────────────
# For every other component in this project the safe direction is DON'T ACT. For this one it is ACT, and two
# project-wide conventions invert for that single reason (stop.rule.md, ref-section-e8k5):
#
#   1. NO REQUIRED DEPENDENCIES, and deliberately NO `set -e`. A missing library that aborted the
#      run, or an unexpected non-zero that abandoned a half-finished kill, would be a stop that did
#      not happen. `set -u` IS used, and it ends a run just as abruptly wherever a name or an
#      argument is read unset -- so a value a caller may legitimately not have passed is defaulted
#      where it is read (confirm_stop). Every library loads behind an inline fallback; the kill
#      path -- enumerate, signal, verify -- runs on bash builtins plus /proc and /sys reads, with
#      `sleep` its only external. That is independence from THIS PROJECT's libraries, not from the
#      base system: `id`, `date`, `logger`, `timeout`, `systemctl` and `sudo` are each outside the
#      kill path or best-effort within it, and the two that can BLOCK -- the attribution calls into
#      the sandbox account's user manager -- run under `timeout`, since a stop that hangs is a stop
#      that did not happen. NO project library is load-bearing here: this helper does not take any
#      input deciding WHICH sessions to stop, so there is no input left for one to gate.
#   2. THE CONFIRMATION DEFAULTS TO YES (messaging.rule.md requires NO). A pipe, a cron run, an
#      absent msg.lib.sh and a bare Enter all proceed; only a deliberate `n` declines. `--dry-run`
#      is how this command is looked at without acting.
#
# ── The mechanism, in one paragraph ──────────────────────────────────────────────────────────
# Sessions are enumerated by walking the sandbox account's per-user cgroup slice: a cgroup is the one container
# a spawned process cannot fall out of (inherited across fork(), surviving setsid(2) and the double fork). A systemd
# UNIT cgroup is the unit of work. The graceful pass is SIGTERM, deepest-first, re-collected each second; the kill pass
# writes `cgroup.kill` (atomic, 5.14+) with a start-time-validated per-pid loop as the older-kernel fallback. Liveness
# is read from the kernel -- `cgroup.events`, `cgroup.procs`, /proc -- and never from systemd, which is used for exactly
# one thing: reading a unit's WorkingDirectory to attribute a session to a project, which is best-effort, is REPORTED
# rather than acted on, and never decides whether something is running or whether it is stopped. Every liveness read
# fails CLOSED. Nothing is spared -- the account's own user manager included -- and it is put back afterwards
# (restore_user_manager) rather than exempted.
#
# ── Why root, and what this command accepts ──────────────────────────────────────────────────
# Signalling the sandbox account's cgroups and writing cgroup.kill is root's to do. The %ai-ops sudoers rule grants this
# helper NOPASSWD in its zero-argument form alone (the drop-in's trailing ""), so `ai-tools stop` runs without a prompt,
# and a flagged form (`ai-tools stop --dry-run`) meets sudo's ordinary password prompt, as a call
# to `ai-tools-{lockdown,reclaim,audit}` does.
#
# IT TAKES NO TARGET AND NO AUTHORIZATION INPUT. What is stopped is decided by one fact a session cannot influence --
# membership of the account's cgroup slice -- and everything this file reads from the account being stopped (a unit's
# WorkingDirectory, its name) is kept away from that decision and used only to label a row. Why neither of them can
# carry a per-project form -- each is written by the account being stopped -- is in stop.rule.md (ref-section-r5r9).
#
# The routine way to FINISH a session is `/exit` inside it, which lets it run its own session-end handback. This command
# TERMINATES instead: it kills the process tree, so no handback runs and the last turn's writes may still be
# sandbox-owned (which is why a run names the reclaim per project).
#
# Usage:  ai-tools-stop [--dry-run] [-y|--yes] [--force] [--all]
#
# `--all` is accepted and inert. A PATH is refused (exit 2) rather than ignored -- see
# refuse_positional_argument.
#
# WHAT A SUCCESSFUL EXIT MEANS, stated exactly rather than generously. Exit 0 means: every session that existed
# at ENUMERATION was stopped, and a final re-enumeration found no process still live. It does NOT mean no session can
# exist afterwards -- the launch/stop window is a stated residual. It does not describe the user manager,
# whose restoration is reported separately and never folded into this status.
#
# Exit:   0 stopped and verified gone (or no session was running)
#         1 something survived SIGKILL -- the only outcome that is not a stop
#         2 usage (an unknown option, or a path -- this command does not take a target)
#         5 this helper could not run (no cgroup2 hierarchy, no sandbox uid) -- distinct from 1,
#           so a caller can tell a broken tool from a surviving process
#         6 declined at the confirmation (a deliberate `n`; never a degraded path) -- the code
#           ai-tools(1) reserves for an operator's explicit decline, spelled here rather than read
#           from a library, since no project library is load-bearing in this helper
#
# Installed 750 root:root, so only root runs it. Its domain rule is stop.rule.md.

# NOT `set -e`: see inverted convention 1. An unexpected non-zero must never abandon a half-finished kill.
set -uo pipefail

# A fixed PATH, set before anything is resolved. This helper runs as root and is reachable directly as well
# as through the CLI, so it must not resolve `sleep`, `id` or `realpath` through a PATH an invoker chose. sudoers
# `secure_path` normally covers the sudo route; this covers the direct one too, at no cost.
PATH=/usr/sbin:/usr/bin:/sbin:/bin
export PATH

# The cgroup walk must see EVERY child directory, and `*/` alone does not: a name beginning with a dot is skipped
# by default globbing. Every name inside the delegated subtree is the DELEGATEE's to choose (see the delegation note),
# so without `dotglob` a session could place itself in a cgroup called `.hidden` and drop out of the enumeration --
# including under `--all`, the form that must hold against a hostile session. `nullglob` makes a childless cgroup expand
# to an empty list rather than the unexpanded pattern. Set once, at file scope: every walk here depends on it.
shopt -s dotglob nullglob

readonly SANDBOX_USER='ai-tools'

# How long the graceful pass gets before the kill. Short and owned here rather than left to a unit's TimeoutStopSec (90s
# by default): an operator waiting that long per session will reach for kill -9 by hand and lose the record of having
# done so.
readonly GRACE_SECONDS=10
# How long the kernel gets to reap after SIGKILL before a process is called unkillable. A task only outlives SIGKILL
# while blocked in an uninterruptible syscall, which resolves well inside this or not at all.
readonly REAP_SECONDS=5

STOP_EXIT_REACHED=false

# ── Optional libraries, every one behind a fallback ──────────────────────────────────────────
# Loaded for quality of output ONLY -- a logger and a message renderer. Neither gates anything, and neither is allowed
# to prevent a stop. safe-paths.lib.sh and operator.lib.sh were loaded to vet and authorize a caller-supplied target;
# with no target to take, this helper has no input for them to decide and does not load them at all.
AI_TOOLS_LOG_TAG="ai-tools-stop"
AI_TOOLS_LOG_FILE="stop.log"
# shellcheck source=SCRIPTDIR/../../lib/ai-tools/log.lib.sh
source /usr/local/lib/ai-tools/log.lib.sh 2>/dev/null || true
# shellcheck source=SCRIPTDIR/../../lib/ai-tools/msg.lib.sh
source /usr/local/lib/ai-tools/msg.lib.sh 2>/dev/null || true

# sanitize <text> -- reduce to printable ASCII before anything agent-influenced reaches a terminal or the trail. Prefers
# the shared reducer; the fallback is the same allowlist of bytes, inline, because a missing logger must not mean
# an unsanitized path.
sanitize() {
    if declare -F ai_tools_log__sanitize >/dev/null 2>&1; then
        ai_tools_log__sanitize "$1"; return 0
    fi
    local LC_ALL=C
    printf '%s' "${1//[^[:print:]]/?}"
}

# log_event <level> <message> [FIELD=value ...] -- record to journald and, as root, to the file sink. Prefers the shared
# structured logger; falls back to logger(1) and a direct append, and to silence if even those are unavailable. Logging
# never changes what happens.
log_event() {
    local level="$1" message="$2"; shift 2
    if declare -F ai_tools_log__structured >/dev/null 2>&1; then
        ai_tools_log__structured "${level}" "${message}" "$@"
        return 0
    fi
    message="$(sanitize "${message}")"
    logger -t ai-tools-stop -p "daemon.${level}" -- "${message}" 2>/dev/null || true
    ( umask 077
      printf '%s %-7s [%d] %s\n' "$(date -Is 2>/dev/null)" "${level^^}" "$$" "${message}" \
          >> /var/log/ai-tools/stop.log
    ) 2>/dev/null || true
}

# say_error / say_warn / say_notice [<message code>] <line...> -- framed through msg.lib.sh when it loaded, plain
# otherwise. Output formatting is the most expendable thing here.
#
# THE EMITTERS TAKE LINES ONLY, NOT A LEADING FD -- unlike ai_tools_msg__headline, whose signature IS <title> <fd>
# <line...>. The two shapes sit next to each other, so passing the headline's fd to an emitter reads as consistent
# and is not: ai_tools_msg__error bakes in fd 2 already, so a leading `2` becomes the message's FIRST LINE and every
# refusal prints a stray digit ahead of itself. It is invisible in the boxed path and obvious only when captured.
#
# THE MESSAGE CODE IS SPLIT OFF HERE RATHER THAN PASSED STRAIGHT THROUGH, because these emitters are the only two-branch
# ones in the project: the library renders a code itself, while the fallback has to render it the way plain mode does --
# on its own leading line, ahead of the prefixed message -- or the one path a reader most needs a searchable token on is
# the one path that drops it. The form is matched inline, against the same anchored expression msg.lib.sh pins
# (tests/unit/msg.sh holds every inline copy to it), since the branch that needs the match is the branch
# where that library is absent.
#
# THE `ai-tools-stop: ` PREFIX IS THE EMITTER'S, so no message text carries one of its own: the code is the identity
# of the situation and the prefix names the component that raised it, and neither is stated twice.
say_error() {
    local code=""
    if [[ "${1-}" =~ ^MSG-[A-Z][0-9][A-Z][0-9]$ ]]; then code="$1"; shift; fi
    if declare -F ai_tools_msg__error >/dev/null 2>&1; then
        ai_tools_msg__error ${code:+"${code}"} "$@"
    else
        [[ -n "${code}" ]] && printf '%s\n' "${code}" >&2
        printf 'ai-tools-stop: %s\n' "$@" >&2
    fi
}
say_warn() {
    local code=""
    if [[ "${1-}" =~ ^MSG-[A-Z][0-9][A-Z][0-9]$ ]]; then code="$1"; shift; fi
    if declare -F ai_tools_msg__warn >/dev/null 2>&1; then
        ai_tools_msg__warn ${code:+"${code}"} "$@"
    else
        [[ -n "${code}" ]] && printf '%s\n' "${code}" >&2
        printf 'ai-tools-stop: %s\n' "$@" >&2
    fi
}
say_notice() {
    local code=""
    if [[ "${1-}" =~ ^MSG-[A-Z][0-9][A-Z][0-9]$ ]]; then code="$1"; shift; fi
    if declare -F ai_tools_msg__notice >/dev/null 2>&1; then
        ai_tools_msg__notice ${code:+"${code}"} "$@"
    else
        [[ -n "${code}" ]] && printf '%s\n' "${code}"
        printf '%s\n' "$@"
    fi
}
say_headline() {
    local title="$1"; shift
    if declare -F ai_tools_msg__headline >/dev/null 2>&1; then ai_tools_msg__headline "${title}" 1 "$@"
    else printf '\n== %s ==\n%s\n' "${title}" "$*"; fi
}

# ── Arguments ────────────────────────────────────────────────────────────────────────────────
DRY_RUN=false
ASSUME_YES=false
FORCE_KILL=false

# refuse_positional_argument <argument> -- refuse anything that is not an option, and exit 2.
#
# WHY THIS IS AN ERROR RATHER THAN AN IGNORED ARGUMENT. Someone typing a path after `stop` believes they are NARROWING
# the command. Proceeding would do the opposite of that belief -- end every session on the host -- and the confirmation
# defaults YES, so a reflexive Enter completes it. A refusal costs one corrected command; the alternative costs every
# running session.
#
# This is not inverted convention 1 being violated. That convention says to act where the operator's intent is KNOWN
# and something environmental is in the way -- no terminal, a missing library, a wedged manager. An unexpected argument
# is ambiguity about what was ASKED FOR, and guessing the most destructive reading of it is not degrading toward
# stopping. Nothing is left running either: the operator is one keystroke away, and the refusal says which.
#
# AND IT KEEPS A LATER EXTENSION NON-BREAKING. If per-target stopping is ever built -- which needs a session-to-project
# mapping the session cannot influence, i.e. something root records at launch, NOT the user manager's WorkingDirectory
# -- then `stop <path>` moves from an error to an accepted, narrower request. That is a pure widening and no existing
# command line changes meaning. Had it meant "stop everything, ignoring your path", the identical line would silently
# begin doing something different, which is the one outcome that cannot be rolled out safely.
#
# THE TEXT HERE IS A DELIBERATE TWIN of the CLI's refusal in cmd_stop, which is the copy an operator normally meets --
# this one is the last line, reached by a direct root call. They cannot be single-sourced: different processes, and this
# file is 750 root:root. They must say the same thing and offer the same four commands -- change one, change both.
#
# SO THE TWO SHARE ONE CODE. It is the same situation reported by whichever process met it first, and a reader searching
# the code has one thing to find. The code is emitted here and named nowhere else, so this site CITES the message
# the CLI defines rather than declaring a second one.
refuse_positional_argument() {
    printf 'MSG-A3M9\nai-tools-stop: stop takes no path: %s\n' "$1" >&2
    printf '%s' '
  ai-tools stop TERMINATES every agent session on this host, and has no per-project
  form. It is not the way to end a session you are finished with -- it kills the process
  tree, so the session cannot run its own session-end handback.
  A session is attributed to a project by asking the sandbox account'"'"'s own user manager
  -- the account being stopped -- so that attribution is reported, never trusted to
  decide what a stop reaches.

  Terminate every session:    ai-tools stop
  See what is running first:  ai-tools stop --dry-run
  End one session cleanly:    /exit inside it, which runs its session-end handback
  Terminate one by hand:      sudo systemctl --user -M ai-tools@.host stop <unit>
' >&2
    exit 2
}

parse_command_line() {
    while (( $# )); do
        case "$1" in
            # Accepted and inert (the header's usage note says why it exists at all). The documented form is
            # `ai-tools stop`.
            --all)        shift ;;
            --dry-run) DRY_RUN=true; shift ;;
            -y|--yes)     ASSUME_YES=true; shift ;;
            --force)      FORCE_KILL=true; shift ;;
            # The CLI refuses this too, before its sudo, and for the same reason the path refusal is twinned:
            # an operator meets whichever side answered. One situation, one code, defined at the CLI's arm and cited
            # here.
            -*) printf 'MSG-B7K4\nai-tools-stop: unknown option for stop: %s\n' "$1" >&2
                printf '  allowed: --all, --dry-run, --yes/-y, --force\n' >&2
                exit 2 ;;
            *)  refuse_positional_argument "$1" ;;
        esac
    done
    readonly DRY_RUN ASSUME_YES FORCE_KILL
}

# resolve_run_context -- establish who is asking and what account is being stopped, and arm the trail's traps.
# Everything here either succeeds or exits; no later step runs on a guess.
resolve_run_context() {
    if [[ "$(id -u)" != "0" ]]; then
        say_error MSG-Z5W3 "This command must run as root: stopping a session means signalling ${SANDBOX_USER}'s cgroups" \
                  "run it as: sudo ai-tools stop"
        exit 5
    fi

    # Who sudo says invoked this -- written by a root process, unreachable by the sandbox account. This is recorded
    # for the TRAIL only: the command does not take an authorization input, so a caller identity has no bearing
    # on what is terminated. It is still cross-checked rather than taken at face value, because "who asked for this" is
    # the line an operator reads first after an incident and a wrong name there is worse than no name.
    #
    # SUDO_USER ALONE IS NOT PROOF OF A SUDO TRANSACTION. A root shell entered with `sudo -i` keeps SUDO_USER set,
    # so its presence does not establish that *this* invocation came through sudo on behalf of that user -- it may be
    # inherited state from an earlier one. SUDO_UID is cross-checked against it: the two are set together by the same
    # sudo run, so a pair that disagrees is inherited or forged environment and is refused rather than trusted. Both,
    # plus the real uid, are recorded separately, so the trail carries what was observed rather than one derived
    # conclusion.
    CALLER="${SUDO_USER:-root}"
    if [[ -n "${SUDO_USER:-}" ]]; then
        local caller_uid_from_name
        caller_uid_from_name="$(id -u "${SUDO_USER}" 2>/dev/null)"
        if [[ -z "${SUDO_UID:-}" || -z "${caller_uid_from_name}" \
              || "${SUDO_UID}" != "${caller_uid_from_name}" ]]; then
            log_event warning \
                "SUDO_USER=${SUDO_USER} does not agree with SUDO_UID=${SUDO_UID:-<unset>} -- recording this as a direct root invocation rather than trusting the name" \
                "AI_TOOLS_SUDO_USER=${SUDO_USER}" "AI_TOOLS_SUDO_UID=${SUDO_UID:-}" "AI_TOOLS_RESULT=identity-rejected"
            CALLER="root"
        fi
    fi
    readonly CALLER

    SANDBOX_UID="$(id -u "${SANDBOX_USER}" 2>/dev/null)"
    if [[ -z "${SANDBOX_UID}" ]]; then
        say_error MSG-Q9Y2 "This host cannot resolve the uid of ${SANDBOX_USER}, so no cgroup can be located" \
                  "the sandbox account is missing -- reprovision with:" \
                  "  sudo ai-tools-admin system bootstrap"
        log_event error "REFUSED: ${SANDBOX_USER} has no uid; cannot locate any session cgroup"
        exit 5
    fi
    readonly SANDBOX_UID

    # An incomplete run must be visible in the trail. The trap covers every exit including a signal that bash can
    # handle; only SIGKILL of this helper escapes it, which is a documented residual.
    trap record_incomplete_run EXIT
    # A signal during a stop is its own event, not merely an incomplete run: an operator who interrupts one mid-way
    # needs the trail to say a kill was in flight when it happened, because the tree is then in whatever state that pass
    # left it. Named separately from the EXIT trap so the two causes are distinguishable. SIGKILL of this helper remains
    # untrappable and is a stated
    # residual.
    trap 'log_event error "stop run by ${CALLER} interrupted by a signal -- some sessions may be partially stopped" "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_RESULT=interrupted"; STOP_EXIT_REACHED=true; exit 130' INT TERM HUP
}

record_incomplete_run() {
    ${STOP_EXIT_REACHED} && return 0
    log_event error "stop run by ${CALLER} ended without completing -- state unverified" \
        "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_RESULT=incomplete"
}

# ── Locating the sandbox account's cgroups ───────────────────────────────────────────────────
# cgroup2_mount -- PRINT the cgroup v2 hierarchy's mount point. Unified hosts mount it at /sys/fs/cgroup; a hybrid host
# puts v1 controllers there and v2 at /sys/fs/cgroup/unified. Read from /proc/mounts rather than assumed, and with no
# external command.
cgroup2_mount() {
    local mount_point fstype
    while read -r _ mount_point fstype _; do
        [[ "${fstype}" == "cgroup2" ]] && { printf '%s' "${mount_point}"; return 0; }
    done < /proc/mounts
    return 1
}

# resolve_cgroup_layout -- fix the three paths the whole enumeration is expressed against, or refuse. Every one is
# derived from the host, never assumed.
resolve_cgroup_layout() {
    CGROUP2_MOUNT="$(cgroup2_mount)"
    if [[ -z "${CGROUP2_MOUNT}" ]]; then
        say_error MSG-K2T7 "This host has no cgroup v2 hierarchy, so sessions cannot be enumerated or stopped reliably." \
                  "Stop them by hand and report the host: sudo systemctl --user -M ${SANDBOX_USER}@.host list-units"
        log_event error "REFUSED: no cgroup2 mount; cannot enumerate sessions for ${CALLER}"
        exit 5
    fi
    readonly CGROUP2_MOUNT
    # The account's own manager slice. A process the operator's sudo spawned under that account lives in the INVOKING
    # session's slice, not this one, so scoping to this path is also what keeps this helper from selecting its own
    # children.
    #
    # The root is the account's WHOLE per-user slice, not just its `user@<uid>.service` manager subtree. systemd places
    # login session scopes (`session-N.scope`) as SIBLINGS of the manager service, under the same per-user slice --
    # so a scan rooted at the manager alone has a blind spot for anything not started by that manager. The sandbox
    # account has no login shell and no password, so no process should ever appear there; scanning the wider root costs
    # one directory level and removes the need for that to be true.
    #
    # It stays scoped to the ACCOUNT's slice rather than widening to uid alone, which is what keeps the scan off this
    # helper's own `sudo -u` children: those share the account's uid but live in the INVOKING user's slice.
    readonly SANDBOX_SLICE="${CGROUP2_MOUNT}/user.slice/user-${SANDBOX_UID}.slice"
    # The manager unit, named exactly rather than by basename. It is descended into but never emitted as a session;
    # every cgroup in the slice is swept. See find_session_cgroups.
    readonly MANAGER_SERVICE="${SANDBOX_SLICE}/user@${SANDBOX_UID}.service"
}

# find_session_cgroups -- PRINT one absolute cgroup directory per live session, deepest first. Every directory
# under the account's manager slice is a candidate except init.scope, which holds `systemd --user` and its (sd-pam)
# helper. Recursion is a bash glob walk, so no `find` is needed and a sandbox that forbids external binaries cannot
# disarm this.
find_session_cgroups() {
    [[ -d "${SANDBOX_SLICE}" ]] || return 0
    local -a stack=("${SANDBOX_SLICE}")
    local current child leaf
    # Tasks placed directly in the manager slice belong to no unit at all. systemd does not put them there, so this is
    # the "something else entirely" case -- reported rather than skipped.
    has_own_tasks "${SANDBOX_SLICE}" && printf '%s\n' "${SANDBOX_SLICE}"
    while (( ${#stack[@]} )); do
        current="${stack[-1]}"; unset 'stack[-1]'
        for child in "${current}"/*/; do
            child="${child%/}"
            [[ -d "${child}" ]] || continue
            leaf="${child##*/}"
            # NOTHING IS EXEMPT -- and that is the guarantee, not an oversight.
            #
            # An exemption is a place to hide. Every name inside the delegated subtree is the DELEGATEE's to choose,
            # and on a DAC-only host a session can reach its own user manager and ask for a unit by any name it likes,
            # so an exemption expressed as a name is one the monitored system can claim for itself. Expressed
            # as an exact path it is still a destination: moving tasks into a spared cgroup is the other DAC-only
            # escape. Sweeping every cgroup closes both at once, without distro-specific knowledge -- no unit name
            # appears in this file, so there is no name here to differ between EL9, EL10 and Fedora.
            #
            # `init.scope` is therefore enumerated and stopped like anything else, the account's own `systemd --user`
            # and its (sd-pam) helper included. What that would otherwise break -- the next launch finding no
            # `--user instance` -- is repaired afterwards by restore_user_manager rather than bought with an exemption.
            #
            # ONE cgroup is still special-cased, and it is not an exemption:
            #   user@<uid>.service -- the manager unit. DESCENDED INTO but never emitted, because
            #                         its name ends in `.service`, so the stop-at-first-unit rule
            #                         would otherwise swallow the whole manager subtree as a
            #                         single "session" and never reach the real units inside it.
            #                         Its contents, init.scope among them, are emitted instead --
            #                         so everything under it is still stopped, and the operator's
            #                         confirmation lists real units rather than one opaque row.
            if [[ "${child}" == "${MANAGER_SERVICE}" ]]; then
                has_own_tasks "${child}" && printf '%s\n' "${child}"
                stack+=("${child}")
                continue
            fi
            if [[ "${leaf}" == *.service || "${leaf}" == *.scope ]]; then
                # A unit is ONE session. Its nested cgroups are part of it -- cgroup_pids counts them and cgroup.kill
                # kills them -- so descending further would list the same processes again under a second heading, double
                # the count the operator confirms against, and (worse) offer a parent SLICE as a stoppable thing,
                # which would take every sibling unit with it.
                printf '%s\n' "${child}"
            else
                has_own_tasks "${child}" && printf '%s\n' "${child}"
                stack+=("${child}")
            fi
        done
    done
}

# has_own_tasks <cgroup-dir> -- succeed when tasks sit in THIS cgroup's own cgroup.procs, ignoring descendants. This is
# what separates "a slice, whose tasks all live in the units under it" from "a cgroup holding processes directly",
# and it is why enumerating units does not lose anything: the only place a task can hide from a unit walk is a slice,
# and a slice with its own tasks is emitted in its own right.
#
# FAIL-CLOSED WHERE A READ ERROR IS DISTINGUISHABLE, which is the whole point. "I could not read cgroup.procs"
# and "there are no tasks" are different facts, and collapsing them is a fail-open on the predicate the guarantee rests
# on. Only one failure means empty -- the file not existing, i.e. the cgroup was removed, which is the successful
# outcome.
#
# Bash cannot tell a failed read(2) from a clean EOF: `read` returns 1 for both, and so does `x="$(< file)"`. Verified,
# not assumed. So each cause is separated by a fact that IS observable:
#
#   permission-unreadable -- `-r` answers it directly, and answers LIVE.
#   THREADED cgroup       -- the kernel's documented case, and the one that matters here: in a
#                            threaded subtree every cgroup.procs under the threaded root fails the
#                            read (EOPNOTSUPP) while the cgroup holds live threads, and the file is
#                            permission-readable, so `-r` does NOT catch it. cgroup.threads is
#                            readable in EVERY cgroup including those, so it is the corroborating
#                            source: no pid but a tid means tasks. In a domain cgroup the two
#                            always agree (a member process' threads are in its own cgroup), so
#                            consulting it costs an ordinary empty cgroup one extra failed open.
#
# What remains indistinguishable -- a cgroup.procs that is present, permission-readable, and errors for some third
# reason with no cgroup.threads beside it -- is not a shape cgroupfs produces.
has_own_tasks() {
    local first_task=""
    [[ -e "$1/cgroup.procs" ]] || return 1          # cgroup gone == no tasks, the good outcome
    # `2>/dev/null` precedes the input redirect: see the note in cgroup_pids.
    read -r first_task 2>/dev/null < "$1/cgroup.procs"
    [[ -n "${first_task}" ]] && return 0
    [[ -r "$1/cgroup.procs" ]] || return 0          # exists, unreadable: assume LIVE
    read -r first_task 2>/dev/null < "$1/cgroup.threads"
    [[ -n "${first_task}" ]]
}

# cgroup_unit_name <cgroup-dir> -- PRINT the systemd unit this cgroup belongs to, or empty.
cgroup_unit_name() {
    local leaf="${1##*/}"
    [[ "${leaf}" == *.service || "${leaf}" == *.scope ]] && printf '%s' "${leaf}"
}

# ── Classification (advisory, exactly like attribution) ──────────────────────────────────────
# The account's slice holds more than agent sessions: its own `systemd --user` and `init.scope`, a dbus broker,
# and a login session scope for every `sudo -u` that crossed pam_systemd. All of them are terminated -- no cgroup is
# exempt -- but calling four such cgroups "4 agent sessions" in the table an operator confirms against, and in the line
# they read first after an incident, is untrue, and untrue in the direction that inflates how much agent work was
# running.
#
# THIS CHANGES A LABEL AND A COUNT, NEVER A TARGET. Like unit_working_directory, it is attribution alone, and carries
# the same caveat: a unit name inside the delegated subtree is the delegatee's to choose, so a session can name itself
# out of the agent class, and does not gain an exemption by it, because both classes are enumerated, listed and killed
# identically. Nothing here is consulted to decide what a stop reaches; that remains cgroup-slice membership alone.

# session_is_agent <unit-name> -- succeed for a unit ai-tools-run started. It names every session
# `<SANDBOX_USER>-<agent>-<pid>.service`, so this matches THIS PROJECT's own prefix rather than any distro's unit names
# -- the file still does not hold a unit name that differs between EL9, EL10 and Fedora.
session_is_agent() {
    [[ "$1" == "${SANDBOX_USER}-"*.service ]]
}

# session_class_note <unit-name> -- PRINT the parenthetical a non-agent row is marked with, or empty for an agent
# session.
#
# ONE MARKER, deliberately, rather than naming the user manager separately. Distinguishing it would take either a path
# test against MANAGER_SERVICE -- which is wrong, since every unit that manager starts is inside its subtree, agent
# sessions included, so the account's dbus broker came out labelled "user manager" -- or an `init.scope` literal,
# which would put a systemd unit name back in a file that deliberately holds none. The headline already states
# that the manager is among these and is restarted afterwards, which is the part an operator acts on.
session_class_note() {
    session_is_agent "$1" && return 0
    printf '(account plumbing)'
}

# cgroup_pids <cgroup-dir> -- PRINT every pid in this cgroup AND its descendants. cgroup.procs is per-cgroup,
# so a nested cgroup's members are not listed by its parent and the subtree is walked. This is the liveness source
# of truth: a task cannot remove itself from a cgroup, so it cannot hide from this by forking, calling setsid(2),
# or being re-parented to PID 1. Emission is DEEPEST FIRST, so a caller signalling in order reaches children
# before their parents and a parent is never left waiting on a child it can still see. The walk is breadth-first,
# which yields the directories in non-decreasing depth, and they are then read in reverse -- rather than a depth-ordered
# sort, which would need an external command in the middle of the kill path.
cgroup_pids() {
    local -a directories=("$1")
    local index=0 current child pid
    while (( index < ${#directories[@]} )); do
        current="${directories[index]}"; index=$(( index + 1 ))
        for child in "${current}"/*/; do
            child="${child%/}"
            [[ -d "${child}" ]] && directories+=("${child}")
        done
    done
    for (( index = ${#directories[@]} - 1; index >= 0; index-- )); do
        # `2>/dev/null` PRECEDES the input redirect deliberately: redirections are applied left to right,
        # so with the order reversed a cgroup that disappeared mid-walk (the normal outcome of a successful kill) writes
        # its failure to the real stderr before the suppression is installed. Verified, not assumed.
        while read -r pid; do
            [[ -n "${pid}" ]] && printf '%s\n' "${pid}"
        done 2>/dev/null < "${directories[index]}/cgroup.procs"
    done
}

# cgroup_is_live <cgroup-dir> -- succeed while any task remains in the subtree. Deliberately not `systemctl is-active`:
# that asks the manager, and the manager is one of the things that can be broken. A cgroup with no tasks is stopped
# whatever any daemon believes. It walks rather than reusing cgroup_pids because this is the VERIFICATION predicate
# and the one place a missing external command would be catastrophic: `cgroup_pids | head -n1` yields an empty string
# when `head` is absent, which reads as "no tasks" and would report a stop as complete while the session is still
# running -- a fail-open on the single check the guarantee rests on. Walking in-shell also exits at the first task
# found, so it is cheaper than collecting every pid to ask a yes/no question.
#
# `cgroup.events` is consulted FIRST, because it is the kernel's own answer to exactly this question: its `populated`
# field is 1 while the cgroup OR ANY DESCENDANT holds a live process, so one read replaces the whole walk and cannot
# disagree with the kernel's view. It is absent on the root cgroup only, and every target here is a non-root cgroup.
# The walk remains as the fallback for a kernel or layout that does not present it.
#
# Both paths fail closed: an unreadable or unparseable answer reports LIVE. A cgroup directory that no longer exists is
# the one failure that reports empty, because that is what a completed kill looks like.
cgroup_is_live() {
    [[ -d "$1" ]] || return 1                        # removed: the successful outcome
    local field value
    if [[ -e "$1/cgroup.events" ]]; then
        if [[ -r "$1/cgroup.events" ]]; then
            while read -r field value; do
                [[ "${field}" == "populated" ]] || continue
                [[ "${value}" == "0" ]] && return 1
                return 0
            done 2>/dev/null < "$1/cgroup.events"
        fi
        # Present but unreadable, or readable with no `populated` line: unprovable, so LIVE.
        return 0
    fi
    local -a directories=("$1")
    local index=0 current child
    while (( index < ${#directories[@]} )); do
        current="${directories[index]}"; index=$(( index + 1 ))
        has_own_tasks "${current}" && return 0
        for child in "${current}"/*/; do
            child="${child%/}"
            [[ -d "${child}" ]] && directories+=("${child}")
        done
    done
    return 1
}

# ── Signalling ───────────────────────────────────────────────────────────────────────────────
# pid_start_time <pid> -- PRINT /proc/<pid>/stat field 22, the process' start time in clock ticks. It is unique per pid
# for that process' lifetime, which makes it the portable guard against signalling a RECYCLED pid: capture it
# when the pid is collected, re-check it immediately before the kill. comm (field 2) may contain spaces and parentheses,
# so the fields are read after the LAST ')' -- field 22 is then index 19.
#
# Failing to read it means the process is GONE: this runs as root, the pid came from a cgroup being torn
# down, and root's only reason to fail on /proc/<pid>/stat is that the entry no longer exists. So a read failure skips
# the pid, which is correct rather than fail-open -- there is no process left to signal. The readability test comes
# first purely to keep the common vanished-pid case off stderr; the read is still checked, because the pid can exit
# between the two.
pid_start_time() {
    local stat_line fields_after_comm
    [[ -r "/proc/$1/stat" ]] || return 1
    stat_line="$(< "/proc/$1/stat")" || return 1
    [[ -n "${stat_line}" ]] || return 1
    fields_after_comm="${stat_line##*) }"
    local -a fields
    read -ra fields <<< "${fields_after_comm}"
    (( ${#fields[@]} > 19 )) || return 1
    printf '%s' "${fields[19]}"
}

# signal_pids_validated <signal> <pid>... -- signal each pid only if its start time still matches what it had when it
# was collected, so a pid recycled in between is skipped rather than signalled blind. `kill` is a bash builtin, so this
# runs without an external binary. A pid that has already exited is not an error -- that is the outcome being aimed at.
signal_pids_validated() {
    local signal="$1"; shift
    local entry pid expected_start current_start
    for entry in "$@"; do
        pid="${entry%%:*}"; expected_start="${entry#*:}"
        current_start="$(pid_start_time "${pid}")" || continue
        [[ "${current_start}" == "${expected_start}" ]] || continue
        kill "-${signal}" "${pid}" 2>/dev/null
    done
}

# collect_pids_with_start_time <cgroup-dir> -- PRINT `<pid>:<start-time>` per live task, which is
# what signal_pids_validated consumes. Collected fresh on every pass: a set read once and signalled twice would miss
# whatever was forked in between.
collect_pids_with_start_time() {
    local pid start
    while read -r pid; do
        start="$(pid_start_time "${pid}")" || continue
        printf '%s:%s\n' "${pid}" "${start}"
    done < <(cgroup_pids "$1")
}

# terminate_gracefully <cgroup-dir> -- the SIGTERM pass. Succeeds when the subtree empties within the grace. Re-collects
# each second so a child forked after the previous pass is signalled too, and cgroup_pids returns deepest-first,
# so children are reached before their parents.
terminate_gracefully() {
    local cgroup_directory="$1" waited=0
    local -a pids
    while (( waited < GRACE_SECONDS )); do
        mapfile -t pids < <(collect_pids_with_start_time "${cgroup_directory}")
        (( ${#pids[@]} )) || return 0
        signal_pids_validated TERM "${pids[@]}"
        sleep 1
        waited=$(( waited + 1 ))
    done
    ! cgroup_is_live "${cgroup_directory}"
}

# kill_outright <cgroup-dir> -- the unrefusable pass. Succeeds when the subtree is empty.
#
# cgroup.kill is the mechanism that makes the guarantee: one write freezes the cgroup and SIGKILLs every member
# INCLUDING descendants, atomically, so there is no window in which a fork can outrun the signal. Where the kernel
# predates it (< 5.14), the fallback re-collects and re-signals in a loop, which narrows that window to one iteration
# but does not close it -- which is exactly why cgroup.kill is preferred rather than treated as an optimisation.
kill_outright() {
    local cgroup_directory="$1" waited=0
    local cgroup_kill_reported=false
    local -a pids
    while (( waited < REAP_SECONDS )); do
        cgroup_is_live "${cgroup_directory}" || return 0
        # Re-asserted every pass, not written once before the loop. The write is idempotent and costs one write,
        # and doing it once would make the whole guarantee rest on a single syscall whose failure is invisible.
        # Re-asserting also covers a nested cgroup created between passes, which the one-shot form would leave
        # to the pid fallback alone.
        #
        # A failure is REPORTED, once per session, and the two reasons are not the same event: an absent file is an old
        # kernel and expected, while a present file that will not take the write means the atomic mechanism is
        # unavailable on a host that should have it, and the run has silently dropped to the racier pid loop.
        # That distinction is exactly what an operator needs after a stop that did not converge, so it must not be
        # swallowed.
        if [[ -e "${cgroup_directory}/cgroup.kill" ]]; then
            if ! printf '1' > "${cgroup_directory}/cgroup.kill" 2>/dev/null; then
                ${cgroup_kill_reported} || log_event warning \
                    "cgroup.kill exists but refused the write for ${cgroup_directory} -- falling back to per-pid signalling, which is not atomic" \
                    "AI_TOOLS_CGROUP=${cgroup_directory}" "AI_TOOLS_MECHANISM=cgroup-kill-failed"
                cgroup_kill_reported=true
            fi
        else
            ${cgroup_kill_reported} || log_event info \
                "no cgroup.kill on this kernel -- using per-pid signalling for ${cgroup_directory}" \
                "AI_TOOLS_CGROUP=${cgroup_directory}" "AI_TOOLS_MECHANISM=cgroup-kill-absent"
            cgroup_kill_reported=true
        fi
        mapfile -t pids < <(collect_pids_with_start_time "${cgroup_directory}")
        (( ${#pids[@]} )) && signal_pids_validated KILL "${pids[@]}"
        sleep 1
        waited=$(( waited + 1 ))
    done
    ! cgroup_is_live "${cgroup_directory}"
}

# end_session <cgroup-dir> -- PRINT which pass ended it: `terminated`, `killed`, or `alive`. The kill pass runs whether
# or not the graceful pass reported success, because its own verification is the only thing trusted.
end_session() {
    local cgroup_directory="$1"
    if ! ${FORCE_KILL}; then
        terminate_gracefully "${cgroup_directory}" && { printf 'terminated'; return 0; }
    fi
    kill_outright "${cgroup_directory}" && { printf 'killed'; return 0; }
    printf 'alive'
}

# ── Attribution (best-effort, and never decides liveness) ────────────────────────────────────
# unit_working_directory <unit> -- PRINT the unit's WorkingDirectory, or empty when the user manager cannot be reached.
# The machine transport is preferred: this runs as root, where the system bus already authorizes it, whereas `sudo -u`
# needs that account's own bus to accept the connection, which a host can refuse while the manager is healthy. A failure
# here costs attribution alone.
#
# BOUNDED IN TIME, because "the user manager is wedged" is not a hypothetical here -- it is one of the states
# an operator reaches for this command IN. A d-bus call to a hung manager blocks indefinitely, and a stop that hangs
# while attributing sessions is a stop that did not happen, which is the one outcome this file exists to prevent. Both
# calls therefore run under a short `timeout`, and every way that can fail -- the manager not answering, `timeout`
# itself absent -- returns an empty WorkingDirectory, which refuses the SCOPED form and sends the operator to `--all`.
# `--all` does not read one at all, so the undeclinable form cannot be delayed by this.
unit_working_directory() {
    local raw
    raw="$(timeout 5 systemctl --user -M "${SANDBOX_USER}@.host" show --property=WorkingDirectory "$1" 2>/dev/null)"
    if [[ -z "${raw}" ]]; then
        raw="$(timeout 5 sudo -n -u "${SANDBOX_USER}" XDG_RUNTIME_DIR="/run/user/${SANDBOX_UID}" \
                   systemctl --user show --property=WorkingDirectory "$1" 2>/dev/null)"
    fi
    raw="${raw#WorkingDirectory=}"
    # Strip systemd's "missing is ok" marker. THE D-BUS PROPERTY RENDERS IT `!`, which is what `show` returns
    # and therefore the only spelling this function actually meets; `-` is the unit-file spelling of the same flag
    # and is stripped too, so neither rendering reaches the comparison. (Observed: dbus-broker.service reports
    # `WorkingDirectory=!/home/<user>`.)
    if [[ "${raw}" == '!'* || "${raw}" == '-'* ]]; then raw="${raw:1}"; fi
    # ONLY AN ABSOLUTE PATH IS A RESULT; anything else yields an empty value and the session reads as `unknown`.
    # Attribution does not decide any part of the sweep here, so this is not a gate -- it is what keeps a value this
    # helper cannot interpret from being printed as though it could be used. The concrete case is
    # print_reclaim_guidance, which turns each attributed directory into a command the operator is invited to run:
    # an unstripped marker emitted `ai-tools projects handback !/opt/ai-tools`, which is not a runnable command
    # and, pasted into an interactive bash, is not even an inert one.
    #
    # The shape is ALLOWLISTED rather than the markers enumerated, so a rendering systemd adds later degrades
    # to `unknown` instead of reaching the operator as a broken command. `~` (WorkingDirectory=~, the account's home)
    # does not carry a path and is correctly refused here.
    [[ "${raw}" == /* ]] || raw=""
    printf '%s' "${raw}"
}

# ── Restoring the user manager ───────────────────────────────────────────────────────────────
# restore_user_manager -- put `user@<uid>.service` back after a stop that necessarily took it down.
#
# WHY IT HAS TO EXIST. The enumeration covers every cgroup, the account's own `systemd --user` included (see
# find_session_cgroups): an exemption is a destination a session can move into, and on a DAC-only host it can also ask
# that manager for a unit outside any subtree we chose to sweep. Covering every cgroup closes both. The price is
# that the manager is gone afterwards -- and SIGKILL leaves `user@<uid>.service` FAILED rather than restarting it,
# so the next launch would find no `--user instance`. This pays that price back instead of buying it with an exemption.
#
# IT RUNS AFTER THE KILL AND AFTER THE VERIFICATION, and cannot affect either. The invariant is that a stop reported
# as done HAS happened; a manager that did not come back is a different and lesser problem -- the host cannot start
# a NEW session until it is fixed, which is nearer to the point of this command than against it. So every failure here
# warns, names the command, and leaves the exit status alone.
restore_user_manager() {
    local unit="user@${SANDBOX_UID}.service"
    # reset-failed first: the manager was SIGKILLed, so the unit is in `failed`, and `start` on a failed unit is not
    # uniformly a reset-and-start across systemd versions.
    timeout 10 systemctl reset-failed "${unit}" >/dev/null 2>&1
    if timeout 30 systemctl start "${unit}" >/dev/null 2>&1; then
        log_event notice "restarted ${unit} after the stop -- the next launch has a --user instance" \
            "AI_TOOLS_UNIT=${unit}" "AI_TOOLS_RESULT=manager-restored"
        return 0
    fi
    say_warn MSG-W8C6 "The sessions were stopped, but ${SANDBOX_USER}'s user manager did not come back, so the next launch has no systemd --user instance to start a session in. Restore it with:" \
             "sudo systemctl reset-failed ${unit} && sudo systemctl start ${unit}"
    log_event error "could not restart ${unit} after the stop -- the next launch will have no --user instance" \
        "AI_TOOLS_UNIT=${unit}" "AI_TOOLS_RESULT=manager-not-restored"
    return 1
}

# ── Confirmation ─────────────────────────────────────────────────────────────────────────────
# confirm_stop <agent-count> <plumbing-count> -- succeed unless the operator deliberately declines. Inverted convention
# 2 in the header: the default is YES, so no terminal, no msg.lib.sh, a pipe, or a bare Enter all proceed, and only
# an explicit `n` stops the stop.
#
# THE QUESTION NAMES BOTH CLASSES, for the reason the table separates them: consent given to "4 sessions" that were one
# session and three units of the account's own plumbing was not informed consent about either number.
#
# THE ARITY IS DEFAULTED, and that is inverted convention 1 rather than defensive habit. This file runs under `set -u`,
# where reading an argument a caller did not pass aborts the shell outright -- here, mid-question, after the table has
# been printed and before anything has been signalled: a stop that was asked for and did not happen, which is the one
# outcome this file exists to prevent. main() always passes both counts, so the default is not an expected path; it is
# the guarantee that a caller's slip costs the wording of a question and never the answer to it.
#
# WHICH PATH GAVE CONSENT IS RECORDED, because the paths are not equally strong and a reader of the trail must not have
# to guess. `--yes` is an operator decision; the library prompt is a real answered question; the raw /dev/tty prompt is
# the same question asked without the shared renderer; and `no-tty` means nobody was asked at all and the default
# carried the run. The last is legitimate -- it is the whole point of defaulting YES -- but it is the one an operator
# would want to see when asking why a cron job stopped a session at 4am.
confirm_stop() {
    local agent_count="${1:-0}" plumbing_count="${2:-0}"
    # Three shapes, none of which names a class that is not there. main() reaches this only with at least one cgroup
    # selected, so "no sessions and no plumbing" cannot occur.
    local question
    if   (( agent_count == 0 )); then question="Terminate the ${plumbing_count} unit(s) of the ${SANDBOX_USER} account's own plumbing listed above?"
    elif (( plumbing_count ));   then question="Terminate the ${agent_count} agent session(s) listed above, and ${plumbing_count} unit(s) of the ${SANDBOX_USER} account's own plumbing with them?"
    else                              question="Terminate the ${agent_count} agent session(s) listed above?"
    fi
    if ${ASSUME_YES}; then
        log_event info "stop confirmed by ${CALLER} via --yes" "AI_TOOLS_CONSENT=flag"
        return 0
    fi
    if declare -F ai_tools_msg__confirm >/dev/null 2>&1; then
        if ai_tools_msg__confirm "${question}" y; then
            log_event info "stop confirmed by ${CALLER} at the prompt" "AI_TOOLS_CONSENT=prompt"
            return 0
        fi
        return 1
    fi
    local answer=""
    if ! read -r -p "${question} [Y/n] (default: Yes): " answer 2>/dev/null < /dev/tty; then
        log_event notice \
            "no terminal to confirm with -- proceeding, since a stop that declines when unattended is a stop that failed" \
            "AI_TOOLS_CONSENT=no-tty"
        return 0
    fi
    if [[ "${answer}" =~ ^[Nn] ]]; then return 1; fi
    log_event info "stop confirmed by ${CALLER} at the fallback prompt (msg.lib.sh unavailable)" \
        "AI_TOOLS_CONSENT=fallback-prompt"
    return 0
}

# ── Reporting ────────────────────────────────────────────────────────────────────────────────
# print_session_table <verb> -- the detail the confirmation is answered against. An operator agreeing to stop "3
# sessions" without seeing which projects they are in is not consenting to
# anything.
#
# Reads the parallel selection arrays directly rather than taking packed records. Nothing in this file joins fields
# into a delimited string any more: a WorkingDirectory is an arbitrary pathname and may contain the delimiter,
# which would split one session into two garbled rows -- in the very table whose job is to be exact. Agent sessions are
# listed FIRST, ahead of the account's plumbing, because the table is read under time pressure and the rows that answer
# "what was running" must not be interleaved with rows that are always there. Order is presentation only; every selected
# cgroup is ended by the loop in main() in its own order, and no cgroup is dropped from either pass.
print_session_table() {
    local verb="$1" index note
    printf '  %-11s %-40s %6s  %s\n' "" "SESSION" "PROCS" "PROJECT"
    for index in "${!selected_cgroups[@]}"; do
        ${selected_is_agent[index]} || continue
        printf '  %-11s %-40s %6s  %s\n' "${verb}" \
            "$(sanitize "${selected_units[index]:-unattributed cgroup}")" \
            "${selected_counts[index]}" \
            "$(sanitize "${selected_dirs[index]:-unknown}")"
    done
    for index in "${!selected_cgroups[@]}"; do
        ${selected_is_agent[index]} && continue
        note="$(session_class_note "${selected_units[index]}")"
        printf '  %-11s %-40s %6s  %s  %s\n' "${verb}" \
            "$(sanitize "${selected_units[index]:-unattributed cgroup}")" \
            "${selected_counts[index]}" \
            "$(sanitize "${selected_dirs[index]:-unknown}")" "${note}"
    done
}

# print_reclaim_guidance <dir>... -- name the reclaim per project actually stopped. A stop cannot run the agent's
# SessionEnd hook, so the in-flight turn's writes may still be sandbox-owned; naming the project beats a pointer
# the operator must translate mid-incident.
#
# CALLED WITH AGENT SESSIONS' DIRECTORIES ONLY. The account's plumbing has no project to hand back, and its
# WorkingDirectory is routinely a path `ai-tools projects handback` would refuse outright: the account's dbus broker
# reports `/opt/ai-tools`, the control plane, which the safe-paths backstop protects. Emitting it produced a remedy
# that cannot run, offered to an operator mid-incident with no marker to distinguish it from the one that can.
print_reclaim_guidance() {
    local directory
    local -A seen=()
    local -a targets=()
    for directory in "$@"; do
        [[ -n "${directory}" ]] || continue
        [[ -n "${seen[${directory}]:-}" ]] && continue
        seen["${directory}"]=1
        targets+=("${directory}")
    done
    (( ${#targets[@]} )) || return 0
    printf '\n  %s\n' "A stopped session cannot run its own session-end handback, so the last turn's"
    printf '  %s\n'   "writes may still be ${SANDBOX_USER}-owned. Hand them back with:"
    for directory in "${targets[@]}"; do
        printf '  %s\n' "    ai-tools projects handback $(sanitize "${directory}")"
    done
}

# ── Run ──────────────────────────────────────────────────────────────────────────────────────
main() {
    # There is one scope and no way to ask for another, so it is a constant rather than a decision: every live cgroup
    # in the account's slice. Kept as a named value because it is what the headline, the trail and the no-sessions
    # message all read.
    local scope="every agent session"

    # Recorded before anything is selected, so a run interrupted part-way still left a record of what was asked
    # for and by whom.
    log_event notice \
        "${CALLER} requested stop (scope=${scope}, force=${FORCE_KILL}, dry-run=${DRY_RUN})" \
        "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_SCOPE=${scope}" \
        "AI_TOOLS_FORCE=${FORCE_KILL}" "AI_TOOLS_DRYRUN=${DRY_RUN}"

    local -a selected_cgroups=() selected_units=() selected_dirs=() selected_counts=() \
             selected_is_agent=()
    local agent_count=0 plumbing_count=0
    local cgroup_directory unit working_directory pid_count
    local -a session_pids
    while read -r cgroup_directory; do
        [[ -n "${cgroup_directory}" ]] || continue
        # Counted in the shell rather than through `wc -l`: the pid list is already being read, so a pipe to an external
        # command does not buy accuracy and puts one more binary on the path.
        mapfile -t session_pids < <(cgroup_pids "${cgroup_directory}")
        pid_count=${#session_pids[@]}
        (( pid_count )) || continue
        unit="$(cgroup_unit_name "${cgroup_directory}")"
        working_directory=""
        [[ -n "${unit}" ]] && working_directory="$(unit_working_directory "${unit}")"
        # NOTHING IS FILTERED. Attribution is read for the table and the reclaim guidance only -- it does not select
        # a target, so a session whose working directory cannot be read is stopped exactly like one whose can,
        # and simply shows as `unknown`. That is what makes the attribution safe to take from the account being stopped:
        # a unit that misreports its project misleads a reader, and cannot buy itself survival.
        selected_cgroups+=("${cgroup_directory}")
        selected_units+=("${unit}")
        selected_dirs+=("${working_directory}")
        selected_counts+=("${pid_count}")
        if session_is_agent "${unit}"; then
            selected_is_agent+=(true);  agent_count=$(( agent_count + 1 ))
        else
            selected_is_agent+=(false); plumbing_count=$(( plumbing_count + 1 ))
        fi
    done < <(find_session_cgroups)

    if (( ${#selected_cgroups[@]} == 0 )); then
        say_headline "Stop" "No agent session is running."
        log_event info "no session matched ${scope} for ${CALLER} -- nothing stopped"
        return 0
    fi

    # The two counts are reported separately everywhere, never summed into one "session" figure. See the classification
    # block for why, and for why the split is advisory.
    #
    # Two wordings, because "go with them" has no antecedent when no agent session was found -- the shape a RERUN always
    # takes, the manager having been restarted by the run before it.
    local plumbing_clause=""
    if (( plumbing_count && agent_count )); then
        plumbing_clause=" ${plumbing_count} unit(s) of the ${SANDBOX_USER} account's own plumbing (marked below) go with them -- nothing in the account's slice is exempt -- and its user manager is restarted afterwards."
    elif (( plumbing_count )); then
        plumbing_clause=" ${plumbing_count} unit(s) of the ${SANDBOX_USER} account's own plumbing (marked below) are stopped regardless -- nothing in the account's slice is exempt -- and its user manager is restarted afterwards."
    fi

    if ${DRY_RUN}; then
        local dry_headline
        if (( agent_count )); then
            dry_headline="${agent_count} agent session(s) would be terminated.${plumbing_clause} Nothing was changed."
        else
            dry_headline="No agent session is running.${plumbing_clause} Nothing was changed."
        fi
        say_headline "Stop (dry run)" "${dry_headline}"
        print_session_table "would stop"
        log_event info "dry run: ${agent_count} agent session(s) and ${plumbing_count} account unit(s) would be stopped in ${scope} for ${CALLER}"
        return 0
    fi

    local headline
    if (( agent_count == 0 )); then
        headline="No agent session is running.${plumbing_clause}"
    elif ${FORCE_KILL}; then
        headline="${agent_count} agent session(s) will be KILLED immediately (--force): no grace period, no session-end handback, and unsaved work in the current turn is lost.${plumbing_clause}"
    else
        headline="${agent_count} agent session(s) will be terminated, with everything they spawned. Each gets ${GRACE_SECONDS}s to exit before it is killed. No session runs its session-end handback, so reclaim the projects named below afterwards.${plumbing_clause}"
    fi
    say_headline "Terminate agent sessions" "${headline}"
    print_session_table "stop"

    if ! confirm_stop "${agent_count}" "${plumbing_count}"; then
        say_notice MSG-J3U9 "Nothing was stopped"
        log_event notice \
            "${CALLER} declined the stop of ${agent_count} agent session(s) and ${plumbing_count} account unit(s) in ${scope} -- nothing stopped" \
            "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_SCOPE=${scope}" "AI_TOOLS_RESULT=declined"
        return 6
    fi

    local index outcome survivors=0 class_note
    local -a survived_cgroups=()
    for index in "${!selected_cgroups[@]}"; do
        cgroup_directory="${selected_cgroups[index]}"
        unit="${selected_units[index]}"
        working_directory="${selected_dirs[index]}"
        # Carried onto the result lines too, not just the table: in a run with several sessions the table has scrolled
        # off by the time these appear, and these are what the operator watches.
        class_note="$(session_class_note "${unit}")"
        [[ -n "${class_note}" ]] && class_note="  ${class_note}"
        outcome="$(end_session "${cgroup_directory}")"
        case "${outcome}" in
            terminated)
                printf '  stopped    %s  (%s)%s\n' \
                    "$(sanitize "${unit:-unattributed cgroup}")" "$(sanitize "${working_directory:-unknown}")" "${class_note}"
                log_event notice \
                    "stopped session ${unit:-<unattributed>} in ${working_directory:-<none>} on SIGTERM for ${CALLER}" \
                    "AI_TOOLS_UNIT=${unit}" "AI_TOOLS_CWD=${working_directory}" \
                    "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_PASS=sigterm" "AI_TOOLS_RESULT=stopped" ;;
            killed)
                local reason
                ${FORCE_KILL} && reason="--force: no grace period" \
                              || reason="did not exit within ${GRACE_SECONDS}s"
                printf '  KILLED     %s  (%s)  %s%s\n' \
                    "$(sanitize "${unit:-unattributed cgroup}")" "$(sanitize "${working_directory:-unknown}")" "${reason}" "${class_note}"
                log_event warning \
                    "killed session ${unit:-<unattributed>} in ${working_directory:-<none>} with SIGKILL for ${CALLER} (${reason}; it flushed nothing)" \
                    "AI_TOOLS_UNIT=${unit}" "AI_TOOLS_CWD=${working_directory}" \
                    "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_PASS=sigkill" "AI_TOOLS_RESULT=killed" ;;
            *)
                survivors=$(( survivors + 1 ))
                survived_cgroups+=("${cgroup_directory}")
                printf '  STILL UP   %s  (%s)%s\n' \
                    "$(sanitize "${unit:-unattributed cgroup}")" "$(sanitize "${working_directory:-unknown}")" "${class_note}" >&2
                log_event error \
                    "session ${unit:-<unattributed>} in ${working_directory:-<none>} survived SIGKILL for ${CALLER}" \
                    "AI_TOOLS_UNIT=${unit}" "AI_TOOLS_CWD=${working_directory}" \
                    "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_PASS=sigkill" "AI_TOOLS_RESULT=alive" ;;
        esac
    done

    # ONE FINAL SWEEP OF THE WHOLE SLICE, after every session has been dealt with individually. Each cgroup was verified
    # empty on its own, which leaves one theoretical gap: a process that moved between two selected cgroups
    # during the run would be verified gone from the one it left and never looked for in the one it joined. Migration
    # needs write access to the destination's cgroup.procs, which the confined session does not have -- so this closes
    # a gap that should be unreachable, and its value is exactly that: if it ever fires, a confinement invariant has
    # broken and that is far more important than the stop it just caught.
    #
    # RE-ENUMERATED, not merely re-checking what was selected: a session that started while this ran sits in a cgroup
    # that was never selected, so a selection-only recheck would report success with a live session on the host.
    # Re-walking catches it. It does not make the command atomic -- a session starting after this sweep is still missed,
    # and closing that needs a launch/stop gate shared with ai-tools-run -- but it converts a silent miss
    # into a reported one, which is the honest bound. See the exit contract at the top of this file for what a success
    # means.
    #
    # The assertion is the simple one, because the sweep covers every cgroup: the account's slice does not hold any live
    # cgroup at all.
    local -a remaining=()
    while read -r cgroup_directory; do
        [[ -n "${cgroup_directory}" ]] && cgroup_is_live "${cgroup_directory}" \
            && remaining+=("${cgroup_directory}")
    done < <(find_session_cgroups)
    # The sweep's verdict is SET-BASED, not a count comparison. `survivors` counts SESSIONS that reported alive;
    # `remaining` counts LIVE CGROUPS afterwards. Those are different units -- one session cgroup holds many tasks,
    # and a cgroup that appeared after the loop was never in `survivors` at all -- so comparing the two numbers does not
    # detect any case reliably: it misses a new cgroup whenever `survivors` was already non-zero, and can fire
    # on a healthy host.
    #
    # So: ANY live cgroup after a run is a failure, full stop. And separately, a live cgroup that is not one the loop
    # already reported alive is the interesting case -- something appeared in, or moved into, a cgroup after it was
    # verified empty. Within a DELEGATED subtree that move is permitted (see the header), so this is a reachable event
    # and not a broken invariant; it is reported because it means a session started while the command was running,
    # or that something re-entered a cgroup after it was verified empty.
    local -a unexpected=()
    local swept known was_known
    for swept in "${remaining[@]}"; do
        was_known=false
        for known in "${survived_cgroups[@]}"; do
            [[ "${swept}" == "${known}" ]] && { was_known=true; break; }
        done
        ${was_known} || unexpected+=("${swept}")
    done

    if (( ${#unexpected[@]} )); then
        log_event error \
            "final sweep found ${#unexpected[@]} live cgroup(s) that the per-session checks had verified empty -- either a new session started during this run, or a process moved between cgroups inside the delegated subtree to evade the stop" \
            "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_SCOPE=${scope}" "AI_TOOLS_RESULT=reappeared"
        say_error MSG-F9U5 "The final sweep found live cgroups that the per-session checks had verified empty, so a session started while this ran or something re-entered a cgroup after it was emptied. Re-running is safe and is the remedy -- this command is idempotent. To see what is there first:" \
                  "sudo ps -o pid,stat,cgroup,cmd -u ${SANDBOX_USER}"
        for swept in "${unexpected[@]}"; do
            printf '  UNEXPECTED %s\n' "$(sanitize "${swept}")" >&2
            log_event error "unexpected live cgroup after the stop: ${swept}" \
                "AI_TOOLS_CGROUP=${swept}" "AI_TOOLS_RESULT=reappeared"
        done
    fi
    # Whatever the cause, a non-empty sweep means the stop did not complete. Take the sweep's count as authoritative
    # over the loop's, since it is the later and broader observation.
    (( ${#remaining[@]} )) && survivors=${#remaining[@]}

    # Unconditional, and deliberately placed after every verification and before the exit-status decision: the manager
    # has to come back whether or not something survived SIGKILL, and it must not be able to influence what this command
    # reports about the stop itself.
    restore_user_manager

    if (( survivors )); then
        log_event error \
            "stop finished with ${survivors} of ${#selected_cgroups[@]} cgroup(s) still present for ${CALLER}" \
            "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_SCOPE=${scope}" "AI_TOOLS_RESULT=survived"
        # The counts follow the literal rather than opening the line: a coded message is read from its first quoted
        # word, so one beginning with an expansion carries a code the index cannot resolve to any message.
        say_error MSG-W7D3 "Some cgroups survived SIGKILL: ${survivors} of ${#selected_cgroups[@]}. A task only outlives SIGKILL while blocked in an uninterruptible kernel call: it holds no CPU, runs no code and can start nothing new, but only the I/O completing or a reboot clears it. Inspect it with:" \
                  "sudo ps -o pid,stat,wchan:20,cmd -u ${SANDBOX_USER}"
        return 1
    fi
    log_event notice \
        "stopped ${agent_count} agent session(s) and ${plumbing_count} account unit(s) in ${scope} for ${CALLER}, verified empty via cgroup.procs and a final slice sweep" \
        "AI_TOOLS_CALLER=${CALLER}" "AI_TOOLS_SCOPE=${scope}" \
        "AI_TOOLS_COUNT=${#selected_cgroups[@]}" "AI_TOOLS_AGENT_COUNT=${agent_count}"
    # Agent sessions only -- the account's plumbing has no project to reclaim. See the function.
    local -a agent_dirs=(); local index_reclaim
    for index_reclaim in "${!selected_cgroups[@]}"; do
        ${selected_is_agent[index_reclaim]} && agent_dirs+=("${selected_dirs[index_reclaim]}")
    done
    (( ${#agent_dirs[@]} )) && print_reclaim_guidance "${agent_dirs[@]}"
    return 0
}

# ── Entry point ──────────────────────────────────────────────────────────────────────────────
# SOURCING THIS FILE IS INERT: it defines the functions and returns here, having parsed no argument, resolved no host
# state, armed no trap and signalled no process. That is what lets the unit suite drive the enumeration and liveness
# predicates -- the two things a reading review has repeatedly failed to get right -- against a FIXTURE cgroup tree,
# on any host, with no session running and no privilege. It is not a mode and not a hook: there is no environment
# variable to set, no branch inside any function, and no input an invoker can reach that changes what a real run does.
# Running the file is unchanged.
if [[ "${BASH_SOURCE[0]}" != "${0}" ]]; then
    return 0
fi

parse_command_line "$@"
resolve_run_context
resolve_cgroup_layout
main
main_status=$?
STOP_EXIT_REACHED=true
exit "${main_status}"
