#!/usr/bin/env bash
# SPDX-License-Identifier: GPL-3.0-or-later
# SPDX-FileCopyrightText: 2026 Martin J. Gallagher

#
# provost — turn command output into data.
#
# PROTOTYPE (proposal 0001). Implements the product's one path — capture a
# command's output, auto-detect its format, present it as a tidy, one-row-per-
# record dataset — on a git-backed store:
#
#     provost run <cmd>        execute a binary, capture its output as a commit
#     cmd | provost            inject stdin the same way
#     provost source <N>       show the original source data behind dataset row #N
#     provost log / restore    list captures / bring a capture's commit back up
#     provost ls / show        list datasets / present one as a tidy table
#     provost stats            statistics on a dataset
#
# The extract engine (all strategies) is bundled in this repo at provost/extract/.
#
# STORE MODEL (Stage A): the workspace is a git repo. By default captures go to
# a GLOBAL store (~/.provost, override with PROVOST_HOME) so you can capture from any
# directory. Every dataset row carries a globally-unique visible index (#N);
# `provost source N` maps it back to the exact captured input that produced it.
# The store may be shared by multiple users (point PROVOST_HOME at a shared,
# group-writable dir); a mkdir-based lock serialises concurrent writers.
#
# CODE vs DATA: executable code (the extract/analyze engines, the transform
# module) is loaded ONLY from provost's own install dir ($PROVOST_ROOT, the
# directory this script really lives in) — never from the data store, which a
# shared store's other members can write to.
#
set -euo pipefail

# The top-level VERSION file is the single source of truth (pyproject reads
# it too, so `provost --version` and the wheel can never disagree); the
# literal is only a fallback for a stripped-down copy shipped without it.
PROVOST_VERSION="0.0.0-unknown"
if [[ -f "${BASH_SOURCE[0]%/*}/VERSION" ]]; then
    read -r PROVOST_VERSION < "${BASH_SOURCE[0]%/*}/VERSION" || true
fi

# ---------------------------------------------------------------------------
# Config (env-overridable; PROVOST_* is the post-rename namespace)
# ---------------------------------------------------------------------------
# Make a path absolute (relative to $PWD) without requiring it to exist.
_abspath () { case "$1" in /*) printf '%s\n' "$1" ;; *) printf '%s/%s\n' "$PWD" "$1" ;; esac; }

# Workspace resolution:
#   1. $PROVOST_WORKSPACE explicit override (used by tests / one-off stores)
#   2. a controlled directory found by walking up from CWD
#   3. else the global store: $PROVOST_HOME, or ~/.provost
_default_home () { _abspath "${PROVOST_HOME:-$HOME/.provost}"; }

# The real directory of this script, following symlinks — so provost finds its
# bundled extract/analyze/transform trees even when invoked via a symlink on
# PATH (e.g. after install.sh). Computed once; used by all path resolvers.
_provost_root () {
    local src="${BASH_SOURCE[0]}" d
    if command -v readlink >/dev/null 2>&1; then
        src="$(readlink -f "$src" 2>/dev/null || echo "$src")"
    fi
    while [[ -L "$src" ]]; do
        d="$(cd -P "$(dirname "$src")" && pwd)"; src="$(readlink "$src")"
        [[ "$src" != /* ]] && src="$d/$src"
    done
    cd -P "$(dirname "$src")" && pwd
}
PROVOST_ROOT="$(_provost_root)"

# A controlled directory holds its instance at <dir>/.provost with a CONTROLLED
# marker (which the global store lacks). Walk up from $start to find one.
_find_controlling_dir () {
    local here="$1"
    here="$(cd "$here" 2>/dev/null && pwd)" || return 1
    while [[ -n "$here" && "$here" != "/" ]]; do
        [[ -f "$here/.provost/CONTROLLED" ]] && { printf '%s\n' "$here"; return 0; }
        here="$(dirname "$here")"
    done
    [[ -f "/.provost/CONTROLLED" ]] && { printf '/\n'; return 0; }
    return 1
}

PROVOST_WORKSPACE_EXPLICIT="${PROVOST_WORKSPACE:+1}"
if [[ -n "$PROVOST_WORKSPACE_EXPLICIT" ]]; then
    PROVOST_WORKSPACE="$(_abspath "$PROVOST_WORKSPACE")"
    PROVOST_STORE_KIND="explicit override"
else
    _ctl="$(_find_controlling_dir "$PWD" 2>/dev/null || true)"
    if [[ -n "$_ctl" ]]; then
        PROVOST_WORKSPACE="$_ctl/.provost"; PROVOST_STORE_KIND="controlled: $_ctl"
    else
        PROVOST_WORKSPACE="$(_default_home)"; PROVOST_STORE_KIND="global"
    fi
    unset _ctl
fi
PROVOST_MIN_CONFIDENCE="${PROVOST_MIN_CONFIDENCE:-50}"
# Quality gate: a parse that kept too little of the input (line coverage) or
# put values under the wrong headers (column-type consistency) is held back for
# triage instead of filed. 0 turns a check off. See _quality_exempt.
PROVOST_MIN_COVERAGE="${PROVOST_MIN_COVERAGE:-80}"
PROVOST_MIN_CONSISTENCY="${PROVOST_MIN_CONSISTENCY:-70}"
PROVOST_MAX_INPUT="${PROVOST_MAX_INPUT:-10485760}"   # 10 MiB
PROVOST_FILES_MAX_SIZE="${PROVOST_FILES_MAX_SIZE:-1048576}"  # a controlled dir's files kept per run up to this size (0 = none)
PROVOST_TEMP_DIR="${TMPDIR:-/tmp}"
PROVOST_LOCK_TIMEOUT="${PROVOST_LOCK_TIMEOUT:-30}"   # seconds to wait for the store lock
PROVOST_LOCK_MAX_AGE="${PROVOST_LOCK_MAX_AGE:-3600}" # seconds after which any lock is stale
PROVOST_RECONCILE_OVERLAP="${PROVOST_RECONCILE_OVERLAP:-60}"  # % column overlap to union vs split a variant
PROVOST_MULTI="${PROVOST_MULTI:-0}"                  # 1 = file every dataset an input holds (--multi)
PROVOST_CONTEXT_COLUMNS="${PROVOST_CONTEXT_COLUMNS:-0}" # 1 = '#% key=value' context becomes row columns (--context-columns)
PROVOST_AUTO_MULTI="${PROVOST_AUTO_MULTI:-1}"        # 1 = segment a parse the quality gate would hold (see _auto_multi)
PROVOST_MODEL="${PROVOST_MODEL:-0}"                  # 1 = try the learned layout model on a parse the gate would hold (see _layout_model)
PROVOST_LLM="${PROVOST_LLM:-0}"                      # 1 = use (or have Claude write) a parsing template for a parse the gate would hold (see _llm_template)
PROVOST_LLM_MODEL="${PROVOST_LLM_MODEL:-claude-opus-5-5}"   # the model that writes templates
PROVOST_QUIET="${PROVOST_QUIET:-0}"                  # 1 = no progress/ok chatter on stderr
PROVOST_ASYNC="${PROVOST_ASYNC:-}"                   # ''/auto = background only when interactive

# ---------------------------------------------------------------------------
# Tiny UI helpers
# ---------------------------------------------------------------------------
_is_tty () { [[ -t 2 ]]; }
if _is_tty; then C_OK=$'\033[32m'; C_DIM=$'\033[2m'; C_ERR=$'\033[31m'; C_RST=$'\033[0m'
else C_OK=""; C_DIM=""; C_ERR=""; C_RST=""; fi
info () { [[ "$PROVOST_QUIET" == 1 ]] || printf '%s\n' "$*" >&2; }
ok ()   { [[ "$PROVOST_QUIET" == 1 ]] || printf '%s ok%s %s\n' "$C_OK" "$C_RST" "$*" >&2; }
warn () { printf '%swarning%s %s\n' "$C_ERR" "$C_RST" "$*" >&2; }
die ()  { printf '%serror%s %s\n' "$C_ERR" "$C_RST" "$*" >&2; exit 1; }
# A usage error: the command line itself is wrong. stderr, exit 2.
usage_die () { printf 'provost: %s\n' "$*" >&2; exit 2; }

# ---------------------------------------------------------------------------
# Validation — every user-controlled value that reaches arithmetic, a path, or
# a git ref is checked first. Bash evaluates arithmetic operands recursively
# (`x='a[$(cmd)]'; (( x ))` runs cmd), so numeric settings MUST be validated
# before they are ever compared or computed with.
# ---------------------------------------------------------------------------
# Locale-proof character classes ([A-Z] ranges are collation-dependent).
_ALNUM='abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
_NAME_RE="^[${_ALNUM}][${_ALNUM}._-]*\$"

# _validate_uint LABEL VALUE MIN [MAX] — whole number in range, else exit 2.
_validate_uint () {
    local label="$1" v="$2" lo="$3" hi="${4:-}"
    [[ "$v" =~ ^[0-9]{1,12}$ ]] || usage_die "$label must be a whole number (got '$v')."
    v=$((10#$v))
    (( v >= lo )) || usage_die "$label must be at least $lo (got $v)."
    if [[ -n "$hi" ]] && (( v > hi )); then usage_die "$label must be at most $hi (got $v)."; fi
}
# Normalise a validated number (drop leading zeros so later $(( )) never
# reads it as octal).
_norm_uint () { printf '%d' "$((10#$1))"; }

_validate_settings () {
    _validate_uint PROVOST_MIN_CONFIDENCE    "$PROVOST_MIN_CONFIDENCE"    0 101
    _validate_uint PROVOST_MIN_COVERAGE      "$PROVOST_MIN_COVERAGE"      0 100
    _validate_uint PROVOST_MIN_CONSISTENCY   "$PROVOST_MIN_CONSISTENCY"   0 100
    _validate_uint PROVOST_RECONCILE_OVERLAP "$PROVOST_RECONCILE_OVERLAP" 0 101
    _validate_uint PROVOST_LOCK_TIMEOUT      "$PROVOST_LOCK_TIMEOUT"      0 86400
    _validate_uint PROVOST_LOCK_MAX_AGE      "$PROVOST_LOCK_MAX_AGE"      1
    _validate_uint PROVOST_MAX_INPUT         "$PROVOST_MAX_INPUT"         1
    _validate_uint PROVOST_FILES_MAX_SIZE    "$PROVOST_FILES_MAX_SIZE"    0
    # (validated above, so these arithmetic reads are safe)
    PROVOST_MIN_CONFIDENCE=$((10#$PROVOST_MIN_CONFIDENCE))
    PROVOST_MIN_COVERAGE=$((10#$PROVOST_MIN_COVERAGE))
    PROVOST_MIN_CONSISTENCY=$((10#$PROVOST_MIN_CONSISTENCY))
    PROVOST_RECONCILE_OVERLAP=$((10#$PROVOST_RECONCILE_OVERLAP))
    PROVOST_LOCK_TIMEOUT=$((10#$PROVOST_LOCK_TIMEOUT))
    PROVOST_LOCK_MAX_AGE=$((10#$PROVOST_LOCK_MAX_AGE))
    PROVOST_MAX_INPUT=$((10#$PROVOST_MAX_INPUT))
    PROVOST_FILES_MAX_SIZE=$((10#$PROVOST_FILES_MAX_SIZE))
    [[ "$PROVOST_QUIET" == 0 || "$PROVOST_QUIET" == 1 ]] || usage_die "PROVOST_QUIET must be 0 or 1 (got '$PROVOST_QUIET')."
    [[ "$PROVOST_MULTI" == 0 || "$PROVOST_MULTI" == 1 ]] || usage_die "PROVOST_MULTI must be 0 or 1 (got '$PROVOST_MULTI')."
    [[ "$PROVOST_CONTEXT_COLUMNS" == 0 || "$PROVOST_CONTEXT_COLUMNS" == 1 ]] || usage_die "PROVOST_CONTEXT_COLUMNS must be 0 or 1 (got '$PROVOST_CONTEXT_COLUMNS')."
    [[ "$PROVOST_AUTO_MULTI" == 0 || "$PROVOST_AUTO_MULTI" == 1 ]] || usage_die "PROVOST_AUTO_MULTI must be 0 or 1 (got '$PROVOST_AUTO_MULTI')."
    [[ "$PROVOST_MODEL" == 0 || "$PROVOST_MODEL" == 1 ]] || usage_die "PROVOST_MODEL must be 0 or 1 (got '$PROVOST_MODEL')."
    [[ "$PROVOST_LLM" == 0 || "$PROVOST_LLM" == 1 ]] || usage_die "PROVOST_LLM must be 0 or 1 (got '$PROVOST_LLM')."
    [[ "$PROVOST_LLM_MODEL" =~ ^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$ ]] || usage_die "PROVOST_LLM_MODEL must be a model id (got '$PROVOST_LLM_MODEL')."
    case "$PROVOST_ASYNC" in ""|auto|0|1) ;; *) usage_die "PROVOST_ASYNC must be 0, 1 or auto (got '$PROVOST_ASYNC')." ;; esac
}

# A dataset name is one path component: [A-Za-z0-9][A-Za-z0-9._-]*, no "..".
_valid_name () { [[ "$1" =~ $_NAME_RE && "$1" != *..* ]]; }
_require_name () { _valid_name "$1" || usage_die "invalid dataset name '$1' (allowed: letters, digits, '.', '_', '-'; must start with a letter or digit; no '..')."; }
_valid_capid () { [[ "$1" =~ ^c[0-9]{1,12}$ ]]; }
# A capture id from what a user types for one: c0007, c7, 7 or capture-c0007
# all mean c0007. Prints it; fails on anything else.
_capid_of () {
    local t="${1#capture-}"
    [[ "$t" =~ ^c?([0-9]{1,12})$ ]] || return 1
    printf 'c%04d\n' "$((10#${BASH_REMATCH[1]}))"
}

# Turn an arbitrary word (a command's basename) into a valid dataset name.
# Prints nothing when nothing usable is left.
_sanitize_name () {
    local s="$1"
    s="${s//[!${_ALNUM}._-]/-}"
    while [[ "$s" == *..* ]]; do s="${s//../.}"; done
    s="${s#"${s%%["${_ALNUM}"]*}"}"          # must start with a letter/digit
    s="${s:0:64}"
    printf '%s' "$s"
}

# ---------------------------------------------------------------------------
# Process-wide cleanup: temp dirs we created and store locks we hold are
# released on ANY exit (normal, die, or signal) — a `die` inside the critical
# section can never leave the store locked.
# ---------------------------------------------------------------------------
_TMPDIRS=()
_LOCKS_HELD=()      # lock dir paths held by this process
_LOCKS_OWNER=()     # the owner record we wrote into each
_cleanup () {
    local i d
    for i in ${_LOCKS_HELD[@]+"${!_LOCKS_HELD[@]}"}; do
        _release_lock_dir "${_LOCKS_HELD[i]}" "${_LOCKS_OWNER[i]}"
    done
    _LOCKS_HELD=(); _LOCKS_OWNER=()
    for d in ${_TMPDIRS[@]+"${_TMPDIRS[@]}"}; do rm -rf "$d" 2>/dev/null || true; done
    _TMPDIRS=()
}
trap _cleanup EXIT
trap 'exit 130' INT
trap 'exit 143' TERM

# _mktempd PREFIX — make a registered temp dir; result in $_MKTEMPD. (Not via
# $(...): the registration must happen in THIS shell.)
_mktempd () {
    _MKTEMPD="$(mktemp -d "$PROVOST_TEMP_DIR/$1.XXXXXX")" \
        || die "cannot create a temp dir in $PROVOST_TEMP_DIR (set TMPDIR)."
    _TMPDIRS+=("$_MKTEMPD")
}

# ---------------------------------------------------------------------------
# Locate the engines + transform — ONLY inside provost's install dir. The data
# store is never searched: a group-writable shared store would otherwise let
# any member plant code that runs as whoever captures next.
# ---------------------------------------------------------------------------
_find_engine () {
    local c="$PROVOST_ROOT/extract/provost-extract"
    [[ -f "$c" ]] && { printf '%s\n' "$c"; return 0; }
    return 1
}
_find_analyze () {
    local c="$PROVOST_ROOT/analyze/provost-analyze"
    [[ -f "$c" ]] && { printf '%s\n' "$c"; return 0; }
    return 1
}

# Run a bundled engine: no GIT_DIR leaking in, and the engine's legacy .provostrc
# config layer switched off (PROVOST_NO_RC=1) so a .provostrc in the CWD, $HOME or
# any parent cannot reconfigure a capture behind provost's back.
_run_engine () { env -u GIT_DIR PROVOST_NO_RC=1 bash "$@"; }

# ---------------------------------------------------------------------------
# Transform layer — the tidy melt (`_core_melt_tidy`) and schema-union helpers
# (`_core_union_header`, `_core_align_rows`) live in provost/transform/melt.sh.
# ---------------------------------------------------------------------------
if [[ -r "$PROVOST_ROOT/transform/melt.sh" ]]; then
    # shellcheck source=transform/melt.sh
    source "$PROVOST_ROOT/transform/melt.sh"
else
    die "transform module not found (expected at $PROVOST_ROOT/transform/melt.sh)."
fi

_ds_dir ()  { printf '%s/datasets\n' "$PROVOST_WORKSPACE"; }
_cap_dir () { printf '%s/captures\n' "$PROVOST_WORKSPACE"; }
_prov_file ()  { printf '%s/provenance.tsv\n' "$PROVOST_WORKSPACE"; }
_rowid_file () { printf '%s/.next_row_id\n' "$PROVOST_WORKSPACE"; }
_capid_file () { printf '%s/.next_capture_id\n' "$PROVOST_WORKSPACE"; }
_faillog ()    { printf '%s/.async-failures.log\n' "$PROVOST_WORKSPACE"; }
_failseen ()   { printf '%s/.async-failures.seen\n' "$PROVOST_WORKSPACE"; }

# git wrapper:
#   * pinned identity + no signing, so commits work in any environment;
#   * safe.directory=<store>, so a shared store owned by another user works
#     instead of tripping git's dubious-ownership check;
#   * hooks and fsmonitor disabled — with safe.directory lifted, a store's own
#     .git/config or hooks (writable by the store's other members) must not be
#     able to run code as us;
#   * the caller's GIT_DIR/GIT_WORK_TREE/GIT_INDEX_FILE can't redirect us.
_WS_REAL=""; _WS_REAL_FOR=""

# Lifting safe.directory hands the store's own .git/config the trust git would
# otherwise withhold, and hooks are not the only way a config runs code: filter
# drivers (filter.*.clean/smudge), diff/merge drivers, core.pager,
# core.sshCommand, include.path and friends all do. So when the store's git dir
# could have been written by someone else — .git or .git/config not owned by
# us, or the config group/world-writable — every key in it must be one git
# itself writes (or `provost import`'s clone does). Anything else: refuse.
_perm_bits () { stat -c %a "$1" 2>/dev/null || stat -f %Lp "$1" 2>/dev/null || true; }
_owner_uid () { stat -c %u "$1" 2>/dev/null || stat -f %u "$1" 2>/dev/null || true; }
_vet_store_config () {
    local gd="$PROVOST_WORKSPACE/.git" cfg me p bad
    cfg="$gd/config"
    [[ -f "$cfg" ]] || return 0
    me="$(id -u 2>/dev/null || echo)"
    p="$(_perm_bits "$cfg")"
    if [[ -n "$me" && "$(_owner_uid "$gd")" == "$me" && "$(_owner_uid "$cfg")" == "$me" \
          && "$p" =~ ^[0-7]*[0-7][0145][0145]$ ]]; then
        return 0                                   # ours alone: git's own trust applies
    fi
    bad="$(env -u GIT_CONFIG_PARAMETERS -u GIT_CONFIG_COUNT \
             git config --file "$cfg" --no-includes --name-only --list 2>/dev/null \
           | LC_ALL=C awk '{ k = tolower($0) }
               k ~ /^core\.(repositoryformatversion|filemode|bare|logallrefupdates|sharedrepository|ignorecase|precomposeunicode|symlinks)$/ { next }
               k ~ /^extensions\.objectformat$/ || k ~ /^init\.defaultbranch$/ { next }
               k ~ /^user\.(name|email)$/ || k ~ /^gc\.[a-z]+$/ { next }
               k ~ /^remote\.origin\.(url|fetch)$/ || k ~ /^branch\.(main|master)\.(remote|merge)$/ { next }
               { print $0 }')"
    [[ -z "$bad" ]] && return 0
    die "refusing to use the store at $PROVOST_WORKSPACE: its .git/config is writable by others and sets ${bad//$'\n'/, } — settings that can make git run programs. Remove them (git config --file $(printf '%q' "$cfg") --unset <key>) if you trust them to be harmless."
}

_git () {
    if [[ "$_WS_REAL_FOR" != "$PROVOST_WORKSPACE" ]]; then
        _vet_store_config
        if _WS_REAL="$(cd -P -- "$PROVOST_WORKSPACE" 2>/dev/null && pwd)"; then
            _WS_REAL_FOR="$PROVOST_WORKSPACE"
        else
            _WS_REAL="$PROVOST_WORKSPACE"
        fi
    fi
    env -u GIT_DIR -u GIT_WORK_TREE -u GIT_INDEX_FILE -u GIT_OBJECT_DIRECTORY \
        -u GIT_COMMON_DIR -u GIT_NAMESPACE ${_GIT_INDEX:+"GIT_INDEX_FILE=$_GIT_INDEX"} \
        git -C "$PROVOST_WORKSPACE" -c safe.directory="$_WS_REAL" \
        -c core.hooksPath=/dev/null -c core.fsmonitor=false \
        -c user.name=provost -c user.email=provost@localhost \
        -c commit.gpgsign=false -c tag.gpgSign=false "$@"
}

# Stage everything and commit. Prints git's own error and exits non-zero if
# git fails — provost never reports "ok" for a capture git did not record.
_commit () {
    local msg="$1" out rc=0
    out="$(_git add -A 2>&1)" || { printf 'git: %s\n' "$out" >&2; die "git add failed in $PROVOST_WORKSPACE — nothing was committed."; }
    _git diff --cached --quiet >/dev/null 2>&1 || rc=$?
    [[ $rc -eq 0 ]] && return 0                    # nothing staged: nothing to do
    out="$(_git commit -q -m "$msg" 2>&1)" || {
        printf 'git: %s\n' "$out" >&2
        die "git commit failed in $PROVOST_WORKSPACE — nothing was committed (the files stay in the store and go into the next successful commit)."
    }
}

# The branch captures are committed on: main (or master in a store created by
# an older git without our rename).
_main_branch () {
    if _git show-ref -q --verify refs/heads/main 2>/dev/null; then echo main
    elif _git show-ref -q --verify refs/heads/master 2>/dev/null; then echo master
    else echo main; fi
}

# Refuse to write while the store is on a restored capture (detached HEAD) or
# some other branch: a commit there would be orphaned or land off-history.
_require_on_branch () {
    [[ -d "$PROVOST_WORKSPACE/.git" ]] || return 0
    local head main
    head="$(_git symbolic-ref -q --short HEAD 2>/dev/null)" \
        || die "store is on a restored capture (detached HEAD) — new captures would be lost. Return with: provost restore latest"
    main="$(_main_branch)"
    [[ "$head" == "$main" ]] \
        || die "store is on branch '$head', not '$main' — refusing to write. Return with: provost restore latest"
}

# --- store lock (a shared PROVOST_HOME may have concurrent writers) ----------
# mkdir is atomic across processes/users; spin up to PROVOST_LOCK_TIMEOUT
# seconds. The lock records "host<TAB>pid<TAB>uid<TAB>epoch" (owner file).
# A lock is STALE only if:
#   * it is older than PROVOST_LOCK_MAX_AGE seconds (any host), or
#   * it was taken on THIS host (and pid namespace) and its pid definitely no
#     longer exists (ESRCH — NOT EPERM: another user's live process is alive).
# Breaking is serialised by a second mkdir mutex (.lock.break) and done by an
# atomic rename, after re-checking staleness under that mutex — so two
# breakers can never delete each other's (or a newcomer's) fresh lock.
# $BASHPID (this process), NOT $$: a detached async ingest runs in a subshell
# where $$ is still the already-exited parent provost.
_LOCK_HOST=""
_lock_host_init () {
    if [[ -z "$_LOCK_HOST" ]]; then
        local h ns=""
        h="${HOSTNAME:-}"; [[ -n "$h" ]] || h="$(uname -n 2>/dev/null || echo unknown)"
        ns="$(readlink /proc/self/ns/pid 2>/dev/null || true)"
        _LOCK_HOST="$h${ns:+/$ns}"
    fi
}
# mtime in epoch seconds; prints nothing if the path is gone.
_mtime () { stat -c %Y "$1" 2>/dev/null || stat -f %m "$1" 2>/dev/null || true; }
# Age in seconds of a path; -1 if it is gone (never "old").
_age () { local m; m="$(_mtime "$1")"; [[ "$m" =~ ^[0-9]+$ ]] || { echo -1; return 0; }; echo $(( $(date +%s) - m )); }
# 0 only when the pid definitely does not exist on this host.
_pid_gone () {
    local pid="$1" err
    [[ "$pid" =~ ^[0-9]{1,10}$ ]] || return 1
    if [[ -d /proc/self ]]; then [[ ! -d "/proc/$pid" ]]; return; fi
    err="$(LC_ALL=C kill -0 "$pid" 2>&1)" && return 1       # alive and ours
    [[ "$err" == *"No such process"* ]]                     # ESRCH; EPERM = alive
}
# 0 = the lock is stale. Sets $_STALE_SEEN to the owner record it judged, so a
# breaker can verify it removed THAT lock and not a newcomer's.
_lock_stale () {
    local lock="$1" owner="" host="" pid="" t="" now age
    _STALE_SEEN=""
    [[ -d "$lock" ]] || return 1
    now="$(date +%s)"
    owner="$(cat "$lock/owner" 2>/dev/null || true)"
    if [[ -z "$owner" ]]; then
        age="$(_age "$lock")"
        if [[ -f "$lock/pid" ]]; then               # a lock from an older provost
            pid="$(cat "$lock/pid" 2>/dev/null || true)"
            _STALE_SEEN="legacy:$pid"
            (( age > PROVOST_LOCK_MAX_AGE )) && return 0
            _pid_gone "$pid" && return 0
            return 1
        fi
        # mkdir'd but no owner yet: a holder is between mkdir and writing it —
        # or died right there. Give it a minute.
        _STALE_SEEN="none"
        (( age > 60 )) && return 0
        return 1
    fi
    _STALE_SEEN="$owner"
    IFS=$'\t' read -r host pid _ t <<< "$owner" || true
    if [[ "$t" =~ ^[0-9]{1,12}$ ]] && (( now - t > PROVOST_LOCK_MAX_AGE )); then return 0; fi
    _lock_host_init
    [[ "$host" == "$_LOCK_HOST" ]] && _pid_gone "$pid" && return 0
    return 1
}
_lock_break () {  # 0 = we removed a stale lock
    local lock="$1" brk="$1.break" rc=1 aside seen got
    if ! mkdir "$brk" 2>/dev/null; then
        # Another process is breaking it. Clear an abandoned break mutex.
        if (( $(_age "$brk") > 60 )); then rmdir "$brk" 2>/dev/null || true; fi
        return 1
    fi
    if _lock_stale "$lock"; then
        seen="$_STALE_SEEN"
        aside="$lock.stale.$BASHPID.$RANDOM"
        if mv "$lock" "$aside" 2>/dev/null; then
            # Verify we moved the lock we judged (not one created meanwhile).
            got="$(cat "$aside/owner" 2>/dev/null || true)"
            if [[ -z "$got" ]]; then
                if [[ -f "$aside/pid" ]]; then got="legacy:$(cat "$aside/pid" 2>/dev/null || true)"; else got="none"; fi
            fi
            if [[ "$got" == "$seen" ]]; then
                rm -rf "$aside" 2>/dev/null || true; rc=0
            else
                mv "$aside" "$lock" 2>/dev/null || true     # not ours to break: put it back
            fi
        fi
    fi
    rmdir "$brk" 2>/dev/null || true
    return $rc
}
# _lock [STORE-DIR] (default: the active store)
_lock () {
    local dir="${1:-$PROVOST_WORKSPACE}" lock tries=0 max owner i
    lock="$dir/.lock"
    for i in ${_LOCKS_HELD[@]+"${!_LOCKS_HELD[@]}"}; do
        [[ "${_LOCKS_HELD[i]}" == "$lock" ]] && return 0
    done
    max=$((PROVOST_LOCK_TIMEOUT * 5))
    mkdir -p "$dir" 2>/dev/null || true
    while ! mkdir "$lock" 2>/dev/null; do
        [[ -d "$dir" ]] || die "cannot create the store directory: $dir"
        if _lock_stale "$lock" && _lock_break "$lock"; then continue; fi
        tries=$((tries + 1))
        if (( tries > max )); then
            owner="$(cat "$lock/owner" 2>/dev/null || cat "$lock/pid" 2>/dev/null || echo unknown)"
            die "store is locked (waited ${PROVOST_LOCK_TIMEOUT}s): $lock — held by ${owner//$'\t'/ }. If that holder is gone, remove the lock dir."
        fi
        sleep 0.2
    done
    _lock_host_init
    local mypid="$BASHPID" myuid; myuid="$(id -u 2>/dev/null || echo "?")"
    printf -v owner "%s\t%s\t%s\t%s" "$_LOCK_HOST" "$mypid" "$myuid" "$(date +%s)"
    printf '%s\n' "$owner" > "$lock/owner.tmp" && mv "$lock/owner.tmp" "$lock/owner"
    _LOCKS_HELD+=("$lock"); _LOCKS_OWNER+=("$owner")
}
# Remove a lock dir only if it still carries OUR owner record.
_release_lock_dir () {
    local lock="$1" mine="$2" cur
    cur="$(cat "$lock/owner" 2>/dev/null || true)"
    [[ "$cur" == "$mine" ]] && rm -rf "$lock" 2>/dev/null
    return 0
}
_unlock () {
    local lock="${1:-$PROVOST_WORKSPACE}/.lock" i keep_l=() keep_o=()
    for i in ${_LOCKS_HELD[@]+"${!_LOCKS_HELD[@]}"}; do
        if [[ "${_LOCKS_HELD[i]}" == "$lock" ]]; then
            _release_lock_dir "$lock" "${_LOCKS_OWNER[i]}"
        else
            keep_l+=("${_LOCKS_HELD[i]}"); keep_o+=("${_LOCKS_OWNER[i]}")
        fi
    done
    _LOCKS_HELD=(${keep_l[@]+"${keep_l[@]}"}); _LOCKS_OWNER=(${keep_o[@]+"${keep_o[@]}"})
}

# ---------------------------------------------------------------------------
# Store initialisation + migrations
# ---------------------------------------------------------------------------
# Runtime files that must never be committed (lock, async failure log, temp
# files, the machine-local instance registry). The first line is a version
# marker: a store whose .gitignore lacks it is migrated.
_GITIGNORE_MARK="# provost store .gitignore v1 — runtime files, never committed"
_write_gitignore () {
    cat > "$PROVOST_WORKSPACE/.gitignore" <<EOF
$_GITIGNORE_MARK
/.lock
/.lock.*
/.async-failures.log
/.async-failures.seen
/instances.tsv
/instances.tsv.*
/datasets/.recon.*
/datasets/.stage.*
/datasets/.rm.*
/datasets/.rmprov.*
/captures/*/.meta.tmp
/context/*.tmp
EOF
}

# Highest capture number ever used in this store: capture dirs, capture tags
# and capture commit subjects (the latter survive an 'rm' purge).
_scan_max_capture_num () {
    local d n max=0
    for d in "$(_cap_dir)"/c*/; do
        [[ -d "$d" ]] || continue
        d="${d%/}"; n="${d##*/c}"
        [[ "$n" =~ ^[0-9]{1,12}$ ]] && (( 10#$n > max )) && max=$((10#$n))
    done
    n="$( { _git tag -l 'capture-c*' 2>/dev/null || true
            _git log --all --format=%s 2>/dev/null || true; } \
          | awk '{ s = $0
                   if (s ~ /^capture-c[0-9]+$/) { sub(/^capture-c/, "", s) }
                   else if (s ~ /^capture c[0-9]+:/) { sub(/^capture c/, "", s); sub(/:.*/, "", s) }
                   else next
                   if (s + 0 > m) m = s + 0 }
                 END { print m + 0 }')"
    [[ "$n" =~ ^[0-9]+$ ]] && (( n > max )) && max=$n
    printf '%s\n' "$max"
}

# Idempotent store upkeep for stores made by an older provost: add the
# .gitignore, untrack a committed .lock (and other runtime files), and seed the
# monotonic capture counter. Only when writable and on the capture branch.
# Store format, recorded (and committed) in .store_format. 2 = datasets carry
# no `total_entries` column (melt stopped emitting the engine's record count
# on 2026-09-25; datasets written before then still had it). 3 = every
# dataset's second column is `capture`, the id of the capture each row came
# from (2026-10-28; earlier rows are filled in from provenance.tsv).
_STORE_FORMAT=3
_store_format_file () { printf '%s/.store_format\n' "$PROVOST_WORKSPACE"; }
_store_format () { local f; f="$(cat "$(_store_format_file)" 2>/dev/null || true)"; [[ "$f" =~ ^[0-9]{1,6}$ ]] && echo "$f" || echo 1; }

# Format 1 -> 2: drop the `total_entries` column from every dataset that has
# one. It is the engine's per-capture record count, not a variable; in a
# reconciled dataset new rows already blank-fill it. Only a column whose cells
# are all integers or empty is dropped — anything else is real data under
# that name and stays.
_migrate_drop_total_entries () {
    local f tmp
    for f in "$(_ds_dir)"/*.tsv; do
        [[ -f "$f" ]] || continue
        tmp="$f.mig.$$"
        if awk -F'\t' -v OFS='\t' '
            NR == 1 { for (i = 1; i <= NF; i++) if ($i == "total_entries") c = i
                      if (!c) exit 3 }
            { L[NR] = $0; if (NR > 1 && $c != "" && $c !~ /^[0-9]+$/) exit 3 }
            END {
                if (!c) exit 3
                for (r = 1; r <= NR; r++) {
                    n = split(L[r], F, "\t"); out = ""; k = 0
                    for (i = 1; i <= n; i++) if (i != c) { out = out (k++ ? OFS : "") F[i] }
                    print out
                }
            }' "$f" > "$tmp"; then
            mv "$tmp" "$f"
        else
            rm -f "$tmp"
        fi
    done
}

# Format 2 -> 3: insert the `capture` column after '#' in every dataset,
# filled from provenance.tsv (# -> capture). One awk over provenance and every
# dataset; a header that does not start with '#' is not a provost dataset and
# is left alone. A data column already named `capture` keeps its name and the
# added one becomes `_capture`, exactly as a new capture names it. A dataset
# that already carries the column (its second column is named so and every
# row's value is that row's capture) is left as it is, so a store whose
# .store_format went missing is not given the column twice.
_migrate_add_capture () {
    local dir prov f skip; dir="$(_ds_dir)"; prov="$(_prov_file)"
    local -a files=()
    for f in "$dir"/*.tsv; do [[ -f "$f" ]] && files+=("$f"); done
    (( ${#files[@]} )) || return 0
    [[ -f "$prov" ]] || prov=/dev/null
    skip="$dir/.mig-skip.$$"; : > "$skip"
    awk -F'\t' -v OFS='\t' -v P="$prov" -v SKIP="$skip" '
        function done_file() { if (out != "") { close(out); if (had) print prev > SKIP }; out = "" }
        FILENAME == P { if (FNR > 1) cap[$1] = $3; next }
        FNR == 1 {
            done_file(); if ($1 != "#") next
            out = FILENAME ".mig"; prev = FILENAME; split("", U)
            had = ($2 ~ /^_*capture$/)
            for (i = 1; i <= NF; i++) U[$i] = 1
            c = "capture"; while (c in U) c = "_" c
        }
        out == "" { next }
        FNR > 1 && had && !($1 in cap && $2 == cap[$1]) { had = 0 }
        { rest = substr($0, length($1) + 2)
          v = (FNR == 1) ? c : (($1 in cap) ? cap[$1] : "")
          if (NF == 1) print $1, v > out; else print $1, v, rest > out }
        END { done_file() }
    ' "$prov" "${files[@]}" || { for f in "${files[@]}"; do rm -f "$f.mig"; done; rm -f "$skip"; return 1; }
    for f in "${files[@]}"; do
        [[ -f "$f.mig" ]] || continue
        if grep -qxF -- "$f" "$skip"; then rm -f "$f.mig"; else mv "$f.mig" "$f"; fi
    done
    rm -f "$skip"
    return 0
}

_migrate_store () {
    local ws="$PROVOST_WORKSPACE" first=""
    [[ -d "$ws/.git" && -w "$ws" ]] || return 0
    [[ -f "$ws/.gitignore" ]] && { IFS= read -r first < "$ws/.gitignore" || true; }
    [[ "$first" == "$_GITIGNORE_MARK" && -f "$(_capid_file)" && "$(_store_format)" -ge $_STORE_FORMAT ]] && return 0
    _git symbolic-ref -q HEAD >/dev/null 2>&1 || return 0      # restored/detached: later
    _lock
    IFS= read -r first < "$ws/.gitignore" 2>/dev/null || first=""
    if [[ "$first" != "$_GITIGNORE_MARK" ]]; then
        _write_gitignore
        _git rm -r -q --cached --ignore-unmatch -- .lock instances.tsv \
            .async-failures.log .async-failures.seen >/dev/null 2>&1 || true
    fi
    if [[ ! -f "$(_capid_file)" ]]; then
        echo "$(( $(_scan_max_capture_num) + 1 ))" > "$(_capid_file)"
    fi
    if (( $(_store_format) < 2 )); then
        _migrate_drop_total_entries
    fi
    if (( $(_store_format) < 3 )); then
        _migrate_add_capture || die "could not add the capture column to the datasets (store format 3)."
    fi
    echo "$_STORE_FORMAT" > "$(_store_format_file)"
    _commit "provost: store maintenance (.gitignore, untrack runtime files, capture counter, store format $_STORE_FORMAT)"
    _unlock
}

# The workspace is a git repo: every capture is one commit (tagged capture-NNNN),
# so a run can be brought back up later with `provost restore`. Initialization is
# serialised with double-checked locking so concurrent first-writers to a shared
# store don't race on `git init`.
_ensure_workspace () {
    _vet_store_config
    mkdir -p "$(_ds_dir)" "$(_cap_dir)" 2>/dev/null \
        || die "cannot create the store at $PROVOST_WORKSPACE"
    if [[ ! -d "$PROVOST_WORKSPACE/.git" || ! -f "$(_rowid_file)" ]]; then
        _lock
        if [[ ! -d "$PROVOST_WORKSPACE/.git" ]]; then
            local out
            out="$(_git init -q 2>&1)" || { printf 'git: %s\n' "$out" >&2; die "git init failed in $PROVOST_WORKSPACE"; }
            _git symbolic-ref HEAD refs/heads/main >/dev/null 2>&1 || true
            _write_gitignore
            printf 'provost workspace (store model %s)\n' "$PROVOST_VERSION" > "$PROVOST_WORKSPACE/.provostrc"
            [[ -f "$(_capid_file)" ]] || echo 1 > "$(_capid_file)"
            echo "$_STORE_FORMAT" > "$(_store_format_file)"
        fi
        [[ -f "$(_rowid_file)" ]] || echo 1 > "$(_rowid_file)"
        _commit "provost: init workspace"
        _unlock
    fi
    _migrate_store
}

# Allocate the next capture id from the monotonic counter (under the lock).
# Ids are never reused — not after an 'rm' purge, not after a 'restore'.
# Result in $_CAPID.
_next_capture_id () {
    local f n; f="$(_capid_file)"
    n="$(cat "$f" 2>/dev/null || true)"
    if [[ ! "$n" =~ ^[0-9]{1,12}$ ]] || (( 10#$n < 1 )); then
        n=$(( $(_scan_max_capture_num) + 1 ))
    fi
    n=$((10#$n))
    # Belt and braces: never land on an id that is already on disk or tagged.
    while [[ -e "$(_cap_dir)/$(printf 'c%04d' "$n")" ]] \
          || _git rev-parse -q --verify "refs/tags/capture-$(printf 'c%04d' "$n")" >/dev/null 2>&1; do
        n=$((n + 1))
    done
    echo "$((n + 1))" > "$f"
    _CAPID="$(printf 'c%04d' "$n")"
}

# Column overlap (Jaccard, as an integer percent) between two tab-separated
# headers passed as the two lines of stdin.
_col_overlap () {
    # A stored header's second column (after '#') is the capture id every
    # dataset carries; it is in both headers by construction, so it is left
    # out rather than let it nudge two different shapes into one dataset.
    awk -F'\t' '
        NR==1 { for(i=1;i<=NF;i++){ if(i==2 && $1=="#") continue; a[$i]=1 }; na=NF }
        NR==2 { for(i=1;i<=NF;i++){ if(i==2 && $1=="#") continue; b[$i]=1; if($i in a) inter++ } }
        END { for(k in a) u[k]=1; for(k in b) u[k]=1; nu=0; for(k in u) nu++;
              if(nu==0){print 100; exit} printf "%d\n", int(inter*100/nu) }'
}

# Append a tidy TSV FILE (header+rows) to a dataset. Store rule:
#   new dataset            → write header + rows;
#   header matches         → append data rows;
#   header differs, overlap ≥ PROVOST_RECONCILE_OVERLAP% → RECONCILE: union the
#                            columns, blank-fill both sides, rewrite in place;
#   overlap below threshold → split into a numbered variant (.v2, .v3, …).
# Sets $_SA_TARGET to the final dataset name. Works on files throughout — a
# large capture is never echoed through a pipe or held in a variable.
_store_append () {
    local name="$1" src="$2" dir; dir="$(_ds_dir)"
    local base="$dir/$name.tsv" target="$name" file new_header="" old_header="" line=""
    IFS= read -r new_header < "$src" || true

    if [[ ! -f "$base" ]]; then
        cp "$src" "$base"
    else
        IFS= read -r old_header < "$base" || true
        if [[ "$old_header" == "$new_header" ]]; then
            tail -n +2 "$src" >> "$base"
        else
            local overlap; overlap="$(printf '%s\n%s\n' "$old_header" "$new_header" | _col_overlap)"
            [[ "$overlap" =~ ^[0-9]+$ ]] || overlap=0
            if (( overlap >= PROVOST_RECONCILE_OVERLAP )); then
                # Reconcile into the base dataset: union header, realign both sides.
                local unified tmp; unified="$(_core_union_header "$old_header" "$new_header")"
                tmp="$(mktemp "$dir/.recon.XXXXXX")"
                if {
                    printf '%s\n' "$unified"
                    tail -n +2 "$base" | _core_align_rows "$old_header" "$unified"
                    tail -n +2 "$src"  | _core_align_rows "$new_header" "$unified"
                } > "$tmp"; then
                    mv "$tmp" "$base"
                else
                    rm -f "$tmp"; return 1
                fi
            else
                # Genuinely different shape → variant file.
                local v=2
                while [[ -f "$dir/$name.v$v.tsv" ]]; do
                    line=""; IFS= read -r line < "$dir/$name.v$v.tsv" || true
                    [[ "$line" == "$new_header" ]] && break
                    v=$((v+1))
                done
                target="$name.v$v"; file="$dir/$target.tsv"
                if [[ -f "$file" ]]; then tail -n +2 "$src" >> "$file"
                else cp "$src" "$file"; fi
            fi
        fi
    fi
    _SA_TARGET="$target"
}

# Render argv as ONE line of shell: plain words bare, others single-quoted,
# and anything holding a control character (tab, newline, ...) in $'...' form
# via printf %q — so the result never contains a raw tab or newline.
_shell_quote () {
    local a out="" q sq="'\\''"
    for a in "$@"; do
        if [[ -n "$a" && "$a" =~ ^[${_ALNUM}_./=:,+@%^-]+$ ]]; then q="$a"
        elif [[ "$a" == *[[:cntrl:]]* ]]; then printf -v q '%q' "$a"
        else q="'${a//\'/$sq}'"; fi
        out="${out:+$out }$q"
    done
    printf '%s' "$out"
}

# One-line, tab/newline-free rendering of a value for a meta.tsv field.
_meta_safe () { local s="$1"; s="${s//$'\t'/\\t}"; s="${s//$'\n'/\\n}"; printf '%s' "$s"; }

# Snapshot a controlled directory's contents as commit context: a manifest of
# path/size/mtime/sha for each file (excluding the .provost instance and .git).
_snapshot_context () {
    local dir="$1" ws="$2" f sz mt sha
    local mani="$ws/context/manifest.tsv"
    mkdir -p "$ws/context"
    {
        printf 'path\tsize\tmtime\tsha1\n'
        while IFS= read -r -d '' f; do
            sz="$(stat -c '%s' "$f" 2>/dev/null || echo 0)"
            mt="$(stat -c '%Y' "$f" 2>/dev/null || echo 0)"
            sha="(skipped)"
            if [[ "$sz" =~ ^[0-9]+$ ]] && (( sz <= 1048576 )) && command -v sha1sum >/dev/null 2>&1; then
                sha="$(sha1sum 2>/dev/null < "$f")" || sha="(unreadable)"
                sha="${sha%% *}"
            fi
            printf '%s\t%s\t%s\t%s\n' "$(_meta_safe "${f#"$dir"/}")" "$sz" "$mt" "$sha"
        done < <(find "$dir" -type f -not -path '*/.provost/*' -not -path '*/.git/*' -print0 2>/dev/null)
    } > "$mani.tmp"
    # sort the rows (header stays first) for a stable manifest
    { IFS= read -r f; printf '%s\n' "$f"; LC_ALL=C sort; } < "$mani.tmp" > "$mani"
    rm -f "$mani.tmp"
    local n=0; n=$(( $(wc -l < "$mani") - 1 ))
    printf '%s\n' "$n"
}

# RUN FILES — a capture in a controlled directory also commits the
# directory's files themselves: a commit of their own, tagged files-<id>, so
# any run's files can be listed, read, compared or written out again later
# ('provost files', 'provost diff'). It is built with a scratch index and the
# controlled directory as the work tree, so the store's own tree never holds
# a copy, and git stores a file that did not change between runs once. The
# directory's .gitignore is honoured; .provost and .git never go in; a file
# over PROVOST_FILES_MAX_SIZE bytes (0 = keep no files) is only listed in
# context/manifest.tsv. Each run's commit has the previous run's as its
# parent, so plain git shows the directory's history across runs
# (git log --stat files-c0007); a run whose files did not change is tagged on
# the previous commit rather than given an empty one. Called under the lock.
_snapshot_files () {
    local dir="$1" id="$2" git_dir idx spec tree parent ptree commit out f
    (( PROVOST_FILES_MAX_SIZE > 0 )) || return 0
    git_dir="$PROVOST_WORKSPACE/.git"
    idx="$git_dir/provost-files.index"; spec="$git_dir/provost-files.pathspec"
    rm -f "$idx" "$spec"
    {
        printf ':/\0:(top,exclude,literal).provost\0'
        while IFS= read -r -d '' f; do printf ':(top,exclude,literal)%s\0' "${f#./}"; done \
            < <(cd "$dir" && find . \( -path ./.provost -o -path ./.git \) -prune -o \
                    -type f -size +"${PROVOST_FILES_MAX_SIZE}"c -print0 2>/dev/null)
    } > "$spec"
    out="$(_GIT_INDEX="$idx" _git --work-tree="$dir" add -A \
            --pathspec-from-file="$spec" --pathspec-file-nul 2>&1)" \
        || { rm -f "$idx" "$spec"; printf 'git: %s\n' "$out" >&2; return 1; }
    tree="$(_GIT_INDEX="$idx" _git write-tree 2>/dev/null)" || { rm -f "$idx" "$spec"; return 1; }
    rm -f "$idx" "$spec"
    parent="$(_git for-each-ref --count=1 --sort=-v:refname --format='%(objectname)' 'refs/tags/files-c*' 2>/dev/null)"
    ptree=""; [[ -n "$parent" ]] && ptree="$(_git rev-parse -q --verify "$parent^{tree}" 2>/dev/null)"
    if [[ -n "$parent" && "$ptree" == "$tree" ]]; then
        commit="$parent"
    else
        commit="$(_git commit-tree "$tree" ${parent:+-p "$parent"} \
                    -m "files of $dir at capture $id" 2>/dev/null)" || return 1
    fi
    _git tag "files-$id" "$commit" >/dev/null 2>&1
}

# _quality_exempt ENGINE STRATEGY — true when the quality gate does not apply.
# Coverage and consistency judge GENERIC parses: a tool-specific (bespoke or
# plugin) parser normalises values, so its output legitimately shares few
# tokens with the input (22 correct corpus winners read under 40% coverage),
# and a strict grammar (json/yaml/xml/toml) validates the whole document
# itself. Only parsers under extract/strategies/{delimited,structured} are
# gated. An LLM template's blocks are exempt too: a template is written for
# the tool's layout and deliberately makes key/value tables whose value
# column mixes types (a type-consistency measure built for homogeneous
# tables reads 30-60% on them); _llm_file's own rules judge its blocks.
_quality_exempt () {
    local sdir; sdir="$(dirname "$1")/strategies"
    case "$2" in json|json-array|jsonl|yaml|xml|toml|har|llm-template) return 0 ;; esac
    [[ -f "$sdir/delimited/$2.sh" || -f "$sdir/structured/$2.sh" ]] && return 1
    return 0
}

# _multi_split FILE DIR — split the engine's --multi output into DIR/ds.N.meta
# (`strategy<TAB>confidence<TAB>coverage<TAB>consistency<TAB>lines`, the
# fields after `__DATASET__<TAB>N`) and DIR/ds.N.wide (that dataset's -f tsv).
_multi_split () {
    rm -f "$2"/ds.* 2>/dev/null || true
    DIR="$2" awk -F'\t' '
        /^__DATASET__/ { n = ($2 ~ /^[0-9]+$/) ? $2 : ""; if (n != "") printf "%s\t%s\t%s\t%s\t%s\n", $3, $4, $5, $6, $7 > (ENVIRON["DIR"] "/ds." n ".meta"); next }
        n != "" { print > (ENVIRON["DIR"] "/ds." n ".wide") }
    ' "$1" 2>/dev/null || true
}

# _tidy_cols FILE — a tidy dataset's column names, one per line, without the
# generated row index (`n`).
_tidy_cols () { head -n 1 "$1" | tr '\t' '\n' | awk 'NR == 1 && $0 == "n" { next } { print }'; }

# AUTO-MULTI (PROVOST_AUTO_MULTI, --auto-multi / --no-auto-multi). Called by
# _ingest when the quality gate would hold the single-winner parse for
# low coverage or low consistency. It runs the engine's --multi segmentation
# and files its datasets INSTEAD of holding the capture only when the split
# is clearly better:
#   1. it is a split: the engine's own choice is not the unsplit input, and
#      it yields at least two datasets;
#   2. every dataset passes the gate itself — confidence >= the threshold
#      and, for a generic parser, its own coverage and consistency (measured
#      by the engine against the dataset's own blocks) >= the gate's;
#   3. together they cover MATERIALLY more of the input: the whole input's
#      coverage by all the datasets is at least the gate's minimum and
#      _AM_GAIN (10) points above the single parse's;
#   4. no dataset is a POSITIONAL table — 2+ rows under nothing but generated
#      A/B/C… column names: a header the split cut off (a rule line between
#      header and rows), or box chrome read as cells. Filing values under
#      positional names is the misplacement the gate exists to stop;
#   5. the parser the gate just rejected does not come back unmeasured: a
#      dataset (3+ lines) read by the same strategy must have a judged
#      column-type consistency (a one-record flatten is not judged);
#   6. it is not shattered into trivial fragments: datasets of 3+ input lines
#      hold at least half the input's lines (a banner, a footer or a prompt
#      may be a one-line dataset of its own; most of the input may not).
# Otherwise it returns 1 with the reason in am_note, and the capture is held
# exactly as before. Works on _ingest's locals (dynamic scope): reads engine,
# eargs, body, tmp, threshold, name, strategy, cov; fills ds_* and am_*.
_AM_GAIN=10
_LLM_CATCHALL_SLACK=20   # _llm_file: points a template may trail a catch-all parse by
_AM_FRAG_LINES=3

# _am_part_check I WIDE TSV — the per-part checks of _auto_multi (rules 2, 4
# and 5) on part I's dstrat / dconf / dcov / dcons / dlines (the caller's
# locals); melts WIDE into TSV. Sets _AMP_NOTE (empty = the part passes) and
# _AMP_ROWS.
_am_part_check () {
    local i="$1" wide="$2" tsv="$3" ncols nsyn
    _AMP_NOTE=""; _AMP_ROWS=0
    if [[ -z "$dstrat" ]]; then _AMP_NOTE="part $i has no strategy"; return 0; fi
    if (( dconf < threshold )); then
        _AMP_NOTE="part $i ($dstrat) is low-confidence ($dconf<$threshold)"; return 0
    fi
    if ! _quality_exempt "$engine" "$dstrat"; then
        if [[ "$dcov" != - ]] && (( PROVOST_MIN_COVERAGE > 0 && dcov < PROVOST_MIN_COVERAGE )); then
            _AMP_NOTE="part $i ($dstrat) keeps $dcov% of its lines"; return 0
        fi
        if [[ "$dcons" != - ]] && (( PROVOST_MIN_CONSISTENCY > 0 && dcons < PROVOST_MIN_CONSISTENCY )); then
            _AMP_NOTE="part $i ($dstrat) columns are $dcons% type-consistent"; return 0
        fi
    fi
    if [[ "$dstrat" == "$strategy" && "$dcons" == - ]] && (( dlines >= _AM_FRAG_LINES )); then
        _AMP_NOTE="part $i is read by the rejected parser ($dstrat) and cannot be judged"; return 0
    fi
    _core_melt_tidy_file "$wide" > "$tsv"
    _AMP_ROWS=$(awk 'NR > 1 && $0 != "" { r++ } END { print r + 0 }' "$tsv")
    read -r ncols nsyn < <(_tidy_cols "$tsv" | awk '{ c++ } /^[A-Z][A-Z]?$/ { s++ } END { print c + 0, s + 0 }')
    if (( _AMP_ROWS >= 2 && ncols >= 2 && nsyn == ncols )); then
        _AMP_NOTE="part $i ($dstrat) is a table with positional columns only"
    fi
    return 0
}

# _model_table INPUT ENGINE_WIDE ENGINE_META TAG — the layout model's table
# for INPUT (the explicit-only layout-model strategy, measured by the engine
# like any parser), arbitrating line by line with the engine's own parse of
# the same lines (ENGINE_WIDE: its -f tsv output; ENGINE_META: strategy,
# confidence, coverage, consistency, tab-separated). Sets _MT_OUT (the -f tsv
# output) and _MT_C / _MT_V / _MT_T (confidence, coverage, consistency; empty
# when unmeasured). Returns 1 when the model produced nothing.
_model_table () {
    local inp="$1" ew="$2" em="$3" tag="$4" lmeta="$tmp/lm.$4.meta" ls=""
    _MT_OUT="$tmp/lm.$tag.tsv"; _MT_C=""; _MT_V=""; _MT_T=""
    PROVOST_LM_ENGINE_WIDE="$ew" PROVOST_LM_ENGINE_META="$em" \
        _run_engine "$engine" --max-input-size "$PROVOST_MAX_INPUT" \
        ${eargs[@]+"${eargs[@]}"} -s layout-model --emit-meta -f tsv --max-strategies 1 "$inp" \
        >"$_MT_OUT" 2>"$lmeta" || true
    IFS=$'\t' read -r ls _MT_C _MT_V _MT_T < <(awk -F'\t' \
        '/^__EXTRACT_META__/ { s = $2; c = $3; v = $4; t = $5 } END { printf "%s\t%s\t%s\t%s\n", s, c, v, t }' "$lmeta" 2>/dev/null) || true
    [[ "$ls" == layout-model && -s "$_MT_OUT" ]] && grep -q . "$_MT_OUT" || return 1
    [[ "$_MT_C" =~ ^[0-9]{1,9}$ ]] || _MT_C=0
    [[ "$_MT_V" =~ ^[0-9]{1,3}$ ]] || _MT_V=""
    [[ "$_MT_T" =~ ^[0-9]{1,3}$ ]] || _MT_T=""
    return 0
}

_auto_multi () {
    local mo="$tmp/am.txt" mm="$tmp/am.meta" kind="" n="" mcov="" mlines="" textdir=""
    if [[ "$PROVOST_MODEL" == 1 ]]; then textdir="$tmp/amtext"; mkdir -p "$textdir"; fi
    PROVOST_MULTI_TEXT_DIR="$textdir" _run_engine "$engine" --max-input-size "$PROVOST_MAX_INPUT" \
        ${eargs[@]+"${eargs[@]}"} --multi --emit-meta -f tsv --max-strategies 1 "$body" \
        >"$mo" 2>"$mm" || true
    IFS=$'\t' read -r kind n mcov mlines < <(awk -F'\t' \
        '/^__MULTI_META__/ { k = $2; n = $3; c = $4; l = $5 } END { printf "%s\t%s\t%s\t%s\n", k, n, c, l }' "$mm" 2>/dev/null) || true
    [[ "$n" =~ ^[0-9]{1,6}$ ]] || n=0
    if [[ "$kind" != split && "$kind" != strict && "$kind" != hdr ]]; then
        am_note="the unsplit input is still the engine's best reading"; return 1
    fi
    (( n >= 2 )) || { am_note="segmenting yields one dataset"; return 1; }
    [[ "$mcov" =~ ^[0-9]{1,3}$ ]] || { am_note="the segmented coverage is not measured"; return 1; }
    local need=$(( ${cov:-0} + _AM_GAIN ))
    (( need < PROVOST_MIN_COVERAGE )) && need=$PROVOST_MIN_COVERAGE
    if (( need > 100 )); then
        # The single parse already keeps (nearly) every line — it was held for
        # its column types — so no split can keep materially more.
        am_note="the single parse already keeps $cov% of the lines"; return 1
    fi
    if (( mcov < need )); then
        am_note="$n parts keep $mcov% of the lines, <$need%"; return 1
    fi

    _multi_split "$mo" "$tmp"
    local i dstrat dconf dcov dcons dlines nrows ncols nsyn tot=0 sub=0
    local -a an=() as=() ac=() af=() acv=() acs=() arows=()
    for (( i = 1; i <= n; i++ )); do
        if [[ ! -f "$tmp/ds.$i.meta" || ! -s "$tmp/ds.$i.wide" ]]; then
            am_note="part $i is empty"; return 1
        fi
        dstrat=""; dconf=""; dcov=""; dcons=""; dlines=""
        IFS=$'\t' read -r dstrat dconf dcov dcons dlines < "$tmp/ds.$i.meta" || true
        [[ "$dconf" =~ ^[0-9]{1,9}$ ]] || dconf=0
        [[ "$dcov" =~ ^[0-9]{1,3}$ ]] || dcov="-"
        [[ "$dcons" =~ ^[0-9]{1,3}$ ]] || dcons="-"
        [[ "$dlines" =~ ^[0-9]{1,9}$ ]] || dlines=0
        _am_part_check "$i" "$tmp/ds.$i.wide" "$tmp/ds.$i.tsv"
        # LAYOUT MODEL, per part: a part the gate rejects is read again by
        # the model on just that part's lines (its text comes from the engine,
        # PROVOST_MULTI_TEXT_DIR), arbitrating with the engine's own parse of
        # the part line by line. The model's table replaces the part only if
        # it passes the very same checks; otherwise the split is declined as
        # before.
        if [[ -n "$_AMP_NOTE" && "$PROVOST_MODEL" == 1 && -s "$tmp/amtext/ds.$i.txt" ]] &&
           _model_table "$tmp/amtext/ds.$i.txt" "$tmp/ds.$i.wide" "$dstrat"$'\t'"$dconf"$'\t'"$dcov"$'\t'"$dcons" "part$i"; then
            local _pnote="$_AMP_NOTE" _ostrat="$dstrat"
            dstrat=layout-model; dconf="$_MT_C"; dcov="$_MT_V"; dcons="${_MT_T:--}"
            _am_part_check "$i" "$_MT_OUT" "$tmp/ds.$i.lm.tsv"
            if [[ -z "$_AMP_NOTE" ]]; then
                cp "$_MT_OUT" "$tmp/ds.$i.wide"; mv "$tmp/ds.$i.lm.tsv" "$tmp/ds.$i.tsv"
                am_model=$((am_model + 1))
            else
                dstrat="$_ostrat"; _AMP_NOTE="$_pnote; the layout model's reading of it fails too"
            fi
        fi
        [[ -z "$_AMP_NOTE" ]] || { am_note="$_AMP_NOTE"; return 1; }
        nrows="$_AMP_ROWS"
        tot=$(( tot + dlines )); (( dlines >= _AM_FRAG_LINES )) && sub=$(( sub + dlines ))
        if _valid_name "$dstrat"; then :; else dstrat="$(_sanitize_name "$dstrat")"; dstrat="${dstrat:-dataset}"; fi
        an+=("${name:+$name.}$dstrat"); as+=("$dstrat"); ac+=("$dconf"); af+=("$tmp/ds.$i.tsv")
        acv+=("$dcov"); acs+=("$dcons"); arows+=("$nrows")
    done
    if (( sub == 0 || sub * 2 < tot )); then
        am_note="the parts are fragments ($sub of $tot lines in parts of $_AM_FRAG_LINES+ lines)"; return 1
    fi

    # Naming. Every part is named as --multi names it: NAME.strategy for
    # 'run' / --name, its strategy for a nameless inject (which is how a
    # nameless inject is always named — after the parser that read it). With
    # a NAME, the part that reads the same columns as the single parse (the
    # store's own reconcile overlap, PROVOST_RECONCILE_OVERLAP) CONTINUES
    # dataset NAME, where that parse would have been filed: a command whose
    # output passes the gate one day and segments the next keeps its rows in
    # one dataset. When several parts qualify, the one with the most rows.
    if [[ -n "$name" && -s "$tmp/wide.tsv" ]]; then
        local best=-1 bi=-1 ov hs hd
        _core_melt_tidy_file "$tmp/wide.tsv" > "$tmp/single.tsv"
        hs="$(_tidy_cols "$tmp/single.tsv" | paste -sd '\t' -)"
        for (( i = 0; i < ${#af[@]}; i++ )); do
            hd="$(_tidy_cols "${af[i]}" | paste -sd '\t' -)"
            ov="$(printf '%s\n%s\n' "$hs" "$hd" | _col_overlap)"
            [[ "$ov" =~ ^[0-9]+$ ]] || ov=0
            if (( ov >= PROVOST_RECONCILE_OVERLAP && arows[i] > best )); then best=${arows[i]}; bi=$i; fi
        done
        (( bi >= 0 )) && an[bi]="$name"
    fi
    ds_name=("${an[@]}"); ds_strategy=("${as[@]}"); ds_conf=("${ac[@]}"); ds_file=("${af[@]}")
    ds_cov=("${acv[@]}"); ds_cons=("${acs[@]}")
    am_cov="$mcov"; am_kind="$kind"
    return 0
}

# LAYOUT MODEL (PROVOST_MODEL, --model / --no-model). Called by _ingest when
# the quality gate would still hold the single parse (after AUTO-MULTI). It
# runs the explicit-only layout-model strategy and adopts its table INSTEAD of
# holding the capture only when the engine's own measures of that table pass
# the gate — coverage >= PROVOST_MIN_COVERAGE, consistency (when judged) >=
# PROVOST_MIN_CONSISTENCY — and it keeps at least as many lines as the held
# parse. Otherwise it returns 1 with the reason in lm_note. Works on
# _ingest's locals (dynamic scope): reads engine, eargs, body, tmp, cov;
# on success replaces wide.tsv and sets strategy, conf, cov, tcons.
_layout_model () {
    local lv lt
    if ! _model_table "$body" "$tmp/wide.tsv" "$strategy"$'\t'"$conf"$'\t'"${cov:--}"$'\t'"${tcons:--}" whole; then
        lm_note="it found nothing to file (it needs python3 and declines inputs over 600 lines, and mixed reports)"; return 1
    fi
    lv="$_MT_V"; lt="$_MT_T"
    [[ -n "$lv" ]] || { lm_note="its line coverage is not measured"; return 1; }
    if (( PROVOST_MIN_COVERAGE > 0 && lv < PROVOST_MIN_COVERAGE )); then
        lm_note="its table keeps $lv% of the lines, <$PROVOST_MIN_COVERAGE%"; return 1
    fi
    if [[ -n "$lt" ]] && (( PROVOST_MIN_CONSISTENCY > 0 && lt < PROVOST_MIN_CONSISTENCY )); then
        lm_note="its columns are $lt% type-consistent, <$PROVOST_MIN_CONSISTENCY%"; return 1
    fi
    if [[ -n "$cov" ]] && (( lv < cov )); then
        lm_note="its table keeps $lv% of the lines, fewer than the held parse ($cov%)"; return 1
    fi
    cp "$_MT_OUT" "$tmp/wide.tsv"
    strategy=layout-model; conf="$_MT_C"; cov="$lv"; tcons="$lt"
    return 0
}

# LAYOUT MODEL, segmenting (after _layout_model declines a whole capture):
# the model cuts a mixed report into uniform runs — a key/value block, a
# table, a list — from its own line roles (layout_model.py --segments), and
# each run's records, made from the whole document's labels, are measured by
# the engine like any parse (PROVOST_LM_RECORDS). The runs are filed as the
# capture's datasets INSTEAD of holding it under AUTO-MULTI's rules: every
# run passes the per-part checks (_am_part_check), the runs hold at least
# _AM_GAIN points more of the input than the held parse (and the gate's
# minimum), and they are not fragments. Otherwise 1, the reason in mm_note.
# Works on _ingest's locals like _auto_multi; fills ds_* and am_cov/am_kind.
_model_multi () {
    local segd="$tmp/lmseg" i f n=0 dstrat dconf dcov dcons dlines tot=0 sub=0 covl=0 alll jcov need
    command -v python3 >/dev/null 2>&1 || { mm_note="python3 is not installed"; return 1; }
    rm -rf "$segd"; mkdir -p "$segd"
    python3 "$PROVOST_ROOT/extract/model/layout_model.py" "$body" --segments "$segd" >/dev/null 2>&1 \
        || { mm_note="it finds no two sections it can read"; return 1; }
    while [[ -f "$segd/seg.$((n + 1)).txt" ]]; do n=$((n + 1)); done
    local -a an=() as=() ac=() af=() acv=() acs=()
    for (( i = 1; i <= n; i++ )); do
        f="$segd/seg.$i.txt"
        PROVOST_LM_RECORDS="$segd/seg.$i.rec" _model_table "$f" "" "" "seg$i" \
            || { mm_note="its section $i yields nothing"; return 1; }
        dstrat=layout-model; dconf="$_MT_C"; dcov="${_MT_V:--}"; dcons="${_MT_T:--}"
        dlines=$(grep -c '[^[:space:]]' "$f" || true)
        _am_part_check "$i" "$_MT_OUT" "$segd/seg.$i.tsv"
        [[ -z "$_AMP_NOTE" ]] || { mm_note="${_AMP_NOTE/#part/section}"; return 1; }
        tot=$(( tot + dlines )); (( dlines >= _AM_FRAG_LINES )) && sub=$(( sub + dlines ))
        [[ "$dcov" == - ]] || covl=$(( covl + dcov * dlines ))
        an+=("${name:+$name.}layout-model"); as+=(layout-model); ac+=("$dconf"); af+=("$segd/seg.$i.tsv")
        acv+=("$dcov"); acs+=("$dcons")
    done
    if (( sub == 0 || sub * 2 < tot )); then
        mm_note="its sections are fragments ($sub of $tot lines in sections of $_AM_FRAG_LINES+ lines)"; return 1
    fi
    alll=$(grep -c '[^[:space:]]' "$body" || true)
    (( alll > 0 )) || { mm_note="the input is empty"; return 1; }
    jcov=$(( covl / alll ))
    need=$(( ${cov:-0} + _AM_GAIN ))
    (( need < PROVOST_MIN_COVERAGE )) && need=$PROVOST_MIN_COVERAGE
    if (( need > 100 )); then mm_note="the held parse already keeps $cov% of the lines"; return 1; fi
    if (( jcov < need )); then mm_note="its $n sections keep $jcov% of the lines, <$need%"; return 1; fi
    ds_name=("${an[@]}"); ds_strategy=("${as[@]}"); ds_conf=("${ac[@]}"); ds_file=("${af[@]}")
    ds_cov=("${acv[@]}"); ds_cons=("${acs[@]}")
    am_cov="$jcov"; am_kind=layout-model
    return 0
}

# _llm_table INPUT TAG — the engine's measure of records an LLM template made
# for INPUT (the explicit-only llm-template strategy emits PROVOST_LLM_RECORDS,
# set by the caller). Sets _MT_OUT / _MT_C / _MT_V / _MT_T like _model_table.
_llm_table () {
    local inp="$1" tag="$2" lmeta="$tmp/llm.$2.meta" ls=""
    _MT_OUT="$tmp/llm.$tag.tsv"; _MT_C=""; _MT_V=""; _MT_T=""
    _run_engine "$engine" --max-input-size "$PROVOST_MAX_INPUT" \
        ${eargs[@]+"${eargs[@]}"} -s llm-template --emit-meta -f tsv --max-strategies 1 "$inp" \
        >"$_MT_OUT" 2>"$lmeta" || true
    IFS=$'\t' read -r ls _MT_C _MT_V _MT_T < <(awk -F'\t' \
        '/^__EXTRACT_META__/ { s = $2; c = $3; v = $4; t = $5 } END { printf "%s\t%s\t%s\t%s\n", s, c, v, t }' "$lmeta" 2>/dev/null) || true
    [[ "$ls" == llm-template && -s "$_MT_OUT" ]] && grep -q . "$_MT_OUT" || return 1
    [[ "$_MT_C" =~ ^[0-9]{1,9}$ ]] || _MT_C=0
    [[ "$_MT_V" =~ ^[0-9]{1,3}$ ]] || _MT_V=""
    [[ "$_MT_T" =~ ^[0-9]{1,3}$ ]] || _MT_T=""
    return 0
}

# _llm_file TEMPLATE — apply one LLM template to the capture and file what it
# yields under the gate's rules: each block the template finds is measured by
# the engine on that block's own lines and must pass the per-part checks
# (_am_part_check); the blocks are not fragments; and together they keep at
# least _AM_GAIN points more of the input than the held parse (and the gate's
# minimum), the lines the template marks as decoration not counted. One block
# replaces the held parse (wide.tsv); several are filed as the capture's
# datasets, named NAME.<block>. Otherwise 1, the reason in llm_note.
_llm_file () {
    local t="$1" py="$PROVOST_ROOT/extract/llm/template.py" bd="$tmp/llmblk" out n=0 skipped=0 i f bname
    local dstrat dconf dcov dcons dlines tot=0 sub=0 covl=0 alll jcov need
    llm_notes=""; rm -rf "$bd"
    out="$(python3 "$py" apply "$t" "$body" --out "$bd" 2>/dev/null)" \
        || { llm_note="it matches nothing in the input"; return 1; }
    skipped="${out##*$'\t'}"; [[ "$skipped" =~ ^[0-9]{1,9}$ ]] || skipped=0
    while [[ -f "$bd/blk.$((n + 1)).rec" ]]; do n=$((n + 1)); done
    (( n >= 1 )) || { llm_note="it matches nothing in the input"; return 1; }
    # a nameless capture's datasets are named after the template's tool
    local dsbase="$name"
    if [[ -z "$dsbase" ]]; then
        dsbase="$(head -n 1 "$bd/tool" 2>/dev/null)"; _valid_name "$dsbase" || dsbase=""
    fi
    local -a an=() as=() ac=() af=() acv=() acs=() aw=()
    for (( i = 1; i <= n; i++ )); do
        f="$bd/blk.$i.txt"; bname="$(head -n 1 "$bd/blk.$i.name" 2>/dev/null)"
        _valid_name "$bname" || bname="block$i"
        PROVOST_LLM_RECORDS="$bd/blk.$i.rec" _llm_table "$f" "blk$i" \
            || { llm_note="its block '$bname' yields nothing"; return 1; }
        # a block's lines are, by construction, the lines the template read
        # into its records (its text is written from them), so its coverage
        # is 100 and jcov below is the share of the input's data lines the
        # template read. The engine's token-recall measure is not used here:
        # it counts a value the template dropped on purpose (a broadcast
        # address, a trailing flag) as a missed line. Its type consistency
        # is recorded, not gated (see _quality_exempt); the shape checks
        # still apply.
        dstrat=llm-template; dconf="$_MT_C"; dcov=100; dcons="${_MT_T:--}"
        dlines=$(grep -c '[^[:space:]]' "$f" || true)
        _am_part_check "$i" "$_MT_OUT" "$bd/blk.$i.tsv"
        [[ -z "$_AMP_NOTE" ]] || { llm_note="${_AMP_NOTE/#part $i/block $bname}"; return 1; }
        tot=$(( tot + dlines )); (( dlines >= _AM_FRAG_LINES )) && sub=$(( sub + dlines ))
        [[ "$dcov" == - ]] || covl=$(( covl + dcov * dlines ))
        an+=("${dsbase:+$dsbase.}$bname"); as+=(llm-template); ac+=("$dconf"); af+=("$bd/blk.$i.tsv")
        acv+=("$dcov"); acs+=("$dcons"); aw+=("$_MT_OUT")
    done
    if (( n > 1 )) && (( sub == 0 || sub * 2 < tot )); then
        llm_note="its blocks are fragments ($sub of $tot lines in blocks of $_AM_FRAG_LINES+ lines)"; return 1
    fi
    alll=$(grep -c '[^[:space:]]' "$body" || true)
    # the lines a template marks as decoration are not counted against it:
    # none is lost, since every one that carries a word or a number is filed
    # in its NAME.notes (template.py notes) and the rest are rules and box
    # glyphs. Against a CATCH-ALL parse the gate passed (text-lines,
    # line-records: the engine's own lossless readings, _FR_COV_CATCHALL)
    # it may trail by _LLM_CATCHALL_SLACK points rather than _AM_GAIN.
    local catchall=0
    case " $strategy " in *" text-lines "*|*" line-records "*) [[ "$llm_floor" == match ]] && catchall=1 ;; esac
    alll=$(( alll - skipped ))
    (( alll > 0 )) || { llm_note="it keeps none of the input's data lines"; return 1; }
    jcov=$(( covl / alll ))
    # against a held parse the template must gain _AM_GAIN points; against a
    # generic parse the gate passed (llm_floor=match) it may trail by that
    # much: it was written for this tool's layout, the heuristic was not, and
    # a title or prompt the heuristic counted as a row is decoration here
    if (( catchall )); then need=$(( ${cov:-0} - _LLM_CATCHALL_SLACK ))
    elif [[ "$llm_floor" == match ]]; then need=$(( ${cov:-0} - _AM_GAIN ))
    else need=$(( ${cov:-0} + _AM_GAIN )); fi
    (( need < PROVOST_MIN_COVERAGE )) && need=$PROVOST_MIN_COVERAGE
    (( need > 100 )) && need=100
    if (( jcov < need )); then
        llm_note="its $n block(s) keep $jcov% of the input's data lines, <$need%"; return 1
    fi
    [[ -s "$bd/notes.tsv" ]] && { llm_notes="$bd/notes.tsv"; llm_notes_base="$dsbase"; }
    if (( n == 1 )); then
        cp "${aw[0]}" "$tmp/wide.tsv"
        strategy=llm-template; conf="${ac[0]}"; cov="$jcov"; tcons="${acs[0]/#-/}"
        [[ -z "$name" && -n "$dsbase" ]] && name="$dsbase"
        llm_multi=0
    else
        ds_name=("${an[@]}"); ds_strategy=("${as[@]}"); ds_conf=("${ac[@]}"); ds_file=("${af[@]}")
        ds_cov=("${acv[@]}"); ds_cons=("${acs[@]}")
        am_cov="$jcov"; am_kind=llm-template; llm_multi=1
    fi
    return 0
}

# LLM TEMPLATES (PROVOST_LLM, --llm / --no-llm). Called by _ingest on a
# capture the quality gate would hold, and on a generic parse it passed. A
# template is a set of regular expressions per kind of line, or a boxed
# parser, that Claude wrote for one tool (extract/llm/). provost looks for one
# in the store (llm-templates/, committed with the captures: the tool's name,
# or any template whose `detect` pattern recognises the input), then in
# PROVOST_LLM_TEMPLATE_DIRS and among the templates shipped with provost, and
# files what it yields (_llm_file). With an API key, a stored template that
# meets a kind of line it has not met before is first sent back to Claude to
# be extended (repair), and when none fits Claude writes one from this
# capture — the steps that send the capture's text to the Anthropic API. A
# new or extended template is kept only when what it yields passes the same
# rules. Every later capture of the tool is then parsed offline. With LEARN=0
# Claude is never asked (only a stored template can fit). Otherwise 1, the
# reason in llm_note.
_llm_template () {
    local learn="${1:-1}" py="$PROVOST_ROOT/extract/llm/template.py" tdir="$PROVOST_WORKSPACE/llm-templates"
    local tool="$name" kf tpl="" found="" new="$tmp/llm-new.json" err prev
    command -v python3 >/dev/null 2>&1 || { llm_note="python3 is not installed"; return 1; }
    [[ -z "$tool" && -n "$cmd" ]] && { tool="${cmd%% *}"; tool="${tool##*/}"; }
    kf="$(python3 "$py" key "$body" --name "$tool" 2>/dev/null)" || { llm_note="the input could not be read"; return 1; }
    # the store's templates, then the user's own libraries
    # (PROVOST_LLM_TEMPLATE_DIRS), then the ones shipped with provost
    local d dirs="$tdir:${PROVOST_LLM_TEMPLATE_DIRS:-}"
    IFS=: read -ra _dl <<< "$dirs"
    for d in "${_dl[@]}"; do [[ -n "$d" && -f "$d/$kf" ]] && { tpl="$d/$kf"; break; }; done
    if [[ -z "$tpl" ]]; then
        found="$(python3 "$py" find "$body" --dirs "$dirs" 2>/dev/null)" && tpl="${found%%$'\t'*}"
    fi
    if [[ -n "$tpl" ]]; then
        # REPAIR: a stored template that meets a kind of line it has not met
        # before (left unread, or dropped by a skip pattern) goes back to
        # Claude once to be extended; the extension is kept only if it still
        # reads the captures the template was written from and passes the
        # same rules. Otherwise the template stays as it was.
        local miss="" rnote="" rstat=""
        if (( learn )) && [[ -n "${PROVOST_LLM_API_KEY:-}${ANTHROPIC_API_KEY:-}" ]] \
            && miss="$(python3 "$py" misses "$tpl" "$body" 2>/dev/null)" && [[ -n "$miss" ]]; then
            if rstat="$(python3 "$py" repair "$tpl" "$body" --out "$new" --tool "$tool" --model "$PROVOST_LLM_MODEL" 2>"$tmp/llm-repair.err")"; then
                if _llm_file "$new"; then
                    # kept even when unchanged: it now holds this capture, so
                    # these lines are not sent again
                    mkdir -p "$tdir" && cp "$new" "$tdir/$kf"
                    if [[ "$rstat" == unchanged ]]; then
                        llm_how="template ${tpl##*/}, which Claude checked for $miss and left as it was (llm-templates/$kf)"
                    else
                        llm_how="template ${tpl##*/}, which Claude extended for $miss (llm-templates/$kf)"
                    fi
                    llm_learned="llm-templates/$kf"
                    return 0
                fi
                rnote="the template Claude extended for $miss fails the gate: $llm_note"
            else
                err="$(cat "$tmp/llm-repair.err" 2>/dev/null)"
                rnote="Claude could not extend it for $miss: ${err#template: }"
            fi
        fi
        if _llm_file "$tpl"; then llm_how="template ${tpl##*/}${rnote:+ ($rnote)}"; return 0; fi
        llm_note="template ${tpl##*/} does not fit: $llm_note${rnote:+; $rnote}"
        # a repair was tried: don't also have Claude start the tool's template over
        [[ -z "$rnote" ]] || return 1
    fi
    (( learn )) || { llm_note="${llm_note:-no stored template fits}"; return 1; }
    if [[ -z "${PROVOST_LLM_API_KEY:-}${ANTHROPIC_API_KEY:-}" ]]; then
        llm_note="${llm_note:-no stored template fits}; no API key to have Claude write one (PROVOST_LLM_API_KEY or ANTHROPIC_API_KEY)"
        return 1
    fi
    prev="$llm_note"
    if ! err="$(python3 "$py" learn "$body" --out "$new" --tool "$tool" --model "$PROVOST_LLM_MODEL" 2>&1 >/dev/null)"; then
        llm_note="${prev:+$prev; }Claude wrote no usable template: ${err#template: }"; return 1
    fi
    if ! _llm_file "$new"; then
        llm_note="${prev:+$prev; }the template Claude wrote fails the gate: $llm_note"; return 1
    fi
    mkdir -p "$tdir" && cp "$new" "$tdir/$kf"
    llm_how="a new template Claude wrote (llm-templates/$kf)"; llm_learned="llm-templates/$kf"
    return 0
}

# Core ingest pipeline shared by `inject` (pipe/file), `run` (a command) and
# `batch` (many inputs, one commit). A capture is two halves:
#   _ingest_parse  extracts → classifies → melts, into a work dir, touching
#                  nothing in the store (LLM templates aside, see _llm_template);
#   _ingest_store  assigns a global #N to each row → stores → records
#                  provenance (#N → capture) → commits, under the store lock.
# The halves meet in the work dir: the dataset files, and state.sh (the parse's
# verdict, as `declare -p`), which _ingest_store sources back into locals.
# `batch` runs the parse halves in parallel and the store halves in one lock.
#   $1 input file  $2 threshold  $3 source (inject|run|batch)  $4 cmd ("" for
#   inject)  $5 rc ("" for inject)  $6 dataset name ("" = the winning
#   strategy's name)
_ingest () {
    local in="$1" threshold="$2" source="$3" cmd="${4:-}" rc="${5:-}" name="${6:-}"
    _mktempd provost; local tmp="$_MKTEMPD"
    _ingest_parse "$in" "$threshold" "$cmd" "$name" "$tmp"
    _ingest_store "$in" "$source" "$cmd" "$rc" "$tmp" 1
    rm -rf "$tmp"
}

# _ingest_parse IN THRESHOLD CMD NAME WORKDIR — the parse half of a capture.
_ingest_parse () {
    local in="$1" threshold="$2" cmd="${3:-}" name="${4:-}" tmp="$5"
    local engine; engine="$(_find_engine)" \
        || die "extract engine not found (expected at $PROVOST_ROOT/extract/provost-extract)."

    # A metadata prologue — a leading run of '#%' lines, as prepended by
    # wrappers like deframe — is capture context, not data: peel it off before
    # detection so it can never corrupt a parse. The prologue is recorded with
    # the capture (context.txt) and 'provost source' still shows the input
    # verbatim; a '#%' line after the first data line is data, not context.
    local ctxf="$tmp/context.txt" body="$in" first2=""
    first2="$(head -c 2 "$in" 2>/dev/null || true)"
    if [[ "$first2" == '#%' ]]; then
        body="$tmp/body.txt"
        CTX="$ctxf" awk \
            '!started && /^#%/ { print > (ENVIRON["CTX"]); next } { started=1; print }' \
            "$in" > "$body"
    fi

    local metaf="$tmp/meta.txt" strategy="" conf="" cov="" tcons=""
    # Extra strategy dir (deframe et al.): --plugin-dir / $PROVOST_PLUGIN_DIR is
    # forwarded to the engine, which already searches it alongside its own
    # strategies/ and the ~/.config/provost/strategies user dir.
    local eargs=()
    if [[ -n "${PROVOST_PLUGIN_DIR:-}" ]]; then
        eargs+=(--plugin-dir "$PROVOST_PLUGIN_DIR")
    fi

    # The datasets this capture will file, each as a content FILE (header +
    # rows). A normal capture yields exactly one; --multi (PROVOST_MULTI=1) can
    # yield several, because some reports genuinely hold more than one table
    # (iostat -x: a uname banner, an avg-cpu kv block and a Device table).
    # Everything downstream — row ids, provenance, the commit — loops over
    # these, so the single-dataset case is just N=1.
    local -a ds_name=() ds_strategy=() ds_conf=() ds_file=() ds_cov=() ds_cons=()
    local reason=""
    # AUTO-MULTI state (see _auto_multi): am_reason = why the gate held the
    # single parse that was segmented instead; am_note = why segmenting was
    # declined (the capture is then held as before); am_cov = the coverage of
    # all the datasets together.
    local am_reason="" am_note="" am_cov="" am_kind="" single_strategy="" single_conf="" am_model=0
    # LAYOUT MODEL state (see _layout_model): lm_reason = why the gate held the
    # engine's parse that the model's table replaced; lm_note = why the model
    # was declined; single_cov / single_cons = the held parse's measures.
    local lm_reason="" lm_note="" single_cov="" single_cons="" mm_note=""
    local llm_reason="" llm_note="" llm_how="" llm_learned="" llm_multi=0 llm_floor=gain llm_notes="" llm_notes_base=""

    if [[ "$PROVOST_MULTI" == 1 ]]; then
        # --multi splits the input into blocks, groups them, and emits each
        # surviving dataset behind a `__DATASET__ n strategy confidence
        # coverage consistency lines` line.
        local multi_out="$tmp/multi.txt"
        _run_engine "$engine" --max-input-size "$PROVOST_MAX_INPUT" \
            ${eargs[@]+"${eargs[@]}"} --multi -f tsv --max-strategies 1 "$body" \
            >"$multi_out" 2>"$metaf" || true
        _multi_split "$multi_out" "$tmp"
        local i=1 dstrat dconf dcov dcons
        while [[ -f "$tmp/ds.$i.meta" ]]; do
            dstrat=""; dconf=""; dcov=""; dcons=""
            IFS=$'\t' read -r dstrat dconf dcov dcons _ < "$tmp/ds.$i.meta" || true
            [[ "$dconf" =~ ^[0-9]{1,9}$ ]] || dconf=0
            [[ "$dcov" =~ ^[0-9]{1,3}$ ]] || dcov="-"
            [[ "$dcons" =~ ^[0-9]{1,3}$ ]] || dcons="-"
            # A dataset below the threshold is dropped rather than queued: it is
            # a fragment of an input whose other parts did match, and the whole
            # input is stored with the capture either way.
            if [[ -n "$dstrat" && -s "$tmp/ds.$i.wide" ]] && (( dconf >= threshold )); then
                _core_melt_tidy_file "$tmp/ds.$i.wide" > "$tmp/ds.$i.tsv"
                if _valid_name "$dstrat"; then :; else dstrat="$(_sanitize_name "$dstrat")"; dstrat="${dstrat:-dataset}"; fi
                ds_name+=("${name:+$name.}$dstrat"); ds_strategy+=("$dstrat"); ds_conf+=("$dconf")
                ds_cov+=("$dcov"); ds_cons+=("$dcons")
                ds_file+=("$tmp/ds.$i.tsv")
            fi
            i=$((i + 1))
        done
    fi

    if [[ ${#ds_file[@]} -eq 0 ]]; then
        # Single-dataset path — also the fallback when --multi found nothing
        # worth filing, so an unmatched input still reaches the triage queue.
        _run_engine "$engine" --max-input-size "$PROVOST_MAX_INPUT" \
            ${eargs[@]+"${eargs[@]}"} \
            --emit-meta -f tsv --max-strategies 1 "$body" >"$tmp/wide.tsv" 2>"$metaf" || true
        # The meta line also carries the winner's LINE COVERAGE (the share of
        # the input's non-blank lines that reached the dataset) and its
        # COLUMN-TYPE CONSISTENCY (the share of typed cells that match their
        # column's type); '-' = not measured, e.g. a very large input.
        IFS=$'\t' read -r strategy conf cov tcons < <(awk -F'\t' \
            '/^__EXTRACT_META__/ { s = $2; c = $3; v = $4; t = $5 } END { printf "%s\t%s\t%s\t%s\n", s, c, v, t }' "$metaf" 2>/dev/null) || true
        [[ "$conf" =~ ^[0-9]{1,9}$ ]] || conf=0
        [[ "$cov" =~ ^[0-9]{1,3}$ ]] || cov=""
        [[ "$tcons" =~ ^[0-9]{1,3}$ ]] || tcons=""

        # Build the content to file (header + data rows), pre-#-index, for both
        # the matched and the held-back (unsorted) paths.
        if [[ -z "$strategy" || ! -s "$tmp/wide.tsv" ]] || ! grep -q . "$tmp/wide.tsv"; then reason="no-match"
        elif (( conf < threshold )); then reason="low-confidence ($conf<$threshold)"
        elif ! _quality_exempt "$engine" "$strategy"; then
            if [[ -n "$cov" ]] && (( PROVOST_MIN_COVERAGE > 0 && cov < PROVOST_MIN_COVERAGE )); then
                reason="low-coverage ($strategy kept $cov% of the lines, <$PROVOST_MIN_COVERAGE%)"
            elif [[ -n "$tcons" ]] && (( PROVOST_MIN_CONSISTENCY > 0 && tcons < PROVOST_MIN_CONSISTENCY )); then
                reason="low-consistency ($strategy columns $tcons% type-consistent, <$PROVOST_MIN_CONSISTENCY%)"
            fi
        fi
        single_strategy="$strategy"; single_conf="$conf"
        # AUTO-MULTI: before holding a parse the quality gate rejects, try
        # segmenting the input (the engine's --multi); file the datasets when
        # every one of them passes the gate and together they keep materially
        # more of the input (see _auto_multi). Only gate failures pay for this
        # second engine run — a parse that passes is never re-run.
        # LLM TEMPLATES, FIRST (PROVOST_LLM=1): a template Claude wrote for this
        # tool — stored, or written now — is tried before the heuristics'
        # fallbacks on a held capture, and before a GENERIC parse the gate
        # passed is trusted (the gate cannot see a title or prompt read as the
        # header; a template written for the tool's layout can). Never over an
        # exact-format or tool-specific parser. See _llm_template, _llm_file.
        if [[ "$PROVOST_LLM" == 1 ]]; then
            single_cov="$cov"; single_cons="$tcons"
            if [[ "$reason" == low-coverage* || "$reason" == low-consistency* ]]; then
                if _llm_template 1; then
                    if (( llm_multi )); then am_reason="$reason"; else llm_reason="$reason"; fi
                    reason=""
                fi
            elif [[ -z "$reason" ]] && ! _quality_exempt "$engine" "$strategy"; then
                local _held="passed over (the engine read it as $strategy, coverage ${cov:--}%): a template"
                llm_floor=match
                if _llm_template 1; then
                    if [[ -n "$llm_learned" ]]; then _held="$_held Claude wrote for this tool fits"
                    else _held="$_held stored for this tool fits"; fi
                    if (( llm_multi )); then am_reason="$_held"; else llm_reason="$_held"; fi
                fi
                llm_floor=gain
            fi
        fi
        if [[ "$PROVOST_AUTO_MULTI" == 1 && ( "$reason" == low-coverage* || "$reason" == low-consistency* ) ]]; then
            if _auto_multi; then am_reason="$reason"; reason=""; fi
        fi
        # LAYOUT MODEL: still held after that? Try the learned layout model's
        # table, filed only if it passes the same gate (see _layout_model).
        if [[ "$PROVOST_MODEL" == 1 && -z "$am_reason" && ( "$reason" == low-coverage* || "$reason" == low-consistency* ) ]]; then
            single_cov="$cov"; single_cons="$tcons"
            if _layout_model; then lm_reason="$reason"; reason=""
            elif _model_multi; then am_reason="$reason"; reason=""
            else lm_note="$lm_note; segmenting it itself: $mm_note"; fi
        fi
        # (a generic parse the gate passed and no template bettered is filed
        # as it was; meta.tsv still records why the template was declined)
        if [[ -z "$am_reason" ]]; then
            local dsname content="$tmp/content.tsv"
            if [[ -n "$reason" ]]; then
                dsname="unsorted"; strategy="${strategy:-none}"
                {
                    printf 'raw\n'
                    if ! base64 -w0 < "$in" 2>/dev/null; then base64 < "$in" | tr -d '\n'; fi
                    printf '\n'
                } > "$content"
            else
                if _valid_name "$strategy"; then dsname="$strategy"
                else dsname="$(_sanitize_name "$strategy")"; dsname="${dsname:-dataset}"; fi
                [[ -n "$name" ]] && dsname="$name"
                _core_melt_tidy_file "$tmp/wide.tsv" > "$content"
            fi
            ds_name=("$dsname"); ds_strategy=("$strategy"); ds_conf=("$conf")
            ds_file=("$content")
        fi
    fi
    # LLM NOTES: the lines a template set aside that still carry a word or a
    # number (a title, a version banner, a summary count, a warning or an
    # error between the rows, a line it could not read) are filed beside its
    # datasets as NAME.notes (line, kind, text): the tables stay clean and
    # nothing that carries information is lost (template.py notes).
    if [[ -z "$reason" && -n "$llm_notes" && -s "$llm_notes" ]] \
        && [[ "$strategy" == llm-template || "$am_kind" == llm-template ]]; then
        local nn="${ds_name[0]}.notes" nk="${#ds_name[@]}" _dn
        if (( llm_multi )); then nn="${llm_notes_base:+$llm_notes_base.}notes"
        else ds_cov[0]="${cov:--}"; ds_cons[0]="${tcons:--}"; fi
        for _dn in "${ds_name[@]}"; do [[ "$_dn" == "$nn" ]] && nn="${nn}_2"; done
        ds_name[nk]="$nn"; ds_strategy[nk]=llm-template; ds_conf[nk]="${ds_conf[0]}"
        ds_file[nk]="$llm_notes"; ds_cov[nk]="-"; ds_cons[nk]="-"
    fi

    # The verdict, for the store half: every variable it reads, as `declare -p`
    # (sourced inside _ingest_store, each comes back as a local there).
    declare -p ds_name ds_strategy ds_conf ds_file ds_cov ds_cons reason \
        strategy conf cov tcons am_reason am_note am_cov am_kind am_model \
        single_strategy single_conf single_cov single_cons lm_reason lm_note \
        llm_reason llm_note llm_how llm_learned ctxf > "$tmp/state.sh"
}

# _ingest_store IN SOURCE CMD RC WORKDIR COMMIT — the store half of a capture
# parsed into WORKDIR. COMMIT=1 takes the store lock, commits and tags, as one
# capture always has; COMMIT=0 is for `batch`, which holds the lock across all
# its captures and makes one commit, and leaves the capture id in $_CAPID.
_ingest_store () {
    local in="$1" source="$2" cmd="${3:-}" rc="${4:-}" tmp="$5" commit="${6:-1}"
    # shellcheck source=/dev/null
    source "$tmp/state.sh"

    # --- critical section: assign #N, append, provenance, commit -----------
    if (( commit )); then
        _lock
        _require_on_branch
    fi

    local id capd ts
    _next_capture_id; id="$_CAPID"; capd="$(_cap_dir)/$id"; mkdir -p "$capd"
    ts="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
    cp "$in" "$capd/output.txt"
    if [[ -s "$ctxf" ]]; then cp "$ctxf" "$capd/context.txt"; fi
    local prov; prov="$(_prov_file)"
    [[ -f "$prov" ]] || printf '#\tdataset\tcapture\n' > "$prov"

    # CONTEXT COLUMNS (--context-columns, PROVOST_CONTEXT_COLUMNS=1): every
    # `key=value` word in the '#%' prologue becomes a column on each of this
    # capture's rows, after `#` and `capture` -- so a wrapper that captures one
    # host's output can say `#% host=web03 rack=r2` and every row carries it.
    # The last value of a repeated key wins; words that are not `key=value`
    # stay context only. A data column of the same name keeps its name and the
    # context one takes a leading `_` (melt's clash rule, as `capture` does);
    # the names are chosen once per capture, against the data columns of all
    # its datasets (and `capture`), and recorded in meta.tsv as
    # `context_columns` (key=column,...), so a reader can tell a context
    # column from a data column of the same name.
    # A held capture (`unsorted`) stays as it is: one raw column.
    local ctx_keys="" ctx_vals="" ctx_cols="" ctx_map=""
    if [[ "$PROVOST_CONTEXT_COLUMNS" == 1 && -s "$ctxf" ]]; then
        { IFS= read -r ctx_keys; IFS= read -r ctx_vals; } < <(awk '
            { sub(/^#%[ \t]*/, "")
              for (i = 1; i <= NF; i++) {
                  e = index($i, "="); if (e < 2) continue
                  k = substr($i, 1, e - 1)
                  if (k !~ /^[A-Za-z_][A-Za-z0-9_.-]*$/) continue
                  if (!(k in V)) O[++n] = k
                  V[k] = substr($i, e + 1)
              } }
            END { for (j = 1; j <= n; j++) { ks = ks (j > 1 ? "\t" : "") O[j]; vs = vs (j > 1 ? "\t" : "") V[O[j]] }
                  print ks; print vs }' "$ctxf") || true
        local -a ctx_files=(); local ci
        for (( ci=0; ci<${#ds_name[@]}; ci++ )); do
            [[ "${ds_name[ci]}" == unsorted ]] || ctx_files+=("${ds_file[ci]}")
        done
        if [[ -n "$ctx_keys" ]] && (( ${#ctx_files[@]} > 0 )); then
            { IFS= read -r ctx_cols; IFS= read -r ctx_map; } < <(CK="$ctx_keys" awk -F'\t' '
                !(FILENAME in H) && length($0) { H[FILENAME] = 1; for (i = 1; i <= NF; i++) U[$i] = 1 }
                END { U["capture"] = 1
                      n = split(ENVIRON["CK"], K, "\t")
                      for (j = 1; j <= n; j++) {
                          m = K[j]; while (m in U) m = "_" m
                          U[m] = 1; cs = cs (j > 1 ? "\t" : "") m; ms = ms (j > 1 ? "," : "") K[j] "=" m
                      }
                      print cs; print ms }' "${ctx_files[@]}") || true
        fi
    fi

    # File each dataset in turn: allocate its own #N range, store it, and record
    # a provenance line per row. Row ids run consecutively across the datasets
    # of one capture, so every row still maps back to this one input.
    local nextid firstid startid nrows nf totalrows=0 fields=0 k final ck cv
    nextid="$(cat "$(_rowid_file)" 2>/dev/null || true)"
    [[ "$nextid" =~ ^[0-9]{1,15}$ ]] || die "corrupt row counter in $(_rowid_file): '$nextid'"
    nextid=$((10#$nextid)); firstid="$nextid"
    local -a out_names=() out_ranges=()
    for (( k=0; k<${#ds_file[@]}; k++ )); do
        startid="$nextid"
        # Prepend the visible, globally-unique '#' index and the capture id
        # (which run each row came from) to the header + each row (blank lines
        # dropped), counting rows and header fields as we go. A data column
        # already named `capture` keeps its name; the added one becomes
        # `_capture` (melt's clash rule), so a by-name union never merges them.
        ck="$ctx_cols"; cv="$ctx_vals"
        [[ "${ds_name[k]}" == unsorted ]] && { ck=""; cv=""; }
        CNTF="$tmp/cnt" CAPID="$id" CK="$ck" CV="$cv" awk -F'\t' -v s="$startid" '
            length($0) == 0 { next }
            !h { h = 1; nf = NF + 1
                 for (i = 1; i <= NF; i++) U[$i] = 1
                 nc = (ENVIRON["CK"] == "") ? 0 : split(ENVIRON["CK"], K, "\t")
                 split(ENVIRON["CV"], W, "\t")
                 ch = ""; cr = ""
                 for (j = 1; j <= nc; j++) {
                     m = K[j]; while (m in U) m = "_" m
                     U[m] = 1; ch = ch m "\t"; cr = cr W[j] "\t"
                 }
                 c = "capture"; while (c in U) c = "_" c
                 print "#\t" c "\t" ch $0; next }
            { print (s + n) "\t" ENVIRON["CAPID"] "\t" cr $0; n++ }
            END { print n + 0, nf + 0 > (ENVIRON["CNTF"]) }' "${ds_file[k]}" > "$tmp/id.tsv"
        read -r nrows nf < "$tmp/cnt"
        [[ $k -eq 0 ]] && fields="$nf"

        _store_append "${ds_name[k]}" "$tmp/id.tsv"
        final="$_SA_TARGET"
        out_names+=("$final")
        if (( nrows > 0 )); then out_ranges+=("#${startid}..$((startid+nrows-1))"); else out_ranges+=("none"); fi

        # One provenance line per row (#N → dataset, capture), in one awk.
        if (( nrows > 0 )); then
            PDS="$final" PID="$id" awk -v s="$startid" -v n="$nrows" 'BEGIN {
                ds = ENVIRON["PDS"]; id = ENVIRON["PID"]
                for (i = 0; i < n; i++) printf "%d\t%s\t%s\n", s + i, ds, id }' >> "$prov"
        fi
        nextid=$((startid + nrows)); totalrows=$((totalrows + nrows))
    done
    echo "$nextid" > "$(_rowid_file)"

    # meta.tsv keeps its single-dataset keys; with several datasets they carry a
    # comma-joined value and `datasets` records how many (plus each one's rows).
    local j all_names="" all_strats="" all_confs="" all_rows="" all_covs="" all_conss=""
    for (( j=0; j<${#out_names[@]}; j++ )); do
        all_names="${all_names:+$all_names,}${out_names[j]}"
        all_strats="${all_strats:+$all_strats,}${ds_strategy[j]}"
        all_confs="${all_confs:+$all_confs,}${ds_conf[j]}"
        all_rows="${all_rows:+$all_rows,}${out_names[j]}:${out_ranges[j]}"
        all_covs="${all_covs:+$all_covs,}${out_names[j]}:${ds_cov[j]:--}"
        all_conss="${all_conss:+$all_conss,}${out_names[j]}:${ds_cons[j]:--}"
    done
    {
        printf 'id\t%s\n' "$id";            printf 'ts\t%s\n' "$ts"
        printf 'source\t%s\n' "$source";    printf 'command\t%s\n' "$(_meta_safe "$cmd")"
        printf 'rc\t%s\n' "$rc";            printf 'strategy\t%s\n' "$(_meta_safe "$all_strats")"
        printf 'confidence\t%s\n' "$all_confs";  printf 'dataset\t%s\n' "$all_names"
        printf 'row_ids\t%s\n' "$( (( totalrows > 0 )) && echo "$firstid-$((nextid-1))" || echo none)"
        if [[ -n "$am_reason" ]]; then
            # Segmented by auto-multi: coverage is that of all the datasets
            # together; the single parse the gate held is kept for the record.
            printf 'coverage\t%s\n' "$am_cov"
            printf 'auto_multi\t%s\n' "$(_meta_safe "segmented ($am_kind): $am_reason")"
            (( am_model > 0 )) && printf 'auto_multi_model_parts\t%s\n' "$am_model"
            printf 'single_strategy\t%s\n' "$(_meta_safe "$single_strategy")"
            printf 'single_confidence\t%s\n' "$single_conf"
            [[ -n "$cov" ]] && printf 'single_coverage\t%s\n' "$cov"
            [[ -n "$tcons" ]] && printf 'single_consistency\t%s\n' "$tcons"
        else
            [[ -n "$cov" && -z "$reason" ]] && printf 'coverage\t%s\n' "$cov"
            [[ -n "$tcons" && -z "$reason" ]] && printf 'consistency\t%s\n' "$tcons"
        fi
        [[ -n "$am_note" ]] && printf 'auto_multi\t%s\n' "$(_meta_safe "declined: $am_note")"
        if [[ -n "$lm_reason" ]]; then
            printf 'layout_model\t%s\n' "$(_meta_safe "filed: the engine's parse was $lm_reason")"
            printf 'single_strategy\t%s\n' "$(_meta_safe "$single_strategy")"
            printf 'single_confidence\t%s\n' "$single_conf"
            [[ -n "$single_cov" ]] && printf 'single_coverage\t%s\n' "$single_cov"
            [[ -n "$single_cons" ]] && printf 'single_consistency\t%s\n' "$single_cons"
        fi
        [[ -n "$lm_note" ]] && printf 'layout_model\t%s\n' "$(_meta_safe "declined: $lm_note")"
        if [[ -n "$llm_how" ]]; then
            printf 'llm_template\t%s\n' "$(_meta_safe "filed by $llm_how")"
            if [[ -n "$llm_reason" ]]; then
                printf 'single_strategy\t%s\n' "$(_meta_safe "$single_strategy")"
                printf 'single_confidence\t%s\n' "$single_conf"
                [[ -n "$single_cov" ]] && printf 'single_coverage\t%s\n' "$single_cov"
            fi
        fi
        [[ -n "$llm_note" && -z "$llm_how" ]] && printf 'llm_template\t%s\n' "$(_meta_safe "declined: $llm_note")"
        if [[ ${#out_names[@]} -gt 1 ]]; then
            printf 'datasets\t%s\n' "${#out_names[@]}"
            printf 'dataset_rows\t%s\n' "$all_rows"
            if [[ ${#ds_cov[@]} -gt 0 ]]; then
                printf 'dataset_coverage\t%s\n' "$all_covs"
                printf 'dataset_consistency\t%s\n' "$all_conss"
            fi
        fi
        if [[ -n "${PROVOST_PLUGIN_DIR:-}" ]]; then
            printf 'plugin_dir\t%s\n' "$(_meta_safe "$PROVOST_PLUGIN_DIR")"
        fi
        if [[ -s "$ctxf" ]]; then printf 'context\tcontext.txt\n'; fi
        if [[ -n "$ctx_map" ]]; then printf 'context_columns\t%s\n' "$ctx_map"; fi
    } > "$capd/meta.tsv"
    startid="$firstid"; nrows="$totalrows"; final="$all_names"; strategy="$all_strats"; conf="$all_confs"

    if (( commit )); then
        # In a controlled directory, snapshot the directory's contents as commit
        # context so the capture is tied to what the tree looked like at the time.
        local ctl_dir=""
        if [[ "${PROVOST_STORE_KIND:-}" == controlled:* ]]; then
            ctl_dir="$(dirname "$PROVOST_WORKSPACE")"
            _snapshot_context "$ctl_dir" "$PROVOST_WORKSPACE" >/dev/null \
                || warn "could not snapshot the controlled directory's context."
        fi
        local msg="capture $id: $source $strategy → $final (#${startid}..$((startid+nrows-1)))"
        [[ -n "$cmd" ]] && msg="$msg [$cmd]"
        _commit "$msg"
        local out
        out="$(_git tag "capture-$id" 2>&1)" || { printf 'git: %s\n' "$out" >&2; die "capture $id was committed but could not be tagged capture-$id."; }
        if [[ -n "$ctl_dir" ]]; then
            _snapshot_files "$ctl_dir" "$id" \
                || warn "could not save the controlled directory's files for $id (the capture itself is filed)."
        fi
        _unlock
    fi
    _CAPID="$id"

    local idrange="#${startid}..$((startid+nrows-1))"; (( nrows == 0 )) && idrange="(no rows)"
    if [[ -n "$reason" ]]; then
        info "$id: $reason — queued in 'unsorted' as $idrange (committed)."
        [[ -n "$am_note" ]] && info "      segmenting it did not help: $am_note."
        [[ -n "$lm_note" ]] && info "      the layout model did not help: $lm_note."
        [[ -n "$llm_note" ]] && info "      no LLM template helped: $llm_note."
    elif [[ ${#out_names[@]} -gt 1 ]]; then
        if [[ -n "$am_reason" ]]; then
            ok "$id: → ${#out_names[@]} datasets, rows $idrange — segmented: the single parse was $am_reason; the ${#out_names[@]} parts keep $am_cov%:"
            (( am_model > 0 )) && info "      the layout model read $am_model of the parts the engine could not."
            [[ "$am_kind" == layout-model ]] && info "      the layout model segmented it (the engine's own segmenting did not help: $am_note)."
            [[ "$am_kind" == llm-template ]] && info "      read by $llm_how${llm_learned:+ (committed with this capture; later captures of the tool are parsed offline)}."
        else
            ok "$id: → ${#out_names[@]} datasets, rows $idrange:"
            [[ -n "$llm_reason" ]] && info "      the engine's parse ($single_strategy) was $llm_reason; read by $llm_how${llm_learned:+ (committed with this capture; later captures of the tool are parsed offline)}."
        fi
        for (( j=0; j<${#out_names[@]}; j++ )); do
            local dq=""
            [[ "${ds_cov[j]:-}" =~ ^[0-9]+$ ]] && dq+=", coverage ${ds_cov[j]}%"
            [[ "${ds_cons[j]:-}" =~ ^[0-9]+$ ]] && dq+=", consistency ${ds_cons[j]}%"
            info "      ${out_names[j]}  ${out_ranges[j]}  (strategy ${ds_strategy[j]}, conf ${ds_conf[j]}$dq)"
        done
    else
        ok "$id: → dataset '$final' rows $idrange (strategy $strategy, conf $conf${cov:+, coverage $cov%}) — $((fields-1)) field(s)."
        [[ -n "$lm_reason" ]] && info "      the engine's parse ($single_strategy) was $lm_reason; the layout model's table passes the gate."
        [[ -n "$llm_reason" ]] && info "      the engine's parse ($single_strategy) was $llm_reason; read by $llm_how${llm_learned:+ (committed with this capture; later captures of the tool are parsed offline)}."
        # Low line coverage: the parse kept only part of the input. Say so —
        # silently filing a fragment is the failure this guards against.
        if [[ -n "$cov" ]] && (( cov < 60 )); then
            info "      only $cov% of the input's lines reached '$final'; the whole input stays with capture $id ('provost source'), and --multi files each section as a dataset of its own."
        fi
    fi
    # (the messages above end in `[[ ]] &&` tests; the capture itself is filed)
    return 0
}

# Background ingest: spool the input and run _ingest in a detached child so the
# prompt returns immediately. Used only when interactive (stderr is a TTY) or
# forced with --async / PROVOST_ASYNC=1; scripts get the synchronous path.
# The spool lives OUTSIDE the workspace (never swept into a commit). On success
# the child removes it. On failure it is kept (input + full log, so the capture
# can be retried) and a one-line record is appended to the store's
# .async-failures.log — every later provost invocation warns about it until
# it is seen with 'provost log --failed' or 'provost doctor'.
_ingest_async () {
    local in="$1" threshold="$2" source="$3" cmd="${4:-}" rc="${5:-}" name="${6:-}"
    local spool
    spool="$(mktemp -d "$PROVOST_TEMP_DIR/provost-async.XXXXXX")" \
        || die "cannot create a spool dir in $PROVOST_TEMP_DIR (set TMPDIR)."
    cp "$in" "$spool/in.txt" || { rm -rf "$spool"; die "cannot spool the input to $spool"; }
    (
        trap '' HUP                                   # survive the shell exiting
        _TMPDIRS=(); _LOCKS_HELD=(); _LOCKS_OWNER=()  # nothing of the parent's
        trap _cleanup EXIT
        set +e
        ( set -e; trap _cleanup EXIT; _ingest "$spool/in.txt" "$threshold" "$source" "$cmd" "$rc" "$name" ) \
            >"$spool/log.txt" 2>&1
        irc=$?
        if [[ $irc -eq 0 ]]; then
            rm -rf "$spool"
        else
            last="$(awk 'NF { l = $0 } END { print l }' "$spool/log.txt" 2>/dev/null)"
            last="${last//$'\033'\[[0-9]m/}"; last="${last//$'\033'\[[0-9][0-9]m/}"
            printf '%s\t%s\t%s\t%s\t%s\n' "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$source" \
                "$(_meta_safe "${cmd:-(stdin)}")" "$(_meta_safe "$spool")" \
                "$(_meta_safe "rc=$irc: ${last:-no output}")" >> "$(_faillog)" 2>/dev/null
        fi
        exit 0
    ) </dev/null >/dev/null 2>&1 &
    disown 2>/dev/null || true
    info "queued (async, pid $!) — capturing in the background; see 'provost log' shortly."
}

# Decide foreground vs background: --async / --sync win, then PROVOST_ASYNC
# (1 / 0), else background only when interactive (stderr is a terminal).
_want_async () {
    case "$1" in
        1) return 0 ;; 0) return 1 ;;
    esac
    case "$PROVOST_ASYNC" in
        1) return 0 ;; 0) return 1 ;;
    esac
    _is_tty
}

# Unacknowledged background failures → one warning line on stderr.
_warn_failures () {
    local log seen=0 n
    log="$(_faillog)"
    [[ -s "$log" ]] || return 0
    n="$(wc -l < "$log")"; n="${n//[!0-9]/}"
    seen="$(cat "$(_failseen)" 2>/dev/null || echo 0)"
    [[ "$seen" =~ ^[0-9]{1,12}$ ]] || seen=0
    if (( n > seen )); then
        warn "$((n - seen)) background capture(s) failed — see 'provost log --failed'."
    fi
}
# Print the failure log and mark every entry seen.
_show_failures () {
    local log n; log="$(_faillog)"
    if [[ ! -s "$log" ]]; then info "no failed background captures."; return 0; fi
    printf '%-20s %-7s %-24s %s\n' "WHEN" "SOURCE" "COMMAND" "ERROR (input + full log kept in SPOOL)" >&2
    awk -F'\t' '{ printf "%-20s %-7s %-24s %s\n      spool: %s\n", $1, $2, $3, $5, $4 }' "$log"
    n="$(wc -l < "$log")"; n="${n//[!0-9]/}"
    printf '%s\n' "$n" > "$(_failseen)" 2>/dev/null || true
}

# ---------------------------------------------------------------------------
# Dataset naming for `run`: the command's basename, looking through common
# wrappers (sudo, doas, env, nice, ionice, nohup, time, command, exec, stdbuf,
# timeout) and their options. `xargs` is NOT a wrapper. Prints a valid dataset
# name, or nothing.
# ---------------------------------------------------------------------------
_run_name () {
    local -a a=("$@") sp=()
    local i=0 n=$# w
    while (( i < n )); do
        w="${a[i]##*/}"
        case "$w" in
            sudo)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -u|-g|-h|-p|-C|-D|-r|-t|-U|-T|--user|--group|--host|--prompt|--close-from|--chdir|--role|--type|--other-user|--command-timeout) i=$((i+2)) ;;
                    -*|*=*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            doas)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -u|-C) i=$((i+2)) ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            env)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -u|--unset|-C|--chdir) i=$((i+2)) ;;
                    -S|--split-string)
                        read -r -a sp <<< "${a[i+1]:-}" || true
                        a=(${sp[@]+"${sp[@]}"} "${a[@]:i+2}"); n=${#a[@]}; i=0; continue 2 ;;
                    --split-string=*|-S?*)
                        w="${a[i]#--split-string=}"; [[ "$w" == "${a[i]}" ]] && w="${a[i]#-S}"
                        read -r -a sp <<< "$w" || true
                        a=(${sp[@]+"${sp[@]}"} "${a[@]:i+1}"); n=${#a[@]}; i=0; continue 2 ;;
                    -*|*=*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            nice)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -n|--adjustment) i=$((i+2)) ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            ionice)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -c|-n|-p|-P|-u|--class|--classdata|--pid|--pgid|--uid) i=$((i+2)) ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            nohup)
                i=$((i+1)); [[ "${a[i]:-}" == "--" ]] && i=$((i+1)) ;;
            time)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -f|-o|--format|--output) i=$((i+2)) ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            command)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            exec)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -a) i=$((i+2)) ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            stdbuf)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -i|-o|-e) i=$((i+2)) ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done ;;
            timeout)
                i=$((i+1))
                while (( i < n )); do case "${a[i]}" in
                    --) i=$((i+1)); break ;;
                    -s|-k|--signal|--kill-after) i=$((i+2)) ;;
                    -*) i=$((i+1)) ;;
                    *) break ;;
                esac; done
                i=$((i+1)) ;;                          # the DURATION
            *)
                w="$(_sanitize_name "$w")"
                [[ "$w" == unsorted ]] && w="unsorted-cmd"   # never the triage bucket
                printf '%s' "$w"; return 0 ;;
        esac
    done
    return 0
}

# ---------------------------------------------------------------------------
# Per-command usage (help to STDOUT, exit 0; usage errors to stderr, exit 2)
# ---------------------------------------------------------------------------
_usage_text () {
    case "$1" in
        run) cat <<'EOF'
usage: provost run [options] [--] <command> [args...]
  Execute a command, capture its combined stdout+stderr as a commit,
  auto-detect the format and file it as a dataset named after the command
  (sudo/env/nice/timeout/... wrappers are looked through): 'provost run df -h'
  files into dataset 'df'. Exits with the command's own exit status.
options:
  --name NAME             file into dataset NAME instead
  --min-confidence N      commit threshold 0-101 (default: $PROVOST_MIN_CONFIDENCE)
  --min-coverage N        quality gate: hold back a generic parse keeping < N% of
                          the lines (default: $PROVOST_MIN_COVERAGE; 0 = off)
  --min-consistency N     quality gate: hold back a generic parse whose columns
                          are < N% type-consistent (default: $PROVOST_MIN_CONSISTENCY; 0 = off)
  --plugin-dir DIR        extra extract strategy directory
  --multi                 file every dataset the output holds (as NAME.<strategy>)
  --no-auto-multi         hold a parse the quality gate rejects for triage
                          without first trying to segment it (--auto-multi: try;
                          the default, PROVOST_AUTO_MULTI)
  --model                 before holding such a parse, try the learned layout
                          model (needs python3; --no-model: don't, the default,
                          PROVOST_MODEL)
  --llm                   parse with a template Claude wrote for this tool, or
                          have Claude write one, before trusting a generic parse
                          or holding the capture (sends the capture to the
                          Anthropic API; needs python3 and an API key; --no-llm:
                          don't, the default, PROVOST_LLM)
  --sync, --wait          capture in the foreground (default when not interactive)
  --async                 capture in the background (default when interactive)
  --show                  show the command's output as it runs, as well
  -q, --silent            no progress output
EOF
            ;;
        inject) cat <<'EOF'
usage: cmd | provost [options]      or      provost [options] FILE
  Inject stdin (or FILE) into a typed dataset named after the winning extract
  strategy (e.g. 'csv', 'free', 'aligned-columns'), or NAME with --name.
options:
  --name NAME             file into dataset NAME instead of the strategy's name
  --min-confidence N      commit threshold 0-101 (default: $PROVOST_MIN_CONFIDENCE)
  --min-coverage N        quality gate: hold back a generic parse keeping < N% of
                          the lines (default: $PROVOST_MIN_COVERAGE; 0 = off)
  --min-consistency N     quality gate: hold back a generic parse whose columns
                          are < N% type-consistent (default: $PROVOST_MIN_CONSISTENCY; 0 = off)
  --plugin-dir DIR        extra extract strategy directory
  --multi                 file every dataset the input holds
  --no-auto-multi         hold a parse the quality gate rejects for triage
                          without first trying to segment it (--auto-multi: try;
                          the default, PROVOST_AUTO_MULTI)
  --model                 before holding such a parse, try the learned layout
                          model (needs python3; --no-model: don't, the default,
                          PROVOST_MODEL)
  --llm                   parse with a template Claude wrote for this tool, or
                          have Claude write one, before trusting a generic parse
                          or holding the capture (sends the capture to the
                          Anthropic API; needs python3 and an API key; --no-llm:
                          don't, the default, PROVOST_LLM)
  --sync, --wait          capture in the foreground (default when not interactive)
  --async                 capture in the background (default when interactive)
  --show                  pass the input through to stdout as well (a tee)
  --context-columns       make each 'key=value' in a '#%' prologue a column
                          on every row (--no-context-columns: don't, the
                          default, PROVOST_CONTEXT_COLUMNS)
  -q, --silent            no progress output
EOF
            ;;
        batch) cat <<'EOF'
usage: provost batch [options] [--name NAME] FILE... [--from LIST]
  File many inputs at once: each is a capture of its own (its own id, its own
  'provost source'), parsed in parallel, and all are filed under the store
  lock in ONE commit, with a capture-<id> tag for each. For a wrapper that
  captures a fleet: one file per host, each opening with '#% host=NAME', and
  --context-columns to put the host on every row.
options:
  --name NAME             file the inputs after it into dataset NAME (it
                          applies until the next --name)
  --from LIST             read more inputs from LIST, one a line: PATH, or
                          NAME<TAB>PATH for an input of its own name; '-' is
                          stdin; blank and '#' lines are skipped
  -j, --jobs N            parse at most N inputs at a time (default: the
                          CPUs, at most 16; --llm parses one at a time)
  --context-columns       make each 'key=value' in a '#%' prologue a column
                          on every row (PROVOST_CONTEXT_COLUMNS)
  --min-confidence N, --min-coverage N, --min-consistency N, --plugin-dir DIR,
  --multi, --no-auto-multi, --model, --llm, -q
                          as for piped input ('provost inject --help')
  An empty input is skipped; one that cannot be read or parsed is reported and
  the rest are filed. Exits 1 if any input was not captured, 0 otherwise.
EOF
            ;;
        source) cat <<'EOF'
usage: provost source <#N | ID>
  The original captured input behind dataset row #N, or behind capture ID
  (c0007 or capture-c0007), with that capture's details.
EOF
            ;;
        control) echo "usage: provost control [<dir>]    make <dir> (default: cwd) its own capture instance" ;;
        uncontrol) echo "usage: provost uncontrol [<dir>]    release a controlled directory (its data is kept)" ;;
        instances) echo "usage: provost instances    list controlled directories" ;;
        export) echo "usage: provost export [<file>]    bundle the whole store into a portable git bundle" ;;
        import) echo "usage: provost import <file.bundle> [<dest-dir>]    reconstitute a bundle to browse" ;;
        log) cat <<'EOF'
usage: provost log [--failed]
  List captures (the git history). --failed lists background captures that
  failed (and marks them seen, silencing the startup warning).
EOF
            ;;
        restore) echo "usage: provost restore <capture-id|latest>    bring a capture's commit back up (detached); 'latest' returns" ;;
        files) cat <<'EOF'
usage: provost files [ID] [PATH] [--to DIR]
  The controlled directory's files as capture ID kept them (default: the
  latest capture that kept any). No PATH: list them (size, path); a PATH:
  print that file (or list that directory); --to DIR: write them all out
  into DIR, a new or empty directory.
EOF
            ;;
        listen) cat <<'EOF'
usage: provost listen [CMD...]
  Capture CMD every time you run it: its output is shown as always and a copy
  is filed (as 'provost run --show CMD' would). Each command becomes a shell
  function in one file, which your shell loads from its startup file; provost
  says how. With no CMD, list the commands listened on.
EOF
            ;;
        unlisten) echo "usage: provost unlisten CMD...    stop capturing CMD every time it runs" ;;
        search) cat <<'EOF'
usage: provost search PATTERN [NAME...] [options]
  Find the dataset rows that match PATTERN (a regular expression, matched
  anywhere in the row, '#' and capture included) with git grep, shown as a
  table per dataset. NAME... limits it to those datasets.
options:
  -i, -F, -w              ignore case / a fixed string / whole words
  --at ID                 search the store as it stood after capture ID
  --output                search the captured outputs instead (capture, line, text)
  --files                 search a controlled directory's files as the latest
                          run kept them (or capture ID's, with --at)
EOF
            ;;
        diff) cat <<'EOF'
usage: provost diff [NAME] [A [B]] [-- PATH...]
  What changed between capture A and capture B: for each dataset, the rows
  only A filed (-) and only B filed (+), compared without the columns
  provost adds; and, in a controlled directory, which files changed. With
  NAME, only that dataset, and the default captures are its own.
  No A: the last two captures; A alone: A against the capture before it.
  -- PATH...              the full diff of those files between A and B
EOF
            ;;
        next) echo "usage: provost next    step the restored store forward one capture (past the last: back to the latest)" ;;
        prev) echo "usage: provost prev    step the store back one capture (from the latest: the last capture)" ;;
        ls) echo "usage: provost ls    list datasets" ;;
        show) cat <<'EOF'
usage: provost show NAME [--capture ID] [--at ID]
  Present a dataset as a table: '#' is each row's index in the store,
  'capture' the capture it came from, 'n' its record number in that capture.
options:
  --capture ID            only the rows capture ID filed (c0007, or 7)
  --at ID                 the dataset as it stood just after capture ID, read
                          from git (the store stays where it is)
EOF
            ;;
        rm) echo "usage: provost rm [-f|--yes] [--no-purge] NAME...    delete dataset(s) + the captures only they used" ;;
        stats) echo "usage: provost stats NAME [analyze-args...]    statistics on a dataset (--all for everything)" ;;
        triage) echo "usage: provost triage    show the held-back bucket (unmatched, low-confidence, failed quality gate), raw input decoded" ;;
        doctor) echo "usage: provost doctor    check the engines and the store; lists failed background captures" ;;
        *) return 1 ;;
    esac
}
_cmd_help () {
    _usage_text "$1" | sed -e "s/\\\$PROVOST_MIN_CONFIDENCE/$PROVOST_MIN_CONFIDENCE/" \
        -e "s/\\\$PROVOST_MIN_COVERAGE/$PROVOST_MIN_COVERAGE/" -e "s/\\\$PROVOST_MIN_CONSISTENCY/$PROVOST_MIN_CONSISTENCY/"
    exit 0
}
_cmd_usage_die () { local u; u="$(_usage_text "$1")"; printf '%s\n' "${u%%$'\n'*}" >&2; exit 2; }

# Parse the options shared by inject and run. Sets: _O_THRESHOLD _O_ASYNC _O_NAME
# and leaves the remaining args in _O_REST.
# _capture_opt VERB ARGS... — apply the one capture option at the front of
# ARGS, and set _CO_N to how many words it took (0: it is not one). Shared by
# _parse_capture_opts (inject, run) and cmd_batch, so the three take the same
# options with the same checks.
_capture_opt () {
    local verb="$1"; shift
    _CO_N=1
    case "$1" in
        -q|--silent) PROVOST_QUIET=1 ;;
        --min-confidence)
            [[ $# -ge 2 ]] || usage_die "--min-confidence needs a number."
            _validate_uint --min-confidence "$2" 0 101; _O_THRESHOLD="$(_norm_uint "$2")"; _CO_N=2 ;;
        --min-confidence=*)
            _validate_uint --min-confidence "${1#*=}" 0 101; _O_THRESHOLD="$(_norm_uint "${1#*=}")" ;;
        --min-coverage|--min-consistency)
            [[ $# -ge 2 ]] || usage_die "$1 needs a number."
            _validate_uint "$1" "$2" 0 100
            if [[ "$1" == --min-coverage ]]; then PROVOST_MIN_COVERAGE="$(_norm_uint "$2")"
            else PROVOST_MIN_CONSISTENCY="$(_norm_uint "$2")"; fi
            _CO_N=2 ;;
        --min-coverage=*)
            _validate_uint --min-coverage "${1#*=}" 0 100; PROVOST_MIN_COVERAGE="$(_norm_uint "${1#*=}")" ;;
        --min-consistency=*)
            _validate_uint --min-consistency "${1#*=}" 0 100; PROVOST_MIN_CONSISTENCY="$(_norm_uint "${1#*=}")" ;;
        --plugin-dir)
            [[ $# -ge 2 ]] || usage_die "--plugin-dir needs a directory."
            PROVOST_PLUGIN_DIR="$2"; _CO_N=2 ;;
        --name)
            [[ $# -ge 2 ]] || usage_die "--name needs a dataset name."
            _O_NAME="$2"; _CO_N=2 ;;
        --name=*) _O_NAME="${1#*=}" ;;
        --multi) PROVOST_MULTI=1 ;;
        --auto-multi) PROVOST_AUTO_MULTI=1 ;;
        --no-auto-multi) PROVOST_AUTO_MULTI=0 ;;
        --model) PROVOST_MODEL=1 ;;
        --no-model) PROVOST_MODEL=0 ;;
        --llm) PROVOST_LLM=1 ;;
        --no-llm) PROVOST_LLM=0 ;;
        --context-columns) PROVOST_CONTEXT_COLUMNS=1 ;;
        --no-context-columns) PROVOST_CONTEXT_COLUMNS=0 ;;
        --async) _O_ASYNC=1 ;;
        --sync|--wait) _O_ASYNC=0 ;;
        --show|--tee) _O_SHOW=1 ;;
        -h|--help) _cmd_help "$verb" ;;
        *) _CO_N=0 ;;
    esac
    return 0
}

_parse_capture_opts () {
    local verb="$1"; shift
    _O_THRESHOLD="$PROVOST_MIN_CONFIDENCE"; _O_ASYNC=""; _O_NAME=""; _O_SHOW=0; _O_REST=()
    while [[ $# -gt 0 ]]; do
        case "$1" in
            --) shift; _O_REST=("$@"); return 0 ;;
            -?*)
                _capture_opt "$verb" "$@"
                (( _CO_N > 0 )) || usage_die "unknown option for '$verb': $1 (see 'provost $verb --help')."
                shift "$_CO_N" ;;
            *) _O_REST=("$@"); return 0 ;;
        esac
    done
    return 0
}
_check_capture_opts () {
    if [[ -n "$_O_NAME" ]]; then
        _require_name "$_O_NAME"
        [[ "$_O_NAME" == unsorted ]] && usage_die "'unsorted' is reserved for the triage bucket; pick another --name."
    fi
    if [[ -n "${PROVOST_PLUGIN_DIR:-}" && ! -d "$PROVOST_PLUGIN_DIR" ]]; then
        die "plugin strategy dir not found: $PROVOST_PLUGIN_DIR"
    fi
    return 0
}

# ---------------------------------------------------------------------------
# Default verb: inject (pipe or file)
# ---------------------------------------------------------------------------
cmd_inject () {
    _parse_capture_opts inject "$@"
    local file=""
    [[ ${#_O_REST[@]} -le 1 ]] || usage_die "inject takes at most one FILE (got: ${_O_REST[*]})."
    [[ ${#_O_REST[@]} -eq 1 ]] && file="${_O_REST[0]}"
    _check_capture_opts
    if [[ -z "$file" && -t 0 ]]; then
        usage_die "no input — pipe data into provost (cmd | provost) or give a FILE; see 'provost help'."
    fi
    if [[ -n "$file" ]]; then [[ -f "$file" ]] || die "no such file: $file"; fi

    _ensure_workspace
    _require_on_branch
    _mktempd provost; local tmp="$_MKTEMPD"
    local in="$tmp/in.txt" size
    if [[ -n "$file" ]]; then
        head -c "$((PROVOST_MAX_INPUT + 1))" < "$file" | tr -d '\r' > "$in"
        (( _O_SHOW )) && cat -- "$file"
    elif (( _O_SHOW )); then
        # --show: pass all of stdin through to stdout (a tee), keep the bound.
        tee "$tmp/full"
        head -c "$((PROVOST_MAX_INPUT + 1))" "$tmp/full" | tr -d '\r' > "$in"; rm -f "$tmp/full"
    else
        head -c "$((PROVOST_MAX_INPUT + 1))" | tr -d '\r' > "$in"
    fi
    [[ -s "$in" ]] || die "no input to provost (empty)."
    size="$(wc -c < "$in")"; size="${size//[!0-9]/}"
    (( size <= PROVOST_MAX_INPUT )) \
        || die "input too large (max $PROVOST_MAX_INPUT bytes; raise PROVOST_MAX_INPUT)."

    if _want_async "$_O_ASYNC"; then _ingest_async "$in" "$_O_THRESHOLD" "inject" "" "" "$_O_NAME"
    else _ingest "$in" "$_O_THRESHOLD" "inject" "" "" "$_O_NAME"; fi
}

# ---------------------------------------------------------------------------
# run — execute a command, capture its output as a commit, and ingest it.
# ---------------------------------------------------------------------------
cmd_run () {
    _parse_capture_opts run "$@"
    set -- ${_O_REST[@]+"${_O_REST[@]}"}
    [[ $# -gt 0 ]] || { _cmd_usage_die run; }
    _check_capture_opts
    command -v -- "$1" >/dev/null 2>&1 || { printf 'provost: command not found: %s\n' "$1" >&2; exit 127; }

    _ensure_workspace
    _require_on_branch
    _mktempd provost; local tmp="$_MKTEMPD"
    local in="$tmp/in.txt" rc=0 size
    # Capture combined stdout+stderr (the thing you'd otherwise eyeball),
    # bounded by PROVOST_MAX_INPUT exactly as inject is.
    local -a st=()
    set +e
    if (( _O_SHOW )); then
        # --show: the output goes to the terminal as it comes, as well.
        "$@" 2>&1 | tee "$in.full"
        st=("${PIPESTATUS[@]}")
        head -c "$((PROVOST_MAX_INPUT + 1))" "$in.full" > "$in.raw"; rm -f "$in.full"
    else
        "$@" 2>&1 | head -c "$((PROVOST_MAX_INPUT + 1))" > "$in.raw"
        st=("${PIPESTATUS[@]}")
    fi
    set -e
    rc="${st[0]}"
    size="$(wc -c < "$in.raw")"; size="${size//[!0-9]/}"
    if (( size > PROVOST_MAX_INPUT )); then
        die "command output too large (more than $PROVOST_MAX_INPUT bytes; raise PROVOST_MAX_INPUT) — nothing captured."
    fi
    tr -d '\r' < "$in.raw" > "$in"; rm -f "$in.raw"
    # The command as a shell-quoted, TSV-safe string (printf %q never emits a
    # raw tab or newline), so meta.tsv and 'provost log' show it faithfully.
    local cmd; cmd="$(_shell_quote "$@")"
    [[ -s "$in" ]] || { info "command produced no output (rc=$rc) — nothing captured."; return "$rc"; }
    local name="$_O_NAME"
    [[ -n "$name" ]] || name="$(_run_name "$@")"
    if _want_async "$_O_ASYNC"; then _ingest_async "$in" "$_O_THRESHOLD" "run" "$cmd" "$rc" "$name"
    else _ingest "$in" "$_O_THRESHOLD" "run" "$cmd" "$rc" "$name"; fi
    return "$rc"
}

# ---------------------------------------------------------------------------
# batch — many inputs, one commit. Each input is a capture of its own (its own
# id, its own `provost source`), the parse halves run in parallel, and the
# store halves run in input order under one lock, ending in ONE commit with a
# tag per capture. For wrappers that capture a fleet: one input per host,
# with `#% host=...` context and --context-columns.
# ---------------------------------------------------------------------------
_ncpu () {
    local n; n="$(getconf _NPROCESSORS_ONLN 2>/dev/null || nproc 2>/dev/null || echo 4)"
    [[ "$n" =~ ^[0-9]+$ ]] && (( n >= 1 )) || n=4
    (( n > 16 )) && n=16
    echo "$n"
}

cmd_batch () {
    _O_THRESHOLD="$PROVOST_MIN_CONFIDENCE"; _O_ASYNC=""; _O_NAME=""; _O_SHOW=0
    local jobs="" line lname lpath
    local -a files=() names=()
    while [[ $# -gt 0 ]]; do
        case "$1" in
            -j|--jobs)
                [[ $# -ge 2 ]] || usage_die "--jobs needs a number."
                _validate_uint --jobs "$2" 1 1024; jobs="$(_norm_uint "$2")"; shift 2 ;;
            --jobs=*)
                _validate_uint --jobs "${1#*=}" 1 1024; jobs="$(_norm_uint "${1#*=}")"; shift ;;
            --from)
                # A list of inputs, one a line: PATH, or NAME<TAB>PATH for an
                # input of its own name; '-' reads the list from stdin. For
                # more inputs than a command line holds.
                [[ $# -ge 2 ]] || usage_die "--from needs a file (or - for stdin)."
                [[ "$2" == - || -f "$2" ]] || die "no such list: $2"
                while IFS= read -r line || [[ -n "$line" ]]; do
                    line="${line%$'\r'}"
                    [[ -z "$line" || "$line" == '#'* ]] && continue
                    if [[ "$line" == *$'\t'* ]]; then lname="${line%%$'\t'*}"; lpath="${line#*$'\t'}"
                    else lname="$_O_NAME"; lpath="$line"; fi
                    files+=("$lpath"); names+=("$lname")
                done < <(if [[ "$2" == - ]]; then cat; else cat -- "$2"; fi)
                shift 2 ;;
            --) shift
                while [[ $# -gt 0 ]]; do files+=("$1"); names+=("$_O_NAME"); shift; done ;;
            -?*)
                _capture_opt batch "$@"
                (( _CO_N > 0 )) || usage_die "unknown option for 'batch': $1 (see 'provost batch --help')."
                shift "$_CO_N" ;;
            *) files+=("$1"); names+=("$_O_NAME"); shift ;;
        esac
    done
    [[ -z "$_O_ASYNC" && "$_O_SHOW" == 0 ]] \
        || usage_die "batch files in the foreground and shows nothing: --async, --sync and --show do not apply."
    (( ${#files[@]} > 0 )) || _cmd_usage_die batch
    local i missing=()
    for (( i = 0; i < ${#files[@]}; i++ )); do
        [[ -f "${files[i]}" ]] || missing+=("${files[i]}")
        if [[ -n "${names[i]}" ]]; then
            _require_name "${names[i]}"
            [[ "${names[i]}" == unsorted ]] && usage_die "'unsorted' is reserved for the triage bucket; pick another --name."
        fi
    done
    (( ${#missing[@]} == 0 )) || die "no such file: ${missing[*]} -- nothing was captured."
    if [[ -n "${PROVOST_PLUGIN_DIR:-}" && ! -d "$PROVOST_PLUGIN_DIR" ]]; then
        die "plugin strategy dir not found: $PROVOST_PLUGIN_DIR"
    fi
    [[ -n "$jobs" ]] || jobs="$(_ncpu)"
    # --llm may have Claude write a template from the first input that needs
    # one; parsed in parallel, every input would ask for the same template.
    if [[ "$PROVOST_LLM" == 1 && "$jobs" -gt 1 ]]; then
        info "batch: --llm parses one input at a time, so a template is written once, not once per input."
        jobs=1
    fi

    _ensure_workspace
    _require_on_branch
    _mktempd batch; local broot="$_MKTEMPD" d size failed=0 skipped=0
    local -a todo=()
    # Stage each input as inject does: bounded by PROVOST_MAX_INPUT, CRs
    # stripped. An empty one is said and skipped, as `run` skips a command
    # that printed nothing; an oversized one is an error for that input.
    for (( i = 0; i < ${#files[@]}; i++ )); do
        d="$broot/$i"; mkdir -p "$d"
        head -c "$((PROVOST_MAX_INPUT + 1))" < "${files[i]}" | tr -d '\r' > "$d/in.txt"
        if [[ ! -s "$d/in.txt" ]]; then
            info "${files[i]}: empty -- nothing captured."; skipped=$((skipped + 1)); continue
        fi
        size="$(wc -c < "$d/in.txt")"; size="${size//[!0-9]/}"
        if (( size > PROVOST_MAX_INPUT )); then
            warn "${files[i]}: too large (max $PROVOST_MAX_INPUT bytes; raise PROVOST_MAX_INPUT) -- not captured."
            failed=$((failed + 1)); continue
        fi
        todo+=("$i")
    done

    # The parse halves, at most $jobs at a time. Each child starts with no
    # temp dirs or locks of the parent's to clean up, as _ingest_async's does:
    # its EXIT trap must never remove the batch's own work dirs.
    local -a pids=() parsed=()
    local k head=0 rc
    for (( k = 0; k < ${#todo[@]}; k++ )); do
        i="${todo[k]}"; d="$broot/$i"
        (
            _TMPDIRS=(); _LOCKS_HELD=(); _LOCKS_OWNER=()
            trap _cleanup EXIT
            _ingest_parse "$d/in.txt" "$_O_THRESHOLD" "" "${names[i]}" "$d"
        ) > "$d/parse.log" 2>&1 &
        pids[k]=$!
        while (( k + 1 - head >= jobs )); do
            rc=0; wait "${pids[head]}" || rc=$?
            if (( rc == 0 )); then parsed+=("${todo[head]}")
            else
                warn "${files[${todo[head]}]}: not captured: $(awk 'NF { l = $0 } END { print l }' "$broot/${todo[head]}/parse.log")"
                failed=$((failed + 1))
            fi
            head=$((head + 1))
        done
    done
    while (( head < ${#todo[@]} )); do
        rc=0; wait "${pids[head]}" || rc=$?
        if (( rc == 0 )); then parsed+=("${todo[head]}")
        else
            warn "${files[${todo[head]}]}: not captured: $(awk 'NF { l = $0 } END { print l }' "$broot/${todo[head]}/parse.log")"
            failed=$((failed + 1))
        fi
        head=$((head + 1))
    done
    # (waited in launch order, so parsed[] is in input order: capture ids are
    # given out in the order the inputs were named, whatever finished first)

    if (( ${#parsed[@]} > 0 )); then
        _lock
        _require_on_branch
        local -a ids=()
        local rows0 rows1
        rows0="$(cat "$(_rowid_file)" 2>/dev/null || echo 1)"
        for i in "${parsed[@]}"; do
            _ingest_store "$broot/$i/in.txt" batch "" "" "$broot/$i" 0
            ids+=("$_CAPID")
            printf 'input\t%s\n' "$(_meta_safe "${files[i]}")" >> "$(_cap_dir)/$_CAPID/meta.tsv"
        done
        rows1="$(cat "$(_rowid_file)" 2>/dev/null || echo 1)"
        local first="${ids[0]}" last="${ids[${#ids[@]}-1]}" id
        for id in "${ids[@]}"; do
            printf 'batch\t%s..%s\n' "$first" "$last" >> "$(_cap_dir)/$id/meta.tsv"
        done
        local ctl_dir=""
        if [[ "${PROVOST_STORE_KIND:-}" == controlled:* ]]; then
            ctl_dir="$(dirname "$PROVOST_WORKSPACE")"
            _snapshot_context "$ctl_dir" "$PROVOST_WORKSPACE" >/dev/null \
                || warn "could not snapshot the controlled directory's context."
        fi
        local range="$first..$last"; (( ${#ids[@]} == 1 )) && range="$first"
        _commit "batch: ${#ids[@]} capture(s) $range ($((rows1 - rows0)) rows)"
        local out
        for id in "${ids[@]}"; do
            out="$(_git tag "capture-$id" 2>&1)" || { printf 'git: %s\n' "$out" >&2; die "batch $first..$last was committed but $id could not be tagged capture-$id."; }
            if [[ -n "$ctl_dir" ]]; then
                _snapshot_files "$ctl_dir" "$id" \
                    || warn "could not save the controlled directory's files for $id (the capture itself is filed)."
            fi
        done
        _unlock
        local what="captures $first..$last"; (( ${#ids[@]} == 1 )) && what="capture $first"
        ok "batch: ${#ids[@]} input(s) -> $what, one commit$( (( skipped )) && printf '; %d empty' "$skipped")$( (( failed )) && printf '; %d not captured' "$failed")."
    else
        info "batch: nothing captured ($skipped empty, $failed not captured)."
    fi
    rm -rf "$broot"
    (( failed == 0 ))
}

# ---------------------------------------------------------------------------
# log / restore — the git core: list captures, bring one back up.
# ---------------------------------------------------------------------------
cmd_log () {
    local failed=0
    while [[ $# -gt 0 ]]; do
        case "$1" in
            --failed) failed=1; shift ;;
            -h|--help) _cmd_help log ;;
            *) usage_die "unknown argument for 'log': $1 (see 'provost log --help')." ;;
        esac
    done
    if [[ $failed -eq 1 ]]; then _show_failures; return 0; fi
    local capdir cur=""; capdir="$(_cap_dir)"
    # Restored to an earlier capture: the store's files stop at that capture,
    # so list every capture from the latest's tree, read from git.
    if [[ -d "$PROVOST_WORKSPACE/.git" ]] && cur="$(_current_capture)" && [[ -n "$cur" ]]; then
        _mktempd log
        _git archive "$(_main_branch)" -- 'captures/*/meta.tsv' 2>/dev/null | tar -x -C "$_MKTEMPD" 2>/dev/null || true
        capdir="$_MKTEMPD/captures"
    fi
    local d b keys=() files=() k
    shopt -s nullglob
    for d in "$capdir"/c*/; do
        [[ -f "${d}meta.tsv" ]] || continue
        b="${d%/}"; b="${b##*/}"
        if [[ "$b" =~ ^c([0-9]{1,12})$ ]]; then keys+=("$((10#${BASH_REMATCH[1]}))"$'\t'"$b"); fi
    done
    shopt -u nullglob
    if [[ ${#keys[@]} -eq 0 ]]; then
        info "no captures yet — run 'provost run <cmd>' or pipe into 'provost'."; return 0
    fi
    # Numeric order (c10000 after c9999), each dir's real name kept.
    while IFS=$'\t' read -r k b; do
        files+=("$capdir/$b/meta.tsv")
    done < <(printf '%s\n' "${keys[@]}" | sort -n -k1,1)
    printf '%-6s %-20s %-7s %-14s %-12s %s\n' "ID" "WHEN" "SOURCE" "STRATEGY" "DATASET" "COMMAND" >&2
    # One awk over every meta.tsv (not six per capture).
    awk -F'\t' '
        function emit() {
            if (!have) return
            printf "%-6s %-20s %-7s %-14s %-12s %s\n", v["id"], v["ts"], v["source"], v["strategy"], v["dataset"], v["command"]
            split("", v); have = 0
        }
        FNR == 1 { emit() }
        { k = $1; val = substr($0, length(k) + 2); v[k] = val; have = 1 }
        END { emit() }' "${files[@]}"
    [[ -z "$cur" ]] || info "the store is restored to $cur — step with: provost next / prev; return with: provost restore latest"
}

cmd_restore () {
    [[ $# -ge 1 ]] || _cmd_usage_die restore
    case "$1" in -h|--help) _cmd_help restore ;; esac
    [[ $# -eq 1 ]] || _cmd_usage_die restore
    _ensure_workspace
    local target="$1" out main
    if [[ "$target" == "latest" ]]; then
        main="$(_main_branch)"
        _lock
        out="$(_git checkout -q "$main" 2>&1)" || { printf 'git: %s\n' "$out" >&2; die "could not return to latest ($main)."; }
        _unlock
        ok "restored workspace to latest ($main)."
        return 0
    fi
    target="$(_capid_of "$target")" || usage_die "invalid capture id '$1' (expected e.g. c0003, 3 or latest)."
    _git rev-parse -q --verify "refs/tags/capture-$target" >/dev/null 2>&1 \
        || die "no such capture: $target (see 'provost log')."
    # Bring that capture's committed tree back up (detached).
    _lock
    out="$(_git checkout -q "capture-$target" 2>&1)" || { printf 'git: %s\n' "$out" >&2; die "checkout failed for capture-$target."; }
    _unlock
    ok "brought up $target (detached; new captures are refused until you return). Step with: provost next / prev; return with: provost restore latest"
    local capd; capd="$(_cap_dir)/$target"
    if [[ -f "$capd/meta.tsv" ]]; then echo; sed 's/\t/: /' "$capd/meta.tsv"; fi
}

# Every capture id, oldest first (from the capture-<id> tags).
_capture_ids () {
    _git for-each-ref --sort=v:refname --format='%(refname:short)' 'refs/tags/capture-c*' 2>/dev/null \
        | sed -n 's/^capture-\(c[0-9]*\)$/\1/p'
}
# The capture the store is restored to; empty when it is on the latest.
_current_capture () {
    _git symbolic-ref -q HEAD >/dev/null 2>&1 && return 0
    _git describe --tags --exact-match --match 'capture-c*' HEAD 2>/dev/null | sed 's/^capture-//'
}

# next / prev — step the restored store one capture forward or back, as
# 'restore' does: from the latest, 'prev' brings up the last capture (or the
# one before it, when the latest is that capture's own commit); 'next' past
# the last capture returns to the latest.
cmd_step () {
    local verb="$1"; shift
    case "${1:-}" in -h|--help) _cmd_help "$verb" ;; esac
    [[ $# -eq 0 ]] || _cmd_usage_die "$verb"
    _ensure_workspace
    local -a ids=(); local x cur pos=-1 n i
    while IFS= read -r x; do [[ -n "$x" ]] && ids+=("$x"); done < <(_capture_ids)
    n=${#ids[@]}
    (( n > 0 )) || die "no captures yet — nothing to step through."
    cur="$(_current_capture)"
    if [[ -z "$cur" ]]; then
        [[ "$verb" == next ]] && { info "already at the latest (step back with: provost prev)."; return 0; }
        pos=$n
        [[ "$(_git rev-parse HEAD 2>/dev/null)" == "$(_git rev-parse "capture-${ids[n-1]}^{commit}" 2>/dev/null)" ]] && pos=$((n - 1))
    else
        for i in "${!ids[@]}"; do [[ "${ids[i]}" == "$cur" ]] && pos=$i; done
        (( pos >= 0 )) || die "the store is at a commit that is not a capture — return with: provost restore latest"
    fi
    if [[ "$verb" == next ]]; then
        if (( pos + 1 >= n )); then cmd_restore latest; else cmd_restore "${ids[pos+1]}"; fi
    else
        (( pos > 0 )) || { info "already at the first capture (${ids[0]})."; return 0; }
        cmd_restore "${ids[pos-1]}"
    fi
}

# ---------------------------------------------------------------------------
# files — a controlled directory's files as a capture kept them (files-<id>).
# ---------------------------------------------------------------------------
# The newest capture whose files were kept ('' when none).
_latest_files_id () {
    _git for-each-ref --count=1 --sort=-v:refname --format='%(refname:short)' 'refs/tags/files-c*' 2>/dev/null \
        | sed -n 's/^files-\(c[0-9]*\)$/\1/p'
}

cmd_files () {
    local id="" path="" to="" a v
    while [[ $# -gt 0 ]]; do
        a="$1"
        case "$a" in
            -h|--help) _cmd_help files ;;
            --to) [[ $# -ge 2 ]] || usage_die "--to needs a directory."; to="$2"; shift 2; continue ;;
            --) shift; [[ $# -le 1 ]] || _cmd_usage_die files; path="${1:-}"; shift $#; continue ;;
            -?*) usage_die "unknown option for 'files': $a (see 'provost files --help')." ;;
        esac
        if [[ -z "$id" && -z "$path" ]] && v="$(_capid_of "$a")"; then id="$v"
        elif [[ -z "$path" ]]; then path="${a#./}"
        else _cmd_usage_die files; fi
        shift
    done
    [[ -d "$PROVOST_WORKSPACE/.git" ]] || die "no store yet — capture something first."
    if [[ -z "$id" ]]; then
        id="$(_latest_files_id)"
        [[ -n "$id" ]] || die "no run files yet — they are kept for captures made in a controlled directory (see 'provost control')."
    fi
    local ref="files-$id"
    if ! _git rev-parse -q --verify "refs/tags/$ref" >/dev/null 2>&1; then
        _git rev-parse -q --verify "refs/tags/capture-$id" >/dev/null 2>&1 \
            || die "no such capture: $id (see 'provost log')."
        die "capture $id kept no files (it was not made in a controlled directory, or PROVOST_FILES_MAX_SIZE was 0)."
    fi
    if [[ -n "$to" ]]; then
        [[ -z "$path" ]] || usage_die "--to writes out every file; leave out the path."
        if [[ -e "$to" ]]; then
            [[ -d "$to" && -z "$(ls -A "$to" 2>/dev/null)" ]] || die "$to exists and is not an empty directory."
        else
            mkdir -p "$to" || die "cannot create $to"
        fi
        _git archive "$ref" | tar -x -C "$to" || die "could not write the files of $id to $to."
        ok "wrote the files of $id to $to."
        return 0
    fi
    local kind=tree
    if [[ -n "$path" ]]; then
        path="${path%/}"
        kind="$(_git cat-file -t "$ref:$path" 2>/dev/null)" || die "no file '$path' in the files of $id (see 'provost files $id')."
    fi
    if [[ "$kind" == blob ]]; then
        _git cat-file blob "$ref:$path"
    else
        info "files of $id${path:+ under $path} (size, path):"
        _git ls-tree -r -l --full-tree "$ref" ${path:+-- "$path"} \
            | awk -F'\t' '{ split($1, m, " "); printf "%10s  %s\n", m[4], $2 }'
    fi
}

# ---------------------------------------------------------------------------
# search — git grep over the store: dataset rows (default), the captured
# outputs (--output), or a controlled directory's run files (--files); as the
# store stands, or as it stood after a capture (--at).
# ---------------------------------------------------------------------------
cmd_search () {
    local pat="" what=rows at="" v
    local -a gopt=() names=()
    while [[ $# -gt 0 ]]; do
        case "$1" in
            -h|--help) _cmd_help search ;;
            -i|--ignore-case) gopt+=(-i); shift ;;
            -F|--fixed-strings) gopt+=(-F); shift ;;
            -w|--word-regexp) gopt+=(-w); shift ;;
            --output) what=output; shift ;;
            --files) what=files; shift ;;
            --at) [[ $# -ge 2 ]] || usage_die "--at needs a capture id (e.g. c0007)."
                  at="$(_capid_of "$2")" || usage_die "invalid capture id '$2' (expected e.g. c0007 or 7)."
                  shift 2 ;;
            -e) [[ $# -ge 2 && -z "$pat" ]] || _cmd_usage_die search; pat="$2"; shift 2 ;;
            --) shift; if [[ -z "$pat" && $# -gt 0 ]]; then pat="$1"; shift; fi; names+=("$@"); break ;;
            -?*) usage_die "unknown option for 'search': $1 (see 'provost search --help')." ;;
            *) if [[ -z "$pat" ]]; then pat="$1"; else _require_name "$1"; names+=("$1"); fi; shift ;;
        esac
    done
    [[ -n "$pat" ]] || _cmd_usage_die search
    [[ -d "$PROVOST_WORKSPACE/.git" ]] || die "no store yet — capture something first."
    local rev=HEAD
    if [[ "$what" == files ]]; then
        (( ${#names[@]} == 0 )) || usage_die "--files searches every file a run kept; leave out dataset names."
        if [[ -z "$at" ]]; then at="$(_latest_files_id)"
            [[ -n "$at" ]] || die "no run files yet — they are kept for captures made in a controlled directory (see 'provost control')."
        fi
        _git rev-parse -q --verify "refs/tags/files-$at" >/dev/null 2>&1 || die "capture $at kept no files (see 'provost files')."
        info "files of $at matching '$pat' (path:line:text):"
        _git grep -I -n "${gopt[@]}" -e "$pat" "files-$at" -- 2>/dev/null | sed "s/^files-$at://"
        return 0
    fi
    if [[ -n "$at" ]]; then
        _git rev-parse -q --verify "refs/tags/capture-$at" >/dev/null 2>&1 || die "no such capture: $at (see 'provost log')."
        rev="capture-$at"
    fi
    _mktempd search; local hits="$_MKTEMPD/hits"
    if [[ "$what" == output ]]; then
        _git grep -I -n "${gopt[@]}" -e "$pat" "$rev" -- 'captures/*/output.txt' > "$hits" 2>/dev/null || true
        [[ -s "$hits" ]] || { info "no captured output matches '$pat'."; return 0; }
        # REV:captures/<id>/output.txt:LINE:TEXT -> capture, line, text
        R="$rev:" awk 'BEGIN { r = ENVIRON["R"]; print "capture\tline\ttext" }
            { s = substr($0, length(r) + 1); i = index(s, ":"); p = substr(s, 1, i - 1); s = substr(s, i + 1)
              j = index(s, ":"); split(p, P, "/"); print P[2] "\t" substr(s, 1, j - 1) "\t" substr(s, j + 1) }' "$hits" | _table
        return 0
    fi
    local -a spec=()
    if (( ${#names[@]} )); then for v in "${names[@]}"; do spec+=("datasets/$v.tsv"); done
    else spec=(datasets ':(exclude)datasets/unsorted.tsv'); fi
    _git grep -I -n "${gopt[@]}" -e "$pat" "$rev" -- "${spec[@]}" > "$hits" 2>/dev/null || true
    # REV:datasets/NAME.tsv:LINE:ROW — line 1 is the header, not a row.
    R="$rev:" awk -v OFS='\t' 'BEGIN { r = ENVIRON["R"] }
        { s = substr($0, length(r) + 1); i = index(s, ":"); p = substr(s, 1, i - 1); s = substr(s, i + 1)
          j = index(s, ":"); if (substr(s, 1, j - 1) == 1) next
          sub(/^datasets\//, "", p); sub(/\.tsv$/, "", p); print p, substr(s, j + 1) }' "$hits" > "$hits.rows"
    [[ -s "$hits.rows" ]] || { info "no dataset rows match '$pat'${at:+ at $at}."; return 0; }
    local ds prev="" first=1
    while IFS= read -r ds; do
        (( first )) || echo; first=0
        printf '%s (%s row%s)\n' "$ds" "$(awk -F'\t' -v d="$ds" '$1 == d { n++ } END { print n }' "$hits.rows")" \
            "$(awk -F'\t' -v d="$ds" '$1 == d { n++ } END { print (n == 1 ? "" : "s") }' "$hits.rows")"
        { _git show "$rev:datasets/$ds.tsv" 2>/dev/null | head -n 1
          awk -F'\t' -v d="$ds" '$1 == d { print substr($0, length(d) + 2) }' "$hits.rows"; } | _table
    done < <(cut -f1 "$hits.rows" | awk '!seen[$0]++')
}

# ---------------------------------------------------------------------------
# diff — what changed between two captures: the rows each dataset got from
# them, and (in a controlled directory) the files.
# ---------------------------------------------------------------------------
cmd_diff () {
    local name="" a="" b="" v x
    local -a paths=()
    while [[ $# -gt 0 ]]; do
        case "$1" in
            -h|--help) _cmd_help diff ;;
            --) shift; paths=("$@"); break ;;
            -?*) usage_die "unknown option for 'diff': $1 (see 'provost diff --help')." ;;
        esac
        if v="$(_capid_of "$1")"; then
            if [[ -z "$a" ]]; then a="$v"; elif [[ -z "$b" ]]; then b="$v"; else _cmd_usage_die diff; fi
        elif [[ -z "$name" && -z "$a" ]]; then
            _require_name "$1"; name="$1"
        else _cmd_usage_die diff; fi
        shift
    done
    [[ -d "$PROVOST_WORKSPACE/.git" ]] || die "no store yet — capture something first."
    local main; main="$(_main_branch)"
    _mktempd diff; local tmp="$_MKTEMPD"
    # Default: the last two captures (that filed into NAME, when one is given);
    # one capture given: it against the one before it.
    local -a ids=()
    if [[ -n "$name" ]]; then
        _git show "$main:datasets/$name.tsv" > "$tmp/name.tsv" 2>/dev/null || die "no such dataset: $name (try 'provost ls')."
        _git show "$main:provenance.tsv" > "$tmp/name.prov" 2>/dev/null || : > "$tmp/name.prov"
        while IFS= read -r x; do [[ -n "$x" ]] && ids+=("$x"); done < <(awk -F'\t' -v P="$tmp/name.prov" '
            FILENAME == P { if (FNR > 1) pc[$1] = $3; next }
            FNR == 1 { col = ($1 == "#" && $2 ~ /^_*capture$/); next }
            { c = col ? $2 : pc[$1]; if (c != "" && !(c in seen)) { seen[c] = 1; print c } }' "$tmp/name.prov" "$tmp/name.tsv")
    else
        while IFS= read -r x; do [[ -n "$x" ]] && ids+=("$x"); done < <(_capture_ids)
    fi
    local n=${#ids[@]} i
    if [[ -z "$a" ]]; then
        (( n >= 2 )) || die "need two captures${name:+ of '$name'} to compare (found $n)."
        a="${ids[n-2]}"; b="${ids[n-1]}"
    elif [[ -z "$b" ]]; then
        b="$a"; a=""
        for i in "${!ids[@]}"; do [[ "${ids[i]}" == "$b" && $i -gt 0 ]] && a="${ids[i-1]}"; done
        [[ -n "$a" ]] || die "no capture${name:+ of '$name'} before $b to compare it with."
    fi
    for x in "$a" "$b"; do
        _git rev-parse -q --verify "refs/tags/capture-$x" >/dev/null 2>&1 || die "no such capture: $x (see 'provost log')."
    done
    # Read both captures' rows from the later capture's tree (it holds the
    # earlier one's rows too), so the store's own checkout does not matter.
    local late="$b"
    (( 10#${a#c} > 10#${b#c} )) && late="$a"
    mkdir -p "$tmp/t"
    _git archive "capture-$late" -- datasets provenance.tsv 2>/dev/null | tar -x -C "$tmp/t" 2>/dev/null || true
    local cmd_a cmd_b
    cmd_a="$(_git show "capture-$a:captures/$a/meta.tsv" 2>/dev/null | awk -F'\t' '$1 == "command" { print substr($0, 9) }')"
    cmd_b="$(_git show "capture-$b:captures/$b/meta.tsv" 2>/dev/null | awk -F'\t' '$1 == "command" { print substr($0, 9) }')"
    printf '%s%s → %s%s\n' "$a" "${cmd_a:+ ($cmd_a)}" "$b" "${cmd_b:+ ($cmd_b)}"
    local f ds any=0 prov="$tmp/t/provenance.tsv"
    [[ -f "$prov" ]] || prov=/dev/null
    for f in "$tmp"/t/datasets/*.tsv; do
        [[ -f "$f" ]] || continue
        ds="$(basename "$f" .tsv)"
        [[ -z "$name" || "$ds" == "$name" ]] || continue
        # Rows from A and from B, without the columns provost adds ('#',
        # capture, n): a row is the same when its data is. Removed rows in A's
        # order, added rows in B's, then a count of the rows both have.
        awk -F'\t' -v OFS='\t' -v P="$prov" -v A="$a" -v B="$b" -v DS="$ds" '
            FILENAME == P { if (FNR > 1) pc[$1] = $3; next }
            FNR == 1 {
                col = ($1 == "#" && $2 ~ /^_*capture$/); first = col ? 3 : 2
                if ($first == "n" || $first ~ /^_+n$/) first++
                hdr = ""; for (i = first; i <= NF; i++) hdr = hdr (i > first ? OFS : "") $i
                next
            }
            {
                c = col ? $2 : pc[$1]
                if (c != A && c != B) next
                r = ""; for (i = first; i <= NF; i++) r = r (i > first ? OFS : "") $i
                if (c == A) { ra[++na] = r; ca[r]++ } else { rb[++nb] = r; cb[r]++ }
            }
            END {
                if (!na && !nb) exit
                print ""
                if (!nb) { printf "%s: only in %s (%d row%s)\n", DS, A, na, na == 1 ? "" : "s"; exit }
                if (!na) { printf "%s: only in %s (%d row%s)\n", DS, B, nb, nb == 1 ? "" : "s"; exit }
                for (r in cb) left[r] = cb[r]
                out = ""; same = 0
                for (i = 1; i <= na; i++) { r = ra[i]; if (left[r] > 0) { left[r]--; same++ } else out = out "  - " r "\n" }
                for (r in ca) lefta[r] = ca[r]
                for (i = 1; i <= nb; i++) { r = rb[i]; if (lefta[r] > 0) lefta[r]-- ; else out = out "  + " r "\n" }
                if (out == "") { printf "%s: no change (%d row%s)\n", DS, na, na == 1 ? "" : "s"; exit }
                printf "%s:\n    %s\n%s", DS, hdr, out
                if (same) printf "  (%d row%s the same)\n", same, same == 1 ? "" : "s"
            }' "$prov" "$f" > "$tmp/out" || true
        if [[ -s "$tmp/out" ]]; then any=1; cat "$tmp/out"; fi
    done
    (( any )) || { echo; echo "no dataset rows from $a or $b${name:+ in '$name'}."; }
    # Files: only when both captures kept them.
    local fa=0 fb=0
    _git rev-parse -q --verify "refs/tags/files-$a" >/dev/null 2>&1 && fa=1
    _git rev-parse -q --verify "refs/tags/files-$b" >/dev/null 2>&1 && fb=1
    if (( fa && fb )); then
        echo
        if (( ${#paths[@]} )); then
            _git diff "files-$a" "files-$b" -- "${paths[@]}"
        else
            local st; st="$(_git diff --stat=100 "files-$a" "files-$b" 2>/dev/null)"
            if [[ -n "$st" ]]; then echo "files:"; printf '%s\n' "$st" | sed 's/^/  /'
                echo "  (the changes themselves: provost diff $a $b -- PATH)"
            else echo "files: no change"; fi
        fi
    elif (( ${#paths[@]} )); then
        die "$a and $b did not both keep the controlled directory's files."
    fi
}

# ---------------------------------------------------------------------------
# listen / unlisten — capture a command every time you run it. Each listened
# command is a shell function in one file the user loads from their shell's
# startup file; the function runs it through 'provost run --show', so its
# output is on screen as always and a copy is filed (in the background at a
# terminal). Commands go to the store the directory routes to, as 'run' does.
# ---------------------------------------------------------------------------
_listen_file () { printf '%s/provost/listen.sh\n' "${XDG_CONFIG_HOME:-$HOME/.config}"; }
_listen_names () { local f; f="$(_listen_file)"; [[ -f "$f" ]] && sed -n 's/^\([A-Za-z_][A-Za-z0-9_.-]*\) () { command .* # provost listen$/\1/p' "$f"; return 0; }
_listen_write () {   # NAME... — rewrite the file with exactly these commands
    local f tmpf n prov; f="$(_listen_file)"
    mkdir -p "$(dirname "$f")" || die "cannot create $(dirname "$f")"
    # How the functions call provost: by name when it is on PATH (so an
    # upgrade or move is picked up), else by this script's own path.
    prov="provost"
    command -v provost >/dev/null 2>&1 || prov="$(printf '%q' "$PROVOST_ROOT/provost")"
    tmpf="$f.tmp.$$"
    {
        printf '# provost listen: commands whose output provost files every time they run.\n'
        printf '# Load from your shell startup file:  . %s\n' "$(printf '%q' "$f")"
        printf "# Change it with 'provost listen CMD' / 'provost unlisten CMD'.\n"
        for n in "$@"; do
            printf '%s () { command %s run --show -- %s "$@"; } # provost listen\n' "$n" "$prov" "$n"
        done
    } > "$tmpf" && mv "$tmpf" "$f" || { rm -f "$tmpf"; die "cannot write $f"; }
}
_listen_hint () {
    local f q rc found=0; f="$(_listen_file)"; q="$(printf '%q' "$f")"
    for rc in "$HOME/.bashrc" "$HOME/.zshrc" "$HOME/.profile" "$HOME/.bash_profile"; do
        [[ -f "$rc" ]] && F="$f" awk 'index($0, ENVIRON["F"]) { f = 1 } END { exit !f }' "$rc" 2>/dev/null && found=1
    done
    if (( found )); then info "takes effect in new shells; in this one, run:  . $q"
    else info "load it in every shell by adding this line to ~/.bashrc (or ~/.zshrc):  . $q"
         info "and in this shell now:  . $q"; fi
}

cmd_listen () {
    case "${1:-}" in -h|--help) _cmd_help listen ;; esac
    local -a have=() n x
    while IFS= read -r x; do [[ -n "$x" ]] && have+=("$x"); done < <(_listen_names)
    if [[ $# -eq 0 ]]; then
        if (( ${#have[@]} == 0 )); then info "listening on nothing — 'provost listen CMD' to start."; return 0; fi
        printf '%s\n' "${have[@]}"
        return 0
    fi
    local -a add=()
    for n in "$@"; do
        [[ "$n" =~ ^[A-Za-z_][A-Za-z0-9_.-]{0,63}$ ]] || usage_die "'$n' is not a command name provost can listen on."
        [[ "$n" == provost ]] && usage_die "provost cannot listen on itself."
        case "$(type -t -- "$n" 2>/dev/null)" in
            builtin|keyword) usage_die "'$n' is a shell builtin; it has to run in your shell, so provost cannot run it for you." ;;
        esac
        command -v -- "$n" >/dev/null 2>&1 || warn "'$n' is not on PATH here; listening anyway."
        for x in ${have[@]+"${have[@]}"} ${add[@]+"${add[@]}"}; do [[ "$x" == "$n" ]] && continue 2; done
        add+=("$n")
    done
    if (( ${#add[@]} == 0 )); then info "already listening on: $*"; return 0; fi
    _listen_write ${have[@]+"${have[@]}"} "${add[@]}"
    ok "listening on: ${add[*]} — its output is shown as always and filed each time it runs."
    _listen_hint
}

cmd_unlisten () {
    case "${1:-}" in -h|--help) _cmd_help unlisten ;; esac
    [[ $# -ge 1 ]] || _cmd_usage_die unlisten
    local -a have=() keep=() gone=() x n
    while IFS= read -r x; do [[ -n "$x" ]] && have+=("$x"); done < <(_listen_names)
    for x in ${have[@]+"${have[@]}"}; do
        for n in "$@"; do [[ "$x" == "$n" ]] && { gone+=("$x"); continue 2; }; done
        keep+=("$x")
    done
    (( ${#gone[@]} )) || { info "not listening on: $*"; return 0; }
    _listen_write ${keep[@]+"${keep[@]}"}
    ok "stopped listening on: ${gone[*]}."
    info "in a shell that already loaded it, also run:  unset -f ${gone[*]}"
}

# ---------------------------------------------------------------------------
# control / uncontrol / instances — a directory as its own capture instance.
# ---------------------------------------------------------------------------
_registry () { printf '%s/instances.tsv\n' "$(_default_home)"; }

# Add DIR to the global registry once (dedup on the directory column),
# under the global store's lock.
_registry_add () {
    local dir="$1" reg home; reg="$(_registry)"; home="$(_default_home)"
    mkdir -p "$home"
    _lock "$home"
    [[ -f "$reg" ]] || printf 'directory\tcontrolled_at\n' > "$reg"
    if ! D="$dir" awk -F'\t' '$1 == ENVIRON["D"] { f = 1 } END { exit !f }' "$reg" 2>/dev/null; then
        printf '%s\t%s\n' "$dir" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$reg"
    fi
    _unlock "$home"
}
_registry_remove () {
    local dir="$1" reg home; reg="$(_registry)"; home="$(_default_home)"
    [[ -f "$reg" ]] || return 0
    _lock "$home"
    local tmpf="$reg.tmp.$$.$RANDOM"
    if D="$dir" awk -F'\t' '$1 != ENVIRON["D"]' "$reg" > "$tmpf"; then
        mv "$tmpf" "$reg"
    else
        rm -f "$tmpf"
    fi
    _unlock "$home"
}

cmd_control () {
    case "${1:-}" in -h|--help) _cmd_help control ;; esac
    [[ $# -le 1 ]] || _cmd_usage_die control
    local dir="${1:-$PWD}"
    dir="$(cd "$dir" 2>/dev/null && pwd)" || die "no such directory: ${1:-$PWD}"
    local ws="$dir/.provost"
    if [[ -f "$ws/CONTROLLED" ]]; then _registry_add "$dir"; info "already controlled: $dir"; return 0; fi
    # Point this invocation's store at the new instance and initialise it.
    PROVOST_WORKSPACE="$ws"; PROVOST_STORE_KIND="controlled: $dir"; PROVOST_WORKSPACE_EXPLICIT=1
    _ensure_workspace
    _require_on_branch
    _lock
    printf '%s\n' "$dir" > "$ws/CONTROLLED"
    local n; n="$(_snapshot_context "$dir" "$ws")"
    _commit "provost: control $dir (context: $n file(s))"
    _unlock
    # Register in the global store so `provost instances` (run anywhere) sees it.
    _registry_add "$dir"
    ok "controlling '$dir' — captures run here now route to $ws (context: $n file(s))."
}

cmd_uncontrol () {
    case "${1:-}" in -h|--help) _cmd_help uncontrol ;; esac
    [[ $# -le 1 ]] || _cmd_usage_die uncontrol
    local dir="${1:-$PWD}"
    dir="$(cd "$dir" 2>/dev/null && pwd)" || die "no such directory: ${1:-$PWD}"
    [[ -f "$dir/.provost/CONTROLLED" ]] || die "not a controlled directory: $dir"
    local saved="$PROVOST_WORKSPACE"
    PROVOST_WORKSPACE="$dir/.provost"
    _lock
    rm -f "$dir/.provost/CONTROLLED"
    if _git symbolic-ref -q HEAD >/dev/null 2>&1; then
        _commit "provost: uncontrol $dir"
    fi
    _unlock
    PROVOST_WORKSPACE="$saved"
    _registry_remove "$dir"
    ok "released '$dir' (its .provost data is kept; captures here revert to the global store)."
}

cmd_instances () {
    case "${1:-}" in -h|--help) _cmd_help instances ;; esac
    [[ $# -eq 0 ]] || _cmd_usage_die instances
    local reg; reg="$(_registry)"
    [[ -f "$reg" ]] || { info "no controlled directories yet — 'provost control <dir>'."; return 0; }
    local n; n=$(( $(wc -l < "$reg") - 1 )); (( n < 0 )) && n=0
    (( n > 0 )) || { info "no controlled directories yet — 'provost control <dir>'."; return 0; }
    printf '%-40s %-20s %s\n' "DIRECTORY" "CONTROLLED_AT" "ACTIVE" >&2
    local d t
    while IFS=$'\t' read -r d t; do
        [[ "$d" == "directory" ]] && continue
        printf '%-40s %-20s %s\n' "$d" "$t" "$([[ -f "$d/.provost/CONTROLLED" ]] && echo yes || echo released)"
    done < "$reg"
}

# ---------------------------------------------------------------------------
# export / import — a portable bundle of the whole store (datasets + commits).
# ---------------------------------------------------------------------------
cmd_export () {
    case "${1:-}" in -h|--help) _cmd_help export ;; esac
    [[ $# -le 1 ]] || _cmd_usage_die export
    _ensure_workspace
    local out="${1:-}" gout
    if [[ -z "$out" ]]; then out="$PWD/provost-export-$(date -u +%Y%m%dT%H%M%SZ).bundle"; fi
    _git rev-parse HEAD >/dev/null 2>&1 || die "nothing to export — store has no commits yet."
    _lock
    gout="$(_git bundle create "$(_abspath "$out")" --all 2>&1)" || { printf 'git: %s\n' "$gout" >&2; die "export failed."; }
    _unlock
    [[ -f "$out" ]] || die "export failed."
    local sz; sz=$(stat -c '%s' "$out" 2>/dev/null || echo '?')
    ok "exported store → $out ($sz bytes). Import elsewhere with: provost import $(printf '%q' "$out") <dest>"
}

cmd_import () {
    case "${1:-}" in -h|--help) _cmd_help import ;; esac
    [[ $# -ge 1 && $# -le 2 ]] || _cmd_usage_die import
    local file="$1" out
    [[ -f "$file" ]] || die "no such export file: $file"
    local dest="${2:-}"
    [[ -z "$dest" ]] && dest="$PWD/provost-import-$(basename "$file" .bundle)"
    dest="$(_abspath "$dest")"
    file="$(_abspath "$file")"
    # 'git bundle verify' needs a repository to check the bundle's
    # prerequisites against, and refuses to run outside one -- which made an
    # import from any ordinary directory fail as "not a valid bundle". An
    # export is --all and has no prerequisites, so verify it in an empty
    # scratch repository, not whatever one the caller happens to stand in.
    _mktempd import
    git init -q "$_MKTEMPD" >/dev/null 2>&1 || die "cannot create a scratch repository in $PROVOST_TEMP_DIR (set TMPDIR)."
    git -C "$_MKTEMPD" bundle verify "$file" >/dev/null 2>&1 || die "not a valid provost export bundle: $file"
    # mkdir is the atomic claim on the destination (two concurrent imports to
    # the same place cannot both win); the clone then fills the empty dir.
    [[ -e "$dest" ]] && die "destination already exists: $dest"
    mkdir "$dest" 2>/dev/null || die "destination already exists (or cannot be created): $dest"
    if ! out="$(git clone -q "$file" "$dest" 2>&1)"; then
        rm -rf "$dest"; printf 'git: %s\n' "$out" >&2; die "import (clone) failed."
    fi
    git -C "$dest" checkout -q main >/dev/null 2>&1 || git -C "$dest" checkout -q master >/dev/null 2>&1 || true
    local nds ncap
    nds=$(find "$dest/datasets" -name '*.tsv' 2>/dev/null | wc -l | tr -d ' ')
    ncap=$(find "$dest/captures" -mindepth 1 -maxdepth 1 -type d 2>/dev/null | wc -l | tr -d ' ')
    ok "imported → $dest ($nds dataset(s), $ncap capture(s))."
    info "browse it with: PROVOST_WORKSPACE=$(printf '%q' "$dest") provost ls   (or show/log/source/restore)"
}

# ---------------------------------------------------------------------------
# source — given a dataset row's #N index, show the original captured input.
# ---------------------------------------------------------------------------
cmd_source () {
    [[ $# -ge 1 ]] || _cmd_usage_die source
    case "$1" in -h|--help) _cmd_help source ;; esac
    [[ $# -eq 1 ]] || _cmd_usage_die source
    # A capture id (c0007, capture-c0007) shows that whole capture.
    if [[ "$1" =~ ^(capture-)?c[0-9]{1,12}$ ]]; then _source_capture "$(_capid_of "$1")"; return; fi
    local n="${1#\#}"                             # tolerate a leading '#'
    [[ "$n" =~ ^[0-9]{1,15}$ ]] || usage_die "row index must be a number (got '$1')."
    local prov; prov="$(_prov_file)"
    [[ -f "$prov" ]] || die "no provenance yet — capture something first."
    local ds="" cap=""
    IFS=$'\t' read -r ds cap < <(awk -F'\t' -v n="$n" 'NR > 1 && $1 == n { print $2 "\t" $3; exit }' "$prov") || true
    [[ -n "$cap" ]] || die "no row #$n (see 'provost ls' / 'provost show <dataset>')."
    _valid_capid "$cap" || die "provenance for row #$n names an invalid capture id — the store may be corrupt."
    local capd; capd="$(_cap_dir)/$cap"

    info "row #$n — dataset '$ds', from capture $cap"
    if [[ -f "$capd/meta.tsv" ]]; then
        echo; sed 's/\t/: /' "$capd/meta.tsv"
    fi
    if [[ -f "$capd/output.txt" ]]; then
        echo; printf '%s----- original source (capture %s) -----%s\n' "$C_DIM" "$cap" "$C_RST"
        cat "$capd/output.txt"
    else
        die "capture $cap has no stored output (workspace may be detached — try 'provost restore latest')."
    fi
}

# _source_capture ID — a capture's meta and original output: from the store's
# files when they are there, else from the capture's own commit (a capture
# after the one the store is restored to, or one whose files 'rm' purged).
_source_capture () {
    local id="$1" capd meta out
    capd="$(_cap_dir)/$id"
    if [[ -f "$capd/output.txt" ]]; then
        meta="$(cat "$capd/meta.tsv" 2>/dev/null || true)"; out="$capd/output.txt"
    else
        _git rev-parse -q --verify "refs/tags/capture-$id" >/dev/null 2>&1 \
            || die "no such capture: $id (see 'provost log')."
        meta="$(_git show "capture-$id:captures/$id/meta.tsv" 2>/dev/null || true)"
        _mktempd source; out="$_MKTEMPD/output.txt"
        _git show "capture-$id:captures/$id/output.txt" > "$out" 2>/dev/null \
            || die "capture $id has no stored output."
    fi
    info "capture $id"
    if [[ -n "$meta" ]]; then echo; printf '%s\n' "$meta" | sed 's/\t/: /'; fi
    echo; printf '%s----- original source (capture %s) -----%s\n' "$C_DIM" "$id" "$C_RST"
    cat "$out"
}

# ---------------------------------------------------------------------------
# stats — statistics on a dataset (the bundled analyze engine).
# ---------------------------------------------------------------------------
cmd_stats () {
    [[ $# -ge 1 ]] || _cmd_usage_die stats
    case "$1" in -h|--help) _cmd_help stats ;; esac
    local name="$1"; shift
    _require_name "$name"
    local f; f="$(_ds_dir)/$name.tsv"
    [[ -f "$f" ]] || die "no such dataset: $name (try 'provost ls')."
    local analyze; analyze="$(_find_analyze)" \
        || die "analyze engine not found (expected at $PROVOST_ROOT/analyze/provost-analyze)."
    # Datasets are TSV; suppress the gawk-preference banner. Extra args pass
    # straight through (e.g. --all, a numbered analysis list, --format json).
    PROVOST_ANALYZE_SUPPRESS_AWK_WARNING=1 _run_engine "$analyze" "$f" "$@"
}

# ---------------------------------------------------------------------------
# ls / show / triage / doctor
# ---------------------------------------------------------------------------
cmd_ls () {
    case "${1:-}" in -h|--help) _cmd_help ls ;; esac
    [[ $# -eq 0 ]] || _cmd_usage_die ls
    local dir; dir="$(_ds_dir)"
    [[ -d "$dir" ]] || { info "no datasets yet — pipe something into 'provost'."; return 0; }
    shopt -s nullglob
    local f name rows any=0
    printf '%-24s %8s\n' "DATASET" "ROWS" >&2
    for f in "$dir"/*.tsv; do
        any=1; name="$(basename "$f" .tsv)"
        rows=$(( $(wc -l < "$f") - 1 )); (( rows < 0 )) && rows=0
        printf '%-24s %8d\n' "$name" "$rows"
    done
    shopt -u nullglob
    [[ $any -eq 1 ]] || info "no datasets yet — pipe something into 'provost'."
}

# The 'unsorted' dataset holds each held-back capture as a single base64-encoded
# 'raw' column (see _ingest). Decode it back to text for display — in ONE awk
# (mawk-safe; LC_ALL=C so %c emits single bytes), not a base64 process per row —
# escaping embedded tabs/newlines so each captured input stays on one line.
# A value that is not valid base64 is shown as-is.
# 'provost source #N' remains the way back to the exact original bytes.
_decode_unsorted_tsv () {
    LC_ALL=C awk -F'\t' '
        BEGIN {
            b = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
            for (i = 1; i <= 64; i++) val[substr(b, i, 1)] = i - 1
            for (i = 1; i < 256; i++) ch[i] = sprintf("%c", i)
            ch[9] = "\\t"; ch[10] = "\\n"; ch[13] = "\\r"
        }
        # The raw column by name (a store made before format 3 has no
        # capture column); the capture id, when there is one, is shown too.
        NR == 1 { r = 2; for (i = 2; i <= NF; i++) { if ($i == "raw") r = i }
                  cc = (r > 2) ? 2 : 0
                  print "#\t" (cc ? $cc "\t" : "") "raw"; next }
        {
            s = $r; pre = $1 "\t" (cc ? $cc "\t" : "")
            if (s ~ /[^A-Za-z0-9+\/=]/) { print pre s; next }
            printf "%s", pre
            out = ""; acc = 0; bits = 0; n = length(s); cnt = 0
            for (i = 1; i <= n; i++) {
                c = substr(s, i, 1)
                if (!(c in val)) continue
                acc = acc * 64 + val[c]; bits += 6
                if (bits >= 8) {
                    bits -= 8; p = 2 ^ bits
                    byte = int(acc / p); acc -= byte * p
                    if (byte > 0) { out = out ch[byte]; cnt++ }
                    if (cnt >= 4096) { printf "%s", out; out = ""; cnt = 0 }
                }
            }
            printf "%s\n", out
        }' "$1"
}

_table () { if command -v column >/dev/null 2>&1; then column -t -s $'\t'; else cat; fi; }

cmd_show () {
    [[ $# -ge 1 ]] || _cmd_usage_die show
    local name="" at="" cap="" v
    while [[ $# -gt 0 ]]; do
        case "$1" in
            -h|--help) _cmd_help show ;;
            --at|--capture)
                [[ $# -ge 2 ]] || usage_die "$1 needs a capture id (e.g. c0007)."
                v="$(_capid_of "$2")" || usage_die "invalid capture id '$2' (expected e.g. c0007 or 7)."
                if [[ "$1" == --at ]]; then at="$v"; else cap="$v"; fi
                shift 2 ;;
            -?*) usage_die "unknown option for 'show': $1 (see 'provost show --help')." ;;
            *) [[ -z "$name" ]] || _cmd_usage_die show; name="$1"; shift ;;
        esac
    done
    [[ -n "$name" ]] || _cmd_usage_die show
    _require_name "$name"
    local f prov
    if [[ -n "$at" ]]; then
        # The dataset as it stood just after capture AT, read from git: no
        # checkout, so the store stays where it is.
        _git rev-parse -q --verify "refs/tags/capture-$at" >/dev/null 2>&1 \
            || die "no such capture: $at (see 'provost log')."
        _mktempd show; f="$_MKTEMPD/ds.tsv"; prov="$_MKTEMPD/prov.tsv"
        _git show "capture-$at:datasets/$name.tsv" > "$f" 2>/dev/null \
            || die "no dataset '$name' at capture $at."
        _git show "capture-$at:provenance.tsv" > "$prov" 2>/dev/null || : > "$prov"
    else
        f="$(_ds_dir)/$name.tsv"; prov="$(_prov_file)"
        [[ -f "$f" ]] || die "no such dataset: $name (try 'provost ls')."
    fi
    if [[ -n "$cap" ]]; then
        # Only the rows capture CAP filed: by the capture column, or, in a
        # store from before it (format 2), by provenance.
        _mktempd show; local only="$_MKTEMPD/only.tsv"
        [[ -f "$prov" ]] || prov=/dev/null
        awk -F'\t' -v P="$prov" -v C="$cap" '
            FILENAME == P { if (FNR > 1) pc[$1] = $3; next }
            FNR == 1 { col = ($1 == "#" && $2 ~ /^_*capture$/); print; next }
            (col ? $2 : pc[$1]) == C' "$prov" "$f" > "$only"
        f="$only"
        [[ "$(wc -l < "$f")" -gt 1 ]] || { info "no rows in '$name' from capture $cap."; return 0; }
    fi
    if [[ "$name" == "unsorted" ]]; then _decode_unsorted_tsv "$f" | _table
    else _table < "$f"; fi
}

cmd_triage () {
    case "${1:-}" in -h|--help) _cmd_help triage ;; esac
    [[ $# -eq 0 ]] || _cmd_usage_die triage
    local f; f="$(_ds_dir)/unsorted.tsv"
    [[ -f "$f" ]] || { info "triage queue empty — nothing unmatched."; return 0; }
    info "unsorted rows (no-match / low-confidence / failed quality gate), raw input decoded ('provost source #N' for exact bytes):"
    _decode_unsorted_tsv "$f" | _table
}

# ---------------------------------------------------------------------------
# rm — delete one or more datasets, provenance-aware, tracked as a commit.
#   provost rm [-f|--yes] [--no-purge] <dataset> [<dataset>...]
# Removes each dataset file (only the named dataset — variants like NAME.v2 are
# their own datasets, remove them by name), prunes the matching rows from
# provenance.tsv, and — BY DEFAULT — purges each captures/<id>/ dir that fed
# ONLY removed dataset(s) (a --multi capture that also fed a surviving dataset
# is kept, and its meta.tsv updated to drop the removed names). --no-purge /
# --keep-captures keeps every capture. The deletion is committed, so the git
# history (and 'provost restore') still holds the pre-deletion state. Row
# indices (#N) and capture ids are never reused.
# ---------------------------------------------------------------------------
cmd_rm () {
    local yes=0 purge=1 names=()
    while [[ $# -gt 0 ]]; do
        case "$1" in
            -f|--yes|--force)        yes=1; shift ;;
            --purge)                 purge=1; shift ;;
            --no-purge|--keep-captures) purge=0; shift ;;
            -h|--help)               _cmd_help rm ;;
            --)                      shift; while [[ $# -gt 0 ]]; do names+=("$1"); shift; done ;;
            -?*)                     usage_die "unknown option for 'rm': $1 (see 'provost rm --help')." ;;
            *)                       names+=("$1"); shift ;;
        esac
    done
    [[ ${#names[@]} -gt 0 ]] || _cmd_usage_die rm
    local n
    for n in "${names[@]}"; do _require_name "$n"; done
    _ensure_workspace
    _require_on_branch

    local dir; dir="$(_ds_dir)"
    # Validate every name up front — refuse the whole operation if any is missing.
    local missing=()
    for n in "${names[@]}"; do
        [[ -f "$dir/$n.tsv" ]] || missing+=("$n")
    done
    [[ ${#missing[@]} -eq 0 ]] || die "no such dataset: ${missing[*]} (try 'provost ls')."

    # Confirm (unless -f), reporting how many rows go with the datasets.
    local total=0 rows
    for n in "${names[@]}"; do
        rows=$(( $(wc -l < "$dir/$n.tsv") - 1 )); (( rows < 0 )) && rows=0
        total=$((total + rows))
    done
    if [[ $yes -ne 1 ]]; then
        printf 'About to delete %d dataset(s), %d row(s): %s\n' "${#names[@]}" "$total" "${names[*]}" >&2
        if [[ $purge -eq 1 ]]; then
            printf 'Also purging the captures that fed only them — originals removed (--no-purge to keep them).\n' >&2
        else
            printf 'Keeping the source captures (--no-purge).\n' >&2
        fi
        printf '%s' 'Proceed? [y/N] ' >&2
        local ans=""; read -r ans || ans=""
        case "$ans" in y|Y|yes|YES|Yes) ;; *) info "aborted — nothing deleted."; return 0 ;; esac
    fi

    _lock
    _require_on_branch

    # The removed names, newline-joined, drive the awk membership tests.
    local rmset; rmset="$(printf '%s\n' "${names[@]}")"

    # Drop the dataset files.
    for n in "${names[@]}"; do rm -f "$dir/$n.tsv"; done

    # Prune provenance rows whose dataset was removed (keep the header).
    local prov ptmp; prov="$(_prov_file)"
    if [[ -f "$prov" ]]; then
        ptmp="$(mktemp "$dir/.rmprov.XXXXXX")"
        if RMSET="$rmset" awk -F'\t' '
                BEGIN { n = split(ENVIRON["RMSET"], r, "\n"); for (i = 1; i <= n; i++) del[r[i]] = 1 }
                FNR == 1 { print; next }
                !($2 in del) { print }' "$prov" > "$ptmp"; then
            mv "$ptmp" "$prov"
        else
            rm -f "$ptmp"; die "could not prune provenance.tsv — nothing committed."
        fi
    fi

    # --purge: a capture goes only when NONE of the datasets it fed survive
    # (a --multi capture's meta 'dataset' is a comma-joined list). A capture
    # that still feeds a surviving dataset stays, with the removed names
    # dropped from its meta.tsv.
    local purged=0 updated=0 d cid cds part survivors any_removed
    local -a parts=()
    if [[ $purge -eq 1 ]]; then
        shopt -s nullglob
        for d in "$(_cap_dir)"/c*/; do
            [[ -f "$d/meta.tsv" ]] || continue
            cds="$(awk -F'\t' '$1 == "dataset" { print substr($0, 9); exit }' "$d/meta.tsv")"
            IFS=',' read -r -a parts <<< "$cds" || true
            survivors=0; any_removed=0
            for part in ${parts[@]+"${parts[@]}"}; do
                if grep -qxF -- "$part" <<< "$rmset"; then any_removed=1
                elif _valid_name "$part" && [[ -f "$dir/$part.tsv" ]]; then survivors=$((survivors + 1)); fi
            done
            [[ $any_removed -eq 1 ]] || continue
            cid="$(awk -F'\t' '$1 == "id" { print $2; exit }' "$d/meta.tsv")"
            if [[ $survivors -eq 0 ]]; then
                rm -rf "$d"
                if _valid_capid "$cid"; then _git tag -d "capture-$cid" >/dev/null 2>&1 || true; fi
                purged=$((purged + 1))
            else
                RMSET="$rmset" awk -F'\t' -v OFS='\t' '
                    BEGIN { n = split(ENVIRON["RMSET"], r, "\n"); for (i = 1; i <= n; i++) del[r[i]] = 1 }
                    { key[NR] = $1; val[NR] = substr($0, length($1) + 2); if ($1 == "dataset") dl = val[NR] }
                    END {
                        nd = split(dl, dn, ","); removed = ""; kept = 0
                        for (i = 1; i <= nd; i++) if (dn[i] in del) { drop[i] = 1; removed = removed (removed == "" ? "" : ",") dn[i] } else kept++
                        for (ln = 1; ln <= NR; ln++) {
                            k = key[ln]; v = val[ln]
                            if (k == "dataset" || k == "strategy" || k == "confidence" || k == "coverage" || k == "consistency") {
                                m = split(v, p, ","); v = ""
                                for (i = 1; i <= m; i++) if (!(i in drop)) v = v (v == "" ? "" : ",") p[i]
                            } else if (k == "dataset_rows") {
                                m = split(v, p, ","); v = ""
                                for (i = 1; i <= m; i++) { nm = p[i]; sub(/:.*/, "", nm); if (!(nm in del)) v = v (v == "" ? "" : ",") p[i] }
                            } else if (k == "datasets") v = kept
                            print k, v
                        }
                        print "removed_datasets", removed
                    }' "$d/meta.tsv" > "$d/.meta.tmp" \
                    || { rm -f "$d/.meta.tmp"; die "could not update $d/meta.tsv — nothing committed."; }
                mv "$d/.meta.tmp" "$d/meta.tsv"
                updated=$((updated + 1))
            fi
        done
        shopt -u nullglob
    fi

    local msg="rm: deleted ${#names[@]} dataset(s) [${names[*]}], pruned $total row(s)"
    [[ $purge -eq 1 ]] && msg="$msg, purged $purged capture(s)"
    (( updated > 0 )) && msg="$msg, updated $updated shared capture(s)"
    _commit "$msg"
    _unlock

    local extra=""
    (( updated > 0 )) && extra=" $updated capture(s) kept because they also feed a surviving dataset."
    if [[ $purge -eq 1 ]]; then
        ok "deleted ${names[*]} (${total} row(s)); purged $purged capture(s).$extra History kept — 'provost log'/'restore' still reach the pre-deletion state."
    else
        ok "deleted ${names[*]} (${total} row(s)); source captures kept (--no-purge). 'provost source #N' still resolves surviving rows."
    fi
}

cmd_doctor () {
    case "${1:-}" in -h|--help) _cmd_help doctor ;; esac
    local engine rc=0
    if engine="$(_find_engine)"; then
        printf '%s ok%s extract engine: %s\n' "$C_OK" "$C_RST" "$engine"
    else
        printf '%serror%s extract engine not found (expected at %s/extract/)\n' "$C_ERR" "$C_RST" "$PROVOST_ROOT"; rc=1
    fi
    local analyze awkv=""
    if analyze="$(_find_analyze)"; then
        printf '%s ok%s analyze engine: %s\n' "$C_OK" "$C_RST" "$analyze"
        if ! command -v gawk >/dev/null 2>&1; then
            awkv="$(awk --version 2>/dev/null || awk -W version 2>/dev/null || echo awk)"
            printf '   note: gawk not found; analyze uses %s (some analyses may degrade)\n' "${awkv%%$'\n'*}"
        fi
    else
        printf '   analyze engine not found (expected at %s/analyze/)\n' "$PROVOST_ROOT"
    fi
    # Engine/transform copies inside a store are never used (code comes only
    # from the install dir) — say so, so nobody expects an override to work.
    local sub
    for sub in extract analyze transform; do
        if [[ -e "$PROVOST_WORKSPACE/$sub" || -e "$(_default_home)/$sub" ]]; then
            printf '   %snote%s a %s/ tree inside the store is IGNORED — code loads only from %s\n' "$C_ERR" "$C_RST" "$sub" "$PROVOST_ROOT"
        fi
    done
    printf '   store: %s (%s)' "$PROVOST_WORKSPACE" "${PROVOST_STORE_KIND:-global}"
    [[ -d "$PROVOST_WORKSPACE/.git" ]] && printf ' [git-backed]'; printf '\n'
    if [[ -d "$PROVOST_WORKSPACE/.git" ]]; then
        local head
        if head="$(_git symbolic-ref -q --short HEAD 2>/dev/null)"; then
            printf '   branch: %s\n' "$head"
        else
            printf '   %swarning%s store is on a restored capture (detached) — captures are refused; provost restore latest\n' "$C_ERR" "$C_RST"
        fi
    fi
    if [[ -f "$(_rowid_file)" ]]; then
        printf '   next row index: #%s\n' "$(cat "$(_rowid_file)")"
    fi
    if [[ -f "$(_capid_file)" ]]; then
        local nc; nc="$(cat "$(_capid_file)" 2>/dev/null || true)"
        if [[ "$nc" =~ ^[0-9]{1,12}$ ]]; then printf '   next capture id: c%04d\n' "$((10#$nc))"
        else printf '   %swarning%s capture counter is corrupt: %s\n' "$C_ERR" "$C_RST" "$nc"; fi
    fi
    if [[ -d "$PROVOST_WORKSPACE/.lock" ]]; then
        printf '   note: store lock held by: %s\n' "$(tr '\t' ' ' < "$PROVOST_WORKSPACE/.lock/owner" 2>/dev/null || echo unknown)"
    fi
    printf '   min-confidence: %s\n' "$PROVOST_MIN_CONFIDENCE"
    if [[ -s "$(_faillog)" ]]; then
        printf '   %swarning%s failed background captures:\n' "$C_ERR" "$C_RST"
        _show_failures 2>/dev/null | sed 's/^/     /'
    else
        printf '   failed background captures: none\n'
    fi
    return $rc
}

cmd_help () {
    cat <<EOF
provost $PROVOST_VERSION — turn command output into data

Give provost a command's output: it works out the format and files it as a
table you can list, show, search and compare. Every capture is a git commit,
so every run stays viewable.

CAPTURE
    provost run CMD [ARGS...]     run CMD and file its output (dataset CMD)
    CMD | provost                 file piped output
    provost FILE                  file output saved in FILE
    provost batch FILE...         file many outputs at once, one commit
    provost listen CMD            file CMD's output every time you run it
    provost control               give this directory its own store; each run
                                  here also keeps the directory's files

LOOK
    provost ls                    list datasets
    provost show NAME             show a dataset as a table
    provost search PATTERN        find matching rows in every dataset
    provost log                   list captures (c0001, c0002, ...)
    provost source ID             a capture's original output
    provost diff [NAME]           what changed between the last two captures
    provost files [ID]            a controlled directory's files at a capture
    provost stats NAME            statistics on a dataset
    provost triage                output provost could not file, and why

GO BACK
    provost prev / provost next   step the store back / forward one capture
    provost restore latest        return to the present

EXAMPLE
    provost run df -h             capture 'df -h' into dataset 'df'
    provost show df               see it: '#' numbers the row in the store,
                                  'capture' names the run it came from, 'n'
                                  its place in that run

OPTIONS (for run, piped or file input)
    --name NAME                   file into dataset NAME
    --show                        show the output as well as filing it
    --sync                        wait until it is filed
    --llm                         have Claude write a parser for output
                                  provost reads badly (sends that output to
                                  the Anthropic API; needs an API key)

MORE
    provost COMMAND --help        one command's usage (provost show --help)
    provost help advanced         every command, option and setting, and how
                                  provost decides what to file
EOF
}

cmd_help_advanced () {
    cat <<EOF
provost $PROVOST_VERSION — advanced help

COMMANDS
  Capture
    provost run [OPTIONS] [--] CMD [ARGS...]
        Run CMD and file its stdout+stderr as dataset CMD (sudo, env, timeout
        and other wrappers are looked through). Exits with CMD's status.
    CMD | provost [OPTIONS]        provost [OPTIONS] FILE
        File piped input, or a file, as a dataset named after the parser that
        read it (csv, free, aligned-columns, ...).
    provost batch [OPTIONS] FILE... [--from LIST]
        File many inputs as one commit, a capture each, parsed in parallel
        (BATCH CAPTURE below).
    provost listen [CMD...]        provost unlisten CMD...
        Capture CMD every time it runs (a shell function per command, in
        \${XDG_CONFIG_HOME:-~/.config}/provost/listen.sh; provost says how to
        load it). No CMD: list them.
  Look
    provost ls                     list datasets and their row counts
    provost show NAME [--capture ID] [--at ID]
        A dataset as a table; --capture: only the rows capture ID filed;
        --at: the dataset as it stood just after capture ID.
    provost search PATTERN [NAME...] [-i -F -w] [--at ID] [--output | --files]
        git grep over the rows (or the captured outputs, or a controlled
        directory's run files), shown per dataset under its header.
    provost source N | ID          the input behind row #N, or a capture's
    provost diff [NAME] [A [B]] [-- PATH...]
        Rows only A filed (-) and only B filed (+), per dataset, and the run
        files that changed (with PATH: their full diff). Default: the last
        two captures, or NAME's.
    provost stats NAME [ARGS]      statistics (ARGS go to the analyze engine;
                                   --all for every analysis)
    provost triage                 the 'unsorted' bucket: what was held, why
  History
    provost log [--failed]         list captures (--failed: failed background)
    provost restore ID | latest    check out capture ID (read-only) / return to
                                   the latest
    provost prev                   step the store back one capture
    provost next                   step it forward one (past the last: the latest)
  Directories and stores
    provost control [DIR]          give DIR (default: here) its own store
    provost uncontrol [DIR]        release it (its data is kept)
    provost instances              list controlled directories
    provost files [ID] [PATH] [--to DIR]
        A controlled directory's files as capture ID kept them: list them,
        print PATH, or write them all out into DIR.
    provost export [FILE]          the whole store as one portable git bundle
    provost import FILE [DEST]     open an exported store elsewhere to browse
  Upkeep
    provost rm [-f] [--no-purge] NAME...
        Delete datasets, and the captures only they used (--no-purge: keep
        those). The deletion is a commit, so 'restore' still reaches it.
    provost doctor                 check the engines and the store
    provost help [advanced]        the short help / this text
    provost COMMAND --help         one command's usage

  An ID is a capture id: c0007, 7 or capture-c0007.

CAPTURE OPTIONS (run, piped input, FILE)
    --name NAME             file into dataset NAME
    --show, --tee           show the output (or pass the input through) as well
    --sync, --wait          file in the foreground (the default when not at a
                            terminal)
    --async                 file in the background (the default at a terminal)
    -q, --silent            no progress output
    --multi                 file every dataset the input holds (MULTIPLE
                            DATASETS below)
    --llm, --no-llm         try a template Claude wrote for the tool first
                            (LLM TEMPLATES below; default off)
    --model, --no-model     try the learned layout model before holding a
                            capture (LAYOUT MODEL below; default off)
    --auto-multi, --no-auto-multi
                            try segmenting before holding a capture
                            (AUTO-MULTI below; default on)
    --min-confidence N      file only at confidence N or more (0-101; 101
                            holds every capture for triage)
    --min-coverage N        quality gate: line coverage (0-100; 0 = off)
    --min-consistency N     quality gate: type consistency (0-100; 0 = off)
    --plugin-dir DIR        an extra directory of extract strategies
    --context-columns, --no-context-columns
                            make a '#%' prologue's key=value words columns
                            (METADATA PROLOGUE below; default off)

THE STORE
    Captures go to a GLOBAL git store, so you can capture from any directory:
    \$PROVOST_HOME, or ~/.provost. Point PROVOST_HOME at a shared,
    group-writable directory to collaborate; a lock serialises writers. Only
    data lives in the store: provost's engines always load from its own
    install directory ($PROVOST_ROOT).

    'provost control DIR' gives DIR its own store (DIR/.provost): captures
    made in DIR or below it go there. Which store a capture uses:
    PROVOST_WORKSPACE if set, else the nearest controlled directory above the
    current one, else the global store.

    Every capture is one commit, tagged capture-<id>. Capture ids and row
    numbers are never reused. Every dataset row carries:
      #        its number, unique in the store ('provost source N')
      capture  the capture it came from (c0007; 'provost source c0007')
      n        its record number within that capture
    In a controlled directory, each capture also lists the directory's files
    (context/manifest.tsv) and commits the files themselves, tagged
    files-<id>: the directory's .gitignore is honoured, and a file over
    PROVOST_FILES_MAX_SIZE is only listed. Each run's files commit has the
    previous run's as its parent, so git shows the directory's history
    across runs (git log --stat files-c0007, in DIR/.provost).

    'restore ID', 'prev' and 'next' check out an earlier capture: ls, show
    and source then show the store as it was, and new captures are refused
    until 'provost restore latest'. 'show --at', 'source ID', 'search --at',
    'files' and 'diff' read any capture straight from git without moving the
    store. 'export' bundles everything, tags included; 'import' opens it.

DATASET NAMES
    'provost run CMD ...' files into a dataset named after CMD's basename
    (sudo, doas, env, nice, ionice, nohup, time, command, exec, stdbuf and
    timeout are skipped, with their options). Piped input is named after the
    parser that read it (csv, free, aligned-columns, ...). --name sets it.
    Names: letters, digits, '.', '_' and '-', starting with a letter or
    digit, no '..'. A capture of the same name with different columns
    RECONCILES into the dataset (columns unioned) when at least
    PROVOST_RECONCILE_OVERLAP% of the columns are shared, else it goes to
    NAME.v2, NAME.v3, ...

FOREGROUND AND BACKGROUND
    At a terminal a capture is filed in the BACKGROUND and the prompt returns
    at once; from a script or a pipeline (stderr not a terminal) it is filed
    in the FOREGROUND, so 'cmd | provost; provost show x' is race-free.
    --async / --sync (or PROVOST_ASYNC=1 / 0) force either. A failed
    background capture is logged in the store, and every later command warns
    until you look at it with 'provost log --failed' or 'provost doctor'.

HOW A CAPTURE IS FILED
    1. A leading run of '#%' lines is set aside as context (METADATA PROLOGUE).
    2. Every extract strategy scores the input; exact formats and
       tool-specific parsers outrank generic ones. The engine may swap its
       top generic parse for one that keeps more lines (LINE COVERAGE) or
       types its columns better (COLUMN-TYPE CONSISTENCY).
    3. No match, or below --min-confidence: held in 'unsorted', with the
       reason.
    4. A generic parse is checked by the QUALITY GATE. With --llm, a template
       for the tool is tried first (LLM TEMPLATES). One the gate would hold
       is then tried segmented (AUTO-MULTI), then by the layout model
       (--model); if nothing passes, it is held in 'unsorted', with the
       reason.
    The whole input is kept with every capture, filed or held
    ('provost source').

LINE COVERAGE
    A capture reports its coverage: the share of the input's non-blank lines
    that reached the dataset (shown after the confidence, recorded as
    'coverage' in meta.tsv; not measured above 128 KB). Confidence says how
    sure the engine is of the FORMAT; coverage says how much of the input the
    parse KEPT. When the parse ranked first keeps under 60% of the lines (one
    section of a multi-section report, say) the engine tries a lossless
    key/value parse of the whole report (one row per key, list item and
    section) and files that instead if it keeps materially more. A capture
    still filed under 60% says so.

COLUMN-TYPE CONSISTENCY
    A parse can keep every line and still put values under the wrong headers:
    a date split over three columns, a multi-word name spread over Name / Id /
    Version, a key: value file read as a two-column table. The engine types
    every cell (number, size, date/time, address, version, path, word, ...)
    and records the share of cells that match their column's type as
    'consistency' in meta.tsv (not measured above 128 KB; a word cut in half
    counts as a mismatch). When the generic parse ranked first is below 95%
    and another generic parse within 20 confidence points keeps as many
    lines, types its columns better (95% or more) and mixes fewer columns,
    the capture is filed with that parse instead.

QUALITY GATE
    A GENERIC parse that keeps under --min-coverage % of the input's lines
    (default 80) or whose columns are under --min-consistency % type-
    consistent (default 70) is held in 'unsorted', with the reason, instead
    of filed as a table: a likely-wrong table is worse than none. On the
    rated benchmark sets this held 18 of 48 wrong parses and no fully right
    one. Tool-specific parsers and strict grammars (json, yaml, xml, toml)
    are exempt; a measurement missing on a very large input never holds a
    capture. Set either to 0 to turn that check off.

AUTO-MULTI (segmenting a weak parse)
    When the gate would hold a parse, provost first asks the engine to
    segment the input, as --multi does, and files the parts instead when the
    split is clearly better:
      - it really is a split (2 or more datasets);
      - every part passes the gate on its own (confidence, coverage,
        consistency);
      - together the parts keep at least --min-coverage % of the lines, and
        10 points more than the single parse;
      - no part is a table under positional A/B/C... columns only;
      - a part read by the parser the gate rejected has judged column types;
      - parts of 3+ lines hold at least half the lines (no shattering into
        fragments).
    Otherwise the capture is held as before, and why segmenting did not help
    is printed and recorded ('auto_multi declined: ...' in meta.tsv). A
    segmented capture says so in its ok line and records 'auto_multi' (why
    the single parse was held), 'single_strategy' / 'single_coverage' /
    'single_consistency', the parts' joint 'coverage', and each part's
    'dataset_coverage' / 'dataset_consistency'. Names follow --multi, except
    that with a NAME the part whose columns match the single parse's keeps
    dataset NAME, so repeated captures of one command keep their rows in one
    dataset whether a given capture segments or not. Only captures the gate
    would hold pay for the second engine run.

LAYOUT MODEL (--model, PROVOST_MODEL=1)
    A capture the gate would still hold after AUTO-MULTI gets one more try:
    the learned layout model (extract/model/, the explicit-only
    'layout-model' strategy). It labels each line's role and where its fields
    start, decides each table's columns once for all its rows, and cuts every
    value from the input. Its table is filed (strategy 'layout-model') only
    when it passes the gate, keeps at least as many lines as the parse the
    gate held, and the input is one table or one key/value record. Where the
    engine also read the lines, an arbiter keeps the engine's field split on
    the lines it trusts more. Before that, AUTO-MULTI asks the model to
    re-read any part of its split the gate rejects; and when the
    whole-capture table is declined, the model cuts the report into its own
    uniform sections (a key/value block, a table, a list), filed under
    AUTO-MULTI's rules. meta.tsv records 'layout_model' (filed or declined,
    and why). Needs python3 (no packages); declines inputs over 600 lines.

LLM TEMPLATES (--llm, PROVOST_LLM=1)
    Claude writes a parser for a tool once; provost reuses it offline.
    A template is either regular expressions per kind of line (table rows,
    key/value lines, decoration to skip; format 1), or, for records that span
    lines, a small Python parser (format 2) run in a box: a separate
    interpreter, a few pure modules, no files or network, CPU and memory
    limits.

    Where templates come from: the store's llm-templates/ (NAME.json for
    --name or the run command, else any template whose detect pattern
    recognises the input), then PROVOST_LLM_TEMPLATE_DIRS, then the
    templates shipped with provost. When none fits and PROVOST_LLM_API_KEY
    or ANTHROPIC_API_KEY is set, Claude (PROVOST_LLM_MODEL) writes one from
    this capture: THAT SENDS THE CAPTURE'S TEXT TO THE ANTHROPIC API (at most
    400 lines / 64 KB), once per tool. A new template is kept, and committed
    with the capture, only when it passes; from then on every capture of the
    tool is parsed offline, deterministically and at no cost.

    When it is used: FIRST, on a capture the gate would hold (before
    segmenting and the layout model), and on a GENERIC parse the gate passed,
    before that parse is trusted (the gate cannot see a title or prompt read
    as the header; a template written for the tool's layout can). Never in
    place of an exact-format or tool-specific parser. Every block it yields
    is measured and filed under the gate's rules: one block as the capture's
    dataset, several as NAME.<block>, AUTO-MULTI's way. Against a generic
    parse the gate passed, it must keep that parse's share of the data lines
    (within 10 points; 20 against text-lines or line-records; lines it marks
    as decoration not counted).

    NOTES: lines a template sets aside (decoration, or lines it cannot read)
    that still carry a word or a number are filed beside its datasets as
    NAME.notes (line, kind, text): a title, a version banner, a summary count
    or an error between the rows is kept, never lost; only rules and box
    glyphs are dropped. A tab or newline inside a value becomes a space.

    REPAIR: a template keeps the captures it was written from (its
    "examples"). With a key, when a stored template meets a kind of line it
    has not met before (one it leaves unread, or one a skip pattern drops),
    provost sends the template, those captures and this one to Claude once,
    to extend it. The extension is kept, with this capture added to its
    examples, only if it still reads every line of the earlier captures the
    template read and passes the same rules; otherwise the template is used
    as it was. A capture whose lines the template has all met is never sent.

    meta.tsv records 'llm_template' (filed by which template, or declined and
    why). PROVOST_LLM_LOG names a file that gets one JSON line per API reply.
    Needs python3 (no packages).

MULTIPLE DATASETS (--multi)
    Some reports hold more than one kind of data: a key/value block, a table,
    and a banner best left alone. The default keeps whichever part the engine
    ranks first; --multi splits the input, parses each part on its own terms
    (a kv block and a table can even use different delimiters), and files
    each as its own dataset under ONE capture. Datasets are named by their
    strategy (prefixed NAME. for 'run' or --name, e.g. iostat.aligned-columns),
    so two parts that parse the same way but differ in shape land in X and
    X.v2. Row numbers run on across them, and 'provost source N' still
    returns the whole input. meta.tsv records 'datasets' (how many) and
    'dataset_rows' (each one's range).

METADATA PROLOGUE
    A leading run of '#%' lines on piped input is capture context, not data:
    it is set aside before format detection (so it cannot corrupt a parse),
    stored with the capture (context.txt), and shown by 'provost source'.
    Wrappers can prepend context unconditionally.

    With --context-columns (PROVOST_CONTEXT_COLUMNS=1) every key=value word
    in it also becomes a column on each of the capture's rows, after '#' and
    'capture': '#% host=web03 rack=r2' puts host and rack on every row, so
    one host's answers stay labelled in a dataset many hosts feed. Keys are
    letters, digits, '_', '.' and '-', starting with a letter or '_'; the
    last value of a repeated key wins, and other words stay context only. A
    data column of the same name keeps it, and the context column becomes
    _NAME; meta.tsv records which columns came from the context
    ('context_columns', key=column,...). A capture held in 'unsorted' is
    kept as it came.

BATCH CAPTURE (provost batch)
    'provost batch FILE...' files each FILE as a capture of its own, as
    'provost FILE' would, but parses them in parallel (--jobs, default the
    CPUs, at most 16) and files them all under one hold of the store lock as
    ONE commit, tagged capture-<id> for each. Capture ids follow the order
    the files were named; meta.tsv records each one's 'input' path and its
    'batch' (the id range). --name applies to the files after it; --from
    LIST reads files one a line (NAME<TAB>PATH sets the name per file).
    Every file is checked before anything is filed; an empty one is skipped,
    and one that fails to parse is reported while the rest are filed (exit
    1). --llm parses one file at a time, so a new template is written once.

SETTINGS (environment)
    PROVOST_HOME              the global store (default: ~/.provost)
    PROVOST_WORKSPACE         use this store, whatever the directory
    PROVOST_ASYNC             1 = always background, 0 = always foreground,
                              unset/auto = background only at a terminal
    PROVOST_QUIET             1 = no progress / ok messages (as --silent)
    PROVOST_MIN_CONFIDENCE    file only at this confidence or more (0-101,
                              default: 50; 101 holds every capture)
    PROVOST_MIN_COVERAGE      quality gate: min line coverage % for a generic
                              parse (default: 80; 0 = off)
    PROVOST_MIN_CONSISTENCY   quality gate: min column-type consistency % for a
                              generic parse (default: 70; 0 = off)
    PROVOST_RECONCILE_OVERLAP % of columns shared to reconcile a changed shape
                              into the same dataset, not NAME.vN (default: 60)
    PROVOST_MULTI             1 = file every dataset an input holds (--multi)
    PROVOST_CONTEXT_COLUMNS   1 = a '#%' prologue's key=value words become row
                              columns (as --context-columns)
    PROVOST_AUTO_MULTI        1 = segment a parse the gate would hold (default;
                              0 = off, as --no-auto-multi)
    PROVOST_MODEL             1 = try the learned layout model (as --model;
                              needs python3)
    PROVOST_LLM               1 = use / have Claude write a template (as --llm)
    PROVOST_LLM_MODEL         the model that writes templates (default:
                              claude-opus-5-5)
    PROVOST_LLM_API_KEY       Anthropic API key for --llm (else
                              ANTHROPIC_API_KEY; use this one where that is
                              taken, e.g. inside Claude Code)
    PROVOST_LLM_BASE_URL      API base URL (default: https://api.anthropic.com)
    PROVOST_LLM_TEMPLATE_DIRS colon-separated template directories, looked in
                              after the store's and before provost's own
    PROVOST_LLM_LOG           a file to append each API reply's usage to (one
                              JSON line: purpose, tool, model, tokens)
    PROVOST_PLUGIN_DIR        an extra strategy directory (as --plugin-dir)
    PROVOST_MAX_INPUT         max input bytes, piped or run (default: 10485760)
    PROVOST_FILES_MAX_SIZE    in a controlled directory, the largest file kept
                              with each run (bytes; default: 1048576; 0 = none)
    PROVOST_LOCK_TIMEOUT      seconds to wait for the store lock (default: 30)
    PROVOST_LOCK_MAX_AGE      seconds after which a lock is presumed stale even
                              if its owner cannot be checked (default: 3600)
    TMPDIR                    temp and background spool directories (default:
                              /tmp)
    XDG_CONFIG_HOME           where 'listen' keeps its file (default ~/.config)
    Numeric settings must be whole numbers; anything else is refused (exit 2).

See docs/0001-rename-and-simplification.md for the design.
EOF
}

# "did you mean" for a mistyped subcommand (edit distance ≤ 2, or a prefix).
_suggest_verb () {
    W="$1" awk 'function lev(a, b,    la, lb, i, j, d, c, x, y, z) {
            la = length(a); lb = length(b)
            for (i = 0; i <= la; i++) d[i, 0] = i
            for (j = 0; j <= lb; j++) d[0, j] = j
            for (i = 1; i <= la; i++) for (j = 1; j <= lb; j++) {
                c = (substr(a, i, 1) == substr(b, j, 1)) ? 0 : 1
                x = d[i-1, j] + 1; y = d[i, j-1] + 1; z = d[i-1, j-1] + c
                d[i, j] = (x < y ? (x < z ? x : z) : (y < z ? y : z))
            }
            return d[la, lb]
        }
        BEGIN {
            w = ENVIRON["W"]; n = split("run inject batch listen unlisten source control uncontrol instances export import log restore next prev files diff search ls show rm stats triage doctor help", v, " ")
            best = ""; bd = 3
            for (i = 1; i <= n; i++) {
                dist = lev(w, v[i])
                if (length(w) >= 2 && index(v[i], w) == 1 && dist > 1) dist = 1
                if (dist < bd) { bd = dist; best = v[i] }
            }
            if (best != "" && bd < length(w)) print best
        }'
}

# ---------------------------------------------------------------------------
# Dispatch — bare invocation (or a leading option / FILE) means inject.
# ---------------------------------------------------------------------------
main () {
    local verb="${1:-}"
    case "$verb" in
        help|-h|--help)
            case "${2:-}" in advanced|all|--all|-a) cmd_help_advanced; exit 0 ;; esac
            if [[ $# -ge 2 ]] && _usage_text "$2" >/dev/null 2>&1; then _cmd_help "$2"; fi
            cmd_help; exit 0 ;;
        -V|--version)
            printf 'provost %s\nCopyright (C) 2026 Martin J. Gallagher\nLicense: GPL-3.0-or-later <https://www.gnu.org/licenses/gpl-3.0.html>\nThis is free software: you are free to change and redistribute it.\nThere is no warranty, to the extent permitted by law.\n' "$PROVOST_VERSION"
            exit 0 ;;
    esac
    _validate_settings
    _vet_store_config
    # Bare 'provost' on a terminal: there is nothing piped in — show help
    # instead of silently waiting on stdin.
    if [[ $# -eq 0 && -t 0 ]]; then cmd_help; exit 0; fi
    case "$verb" in
        doctor|log) ;;                            # these show the failures themselves
        *) _warn_failures ;;
    esac
    case "$verb" in
        run)           shift; cmd_run "$@" ;;
        batch)         shift; cmd_batch "$@" ;;
        source|src)    shift; cmd_source "$@" ;;
        control)       shift; cmd_control "$@" ;;
        uncontrol)     shift; cmd_uncontrol "$@" ;;
        instances)     shift; cmd_instances "$@" ;;
        export)        shift; cmd_export "$@" ;;
        import)        shift; cmd_import "$@" ;;
        log)           shift; cmd_log "$@" ;;
        restore)       shift; cmd_restore "$@" ;;
        next|prev)     cmd_step "$@" ;;
        files)         shift; cmd_files "$@" ;;
        diff)          shift; cmd_diff "$@" ;;
        search|find)   shift; cmd_search "$@" ;;
        listen)        shift; cmd_listen "$@" ;;
        unlisten)      shift; cmd_unlisten "$@" ;;
        ls)            shift; cmd_ls "$@" ;;
        show)          shift; cmd_show "$@" ;;
        rm|delete|drop) shift; cmd_rm "$@" ;;
        stats)         shift; cmd_stats "$@" ;;
        triage)        shift; cmd_triage "$@" ;;
        doctor)        shift; cmd_doctor "$@" ;;
        inject)        shift; cmd_inject "$@" ;;   # explicit alias
        ""|-*)         cmd_inject "$@" ;;          # default verb (options only)
        *)
            # A lone word: a FILE to inject if it exists (or looks like a path),
            # otherwise a mistyped subcommand.
            if [[ -e "$verb" || "$verb" == */* ]]; then cmd_inject "$@"; return; fi
            local s; s="$(_suggest_verb "$verb")"
            if [[ -n "$s" ]]; then
                usage_die "unknown command '$verb' (did you mean '$s'?)"
            fi
            usage_die "unknown command '$verb' (and no such file) — see 'provost help'."
            ;;
    esac
}
main "$@"
