session_kit_common

#!/usr/bin/env bash
# Shared helpers for session-kit commands. Source this file; do not execute it.

SESSION_KIT_BIN_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)
SESSION_KIT_RELEASE_DIR=$(cd -- "$SESSION_KIT_BIN_DIR/.." && pwd -P)

# A release directory is installed read-only, so Python's bytecode cache can
# never be written there: every import attempts to create __pycache__ and is
# denied. Nothing is lost by not asking, since nothing was ever cached, and it
# saves a failed syscall per module per command. This only disables writing, so
# a .pyc precompiled by a future install step is still read.
export PYTHONDONTWRITEBYTECODE=1

SK_STATE_DIR=${SESSION_KIT_STATE_DIR:-${XDG_STATE_HOME:-"$HOME/.local/state"}/session-kit}
SK_JOURNAL_DIR=${SESSION_KIT_JOURNAL_DIR:-${XDG_STATE_HOME:-"$HOME/.local/state"}/shpool-journal}
SK_ARCHIVE_DIR=${SESSION_KIT_ARCHIVE_DIR:-${XDG_STATE_HOME:-"$HOME/.local/state"}/shpool-archive}
SK_RECOVERY_DIR=${SESSION_KIT_JOURNAL_RECOVERY_DIR:-${XDG_STATE_HOME:-"$HOME/.local/state"}/shpool-journal-recovery}
SK_START_DIR=${SESSION_KIT_START_DIR:-${XDG_STATE_HOME:-"$HOME/.local/state"}/shpool-start}
SK_INTEGRATION_MARKER=${SESSION_KIT_INTEGRATION_MARKER:-"$SK_STATE_DIR/integration-ready-v1"}
SK_PROJECTS_FILE=${SESSION_KIT_PROJECTS_FILE:-${XDG_CONFIG_HOME:-"$HOME/.config"}/session-kit/projects.tsv}
SK_CONFIG_FILE=${SESSION_KIT_CONFIG:-${XDG_CONFIG_HOME:-"$HOME/.config"}/session-kit/inventory.json}
SK_SHPOOL=${SESSION_KIT_SHPOOL_CMD:-}
SK_INVENTORY_CORE=${SESSION_KIT_INVENTORY_CORE:-"$SESSION_KIT_RELEASE_DIR/lib/session_inventory.py"}
SK_PROJECTS_CORE=${SESSION_KIT_PROJECTS_CORE:-"$SESSION_KIT_RELEASE_DIR/lib/sessionkit_projects/cli.py"}
SK_ACTION_LOG="$SK_STATE_DIR/action-events.jsonl"
# Internal dynamic-scope proof set only after one exact shell death. Never
# inherit a caller-supplied value that could weaken restore duplicate checks.
SK_RESTORE_IGNORE_STALE_ID=

if [[ -z $SK_SHPOOL ]]; then
  SK_SHPOOL=$(command -v shpool 2>/dev/null || true)
fi
if [[ -z $SK_SHPOOL ]]; then
  for sk_candidate in "$HOME/.cargo/bin/shpool" /opt/homebrew/bin/shpool /usr/local/bin/shpool; do
    if [[ -x $sk_candidate ]]; then
      SK_SHPOOL=$sk_candidate
      break
    fi
  done
fi
unset sk_candidate

sk_die() {
  printf 'session-kit: %s\n' "$*" >&2
  return 1
}

sk_require_shpool() {
  [[ -n $SK_SHPOOL && -x $SK_SHPOOL ]] || sk_die "shpool executable unavailable"
}

# Reads a test-only hook variable. Returns 0 and sets SK_TEST_HOOK to its value
# only when the isolated test harness armed it; otherwise returns 1 and leaves
# SK_TEST_HOOK empty.
#
# Every one of these variables replaces evidence the kit would otherwise gather
# from the machine -- the platform, whether a provider process really started.
# Setting an environment variable is not a privilege, so honouring one outside
# the harness would let any process that can reach this shell's environment
# feed session-kit a fabricated answer. The single SESSION_KIT_TESTING gate
# lives here so no caller can forget it, and an ungated attempt is reported
# once rather than silently swallowed.
sk_test_hook() {
  local name=$1
  SK_TEST_HOOK=${!name:-}
  [[ -n $SK_TEST_HOOK ]] || return 1
  if [[ ${SESSION_KIT_TESTING:-0} != 1 ]]; then
    SK_TEST_HOOK=
    if [[ ${SK_TEST_HOOK_NOTED:-} != *"|$name|"* ]]; then
      SK_TEST_HOOK_NOTED="${SK_TEST_HOOK_NOTED:-|}$name|"
      printf 'session-kit: ignoring %s: reserved for isolated tests\n' "$name" >&2
    fi
    return 1
  fi
  return 0
}

sk_platform() {
  if sk_test_hook SESSION_KIT_TEST_PLATFORM; then
    printf '%s\n' "$SK_TEST_HOOK"
  else
    command uname -s 2>/dev/null
  fi
}

sk_boot_id() {
  if [[ -n ${SESSION_KIT_BOOT_ID_FILE:-} ]]; then
    command cat -- "$SESSION_KIT_BOOT_ID_FILE"
  elif [[ $(sk_platform) == Darwin ]]; then
    python3 "$SK_INVENTORY_CORE" platform boot-id
  else
    command cat -- /proc/sys/kernel/random/boot_id
  fi
}

sk_now_unix_ms() {
  if [[ $(sk_platform) == Darwin ]]; then
    python3 -c 'import time; print(time.time_ns() // 1_000_000)'
  else
    date +%s%3N
  fi
}

# Linux keeps its established flock command. Darwin uses the native flock(2)
# binding through Python, holding the lock on the inherited Bash descriptor so
# a killed command releases it with the descriptor instead of leaving stale
# lock state behind.
#
# The wait is bounded on both. An unbounded acquire meant one holder wedged
# inside the daemon stopped every attach, create and close on the machine for
# as long as it stayed wedged, with no message and no way out but finding the
# process by hand. The bound is generous — an honest create holds this lock
# through an attach and its generation proof — so reaching it means the holder
# is stuck, not merely slow.
SK_LOCK_WAIT_SECONDS=${SESSION_KIT_LOCK_WAIT_SECONDS:-120}
[[ $SK_LOCK_WAIT_SECONDS =~ ^[0-9]+$ && $SK_LOCK_WAIT_SECONDS -ge 1 ]] ||
  SK_LOCK_WAIT_SECONDS=120
sk_lock_acquire() {
  local fd=$1 path=$2 seconds=${3:-$SK_LOCK_WAIT_SECONDS}
  [[ $seconds =~ ^[0-9]+$ && $seconds -ge 1 ]] || seconds=$SK_LOCK_WAIT_SECONDS
  if [[ $(sk_platform) != Darwin ]]; then
    command -v flock >/dev/null 2>&1 || {
      sk_die "flock is required"
      return 1
    }
    flock -x -w "$seconds" "$fd" && return 0
    sk_die "another session-kit action has held $(basename -- "$path") for ${seconds}s; nothing changed"
    return 1
  fi
  python3 - "$fd" "$seconds" <<'PY' && return 0
import fcntl
import sys
import time

descriptor, seconds = int(sys.argv[1]), float(sys.argv[2])
deadline = time.monotonic() + seconds
while True:
    try:
        fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
        break
    except BlockingIOError:
        if time.monotonic() >= deadline:
            raise SystemExit(1)
        time.sleep(0.1)
PY
  sk_die "another session-kit action has held $(basename -- "$path") for ${seconds}s; nothing changed"
  return 1
}

sk_lock_release() {
  local fd=$1
  eval "exec ${fd}>&-"
}

sk_timeout() {
  local seconds=$1
  shift
  # Keep the caller's stdin attached to the child. The former here-doc
  # replaced pipeline input with Python source, so the hidden Claude color
  # bootstrap never received its /color command.
  # This is Python source, so shell expansion is deliberately disabled.
  # shellcheck disable=SC2016
  python3 -c '
import ctypes
import os
import signal
import subprocess
import sys
import time

seconds = float(sys.argv[1])
if not 0 < seconds <= 3600 or len(sys.argv) < 3:
    raise SystemExit(2)

# Opt-in: the child must not outlive this wrapper under ANY death, and one
# Ctrl-C must take it with us. Only callers that start a REAL provider ask for
# this; the picker refresh deliberately does not (see _forward).
tied = os.environ.get("SK_TIMEOUT_TIE_CHILD") == "1"


def _tie_to_parent():
    # PR_SET_PDEATHSIG(1): the kernel signals this child the moment its parent
    # dies, including a SIGKILL nothing can trap. Without it a wrapper killed
    # hard leaves a detached provider running forever -- a four-day-old
    # throwaway pair was found alive on the operator box, holding ~250 MB and
    # an account profile, with nothing on the machine able to reap it.
    try:
        ctypes.CDLL("libc.so.6", use_errno=True).prctl(1, int(signal.SIGTERM))
    except OSError:
        pass


process = subprocess.Popen(
    sys.argv[2:],
    start_new_session=True,
    preexec_fn=_tie_to_parent if tied else None,
)


def _descendants(root):
    # `script` puts the provider in its OWN session, so the provider is NOT in
    # the group this wrapper started: one killpg reached `script` and left the
    # real provider running. Walk the tree instead, deepest last, so a signal
    # goes to every process this wrapper is responsible for.
    children = {}
    for entry in os.listdir("/proc"):
        if not entry.isdigit():
            continue
        try:
            with open(f"/proc/{entry}/stat", "rb") as handle:
                fields = handle.read().rpartition(b")")[2].split()
            parent = int(fields[1])
        except (OSError, IndexError, ValueError):
            continue
        children.setdefault(parent, []).append(int(entry))
    found, queue = [], [root]
    while queue:
        current = queue.pop()
        for child in children.get(current, ()):
            if child not in found:
                found.append(child)
                queue.append(child)
    return found


def _signal_everything(number, targets):
    try:
        os.killpg(process.pid, number)
    except (ProcessLookupError, PermissionError):
        pass
    for pid in targets:
        try:
            os.kill(pid, number)
            os.killpg(pid, number)
        except (ProcessLookupError, PermissionError):
            pass


def _forward(number, _frame):
    # The child owns its own process group and session, so nothing that signals
    # this wrapper reaches the real work: a terminated collector left a full
    # status collection running detached, ~150 MB, until it finished by itself.
    # SIGINT is handled ONLY when the caller tied the child to us: bash sets it
    # to ignore for background jobs, and taking it over unasked would make one
    # Ctrl-C fail the live refresh the picker depends on -- but for a real
    # provider, a Ctrl-C the child never receives is exactly how an orphan is
    # made.
    # Collected BEFORE anything is signalled. `script` dies first and the
    # provider is reparented to init within milliseconds, so a tree walked
    # after the first signal no longer contains the process that matters --
    # which is how the provider survived a wrapper that thought it had killed
    # everything.
    targets = _descendants(process.pid)
    _signal_everything(signal.SIGTERM, targets)
    if tied:
        # One SIGTERM and an immediate exit is not cleanup: a provider that
        # ignores TERM simply outlives the Ctrl-C that was meant to stop it,
        # and the caller is already dying, so nothing later gets to try again.
        # Same escalation the timeout path uses, with time to flush first --
        # a provider SIGKILLed mid-write is what leaves a torn record behind.
        time.sleep(1.5)
        _signal_everything(signal.SIGKILL, targets)
    raise SystemExit(128 + number)


_handled = [signal.SIGTERM, signal.SIGHUP]
if tied:
    _handled.append(signal.SIGINT)
for _number in _handled:
    if signal.getsignal(_number) is not signal.SIG_IGN:
        signal.signal(_number, _forward)
try:
    status = process.wait(timeout=seconds)
except subprocess.TimeoutExpired:
    # The tree is collected first, for the same reason the signal path does
    # it: `script` gives the provider its own session, so signalling only the
    # child process group reaches the wrapper around the provider and not the
    # provider itself, and once the wrapper dies the provider is no longer
    # anywhere a later walk would find it.
    _timed_out = _descendants(process.pid)
    _signal_everything(signal.SIGTERM, _timed_out)
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        pass
    _signal_everything(signal.SIGKILL, _timed_out)
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        pass
    raise SystemExit(124)
raise SystemExit(status if status >= 0 else 128 - status)
' "$seconds" "$@"
}

sk_integration_marker_ok() {
  local release_id=$1
  local marker_value=
  local marker_stat=
  local marker_size=
  if [[ -f $SK_INTEGRATION_MARKER && ! -L $SK_INTEGRATION_MARKER ]]; then
    IFS= read -r marker_value < "$SK_INTEGRATION_MARKER" || true
    if [[ $(sk_platform) == Darwin ]]; then
      marker_stat=$(stat -f '%Lp:%u' "$SK_INTEGRATION_MARKER" 2>/dev/null || true)
      marker_size=$(stat -f '%z' "$SK_INTEGRATION_MARKER" 2>/dev/null || true)
    else
      marker_stat=$(stat -c '%a:%u' "$SK_INTEGRATION_MARKER" 2>/dev/null || true)
      marker_size=$(stat -c '%s' "$SK_INTEGRATION_MARKER" 2>/dev/null || true)
    fi
  fi
  local marker_expected="session-kit-integration-v1 $release_id"
  [[ $release_id =~ ^[0-9a-f]{40}$ &&
     $marker_value == "$marker_expected" &&
     $marker_size == $((${#marker_expected} + 1)) &&
     $marker_stat == "600:$(id -u)" ]]
}

# A marker that is missing or names a stale release while the managed login
# block still exists is an inconsistent state by definition: enabling login
# writes both and disabling removes both. Healing rewrites only that state and
# can never override a deliberate disable, whose bashrc block is gone too.
sk_integration_self_heal() {
  [[ ${SESSION_KIT_NO_SELF_HEAL:-0} != 1 &&
     ! -e $SK_STATE_DIR/self-heal-off ]] || return 1
  local release_id=$1
  local bashrc=${SESSION_KIT_BASHRC:-"$HOME/.bashrc"}
  [[ $release_id =~ ^[0-9a-f]{40}$ ]] || return 1
  [[ -f $bashrc && ! -L $bashrc ]] || return 1
  command grep -qF '# >>> session-kit managed integration >>>' "$bashrc" \
    2>/dev/null || return 1
  [[ ! -L $SK_INTEGRATION_MARKER ]] || return 1
  umask 077
  mkdir -p "$SK_STATE_DIR" || return 1
  printf 'session-kit-integration-v1 %s\n' "$release_id" \
    > "$SK_INTEGRATION_MARKER" || return 1
  chmod 600 "$SK_INTEGRATION_MARKER" || return 1
  sk_log_action marker_self_heal repaired || true
  return 0
}

sk_require_integration() {
  local release_id=${SESSION_KIT_RELEASE_ID:-${SESSION_KIT_RELEASE_DIR##*/}}
  sk_integration_marker_ok "$release_id" && return 0
  if sk_integration_self_heal "$release_id" &&
      sk_integration_marker_ok "$release_id"; then
    return 0
  fi
  sk_die "new-session launch is disabled until shell integration is validated for this release"
}

sk_prepare_state() {
  umask 077
  mkdir -p "$SK_STATE_DIR" "$SK_JOURNAL_DIR" "$SK_ARCHIVE_DIR"
  chmod 700 "$SK_STATE_DIR" "$SK_JOURNAL_DIR" "$SK_ARCHIVE_DIR" 2>/dev/null || true
}

# Keep a small local record of picker actions without recording anything that
# identifies a conversation or its contents. The only variable fields are two
# source-controlled enums. Every append also removes entries older than seven
# days and caps the file at 1,000 records / 256 KiB.
sk_log_action() {
  local action=$1 outcome=$2
  [[ $action =~ ^[a-z][a-z_]{0,31}$ &&
     $outcome =~ ^[a-z][a-z_]{0,31}$ ]] || return 1
  python3 - "$SK_ACTION_LOG" "$action" "$outcome" <<'PY' 2>/dev/null
import fcntl
import json
import os
from pathlib import Path
import re
import stat
import sys
import tempfile
import time

path = Path(sys.argv[1])
action, outcome = sys.argv[2:4]
now_ms = time.time_ns() // 1_000_000
cutoff_ms = now_ms - 7 * 24 * 60 * 60 * 1000
max_records = 1_000
max_bytes = 256 * 1024
enum_pattern = re.compile(r"[a-z][a-z_]{0,31}")

path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
os.chmod(path.parent, 0o700)
lock_path = path.with_suffix(path.suffix + ".lock")
lock_fd = os.open(
    lock_path,
    os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0),
    0o600,
)
try:
    os.fchmod(lock_fd, 0o600)
    fcntl.flock(lock_fd, fcntl.LOCK_EX)
    records = []
    try:
        metadata = path.lstat()
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid != os.geteuid()
            or metadata.st_nlink != 1
            or stat.S_IMODE(metadata.st_mode) != 0o600
            or metadata.st_size > max_bytes
        ):
            raise OSError("unsafe action log")
        payload = path.read_bytes()
    except FileNotFoundError:
        payload = b""
    for raw_line in payload.splitlines():
        try:
            item = json.loads(raw_line)
        except (UnicodeDecodeError, ValueError):
            continue
        if (
            isinstance(item, dict)
            and set(item) in (
                {"action", "at_unix_ms", "outcome", "schema_version"},
                # A record may name the session it is about. An audit line
                # that cannot say WHICH session it describes is why a night of
                # launches that started nothing took hours to explain.
                {"action", "at_unix_ms", "outcome", "schema_version", "session"},
            )
            and (
                "session" not in item
                or (
                    isinstance(item.get("session"), str)
                    and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,63}", item["session"])
                )
            )
            and item.get("schema_version") == 1
            and isinstance(item.get("at_unix_ms"), int)
            and not isinstance(item["at_unix_ms"], bool)
            and cutoff_ms <= item["at_unix_ms"] <= now_ms
            and isinstance(item.get("action"), str)
            and enum_pattern.fullmatch(item["action"])
            and isinstance(item.get("outcome"), str)
            and enum_pattern.fullmatch(item["outcome"])
        ):
            records.append(item)
    records.append(
        {
            "action": action,
            "at_unix_ms": now_ms,
            "outcome": outcome,
            "schema_version": 1,
        }
    )
    encoded = [
        json.dumps(item, sort_keys=True, separators=(",", ":")).encode() + b"\n"
        for item in records[-max_records:]
    ]
    kept = []
    total = 0
    for line in reversed(encoded):
        if total + len(line) > max_bytes:
            break
        kept.append(line)
        total += len(line)
    output = b"".join(reversed(kept))
    descriptor, temporary = tempfile.mkstemp(
        prefix=".action-events.", dir=path.parent
    )
    try:
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "wb") as handle:
            descriptor = -1
            handle.write(output)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    finally:
        if descriptor >= 0:
            os.close(descriptor)
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass
finally:
    os.close(lock_fd)
PY
}

sk_snapshot_file() {
  local destination=$1
  "$SESSION_KIT_BIN_DIR/shpool_status" --json > "$destination"
}

sk_strict_snapshot_file() {
  local destination=$1
  "$SESSION_KIT_BIN_DIR/shpool_status" --strict-json > "$destination"
}

sk_guard_snapshot_file() {
  local destination=$1
  "$SESSION_KIT_BIN_DIR/shpool_status" --guard-json > "$destination"
}

sk_lookup_file() {
  local inventory=$1 selector=$2
  python3 - "$inventory" "$selector" <<'PY'
import json, re, sys
path, selector = sys.argv[1:3]
with open(path, encoding="utf-8") as handle:
    data = json.load(handle)
rows = data.get("sessions", [])
matches = []
if selector.isdigit():
    wanted = int(selector)
    has_terminal_numbers = any(
        isinstance(row, dict) and row.get("terminal_number") is not None
        for row in rows
    )
    selector_field = "terminal_number" if has_terminal_numbers else "row"
    matches = [row for row in rows if row.get(selector_field) == wanted]
else:
    matches = [row for row in rows if row.get("shpool_id") == selector]
if len(matches) != 1:
    raise SystemExit(2)
row = matches[0]
identity = row.get("identity") or {}
daemon = data.get("daemon_generation") or {}
conversation_provider = row.get("provider") or "unknown"
conversation_uuid = identity.get("uuid") or ""
if conversation_provider not in {"claude", "codex"} or not conversation_uuid:
    exited_identity = row.get("exited_identity") or {}
    recovery = row.get("recovery") or {}
    exited_provider = row.get("exited_provider")
    exited_uuid = exited_identity.get("uuid")
    if (
        row.get("provider") == "shell"
        and row.get("agent_status") == "provider exited"
        and exited_provider in {"claude", "codex"}
        and row.get("display_provider") == exited_provider
        and exited_identity.get("confidence") == "historical-exact"
        and isinstance(exited_uuid, str)
        and recovery.get("available") is True
        and recovery.get("provider") == exited_provider
        and recovery.get("uuid") == exited_uuid
    ):
        conversation_provider = exited_provider
        conversation_uuid = exited_uuid
    else:
        conversation_provider = ""
        conversation_uuid = ""
raw_id = row.get("shpool_id_raw")
if not isinstance(raw_id, str) or raw_id != row.get("shpool_id"):
    raise SystemExit(2)
values = [
    raw_id,
    str(row.get("started_at_unix_ms") or ""),
    row.get("shpool_status") or "",
    row.get("availability") or "",
    row.get("provider") or "unknown",
    identity.get("uuid") or "",
    row.get("title") or "",
    row.get("cwd") or "",
    str(identity.get("pid") or ""),
    str(identity.get("process_start_ticks") or ""),
    str((row.get("shpool_shell") or {}).get("pid") or ""),
    str((row.get("shpool_shell") or {}).get("process_start_ticks") or ""),
    str(daemon.get("pid") or ""),
    str(daemon.get("process_start_ticks") or ""),
    "true" if row.get("mutation_allowed") is True else "false",
    row.get("mutation_rejection_reason") or "",
    conversation_provider,
    conversation_uuid,
    str(row.get("terminal_number") or ""),
    row.get("account_alias") or "",
    "true" if row.get("account_switch_capable") is True else "false",
    row.get("model") or "",
]
for value in values:
    value = re.sub(r"[\t\r\n]+", " ", str(value))
    print(value)
PY
}

sk_resolve() {
  local inventory=$1 selector=$2
  local -a fields=()
  while IFS= read -r field; do fields+=("$field"); done < <(sk_lookup_file "$inventory" "$selector") || return 2
  [[ ${#fields[@]} -eq 22 ]] || return 2
  SK_ID=${fields[0]}
  SK_STARTED=${fields[1]}
  SK_STATUS=${fields[2]}
  SK_AVAILABILITY=${fields[3]}
  SK_PROVIDER=${fields[4]}
  SK_UUID=${fields[5]}
  SK_TITLE=${fields[6]}
  SK_CWD=${fields[7]}
  SK_PROVIDER_PID=${fields[8]}
  SK_PROVIDER_START=${fields[9]}
  SK_SHELL_PID=${fields[10]}
  SK_SHELL_START=${fields[11]}
  SK_DAEMON_PID=${fields[12]}
  SK_DAEMON_START=${fields[13]}
  SK_MUTATION_ALLOWED=${fields[14]}
  SK_MUTATION_REASON=${fields[15]}
  SK_CONVERSATION_PROVIDER=${fields[16]}
  SK_CONVERSATION_UUID=${fields[17]}
  SK_NUMBER=${fields[18]}
  SK_ACCOUNT_ALIAS=${fields[19]}
  SK_ACCOUNT_CAPABLE=${fields[20]}
  SK_MODEL=${fields[21]}
}

sk_require_mutation_target() {
  [[ ${SK_MUTATION_ALLOWED:-false} == true ]] || {
    sk_die "session ID is display-only and cannot be mutated (${SK_MUTATION_REASON:-unsafe-id})"
    return 1
  }
}

# Report the exact daemon generation holding the Unix listener used by this
# kit's shpool client. This does not issue a client request of its own, so it
# can bracket the one raw request whose author a caller needs to prove.
sk_observe_shpool_holder() {
  python3 "$SK_INVENTORY_CORE" platform shpool-holder-generation 2>/dev/null
}

# True only when the current socket author is the daemon generation carried by
# a guard-live snapshot. PID alone is never enough because it can be reused.
sk_shpool_holder_is() {
  local expected_pid=$1 expected_start=$2 observed
  observed=$(sk_observe_shpool_holder) || return 1
  [[ $observed == "$expected_pid"$'\t'"$expected_start" ]]
}

# Prove the close itself, independently of a client's exit status or a raw
# list losing a name. The exact old shell generation must be gone while the
# bound daemon generation remains unchanged throughout the observation.
sk_exact_shell_gone() {
  local daemon_pid=$1 daemon_start=$2 shell_pid=$3 shell_start=$4
  python3 "$SK_INVENTORY_CORE" platform exact-shell-gone \
    "$daemon_pid" "$daemon_start" "$shell_pid" "$shell_start"
}

# End one exact PID/start-tick generation. Production delivery is pinned by a
# pidfd and re-reads the identity after the fd is open, immediately before
# TERM and (only after a delivered TERM) bounded KILL.
sk_terminate_exact_process() {
  local pid=$1 start_ticks=$2 outcome status=0
  outcome=$(python3 "$SK_INVENTORY_CORE" platform terminate-exact-process \
    "$pid" "$start_ticks" 2>&1) || status=$?
  if (( status != 0 )); then
    [[ -n $outcome ]] || outcome="exact process terminator unavailable"
    printf '%s\n' "$outcome" >&2
    return "$status"
  fi
  printf '%s\n' "$outcome"
}

# The shell is the destructive act. A name is never used to choose what dies;
# the daemon proof remains a postcondition so a stale snapshot cannot turn a
# successful signal into a false close claim.
sk_terminate_exact_shell() {
  local daemon_pid=$1 daemon_start=$2 shell_pid=$3 shell_start=$4 outcome
  outcome=$(sk_terminate_exact_process "$shell_pid" "$shell_start" 2>&1) || {
    # The picker proof can outlive the shell by a few milliseconds.  Already
    # gone is the desired close state, but only while the exact bound daemon
    # generation still brackets that proof.
    if sk_exact_shell_gone "$daemon_pid" "$daemon_start" \
        "$shell_pid" "$shell_start" >/dev/null 2>&1; then
      return 0
    fi
    printf '%s\n' "$outcome" >&2
    return 1
  }
  sk_exact_shell_gone "$daemon_pid" "$daemon_start" \
    "$shell_pid" "$shell_start"
}

# Read one manager entry only while the Unix listener is still owned by the
# exact daemon generation from the guard snapshot.  Prints `present` or `gone`;
# a reused name, malformed payload, wrong lifecycle state, or changed author
# refuses.
sk_bound_manager_entry_state() {
  local id=$1 started=$2 daemon_pid=$3 daemon_start=$4 payload answer status=0
  sk_shpool_holder_is "$daemon_pid" "$daemon_start" || return 1
  payload=$(sk_timeout 5 "$SK_SHPOOL" list --json 2>/dev/null) || return 1
  sk_shpool_holder_is "$daemon_pid" "$daemon_start" || return 1
  answer=$(SK_MANAGER_LIST_JSON=$payload python3 - "$id" "$started" <<'PY'
import json
import os
import sys

session_id, started_text = sys.argv[1:3]
try:
    started = int(started_text)
    payload = json.loads(os.environ["SK_MANAGER_LIST_JSON"])
except (KeyError, ValueError, json.JSONDecodeError):
    raise SystemExit(2)
rows = payload.get("sessions") if isinstance(payload, dict) else payload
if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows):
    raise SystemExit(2)
matches = [row for row in rows if row.get("name") == session_id]
if not matches:
    print("gone")
    raise SystemExit(0)
if len(matches) != 1:
    raise SystemExit(2)
row = matches[0]
if (
    row.get("started_at_unix_ms") != started
    or str(row.get("status", "")).casefold() not in {"disconnected", "attached"}
):
    raise SystemExit(2)
print("present")
PY
  ) || status=$?
  (( status == 0 )) || return 1
  printf '%s\n' "$answer"
}

# Stock shpool 0.11 aborts `kill` bookkeeping when the child PID is already
# gone.  Reattaching is its supported stale-entry recovery path: select_shell
# replaces the exited generation, and /bin/true exits under the attach handler
# so that handler removes the table entry normally.  The reusable name is used
# only after the old instance timestamp, disconnected state, dead shell, and
# exact daemon author have all been proved under the kit's creation lock.
sk_finalize_dead_shpool_entry() {
  local id=$1 daemon_pid=$2 daemon_start=$3 started=$4 state attach_status=0
  state=$(sk_bound_manager_entry_state \
    "$id" "$started" "$daemon_pid" "$daemon_start") || {
    printf 'session-kit: ERROR: cannot prove manager entry %s is the exact dead instance; it was not removed\n' \
      "$id" >&2
    return 1
  }
  [[ $state == present ]] || return 0
  sk_shpool_holder_is "$daemon_pid" "$daemon_start" || {
    printf 'session-kit: ERROR: manager changed before dead entry %s could be finalized; it was not removed\n' \
      "$id" >&2
    return 1
  }
  sk_timeout 10 "$SK_SHPOOL" attach --cmd /bin/true --dir / "$id" \
    </dev/null >/dev/null 2>&1 || attach_status=$?
  sk_shpool_holder_is "$daemon_pid" "$daemon_start" || {
    printf 'session-kit: ERROR: manager changed while finalizing dead entry %s; removal is unproved\n' \
      "$id" >&2
    return 1
  }
  # The attach client in stock 0.11 can report 1 even after the shell's exit
  # landed.  The table postcondition, not that unreliable status, is truth.
  local attempt
  for attempt in 1 2 3 4 5; do
    state=$(sk_bound_manager_entry_state \
      "$id" "$started" "$daemon_pid" "$daemon_start") || break
    [[ $state == gone ]] && return 0
    sleep 0.1
  done
  printf 'session-kit: ERROR: manager entry %s remains after dead-shell finalization (attach status %s)\n' \
    "$id" "$attach_status" >&2
  return 1
}

sk_cleanup_dead_shpool_entry() {
  local id=$1 daemon_pid=$2 daemon_start=$3 shell_pid=$4 shell_start=$5
  local started=${6:-}
  if ! sk_exact_shell_gone "$daemon_pid" "$daemon_start" \
      "$shell_pid" "$shell_start" >/dev/null 2>&1; then
    printf 'session-kit: ERROR: manager entry %s was not finalized because exact shell death is unproved\n' \
      "$id" >&2
    return 1
  fi
  if [[ ! $started =~ ^[0-9]+$ || $started -le 0 ]]; then
    printf 'session-kit: ERROR: manager entry %s was not finalized because its exact start time is unavailable\n' \
      "$id" >&2
    return 1
  fi
  sk_finalize_dead_shpool_entry "$id" "$daemon_pid" "$daemon_start" "$started"
}

# True only when the running process is the exact one recorded earlier. The
# start ticks make a recycled PID impossible to mistake for the original.
sk_proc_is_exact() {
  local pid=$1 start_ticks=$2 stat tail
  local -a fields=()
  [[ $pid =~ ^[0-9]+$ && $start_ticks =~ ^[0-9]+$ ]] || return 1
  (( pid > 0 && start_ticks > 0 )) || return 1
  if [[ $(sk_platform) == Darwin ]]; then
    local current
    current=$(python3 "$SK_INVENTORY_CORE" platform process-info "$pid" 2>/dev/null) || return 1
    [[ ${current#*$'\t'} == "$start_ticks" ]]
    return
  fi
  [[ -r /proc/$pid/stat ]] || return 1
  stat=$(<"/proc/$pid/stat") || return 1
  tail=${stat##*) }
  read -r -a fields <<<"$tail"
  [[ ${#fields[@]} -ge 20 ]] || return 1
  [[ ${fields[19]} == "$start_ticks" ]]
}

# True when a provider process of the expected kind is running underneath the
# exact proven shell of this session.
#
# This exists because a brand-new Codex session cannot be identified from the
# inventory. Codex identity is derived from an open rollout file, and Codex does
# not write one until the human sends a first message -- so a new Codex session
# reports provider "unknown" until it is used. `sp new codex` used to wait for
# that identity, time out, and abandon a perfectly good session without ever
# attaching the human to it. Claude was unaffected because `claude agents
# --json` lists a new session immediately.
#
# The process tree is the correct evidence for "did the provider start": it is
# exactly what the launch record asked the shell to do.
sk_provider_process_present() {
  local shell_pid=$1 shell_start=$2 provider=$3
  if sk_test_hook SESSION_KIT_PROVIDER_PRESENCE_OVERRIDE; then
    # Test and diagnostic hook only; unset in normal operation. Checked before
    # anything reads /proc, because a fixture's shell PIDs are synthetic. This
    # answer is the whole proof that a launch succeeded, so outside the harness
    # sk_test_hook refuses it and the real process tree decides.
    [[ $SK_TEST_HOOK == present ]]
    return
  fi
  sk_proc_is_exact "$shell_pid" "$shell_start" || return 1
  if [[ $(sk_platform) == Darwin ]]; then
    python3 "$SK_INVENTORY_CORE" platform provider-present \
      "$shell_pid" "$shell_start" "$provider" >/dev/null
    return
  fi
  python3 - "$shell_pid" "$provider" <<'PY'
import os
import re
import sys

wanted_pid, provider = int(sys.argv[1]), sys.argv[2]
patterns = {
    "codex": re.compile(r"(^|/)codex($|\s|\0)"),
    "claude": re.compile(r"(^|/)claude($|\s|\0)"),
}
pattern = patterns.get(provider)
if pattern is None:
    raise SystemExit(1)

parents: dict[int, int] = {}
commands: dict[int, str] = {}
for entry in os.listdir("/proc"):
    if not entry.isdigit():
        continue
    pid = int(entry)
    try:
        with open(f"/proc/{pid}/stat", encoding="utf-8", errors="replace") as handle:
            tail = handle.read().rsplit(") ", 1)[-1].split()
        parents[pid] = int(tail[1])
        with open(f"/proc/{pid}/cmdline", "rb") as handle:
            commands[pid] = handle.read(4096).decode("utf-8", "replace")
    except (OSError, IndexError, ValueError):
        continue

for pid, command in commands.items():
    if not pattern.search(command):
        continue
    # Bounded walk to the session shell; a provider is a close descendant.
    walker, depth = pid, 0
    while walker > 1 and depth < 8:
        if walker == wanted_pid:
            raise SystemExit(0)
        walker = parents.get(walker, 0)
        depth += 1
raise SystemExit(1)
PY
}

# A delegated session was given its own copy of the code; closing it gives the
# copy back. Only a copy the kit cut for that session is a candidate, and the
# core refuses one with uncommitted or unmerged work, one somebody is still
# working in, and anything outside the kit's own worktree root -- it says which
# and keeps the directory. The close already happened, so nothing here can fail
# it: the worst case is a line saying the copy was kept.
sk_release_worktree() {
  local id=$1 verdict= status=0
  [[ -n $id ]] || return 0
  verdict=$(python3 "$SK_INVENTORY_CORE" worktree release --shpool-id "$id" \
    2>/dev/null) || status=$?
  if (( status != 0 )); then
    # The copy is still on disk -- nothing removes one except this verb -- so
    # the only cost is a directory. Saying nothing is the part that is not
    # free: the close would print exactly what it prints when there was no
    # copy at all, and the operator would read "handled".
    printf 'session-kit: could not check the working copy for session %s, so it was left where it is. Run sp worktree list to see it.\n' \
      "$id"
    return 0
  fi
  [[ -n $verdict ]] || return 0
  printf '%s' "$verdict"
  return 0
}

# Ending the session's own shell can leave a provider wedged on a terminal
# write alive and holding the old pty. Only processes
# whose PID *and* start ticks still match the recorded generation are signalled.
sk_reap_session_leftovers() {
  local pid start_ticks
  local -a pairs=("$@")
  local index
  for (( index = 0; index + 1 < ${#pairs[@]}; index += 2 )); do
    pid=${pairs[index]}
    start_ticks=${pairs[index + 1]}
    [[ $pid =~ ^[0-9]+$ && $start_ticks =~ ^[0-9]+$ ]] || continue
    (( pid > 0 && start_ticks > 0 )) || continue
    sk_terminate_exact_process "$pid" "$start_ticks" >/dev/null 2>&1 || true
  done
  return 0
}

sk_snapshot_matches_exact() {
  local inventory=$1 expected_id=$2 expected_started=$3 expected_provider=$4
  local expected_uuid=$5 expected_shell_pid=$6 expected_shell_start=$7
  local expected_provider_pid=$8 expected_provider_start=$9
  local expected_daemon_pid=${10} expected_daemon_start=${11}
  local expected_conversation_provider=${12:-}
  local expected_conversation_uuid=${13:-}
  local result=0
  if ! sk_resolve "$inventory" "$expected_id"; then
    result=1
  elif [[ $SK_ID != "$expected_id" || $SK_STARTED != "$expected_started" ]]; then
    result=1
  elif [[ $SK_PROVIDER != "$expected_provider" || $SK_UUID != "$expected_uuid" ]]; then
    result=1
  elif [[ $SK_SHELL_PID != "$expected_shell_pid" || $SK_SHELL_START != "$expected_shell_start" ]]; then
    result=1
  elif [[ $SK_PROVIDER_PID != "$expected_provider_pid" || $SK_PROVIDER_START != "$expected_provider_start" ]]; then
    result=1
  elif [[ $SK_DAEMON_PID != "$expected_daemon_pid" || $SK_DAEMON_START != "$expected_daemon_start" ]]; then
    result=1
  elif [[ $SK_MUTATION_ALLOWED != true ]]; then
    result=1
  elif [[ -n $expected_conversation_provider ]] &&
       [[ $SK_CONVERSATION_PROVIDER != "$expected_conversation_provider" ||
          $SK_CONVERSATION_UUID != "$expected_conversation_uuid" ]]; then
    result=1
  fi
  return "$result"
}

sk_revalidate() {
  local fresh result=0
  fresh=$(mktemp "$SK_STATE_DIR/revalidate.json.XXXXXX") || return 1
  if ! sk_guard_snapshot_file "$fresh" || ! sk_snapshot_matches_exact "$fresh" "$@"; then
    result=1
  fi
  command rm -- "$fresh"
  return "$result"
}

sk_load_picker_proof() {
  local proof=$1 output
  output=$(python3 - "$proof" "$(id -u)" <<'PY'
import json
import os
import re
import stat
import sys

path, expected_uid_raw = sys.argv[1:3]
expected_uid = int(expected_uid_raw)
expected_keys = {
    "schema_version",
    "proof_type",
    "shpool_id",
    "started_at_unix_ms",
    "provider",
    "uuid",
    "provider_pid",
    "provider_process_start_ticks",
    "shell_pid",
    "shell_process_start_ticks",
    "daemon_pid",
    "daemon_process_start_ticks",
}
id_pattern = re.compile(
    r"(?:main(?:[1-9][0-9]*)?|s[0-9]{8}-[0-9]{6}-[1-9][0-9]*(?:-[1-9][0-9]*)?)"
)
uuid_pattern = re.compile(
    r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
)

def reject_duplicates(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate key")
        result[key] = value
    return result

flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
try:
    descriptor = os.open(path, flags)
except OSError:
    raise SystemExit(2)
try:
    before = os.fstat(descriptor)
    if (
        not stat.S_ISREG(before.st_mode)
        or before.st_uid != expected_uid
        or stat.S_IMODE(before.st_mode) != 0o600
        or before.st_nlink != 1
        or before.st_size <= 0
        or before.st_size > 4096
    ):
        raise SystemExit(2)
    chunks = []
    remaining = before.st_size
    while remaining:
        block = os.read(descriptor, min(remaining, 4096))
        if not block:
            raise SystemExit(2)
        chunks.append(block)
        remaining -= len(block)
    if os.read(descriptor, 1):
        raise SystemExit(2)
    after = os.fstat(descriptor)
    stable_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
    if any(getattr(before, key) != getattr(after, key) for key in stable_fields):
        raise SystemExit(2)
finally:
    os.close(descriptor)

raw = b"".join(chunks)
try:
    text = raw.decode("utf-8", "strict")
    value = json.loads(text, object_pairs_hook=reject_duplicates)
except (UnicodeDecodeError, ValueError):
    raise SystemExit(2)
if not isinstance(value, dict) or set(value) != expected_keys:
    raise SystemExit(2)
canonical = (
    json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n"
).encode("utf-8")
if raw != canonical:
    raise SystemExit(2)
if value["schema_version"] != 1 or value["proof_type"] != "session-kit-picker-session-v1":
    raise SystemExit(2)
if not isinstance(value["shpool_id"], str) or not id_pattern.fullmatch(value["shpool_id"]):
    raise SystemExit(2)
if value["provider"] not in {"claude", "codex", "shell", "unknown"}:
    raise SystemExit(2)
if not isinstance(value["uuid"], str):
    raise SystemExit(2)
if value["provider"] in {"shell", "unknown"}:
    if value["uuid"] != "":
        raise SystemExit(2)
elif not uuid_pattern.fullmatch(value["uuid"]):
    raise SystemExit(2)
integer_keys = expected_keys - {
    "proof_type",
    "shpool_id",
    "provider",
    "uuid",
}
for key in integer_keys:
    item = value[key]
    if isinstance(item, bool) or not isinstance(item, int) or item <= 0:
        raise SystemExit(2)
for key in (
    "shpool_id",
    "started_at_unix_ms",
    "provider",
    "uuid",
    "provider_pid",
    "provider_process_start_ticks",
    "shell_pid",
    "shell_process_start_ticks",
    "daemon_pid",
    "daemon_process_start_ticks",
):
    print(value[key])
PY
  ) || {
    sk_die "invalid private picker proof"
    return 1
  }
  local -a fields=()
  while IFS= read -r field; do fields+=("$field"); done <<<"$output"
  [[ ${#fields[@]} -eq 10 ]] || {
    sk_die "invalid private picker proof"
    return 1
  }
  SK_PROOF_ID=${fields[0]}
  SK_PROOF_STARTED=${fields[1]}
  SK_PROOF_PROVIDER=${fields[2]}
  SK_PROOF_UUID=${fields[3]}
  SK_PROOF_PROVIDER_PID=${fields[4]}
  SK_PROOF_PROVIDER_START=${fields[5]}
  SK_PROOF_SHELL_PID=${fields[6]}
  SK_PROOF_SHELL_START=${fields[7]}
  SK_PROOF_DAEMON_PID=${fields[8]}
  SK_PROOF_DAEMON_START=${fields[9]}
}

sk_picker_proof_matches() {
  local inventory=$1
  sk_resolve "$inventory" "$SK_PROOF_ID" || return 1
  [[ $SK_MUTATION_ALLOWED == true &&
     $SK_ID == "$SK_PROOF_ID" &&
     $SK_STARTED == "$SK_PROOF_STARTED" &&
     $SK_PROVIDER == "$SK_PROOF_PROVIDER" &&
     $SK_UUID == "$SK_PROOF_UUID" &&
     $SK_PROVIDER_PID == "$SK_PROOF_PROVIDER_PID" &&
     $SK_PROVIDER_START == "$SK_PROOF_PROVIDER_START" &&
     $SK_SHELL_PID == "$SK_PROOF_SHELL_PID" &&
     $SK_SHELL_START == "$SK_PROOF_SHELL_START" &&
     $SK_DAEMON_PID == "$SK_PROOF_DAEMON_PID" &&
     $SK_DAEMON_START == "$SK_PROOF_DAEMON_START" ]]
}

# The one allowed close race: the private proof still names the exact old
# shell generation, that generation is now gone, and the same manager instance
# has entered the collector's shell-less quarantine.  No other action accepts
# this weaker shape.
sk_picker_proof_matches_dead_shell() {
  local inventory=$1
  sk_exact_shell_gone "$SK_PROOF_DAEMON_PID" "$SK_PROOF_DAEMON_START" \
    "$SK_PROOF_SHELL_PID" "$SK_PROOF_SHELL_START" >/dev/null 2>&1 || return 1
  sk_resolve "$inventory" "$SK_PROOF_ID" || return 1
  [[ $SK_ID == "$SK_PROOF_ID" &&
     $SK_STARTED == "$SK_PROOF_STARTED" &&
     -z $SK_SHELL_PID && -z $SK_SHELL_START &&
     $SK_DAEMON_PID == "$SK_PROOF_DAEMON_PID" &&
     $SK_DAEMON_START == "$SK_PROOF_DAEMON_START" &&
     $SK_MUTATION_ALLOWED == false &&
     $SK_MUTATION_REASON == missing-shell-generation &&
     ( $SK_STATUS == disconnected || $SK_STATUS == Disconnected ) ]]
}

sk_project_lookup() {
  local alias=$1
  [[ -r $SK_PROJECTS_FILE ]] || return 1
  # An ignore row is not a shortcut, so it never resolves as a launch target.
  awk -F '\t' -v wanted="$alias" '
    $0 !~ /^[[:space:]]*#/ && NF >= 3 && $1 == wanted && $2 != "ignore" {
      if ($2 !~ /^(claude|codex|shell)$/ || $3 !~ /^\//) exit 2
      print $1 "\t" $2 "\t" $3
      found=1
      exit
    }
    END { if (!found) exit 1 }
  ' "$SK_PROJECTS_FILE"
}

sk_project_plan() {
  # One line for `sp new`: root, provider, account, model, startup state.
  # The resolver applies a committed session-kit.toml only for a project on
  # this host's list, so what comes back is already what the host agreed to
  # launch. Flags given here still win: they are passed in and returned
  # unchanged.
  local target=$1 provider=$2 account=$3 model=$4
  local -a options=(--projects-file "$SK_PROJECTS_FILE" --state-dir "$SK_STATE_DIR"
                    launch-plan "$target" --format tsv)
  [[ -z $provider ]] || options+=(--provider "$provider")
  [[ -z $account ]] || options+=(--account "$account")
  [[ -z $model ]] || options+=(--model "$model")
  if [[ -t 0 ]]; then options+=(--interactive); fi
  # Say which of the two things went wrong. Without this the caller's "unknown
  # or invalid project" would blame a perfectly good alias for a missing file.
  [[ -f $SK_PROJECTS_CORE && ! -L $SK_PROJECTS_CORE ]] || {
    printf 'session-kit: the project resolver is missing at %s\n' "$SK_PROJECTS_CORE" >&2
    return 1
  }
  python3 "$SK_PROJECTS_CORE" "${options[@]}"
}

sk_allocate_id() {
  local json base candidate counter=0
  json=$("$SK_SHPOOL" list --json 2>/dev/null) || return 1
  python3 -c '
import json,sys
d=json.load(sys.stdin)
rows=d.get("sessions") if isinstance(d,dict) else None
raise SystemExit(0 if isinstance(rows,list) and all(isinstance(r,dict) and isinstance(r.get("name"),str) for r in rows) else 1)
' <<<"$json" || return 1
  base="s$(date +%Y%m%d-%H%M%S)-$$"
  candidate=$base
  while
    python3 -c '
import json,sys
d=json.load(sys.stdin)
rows=d.get("sessions", d if isinstance(d,list) else [])
raise SystemExit(0 if any(r.get("name")==sys.argv[1] for r in rows) else 1)
' "$candidate" <<<"$json" ||
    [[ -e $SK_START_DIR/$candidate || -L $SK_START_DIR/$candidate ||
       -e $SK_START_DIR/$candidate.expected || -L $SK_START_DIR/$candidate.expected ||
       -e $SK_START_DIR/$candidate.account || -L $SK_START_DIR/$candidate.account ]]
  do
    counter=$((counter + 1))
    candidate="${base}-${counter}"
  done
  printf '%s\n' "$candidate"
}

# True only when a fresh manager list still contains this exact raw name. A
# failed create can have many causes; retrying under a different name is
# justified only when the refused name is now visibly occupied.
sk_shpool_name_exists() {
  local wanted=$1 json
  json=$("$SK_SHPOOL" list --json 2>/dev/null) || return 1
  python3 -c '
import json,sys
try:
    value=json.load(sys.stdin)
except Exception:
    raise SystemExit(2)
rows=value.get("sessions") if isinstance(value,dict) else value
if not isinstance(rows,list):
    raise SystemExit(2)
raise SystemExit(0 if any(isinstance(row,dict) and row.get("name")==sys.argv[1]
                          for row in rows) else 1)
' "$wanted" <<<"$json"
}

# Move every pre-creation record to a newly allocated name without following
# links or leaving a split set if one rename fails.
sk_reassign_launch_records() {
  local old_id=$1 new_id=$2 suffix source destination index rollback
  local -a sources=() destinations=()
  [[ $old_id =~ ^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$ &&
     $new_id =~ ^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$ &&
     $old_id != "$new_id" ]] || return 1
  for suffix in '' .account .launch .prompt; do
    source=$SK_START_DIR/$old_id$suffix
    destination=$SK_START_DIR/$new_id$suffix
    [[ ! -e $source && ! -L $source ]] && continue
    [[ -f $source && ! -L $source && ! -e $destination && ! -L $destination ]] ||
      return 1
    sources+=("$source")
    destinations+=("$destination")
  done
  (( ${#sources[@]} > 0 )) || return 1
  for (( index=0; index<${#sources[@]}; index++ )); do
    if ! mv -- "${sources[index]}" "${destinations[index]}"; then
      for (( rollback=index-1; rollback>=0; rollback-- )); do
        mv -- "${destinations[rollback]}" "${sources[rollback]}" 2>/dev/null || true
      done
      return 1
    fi
  done
}

# After a refused background create, allocate and stage a different name only
# when the refused one is now present as a stale entry. The old entry remains
# visible. No path frees it by destruction.
sk_reallocate_stale_launch_name() {
  local old_id=$1 new_id
  sk_shpool_name_exists "$old_id" || return 1
  new_id=$(sk_allocate_id) || return 1
  [[ $new_id != "$old_id" ]] || return 1
  sk_reassign_launch_records "$old_id" "$new_id" || return 1
  printf 'session-kit: stale shpool entry %s blocked creation; allocated fresh session name %s\n' \
    "$old_id" "$new_id" >&2
  printf '%s\n' "$new_id"
}

# Create under an allocated name, retrying only a proved stale-name collision.
# Timeout remains ambiguous and follows the existing generation-capture path;
# it is never retried because the first session may have been created.
sk_attach_new_unique() {
  local initial_id=$1 cwd=$2 origin=${3:-} status=0 replacement attempts=0
  SK_ATTACHED_ID=$initial_id
  SK_ATTACH_STATUS=0
  while (( attempts < 8 )); do
    status=0
    sk_timeout 30 "$SK_SHPOOL" attach --background --dir "$cwd" \
      "$SK_ATTACHED_ID" || status=$?
    if (( status == 0 || status == 124 || status == 137 )); then
      SK_ATTACH_STATUS=$status
      return 0
    fi
    attempts=$((attempts + 1))
    if (( attempts >= 8 )); then
      SK_ATTACH_STATUS=$status
      return 1
    fi
    replacement=$(sk_reallocate_stale_launch_name "$SK_ATTACHED_ID") || {
      SK_ATTACH_STATUS=$status
      return 1
    }
    SK_ATTACHED_ID=$replacement
    if [[ -n $origin ]] && declare -F sk_record_origin >/dev/null 2>&1; then
      sk_record_origin "$SK_ATTACHED_ID" "$origin"
    fi
  done
  SK_ATTACH_STATUS=$status
  return 1
}

sk_write_start_record() {
  local shpool_id=$1 provider=$2 cwd=$3 uuid=${4:-} launch_mode=${5:-}
  [[ $provider =~ ^(claude|codex|shell)$ && $cwd == /* ]] || return 1
  [[ $cwd != *$'\t'* && $cwd != *$'\r'* && $cwd != *$'\n'* &&
     $cwd != *$'\034'* ]] || return 1
  if [[ -z $launch_mode ]]; then
    if [[ -n $uuid ]]; then launch_mode=resume; else launch_mode=new; fi
  fi
  [[ $launch_mode =~ ^(new|resume|fork)$ ]] || return 1
  if [[ -n $uuid ]]; then
    [[ $uuid =~ ^[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}$ ]] || return 1
  fi
  if [[ $launch_mode == new ]]; then
    [[ -z $uuid ]] || return 1
  else
    [[ -n $uuid && $provider != shell ]] || return 1
  fi
  umask 077
  mkdir -p "$SK_START_DIR" || return 1
  chmod 700 "$SK_START_DIR" 2>/dev/null || true
  [[ ! -e $SK_START_DIR/$shpool_id && ! -L $SK_START_DIR/$shpool_id &&
     ! -e $SK_START_DIR/$shpool_id.expected && ! -L $SK_START_DIR/$shpool_id.expected ]] || return 1
  local tmp="$SK_START_DIR/.${shpool_id}.$$"
  printf '%s\t%s\t%s\t%s\n' "$provider" "$cwd" "$uuid" "$launch_mode" > "$tmp" || return 1
  chmod 600 "$tmp" 2>/dev/null || true
  mv -f "$tmp" "$SK_START_DIR/$shpool_id"
}

sk_write_account_record() {
  local shpool_id=$1 provider=$2 alias=$3
  [[ $provider =~ ^(claude|codex)$ &&
     $alias =~ ^[a-z][a-z0-9_-]{0,11}$ ]] || return 1
  umask 077
  mkdir -p "$SK_START_DIR" || return 1
  chmod 700 "$SK_START_DIR" 2>/dev/null || true
  local path="$SK_START_DIR/$shpool_id.account"
  [[ ! -e $path && ! -L $path ]] || return 1
  local tmp="$SK_START_DIR/.${shpool_id}.account.$$"
  printf '%s\t%s\n' "$provider" "$alias" > "$tmp" || return 1
  chmod 600 "$tmp" 2>/dev/null || true
  mv -f -- "$tmp" "$path"
}

sk_capture_session_generation() {
  local shpool_id=$1 creation_floor_ms=$2
  local attempt fresh
  for attempt in {1..20}; do
    fresh=$(mktemp "$SK_STATE_DIR/generation-proof.json.XXXXXX") || return 1
    if sk_guard_snapshot_file "$fresh" 2>/dev/null &&
       python3 -c '
import json,sys
d=json.load(open(sys.argv[1]))
g=d.get("daemon_generation") or {}
ok=(d.get("source")=="live" and d.get("stale") is False and not d.get("warnings")
    and isinstance(g.get("pid"),int) and isinstance(g.get("process_start_ticks"),int))
raise SystemExit(0 if ok else 1)
' "$fresh" &&
       sk_resolve "$fresh" "$shpool_id" &&
       [[ $SK_MUTATION_ALLOWED == true &&
          $SK_STARTED =~ ^[0-9]+$ && $SK_STARTED -ge $creation_floor_ms &&
          $SK_SHELL_PID =~ ^[0-9]+$ && $SK_SHELL_START =~ ^[0-9]+$ &&
          $SK_DAEMON_PID =~ ^[0-9]+$ && $SK_DAEMON_START =~ ^[0-9]+$ ]]; then
      SK_CREATED_STARTED=$SK_STARTED
      SK_CREATED_SHELL_PID=$SK_SHELL_PID
      SK_CREATED_SHELL_START=$SK_SHELL_START
      SK_CREATED_DAEMON_PID=$SK_DAEMON_PID
      SK_CREATED_DAEMON_START=$SK_DAEMON_START
      SK_CREATED_BOOT_ID=$(sk_boot_id) || {
        command rm -- "$fresh"
        return 1
      }
      SK_CREATED_BOOT_ID=${SK_CREATED_BOOT_ID//$'\r'/}
      SK_CREATED_BOOT_ID=${SK_CREATED_BOOT_ID//$'\n'/}
      [[ -n $SK_CREATED_BOOT_ID && $SK_CREATED_BOOT_ID != *$'\t'* ]] || {
        command rm -- "$fresh"
        return 1
      }
      command rm -- "$fresh"
      return 0
    fi
    command rm -- "$fresh"
    (( attempt == 20 )) || sleep 0.1
  done
  return 1
}

sk_write_generation_record() {
  local shpool_id=$1 provider=$2 cwd=$3 uuid=$4
  local boot_id=$5 started=$6 shell_pid=$7 shell_start=$8 daemon_pid=$9 daemon_start=${10}
  local launch_mode=${11:-}
  if [[ -z $launch_mode ]]; then
    if [[ -n $uuid ]]; then launch_mode=resume; else launch_mode=new; fi
  fi
  [[ $launch_mode =~ ^(new|resume|fork)$ ]] || return 1
  if [[ $launch_mode == new ]]; then
    [[ -z $uuid ]] || return 1
  else
    [[ $provider =~ ^(claude|codex)$ && $uuid =~ ^[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}$ ]] || return 1
  fi
  local tmp="$SK_START_DIR/.${shpool_id}.expected.$$"
  printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \
    "$provider" "$cwd" "$boot_id" "$started" "$shell_pid" "$shell_start" \
    "$daemon_pid" "$daemon_start" "$uuid" "$launch_mode" > "$tmp" || return 1
  chmod 600 "$tmp" 2>/dev/null || true
  if ! mv -f "$tmp" "$SK_START_DIR/$shpool_id.expected"; then
    [[ ! -e $tmp ]] || command rm -- "$tmp"
    return 1
  fi
}

sk_quarantine_start_record() {
  local shpool_id=$1 reason=$2
  local failed_dir="$SK_START_DIR/failed"
  local stamp destination
  umask 077
  mkdir -p "$failed_dir" || return 1
  chmod 700 "$failed_dir" 2>/dev/null || true
  stamp=$(date +%Y%m%d-%H%M%S)
  destination="$failed_dir/$shpool_id.$stamp.$reason"
  if [[ -f $SK_START_DIR/$shpool_id && ! -L $SK_START_DIR/$shpool_id ]]; then
    mv "$SK_START_DIR/$shpool_id" "$destination.start" || return 1
  fi
  if [[ -f $SK_START_DIR/$shpool_id.expected && ! -L $SK_START_DIR/$shpool_id.expected ]]; then
    mv "$SK_START_DIR/$shpool_id.expected" "$destination.expected" || return 1
  fi
  if [[ -f $SK_START_DIR/$shpool_id.account && ! -L $SK_START_DIR/$shpool_id.account ]]; then
    mv "$SK_START_DIR/$shpool_id.account" "$destination.account" || return 1
  fi
  printf '%s\n' "$destination"
}

sk_read_start_expectation() {
  local shpool_id=$1
  local start="$SK_START_DIR/$shpool_id" expected="$SK_START_DIR/$shpool_id.expected"
  [[ -f $start && ! -L $start && -f $expected && ! -L $expected ]] || return 1
  python3 - "$start" "$expected" <<'PY' || return 1
import sys

for path in sys.argv[1:]:
    with open(path, "rb") as handle:
        payload = handle.read(8193)
    if (
        not payload
        or len(payload) > 8192
        or not payload.endswith(b"\n")
        or payload.count(b"\n") != 1
        or b"\r" in payload
    ):
        raise SystemExit(1)
PY
  local start_line expected_line
  IFS= read -r start_line < "$start" || return 1
  IFS= read -r expected_line < "$expected" || return 1
  [[ $start_line != *$'\r'* && $start_line != *$'\n'* && $start_line != *$'\034'* &&
     $expected_line != *$'\r'* && $expected_line != *$'\n'* &&
     $expected_line != *$'\034'* ]] || return 1
  local start_tabs=${start_line//[^$'\t']/}
  local expected_tabs=${expected_line//[^$'\t']/}
  [[ ${#start_tabs} -eq 2 || ${#start_tabs} -eq 3 ]] || return 1
  [[ ${#expected_tabs} -eq 8 || ${#expected_tabs} -eq 9 ]] || return 1
  start_line=${start_line//$'\t'/$'\034'}
  expected_line=${expected_line//$'\t'/$'\034'}
  IFS=$'\034' read -r SK_EXPECT_PROVIDER SK_EXPECT_CWD SK_EXPECT_UUID SK_EXPECT_LAUNCH_MODE <<<"$start_line"
  local side_provider side_cwd side_uuid side_launch_mode
  IFS=$'\034' read -r side_provider side_cwd \
    SK_EXPECT_BOOT_ID SK_EXPECT_STARTED SK_EXPECT_SHELL_PID SK_EXPECT_SHELL_START \
    SK_EXPECT_DAEMON_PID SK_EXPECT_DAEMON_START side_uuid side_launch_mode <<<"$expected_line"
  if [[ -z $SK_EXPECT_LAUNCH_MODE ]]; then
    if [[ -n $SK_EXPECT_UUID ]]; then SK_EXPECT_LAUNCH_MODE=resume; else SK_EXPECT_LAUNCH_MODE=new; fi
  fi
  if [[ -z $side_launch_mode ]]; then
    if [[ -n $side_uuid ]]; then side_launch_mode=resume; else side_launch_mode=new; fi
  fi
  [[ $side_provider == "$SK_EXPECT_PROVIDER" &&
     $side_cwd == "$SK_EXPECT_CWD" &&
     $side_uuid == "$SK_EXPECT_UUID" &&
     $side_launch_mode == "$SK_EXPECT_LAUNCH_MODE" ]] || return 1
  [[ $SK_EXPECT_LAUNCH_MODE =~ ^(new|resume|fork)$ ]] || return 1
  if [[ $SK_EXPECT_LAUNCH_MODE == new ]]; then
    [[ -z $SK_EXPECT_UUID ]] || return 1
  else
    [[ $SK_EXPECT_PROVIDER =~ ^(claude|codex)$ &&
       $SK_EXPECT_UUID =~ ^[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}$ ]] || return 1
  fi
  [[ -n $SK_EXPECT_BOOT_ID && $SK_EXPECT_BOOT_ID != *$'\t'* &&
     $SK_EXPECT_BOOT_ID != *$'\r'* && $SK_EXPECT_BOOT_ID != *$'\n'* &&
     $SK_EXPECT_STARTED =~ ^[0-9]+$ &&
     $SK_EXPECT_SHELL_PID =~ ^[0-9]+$ && $SK_EXPECT_SHELL_START =~ ^[0-9]+$ &&
     $SK_EXPECT_DAEMON_PID =~ ^[0-9]+$ && $SK_EXPECT_DAEMON_START =~ ^[0-9]+$ ]]
}

sk_clear_start_record() {
  local shpool_id=$1 expected_provider=$2 expected_cwd=$3 expected_uuid=$4
  local expected_boot_id=$5 expected_started=$6 expected_shell_pid=$7 expected_shell_start=$8
  local expected_daemon_pid=$9 expected_daemon_start=${10}
  local expected_launch_mode=${11:-}
  if [[ -z $expected_launch_mode ]]; then
    if [[ -n $expected_uuid ]]; then expected_launch_mode=resume; else expected_launch_mode=new; fi
  fi
  local path="$SK_START_DIR/$shpool_id"
  local expected_path="$path.expected"
  exec 8>"$SK_STATE_DIR/create.lock" || return 1
  sk_lock_acquire 8 "$SK_STATE_DIR/create.lock" || { exec 8>&-; return 1; }
  # The managed shell consumes both launch records immediately before it
  # starts the provider.  That is the normal successful path: leaving them in
  # place would let a later `exec bash -i` launch the same request again if the
  # creator died before observing the provider.  Once the exact session
  # generation and provider have been proven by the caller, an already
  # consumed pair is therefore equivalent to clearing it here.
  if [[ ! -e $path && ! -L $path && ! -e $expected_path && ! -L $expected_path ]]; then
    sk_lock_release 8
    return 0
  fi
  sk_read_start_expectation "$shpool_id" || { sk_lock_release 8; return 1; }
  [[ $SK_EXPECT_PROVIDER == "$expected_provider" &&
     $SK_EXPECT_CWD == "$expected_cwd" && $SK_EXPECT_UUID == "$expected_uuid" &&
     $SK_EXPECT_LAUNCH_MODE == "$expected_launch_mode" &&
     $SK_EXPECT_BOOT_ID == "$expected_boot_id" &&
     $SK_EXPECT_STARTED == "$expected_started" &&
     $SK_EXPECT_SHELL_PID == "$expected_shell_pid" &&
     $SK_EXPECT_SHELL_START == "$expected_shell_start" &&
     $SK_EXPECT_DAEMON_PID == "$expected_daemon_pid" &&
     $SK_EXPECT_DAEMON_START == "$expected_daemon_start" ]] || {
       sk_lock_release 8
       return 1
     }
  command rm -- "$path" "$expected_path"
  sk_lock_release 8
}

sk_wait_for_provider() {
  local shpool_id=$1 expected_provider=$2 expected_cwd=$3 expected_uuid=$4
  local expected_boot_id=$5 expected_started=$6 expected_shell_pid=$7 expected_shell_start=$8
  local expected_daemon_pid=$9 expected_daemon_start=${10}
  local expected_launch_mode=${11:-}
  if [[ -z $expected_launch_mode ]]; then
    if [[ -n $expected_uuid ]]; then expected_launch_mode=resume; else expected_launch_mode=new; fi
  fi
  [[ $expected_launch_mode =~ ^(new|resume|fork)$ ]] || return 1
  # Resuming a large conversation on a loaded box takes far longer than the
  # original 15s budget: an 11 MB rollout plus a busy machine regularly needs
  # a minute or more before the provider publishes its identity. Waiting
  # costs nothing when the launch succeeds -- the loop returns the moment the
  # identity matches.
  local attempts=${SESSION_KIT_PROVIDER_PROOF_ATTEMPTS:-480}
  [[ $attempts =~ ^[0-9]+$ && $attempts -ge 1 && $attempts -le 2400 ]] || return 1
  local attempt fresh
  for (( attempt=1; attempt<=attempts; attempt++ )); do
    local current_boot_id=
    current_boot_id=$(sk_boot_id) || return 1
    current_boot_id=${current_boot_id//$'\r'/}
    current_boot_id=${current_boot_id//$'\n'/}
    fresh=$(mktemp "$SK_STATE_DIR/provider-proof.json.XXXXXX") || return 1
    local identity_matches=false
    if sk_guard_snapshot_file "$fresh" 2>/dev/null &&
       sk_resolve "$fresh" "$shpool_id"; then
      case "$expected_launch_mode" in
        # A guard-live inventory already requires an exact provider UUID for
        # Claude/Codex. The fixture guard is intentionally smaller, so new
        # launches need only prove that no source UUID was requested here.
        new) [[ -z $expected_uuid ]] && identity_matches=true ;;
        resume) [[ -n $expected_uuid && $SK_UUID == "$expected_uuid" ]] && identity_matches=true ;;
        fork)
          if [[ -n $expected_uuid && -n $SK_UUID && $SK_UUID != "$expected_uuid" ]] &&
             python3 - "$fresh" "$expected_provider" "$SK_UUID" <<'PY'
import json
import sys

with open(sys.argv[1], encoding="utf-8") as handle:
    inventory = json.load(handle)
wanted_provider = sys.argv[2]
wanted_uuid = sys.argv[3]
rows = list(inventory.get("sessions", ())) + list(inventory.get("outside_agents", ()))
matches = [
    row for row in rows
    if isinstance(row, dict)
    and row.get("provider") == wanted_provider
    and isinstance(row.get("identity"), dict)
    and row["identity"].get("uuid") == wanted_uuid
    and row["identity"].get("confidence") == "exact"
]
raise SystemExit(0 if len(matches) == 1 else 1)
PY
          then
            identity_matches=true
          fi
          ;;
      esac
    fi
    # A new session's provider may not be classifiable yet -- Codex publishes
    # its identity only once it has a rollout, which needs a first message.
    # Accept the process tree as proof of startup in that case, and only in
    # that case: a resume or fork must still match its exact conversation.
    local provider_matches=false
    if [[ $SK_PROVIDER == "$expected_provider" ]]; then
      provider_matches=true
    elif [[ $expected_launch_mode == new && $SK_PROVIDER == unknown ]] &&
         sk_provider_process_present "$SK_SHELL_PID" "$SK_SHELL_START" "$expected_provider"; then
      provider_matches=true
    fi
    if [[ $identity_matches == true &&
          $SK_MUTATION_ALLOWED == true &&
          $provider_matches == true &&
          $current_boot_id == "$expected_boot_id" &&
          $SK_CWD == "$expected_cwd" &&
          $SK_STARTED == "$expected_started" &&
          $SK_SHELL_PID == "$expected_shell_pid" &&
          $SK_SHELL_START == "$expected_shell_start" &&
          $SK_DAEMON_PID == "$expected_daemon_pid" &&
          $SK_DAEMON_START == "$expected_daemon_start" ]]; then
      SK_PROVEN_UUID=$SK_UUID
      command rm -- "$fresh"
      sk_clear_start_record "$shpool_id" "$expected_provider" "$expected_cwd" "$expected_uuid" \
        "$expected_boot_id" "$expected_started" "$expected_shell_pid" "$expected_shell_start" \
        "$expected_daemon_pid" "$expected_daemon_start" "$expected_launch_mode" || return 1
      return 0
    fi
    command rm -- "$fresh"
    # A human is usually watching this wait, and it can run for minutes on a
    # loaded box. Silence reads as a hang; say what is still being waited on.
    # Attempts sleep 0.25s plus inventory time, so attempt/4 is a floor.
    if [[ -t 1 ]] && (( attempt % 40 == 0 )); then
      printf '  still waiting for %s to publish its identity (%ss+ elapsed)\n' \
        "$expected_provider" "$(( attempt / 4 ))"
    fi
    (( attempt == attempts )) || sleep 0.25
  done
  return 1
}

# Provider names as a person reads them. Claude and Codex are product names and
# are always capitalized; a plain shell is `shell`, a common noun, and stays
# lowercase. One helper so no print site has to remember.
sk_provider_name() {
  case "${1:-}" in
    claude) printf 'Claude' ;;
    codex) printf 'Codex' ;;
    *) printf '%s' "${1:-}" ;;
  esac
}

# How a session is named to a person: its session number and its own title,
# or its provider when it has neither. Never an internal ID — no human-readable
# Session Kit output prints one, anywhere.
# The kit owns the terminal tab name (K3), on both providers.
#
# Claude Code's documented off-switch cannot be part of this. The live drill in
# tools/title-ownership-drill (2026-08-13, Claude Code 2.1.229) proved that
# CLAUDE_CODE_DISABLE_TERMINAL_TITLE=1 silences EVERY title write from that
# window, the kit's own hook output included: with the switch unset a hook
# asking for a title produces `ESC]0;<glyph> <kit name>BEL`, and with it set the
# same session writes no title sequence at all, for sessionTitle and for a
# hook-emitted terminalSequence alike. So the kit never sets that variable. It
# writes the tab name itself at the moments no provider hook fires -- opening,
# attaching, and taking over a session -- and lets the provider's in-window hook
# carry renames from there. A restored conversation takes its name when it is
# opened; the name and colour themselves are written into the provider's own
# store before the restore launches (a restore ends at the picker or a prompt,
# not inside the session, so naming the tab there would label a terminal that is
# not in that session).
#
# This lives here rather than beside the attach primitives because both front
# doors need it: `sp` (open, attach, takeover) and the login picker (its own
# open, and the "session kit" name it restores on the way back).
#
# Kill switch: SESSION_KIT_TAB_TITLE=off writes nothing at all, on either
# provider (the Codex launch override in bashrc/shpool.bashrc reads the same
# variable). Rollback is that one variable.
sk_tab_title() {
  local name=${1:-} switch=${SESSION_KIT_TAB_TITLE:-}
  # Read exactly as `session-kit doctor` reads it, or the two disagree about
  # whether the tab name is on: strip, then casefold.
  switch=${switch#"${switch%%[![:space:]]*}"}
  switch=${switch%"${switch##*[![:space:]]}"}
  [[ ${switch,,} != off ]] || return 0
  [[ -t 1 ]] || return 0
  # A session title is provider metadata: it can carry control bytes, and a
  # control byte inside an OSC string is a terminal command. C1 (0x80-0x9F)
  # counts -- 0x9C is ST and 0x9B is CSI, so an untrimmed C1 byte ends the OSC
  # string early and the rest is parsed as a control sequence. LC_ALL=C keeps
  # `tr` working on bytes rather than on the locale's idea of a character.
  name=$(printf '%s' "$name" |
    LC_ALL=C tr -d '\000-\037\177-\237' |
    LC_ALL=C tr -s '[:space:]' ' ')
  name=${name#"${name%%[![:space:]]*}"}
  name=${name%"${name##*[![:space:]]}"}
  [[ -n $name ]] || return 0
  printf '\033]0;%s\007' "${name:0:64}"
}

sk_human_label() {
  local title=${1:-} provider=${2:-} number=${3:-}
  # A title is provider metadata, so it can carry terminal control bytes.
  # Scrub before any of it reaches a screen.
  title=$(printf '%s' "$title" | tr -d '[:cntrl:]' | tr -s '[:space:]' ' ')
  title=${title#"${title%%[![:space:]]*}"}
  title=${title%"${title##*[![:space:]]}"}
  if [[ -n $number && -n $title ]]; then
    printf 'session %s (%s)' "$number" "$title"
  elif [[ -n $number ]]; then
    printf 'session %s' "$number"
  elif [[ -n $title ]]; then
    printf '%s' "$title"
  elif [[ -n $provider ]]; then
    printf 'the %s session' "$(sk_provider_name "$provider")"
  else
    printf 'the session'
  fi
}

# Nothing is ever asked (operator decision, permanent). A person already chose
# an exact session under a frozen proof; the guard-and-revalidate path that
# follows every caller is the safety net, and a question only added friction to
# a decision already made. Mistakes are undone by restore, not prevented by a
# prompt.
#
# The two halves that remain:
#
#   * a person at a terminal is told what is about to happen, by name, and the
#     action proceeds;
#   * automation, which has no one to tell, must name the exact session it is
#     acting on in SESSION_KIT_CONFIRM_ID. That half is a proven safety net —
#     a pipe feeding "y" was never a person reading a label — and it stays.
#
# The exact ID travels as an argument and is never printed back.
sk_confirm_exact() {
  local action=$1 id=$2 title=$3 provider=$4 number=${5:-}
  if [[ ${SESSION_KIT_NONINTERACTIVE:-0} == 1 || ! -t 0 ]]; then
    [[ ${SESSION_KIT_CONFIRM_ID:-} == "$id" ]] || return 1
  fi
  printf '%s %s.\n' "$action" "$(sk_human_label "$title" "$provider" "$number")"
  return 0
}

sk_recovery_journal() {
  local id=$1 map="$SK_RECOVERY_DIR/current-map.tsv"
  [[ -r $map ]] || return 1
  awk -F '\t' -v wanted="$id" '$1 == wanted && $2 ~ /^\// { print $2; exit }' "$map"
}