#!/usr/bin/env bash
#
# 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            (pending) 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.
#
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)
# ---------------------------------------------------------------------------
# Workspace resolution:
#   1. $PROVOST_WORKSPACE explicit override (used by tests / one-off stores)
#   2. else the global store: $PROVOST_HOME, or ~/.provost
# (Stage B will insert "a controlled directory found by walking up from CWD"
#  between these two.)
_default_home () { printf '%s\n' "${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
}

# Resolve the active store:
#   1. $PROVOST_WORKSPACE explicit override
#   2. a controlled directory found by walking up from CWD
#   3. the global store ($PROVOST_HOME / ~/.provost)
PROVOST_WORKSPACE_EXPLICIT="${PROVOST_WORKSPACE:+1}"
if [[ -n "$PROVOST_WORKSPACE_EXPLICIT" ]]; then
    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}"
PROVOST_MAX_INPUT="${PROVOST_MAX_INPUT:-10485760}"   # 10 MiB
PROVOST_TEMP_DIR="${TMPDIR:-/tmp}"
PROVOST_LOCK_TIMEOUT="${PROVOST_LOCK_TIMEOUT:-30}"   # seconds to wait for the store lock
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)

# ---------------------------------------------------------------------------
# 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:-0}" -eq 1 ]] || printf '%s\n' "$*" >&2; }
ok ()   { [[ "${PROVOST_QUIET:-0}" -eq 1 ]] || printf '%s ok%s %s\n' "$C_OK" "$C_RST" "$*" >&2; }
die ()  { printf '%serror%s %s\n' "$C_ERR" "$C_RST" "$*" >&2; exit 1; }

# ---------------------------------------------------------------------------
# Locate the extract engine. It is bundled in this repo at provost/extract/ with
# its shared libs at provost/lib/ (the engine's ../lib). PROVOST_HOME can override.
# ---------------------------------------------------------------------------
_find_engine () {
    local script_dir="$PROVOST_ROOT"
    local candidates=(
        "${PROVOST_HOME:-}/extract/envoy-extract"
        "$script_dir/extract/envoy-extract"       # bundled (canonical)
    )
    local c
    for c in "${candidates[@]}"; do
        [[ -n "$c" && -f "$c" ]] && { printf '%s\n' "$c"; return 0; }
    done
    return 1
}

# Locate the analyze engine (bundled at provost/analyze/). PROVOST_HOME can override.
_find_analyze () {
    local script_dir="$PROVOST_ROOT"
    local candidates=(
        "${PROVOST_HOME:-}/analyze/envoy-analyze"
        "$script_dir/analyze/envoy-analyze"       # bundled (canonical)
    )
    local c
    for c in "${candidates[@]}"; do
        [[ -n "$c" && -f "$c" ]] && { printf '%s\n' "$c"; return 0; }
    done
    return 1
}

# ---------------------------------------------------------------------------
# Transform layer — the tidy melt (`_core_melt_tidy`) and schema-union helpers
# (`_core_union_header`, `_core_align_rows`) live in provost/transform/melt.sh.
# Source it now.
# ---------------------------------------------------------------------------
_source_transform () {
    local script_dir="$PROVOST_ROOT"
    local candidates=(
        "${PROVOST_HOME:-}/transform/melt.sh"
        "$script_dir/transform/melt.sh"           # bundled (canonical)
    )
    local c
    for c in "${candidates[@]}"; do
        [[ -n "$c" && -r "$c" ]] && { source "$c"; return 0; }
    done
    die "transform module not found (expected at $script_dir/transform/melt.sh)."
}
_source_transform

_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"; }

# git wrapper — pinned identity + no signing, so commits work in any environment.
_git () { git -C "$PROVOST_WORKSPACE" -c user.name=provost -c user.email=provost@localhost -c commit.gpgsign=false "$@"; }

# --- store lock (a shared PROVOST_HOME may have concurrent writers) ----------
# mkdir is atomic across processes/users; spin up to PROVOST_LOCK_TIMEOUT seconds.
# Record $BASHPID (this process), NOT $$: a detached async ingest runs in a
# subshell where $$ is still the already-exited parent provost, and the stale-lock
# breaker would then see a "dead" owner and steal a live lock mid-write.
_lock () {
    local lock="$PROVOST_WORKSPACE/.lock" waited=0 pid
    while ! mkdir "$lock" 2>/dev/null; do
        pid="$(cat "$lock/pid" 2>/dev/null || echo)"
        if [[ -n "$pid" ]] && ! kill -0 "$pid" 2>/dev/null; then
            rm -rf "$lock" 2>/dev/null; continue   # break a stale lock
        fi
        waited=$((waited + 1))
        [[ $waited -ge $((PROVOST_LOCK_TIMEOUT * 5)) ]] && die "store is locked (waited ${PROVOST_LOCK_TIMEOUT}s): $lock"
        sleep 0.2
    done
    echo "$BASHPID" > "$lock/pid"
}
_unlock () { rm -rf "$PROVOST_WORKSPACE/.lock" 2>/dev/null || true; }

# 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 () {
    mkdir -p "$(_ds_dir)" "$(_cap_dir)"
    if [[ ! -d "$PROVOST_WORKSPACE/.git" || ! -f "$(_rowid_file)" ]]; then
        _lock
        if [[ ! -d "$PROVOST_WORKSPACE/.git" ]]; then
            _git init -q
            printf 'provost workspace (store model %s)\n' "$PROVOST_VERSION" > "$PROVOST_WORKSPACE/.provostrc"
            _git add -A >/dev/null 2>&1 || true
            _git commit -q -m "provost: init workspace" >/dev/null 2>&1 || true
            _git branch -M main >/dev/null 2>&1 || true
        fi
        [[ -f "$(_rowid_file)" ]] || echo 1 > "$(_rowid_file)"
        _unlock
    fi
}

# Zero-padded next capture id from the count of existing capture dirs.
_next_capture_id () {
    local dir n; dir="$(_cap_dir)"
    n=$(find "$dir" -mindepth 1 -maxdepth 1 -type d 2>/dev/null | wc -l | tr -d ' ')
    printf 'c%04d\n' "$((n + 1))"
}

# Column overlap (Jaccard, as an integer percent) between two tab-separated
# headers passed as the two lines of stdin.
_col_overlap () {
    awk -F'\t' '
        NR==1 { for(i=1;i<=NF;i++){a[$i]=1}; na=NF }
        NR==2 { for(i=1;i<=NF;i++){ 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) }'
}

# (Header union + row alignment come from the imported transform module:
# _core_union_header "$old" "$new"  and  _core_align_rows "$src" "$tgt" < rows.)

# Append tidy TSV (header+rows) to a dataset file. 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, …).
# Echoes "<final-dataset-name>\t<row-count>" (single line) on stdout.
_store_append () {
    local name="$1" tsv="$2" dir; dir="$(_ds_dir)"
    local base="$dir/$name.tsv" target="$name" file="$dir/$name.tsv"
    local new_header; new_header="$(printf '%s\n' "$tsv" | head -1)"

    if [[ ! -f "$base" ]]; then
        printf '%s\n' "$tsv" > "$base"
    else
        local old_header; old_header="$(head -1 "$base")"
        if [[ "$old_header" == "$new_header" ]]; then
            printf '%s\n' "$tsv" | tail -n +2 >> "$base"
        else
            local overlap; overlap="$(printf '%s\n%s\n' "$old_header" "$new_header" | _col_overlap)"
            if [[ "$overlap" -ge "$PROVOST_RECONCILE_OVERLAP" ]]; then
                # Reconcile into the base dataset: union header, realign both sides
                # (imported transform helpers).
                local unified tmp; unified="$(_core_union_header "$old_header" "$new_header")"
                tmp="$(mktemp "$dir/.recon.XXXXXX")"
                {
                    printf '%s\n' "$unified"
                    tail -n +2 "$base"                | _core_align_rows "$old_header" "$unified"
                    printf '%s\n' "$tsv" | tail -n +2 | _core_align_rows "$new_header" "$unified"
                } > "$tmp" && mv "$tmp" "$base" || { rm -f "$tmp"; return 1; }
            else
                # Genuinely different shape → variant file.
                local v=2
                while [[ -f "$dir/$name.v$v.tsv" ]] && \
                      [[ "$(head -1 "$dir/$name.v$v.tsv")" != "$new_header" ]]; do
                    v=$((v+1))
                done
                target="$name.v$v"; file="$dir/$target.tsv"
                if [[ -f "$file" ]]; then printf '%s\n' "$tsv" | tail -n +2 >> "$file"
                else printf '%s\n' "$tsv" > "$file"; fi
            fi
        fi
    fi
    local rows; rows=$(( $(wc -l < "$file") - 1 )); [[ $rows -lt 0 ]] && rows=0
    printf '%s\t%s\n' "$target" "$rows"
}

# Core ingest pipeline shared by `inject` (pipe/file) and `run` (a command).
#   $1 input file  $2 threshold  $3 source (inject|run)  $4 cmd ("" for inject)
#   $5 rc ("" for inject)
# Extracts → classifies → melts → assigns a global #N to each row → stores →
# records provenance (#N → capture) → commits, all under the store lock.
_ingest () {
    local in="$1" threshold="$2" source="$3" cmd="${4:-}" rc="${5:-}"
    local engine; engine="$(_find_engine)" \
        || die "extract engine not found (expected at $PROVOST_ROOT/extract/envoy-extract; set PROVOST_HOME to override)."

    local tmp; tmp="$(mktemp -d "$PROVOST_TEMP_DIR/provost.XXXXXX")"
    # shellcheck disable=SC2064
    trap "rm -rf '$tmp'" RETURN

    # 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"
    if [[ "$(head -c 2 "$in" 2>/dev/null)" == '#%' ]]; then
        body="$tmp/body.txt"
        awk -v ctx="$ctxf" \
            '!started && /^#%/ { print > ctx; next } { started=1; print }' \
            "$in" > "$body"
    fi

    local metaf="$tmp/meta.txt" wide meta strategy conf
    # 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 ~/.mfe/strategies user dir.
    local eargs=()
    if [[ -n "${PROVOST_PLUGIN_DIR:-}" ]]; then
        eargs+=(--plugin-dir "$PROVOST_PLUGIN_DIR")
    fi

    # The datasets this capture will file. 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_content=()
    local content dsname reason=""

    if [[ "${PROVOST_MULTI:-0}" -eq 1 ]]; then
        # --multi splits the input into blocks, groups them, and emits each
        # surviving dataset behind a `__DATASET__ n strategy confidence` line.
        local multi_out="$tmp/multi.txt"
        env -u GIT_DIR bash "$engine" --max-input-size "$PROVOST_MAX_INPUT" \
            ${eargs[@]+"${eargs[@]}"} --multi -f tsv --max-strategies 1 "$body" \
            >"$multi_out" 2>"$metaf" || true
        awk -F'\t' -v dir="$tmp" '
            /^__DATASET__/ { n = $2; printf "%s\t%s\n", $3, $4 > (dir "/ds." n ".meta"); next }
            n != "" { print > (dir "/ds." n ".wide") }
        ' "$multi_out" 2>/dev/null || true
        local i=1 dstrat dconf dwide
        while [[ -f "$tmp/ds.$i.meta" ]]; do
            IFS=$'\t' read -r dstrat dconf < "$tmp/ds.$i.meta" || true
            dwide="$(cat "$tmp/ds.$i.wide" 2>/dev/null || true)"
            [[ "$dconf" =~ ^[0-9]+$ ]] || dconf=0
            # 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" && -n "$dwide" && "$dconf" -ge "$threshold" ]]; then
                ds_name+=("$dstrat"); ds_strategy+=("$dstrat"); ds_conf+=("$dconf")
                ds_content+=("$(_core_melt_tidy "$dwide")")
            fi
            i=$((i + 1))
        done
    fi

    if [[ ${#ds_content[@]} -eq 0 ]]; then
        # Single-dataset path — also the fallback when --multi found nothing
        # worth filing, so an unmatched input still reaches the triage queue.
        wide="$(env -u GIT_DIR bash "$engine" --max-input-size "$PROVOST_MAX_INPUT" \
                ${eargs[@]+"${eargs[@]}"} \
                --emit-meta -f tsv --max-strategies 1 "$body" 2>"$metaf")" || true
        meta="$(grep '^__EXTRACT_META__' "$metaf" 2>/dev/null | tail -1 || true)"
        strategy="$(printf '%s' "$meta" | awk -F'\t' '{print $2}')"
        conf="$(printf '%s' "$meta" | awk -F'\t' '{print $3}')"
        [[ "$conf" =~ ^[0-9]+$ ]] || conf=0

        # Build the content to file (header + data rows), pre-#-index, for both
        # the matched and the held-back (unsorted) paths.
        if [[ -z "$strategy" || -z "$wide" ]]; then reason="no-match"
        elif [[ "$conf" -lt "$threshold" ]]; then reason="low-confidence ($conf<$threshold)"; fi
        if [[ -n "$reason" ]]; then
            dsname="unsorted"; strategy="${strategy:-none}"
            content="$(printf 'raw\n%s' "$(base64 -w0 < "$in" 2>/dev/null || base64 < "$in" | tr -d '\n')")"
        else
            dsname="$strategy"
            content="$(_core_melt_tidy "$wide")"
        fi
        ds_name=("$dsname"); ds_strategy=("$strategy"); ds_conf=("$conf")
        ds_content=("$content")
    fi

    # --- critical section: assign #N, append, provenance, commit -----------
    _lock
    # shellcheck disable=SC2064
    trap "_unlock; rm -rf '$tmp'" RETURN

    local startid nrows fields id capd ts final
    id="$(_next_capture_id)"; 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
    [[ -f "$(_prov_file)" ]] || printf '#\tdataset\tcapture\n' > "$(_prov_file)"

    # 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 totalrows=0 idcontent result k i
    nextid="$(cat "$(_rowid_file)" 2>/dev/null || echo 1)"; firstid="$nextid"
    local -a out_names=() out_ranges=()
    for (( k=0; k<${#ds_content[@]}; k++ )); do
        content="${ds_content[k]}"
        startid="$nextid"
        nrows=$(( $(printf '%s\n' "$content" | grep -c .) - 1 )); [[ $nrows -lt 0 ]] && nrows=0

        # Prepend a visible, globally-unique '#' index to the header + each row.
        idcontent="$(printf '%s\n' "$content" | awk -F'\t' -v OFS='\t' -v s="$startid" '
            NR==1 { print "#", $0; next } { print (s + NR - 2), $0 }')"
        fields=$(printf '%s\n' "$idcontent" | head -1 | awk -F'\t' '{print NF}')

        result="$(_store_append "${ds_name[k]}" "$idcontent")"
        final="${result%%$'\t'*}"
        out_names+=("$final")
        out_ranges+=("$([[ $nrows -gt 0 ]] && echo "#${startid}..$((startid+nrows-1))" || echo none)")

        for (( i=0; i<nrows; i++ )); do
            printf '%s\t%s\t%s\n' "$((startid+i))" "$final" "$id" >> "$(_prov_file)"
        done
        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=""
    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]}"
    done
    {
        printf 'id\t%s\n' "$id";            printf 'ts\t%s\n' "$ts"
        printf 'source\t%s\n' "$source";    printf 'command\t%s\n' "$cmd"
        printf 'rc\t%s\n' "$rc";            printf 'strategy\t%s\n' "$all_strats"
        printf 'confidence\t%s\n' "$all_confs";  printf 'dataset\t%s\n' "$all_names"
        printf 'row_ids\t%s\n' "$([[ $totalrows -gt 0 ]] && echo "$firstid-$((nextid-1))" || echo none)"
        if [[ ${#out_names[@]} -gt 1 ]]; then
            printf 'datasets\t%s\n' "${#out_names[@]}"
            printf 'dataset_rows\t%s\n' "$all_rows"
        fi
        if [[ -n "${PROVOST_PLUGIN_DIR:-}" ]]; then
            printf 'plugin_dir\t%s\n' "$PROVOST_PLUGIN_DIR"
        fi
        if [[ -s "$ctxf" ]]; then printf 'context\tcontext.txt\n'; fi
    } > "$capd/meta.tsv"
    startid="$firstid"; nrows="$totalrows"; final="$all_names"; strategy="$all_strats"; conf="$all_confs"

    # 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.
    if [[ "${PROVOST_STORE_KIND:-}" == controlled:* ]]; then
        # Subshell so a set -e abort inside can't escape past the guard.
        ( _snapshot_context "$(dirname "$PROVOST_WORKSPACE")" "$PROVOST_WORKSPACE" ) >/dev/null 2>&1 || true
    fi
    _git add -A >/dev/null 2>&1 || true
    local msg="capture $id: $source $strategy → $final (#${startid}..$((startid+nrows-1)))"
    [[ -n "$cmd" ]] && msg="$msg [$cmd]"
    _git commit -q -m "$msg" >/dev/null 2>&1 || true
    _git tag -f "capture-$id" >/dev/null 2>&1 || true
    _unlock
    # shellcheck disable=SC2064
    trap "rm -rf '$tmp'" RETURN

    local idrange="#${startid}..$((startid+nrows-1))"; [[ $nrows -eq 0 ]] && idrange="(no rows)"
    if [[ -n "$reason" ]]; then
        info "$id: $reason — queued in 'unsorted' as $idrange (committed)."
    elif [[ ${#out_names[@]} -gt 1 ]]; then
        ok "$id: → ${#out_names[@]} datasets, rows $idrange:"
        for (( j=0; j<${#out_names[@]}; j++ )); do
            info "      ${out_names[j]}  ${out_ranges[j]}  (strategy ${ds_strategy[j]}, conf ${ds_conf[j]})"
        done
    else
        ok "$id: → dataset '$final' rows $idrange (strategy $strategy, conf $conf) — $((fields-1)) field(s)."
    fi
}

# Fire-and-forget ingest: spool the input and run _ingest in a detached child so
# the shell prompt returns immediately. This is the DEFAULT for inject/run;
# --sync/--wait (or PROVOST_ASYNC=0) forces the foreground path that prints the
# resulting dataset + #N range instead of returning at once.
# The spool lives OUTSIDE the workspace (so it is never swept into a git commit,
# and survives the parent's tmp cleanup); the child ingests it, and on success
# removes the spool. On failure the spool — including log.txt — is kept so the
# error is inspectable. Store serialisation is unchanged: the child takes the
# same _lock as any writer, so concurrent async captures stay consistent.
_ingest_async () {
    local in="$1" threshold="$2" source="$3" cmd="${4:-}" rc="${5:-}"
    local spool; spool="$(mktemp -d "$PROVOST_TEMP_DIR/provost-async.XXXXXX")"
    cp "$in" "$spool/in.txt"
    (
        trap '' HUP                                   # survive the shell exiting
        if _ingest "$spool/in.txt" "$threshold" "$source" "$cmd" "$rc" \
               >"$spool/log.txt" 2>&1; then
            rm -rf "$spool"
        fi                                            # keep spool+log on failure
    ) </dev/null >/dev/null 2>&1 &
    disown 2>/dev/null || true
    info "queued (async, pid $!) — capturing in the background; see 'provost log' shortly."
}

# ---------------------------------------------------------------------------
# Default verb: inject (pipe or file)
# ---------------------------------------------------------------------------
cmd_inject () {
    local file="" threshold="$PROVOST_MIN_CONFIDENCE" async="${PROVOST_ASYNC:-1}"
    while [[ $# -gt 0 ]]; do
        case "$1" in
            -q|--silent) PROVOST_QUIET=1; shift ;;
            --min-confidence) threshold="${2:?--min-confidence needs a number}"; shift 2 ;;
            --plugin-dir) PROVOST_PLUGIN_DIR="${2:?--plugin-dir needs a directory}"; shift 2 ;;
            --multi) PROVOST_MULTI=1; shift ;;
            --async) async=1; shift ;;
            --sync|--wait) async=0; shift ;;
            -h|--help) cmd_help; return 0 ;;
            -*) die "unknown option: $1" ;;
            *) file="$1"; shift ;;
        esac
    done
    if [[ -n "${PROVOST_PLUGIN_DIR:-}" && ! -d "$PROVOST_PLUGIN_DIR" ]]; then
        die "plugin strategy dir not found: $PROVOST_PLUGIN_DIR"
    fi

    _ensure_workspace
    local tmp; tmp="$(mktemp -d "$PROVOST_TEMP_DIR/provost.XXXXXX")"
    # shellcheck disable=SC2064
    trap "rm -rf '$tmp'" EXIT INT TERM
    local in="$tmp/in.txt"
    if [[ -n "$file" ]]; then
        [[ -f "$file" ]] || die "no such file: $file"
        tr -d '\r' < "$file" > "$in"
    else
        head -c "$((PROVOST_MAX_INPUT + 1))" | tr -d '\r' > "$in"
    fi
    [[ -s "$in" ]] || die "no input to provost (empty)."
    [[ "$(wc -c < "$in")" -le "$PROVOST_MAX_INPUT" ]] \
        || die "input too large (max $PROVOST_MAX_INPUT bytes; raise PROVOST_MAX_INPUT)."

    if [[ "$async" -eq 1 ]]; then _ingest_async "$in" "$threshold" "inject" "" ""
    else _ingest "$in" "$threshold" "inject" "" ""; fi
}

# ---------------------------------------------------------------------------
# run — execute a command, capture its output as a commit, and ingest it.
# ---------------------------------------------------------------------------
cmd_run () {
    local threshold="$PROVOST_MIN_CONFIDENCE" async="${PROVOST_ASYNC:-1}"
    while [[ $# -gt 0 ]]; do
        case "$1" in
            -q|--silent) PROVOST_QUIET=1; shift ;;
            --min-confidence) threshold="${2:?--min-confidence needs a number}"; shift 2 ;;
            --plugin-dir) PROVOST_PLUGIN_DIR="${2:?--plugin-dir needs a directory}"; shift 2 ;;
            --multi) PROVOST_MULTI=1; shift ;;
            --async) async=1; shift ;;
            --sync|--wait) async=0; shift ;;
            -h|--help) cmd_help; return 0 ;;
            --) shift; break ;;
            -*) die "unknown option: $1" ;;
            *) break ;;
        esac
    done
    [[ $# -gt 0 ]] || die "usage: provost run [--] <command> [args...]"
    if [[ -n "${PROVOST_PLUGIN_DIR:-}" && ! -d "$PROVOST_PLUGIN_DIR" ]]; then
        die "plugin strategy dir not found: $PROVOST_PLUGIN_DIR"
    fi

    _ensure_workspace
    local tmp; tmp="$(mktemp -d "$PROVOST_TEMP_DIR/provost.XXXXXX")"
    # shellcheck disable=SC2064
    trap "rm -rf '$tmp'" EXIT INT TERM
    local in="$tmp/in.txt" rc=0
    # Capture combined stdout+stderr (the thing you'd otherwise eyeball).
    "$@" > "$in" 2>&1 || rc=$?
    tr -d '\r' < "$in" > "$in.clean" && mv "$in.clean" "$in"
    local cmd="$*"
    [[ -s "$in" ]] || { info "command produced no output (rc=$rc) — nothing captured."; return "$rc"; }
    if [[ "$async" -eq 1 ]]; then _ingest_async "$in" "$threshold" "run" "$cmd" "$rc"
    else _ingest "$in" "$threshold" "run" "$cmd" "$rc"; fi
    return "$rc"
}

# ---------------------------------------------------------------------------
# log / restore — the git core: list captures, bring one back up.
# ---------------------------------------------------------------------------
cmd_log () {
    _ensure_workspace
    shopt -s nullglob
    local d id ts src strat ds rc any=0
    printf '%-6s %-20s %-7s %-14s %-12s %s\n' "ID" "WHEN" "SOURCE" "STRATEGY" "DATASET" "COMMAND" >&2
    for d in $(printf '%s\n' "$(_cap_dir)"/*/ | sort); do
        [[ -f "$d/meta.tsv" ]] || continue
        any=1
        id=$(awk -F'\t' '$1=="id"{print $2}' "$d/meta.tsv")
        ts=$(awk -F'\t' '$1=="ts"{print $2}' "$d/meta.tsv")
        src=$(awk -F'\t' '$1=="source"{print $2}' "$d/meta.tsv")
        strat=$(awk -F'\t' '$1=="strategy"{print $2}' "$d/meta.tsv")
        ds=$(awk -F'\t' '$1=="dataset"{print $2}' "$d/meta.tsv")
        local cmd; cmd=$(awk -F'\t' '$1=="command"{print $2}' "$d/meta.tsv")
        printf '%-6s %-20s %-7s %-14s %-12s %s\n' "$id" "$ts" "$src" "$strat" "$ds" "$cmd"
    done
    [[ $any -eq 1 ]] || info "no captures yet — run 'provost run <cmd>' or pipe into 'provost'."
}

cmd_restore () {
    _ensure_workspace
    local target="${1:?usage: provost restore <capture-id|latest>}"
    if [[ "$target" == "latest" ]]; then
        _git checkout -q main 2>/dev/null \
            && ok "restored workspace to latest (main)." \
            || die "could not return to latest."
        return 0
    fi
    [[ "$target" == capture-* ]] || target="capture-$target"
    _git rev-parse -q --verify "refs/tags/$target" >/dev/null 2>&1 \
        || die "no such capture: ${target#capture-} (see 'provost log')."
    # Bring that capture's committed tree back up (detached).
    _git checkout -q "$target" 2>/dev/null \
        || die "checkout failed for $target."
    ok "brought up ${target#capture-} (detached). Return with: provost restore latest"
    local capd; capd="$(_cap_dir)/${target#capture-}"
    [[ -f "$capd/meta.tsv" ]] && { echo; sed 's/\t/: /' "$capd/meta.tsv"; }
}

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

# Snapshot the 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" mani="$ws/context/manifest.tsv" f sz mt sha
    mkdir -p "$ws/context"
    printf 'path\tsize\tmtime\tsha1\n' > "$mani"
    while IFS= read -r f; do
        sz=$(stat -c '%s' "$f" 2>/dev/null || echo 0)
        mt=$(stat -c '%Y' "$f" 2>/dev/null || echo 0)
        if [[ "$sz" -le 1048576 ]] && command -v sha1sum >/dev/null 2>&1; then
            sha=$(sha1sum "$f" 2>/dev/null | cut -d' ' -f1)
        else sha="(skipped)"; fi
        printf '%s\t%s\t%s\t%s\n' "${f#"$dir"/}" "$sz" "$mt" "$sha" >> "$mani"
    done < <(find "$dir" -type f -not -path '*/.provost/*' -not -path '*/.git/*' 2>/dev/null | sort)
    local n; n=$(( $(wc -l < "$mani") - 1 ))
    printf '%s\n' "$n"
}

cmd_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 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
    printf '%s\n' "$dir" > "$ws/CONTROLLED"
    local n; n="$(_snapshot_context "$dir" "$ws")"
    _lock
    _git add -A >/dev/null 2>&1 || true
    _git commit -q -m "provost: control $dir (context: $n file(s))" >/dev/null 2>&1 || true
    _unlock
    # Register in the global store so `provost instances` (run anywhere) sees it.
    mkdir -p "$(_default_home)"
    [[ -f "$(_registry)" ]] || printf 'directory\tcontrolled_at\n' > "$(_registry)"
    grep -qxF "$dir	$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$(_registry)" 2>/dev/null || \
        printf '%s\t%s\n' "$dir" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" >> "$(_registry)"
    ok "controlling '$dir' — captures run here now route to $ws (context: $n file(s))."
}

cmd_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"
    rm -f "$dir/.provost/CONTROLLED"
    if [[ -f "$(_registry)" ]]; then
        grep -v "^$dir	" "$(_registry)" > "$(_registry).tmp" 2>/dev/null && mv "$(_registry).tmp" "$(_registry)" || true
    fi
    ok "released '$dir' (its .provost data is kept; captures here revert to the global store)."
}

cmd_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 -lt 0 ]] && n=0
    [[ $n -gt 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 () {
    _ensure_workspace
    local out="${1:-}"
    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
    local rc=0
    _git bundle create "$out" --all >/dev/null 2>&1 || rc=$?
    _unlock
    [[ $rc -eq 0 && -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 $out <dest>"
}

cmd_import () {
    local file="${1:?usage: provost import <file.bundle> [<dest-dir>]}"
    [[ -f "$file" ]] || die "no such export file: $file"
    local dest="${2:-}"
    [[ -z "$dest" ]] && dest="$PWD/provost-import-$(basename "$file" .bundle)"
    [[ -e "$dest" ]] && die "destination already exists: $dest"
    git bundle verify "$file" >/dev/null 2>&1 || die "not a valid provost export bundle: $file"
    git clone -q "$file" "$dest" >/dev/null 2>&1 || die "import (clone) failed."
    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='$dest' provost ls   (or show/log/source/restore)"
}

# ---------------------------------------------------------------------------
# source — given a dataset row's #N index, show the original captured input.
# ---------------------------------------------------------------------------
cmd_source () {
    _ensure_workspace
    local n="${1:?usage: provost source <#N>}"
    n="${n#\#}"                                   # tolerate a leading '#'
    [[ "$n" =~ ^[0-9]+$ ]] || die "row index must be a number (got '$n')."
    local prov; prov="$(_prov_file)"
    [[ -f "$prov" ]] || die "no provenance yet — capture something first."
    local line; line="$(awk -F'\t' -v n="$n" '$1==n {print; exit}' "$prov")"
    [[ -n "$line" ]] || die "no row #$n (see 'provost ls' / 'provost show <dataset>')."
    local ds cap; ds="$(printf '%s' "$line" | cut -f2)"; cap="$(printf '%s' "$line" | cut -f3)"
    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
}

# ---------------------------------------------------------------------------
# stats — statistics on a dataset (the bundled analyze engine).
# ---------------------------------------------------------------------------
cmd_stats () {
    local name="${1:?usage: provost stats <dataset> [analyze-args...]}"; shift || true
    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/envoy-analyze)."
    # Datasets are TSV; suppress the gawk-preference banner. Extra args pass
    # straight through (e.g. --all, a numbered analysis list, --format json).
    ENVOY_ANALYZE_SUPPRESS_AWK_WARNING=1 env -u GIT_DIR bash "$analyze" "$f" "$@"
}

# ---------------------------------------------------------------------------
# ls / show / triage / doctor
# ---------------------------------------------------------------------------
cmd_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 -lt 0 ]] && rows=0
        printf '%-24s %8d\n' "$name" "$rows"
    done
    [[ $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, escaping any
# embedded tabs/newlines so each captured input stays on one line of the table.
# 'provost source #N' remains the way back to the exact original bytes.
_decode_unsorted_tsv () {
    local f="$1" n b64 decoded
    printf '#\traw\n'
    tail -n +2 "$f" | while IFS=$'\t' read -r n b64; do
        decoded="$(printf '%s' "$b64" | base64 -d 2>/dev/null || printf '%s' "$b64")"
        decoded="${decoded//$'\t'/\\t}"
        decoded="${decoded//$'\n'/\\n}"
        printf '%s\t%s\n' "$n" "$decoded"
    done
}

cmd_show () {
    local name="${1:?usage: provost show NAME}"
    local f; f="$(_ds_dir)/$name.tsv"
    [[ -f "$f" ]] || die "no such dataset: $name (try 'provost ls')."
    if [[ "$name" == "unsorted" ]]; then
        if command -v column >/dev/null 2>&1; then _decode_unsorted_tsv "$f" | column -t -s $'\t'
        else _decode_unsorted_tsv "$f"; fi
    elif command -v column >/dev/null 2>&1; then column -t -s $'\t' "$f"
    else cat "$f"; fi
}

cmd_triage () {
    local f; f="$(_ds_dir)/unsorted.tsv"
    [[ -f "$f" ]] || { info "triage queue empty — nothing unmatched."; return 0; }
    info "unsorted rows (low-confidence / no-match), raw input decoded ('provost source #N' for exact bytes):"
    if command -v column >/dev/null 2>&1; then _decode_unsorted_tsv "$f" | column -t -s $'\t'
    else _decode_unsorted_tsv "$f"; fi
}

# ---------------------------------------------------------------------------
# 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 the captures/<id>/ dirs that fed the
# removed dataset(s) too (--no-purge / --keep-captures keeps them). The deletion
# is committed, so the git history (and 'provost restore') still holds the
# pre-deletion state. Row indices (#N) are never reused.
# ---------------------------------------------------------------------------
cmd_rm () {
    _ensure_workspace
    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; return 0 ;;
            --)                      shift; while [[ $# -gt 0 ]]; do names+=("$1"); shift; done ;;
            -*)                      die "unknown option: $1" ;;
            *)                       names+=("$1"); shift ;;
        esac
    done
    [[ ${#names[@]} -gt 0 ]] || die "usage: provost rm [-f] [--no-purge] <dataset> [<dataset>...]"

    local dir; dir="$(_ds_dir)"
    # Validate every name up front — refuse the whole operation if any is missing.
    local n 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 -lt 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 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
    # shellcheck disable=SC2064
    trap "_unlock" RETURN

    # A one-name-per-line file drives the awk/loop membership tests.
    local tmp; tmp="$(mktemp "$dir/.rm.XXXXXX")"
    printf '%s\n' "${names[@]}" > "$tmp"

    # 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")"
        awk -F'\t' '
            NR==FNR { del[$0]=1; next }
            FNR==1  { print; next }
            !($2 in del) { print }
        ' "$tmp" "$prov" > "$ptmp" && mv "$ptmp" "$prov" || { rm -f "$ptmp"; }
    fi

    # --purge: remove the capture dirs (and their tags) that fed a removed
    # dataset, identified by each capture's own meta 'dataset' field.
    local purged=0 d cid cds
    if [[ $purge -eq 1 ]]; then
        for d in "$(_cap_dir)"/*/; do
            [[ -f "$d/meta.tsv" ]] || continue
            cds=$(awk -F'\t' '$1=="dataset"{print $2}' "$d/meta.tsv")
            if grep -qxF "$cds" "$tmp"; then
                cid=$(awk -F'\t' '$1=="id"{print $2}' "$d/meta.tsv")
                rm -rf "$d"
                [[ -n "$cid" ]] && _git tag -d "capture-$cid" >/dev/null 2>&1 || true
                purged=$((purged + 1))
            fi
        done
    fi
    rm -f "$tmp"

    _git add -A >/dev/null 2>&1 || true
    local msg="rm: deleted ${#names[@]} dataset(s) [${names[*]}], pruned $total row(s)"
    [[ $purge -eq 1 ]] && msg="$msg, purged $purged capture(s)"
    _git commit -q -m "$msg" >/dev/null 2>&1 || true
    _unlock
    # shellcheck disable=SC2064
    trap - RETURN

    if [[ $purge -eq 1 ]]; then
        ok "deleted ${names[*]} (${total} row(s)); purged $purged capture(s). 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 () {
    local engine
    if engine="$(_find_engine)"; then
        printf '%s ok%s extract engine: %s\n' "$C_OK" "$C_RST" "$engine"
        # The bundled engine is the canonical one. If we resolved a different one
        # (a $PROVOST_HOME/extract/ override), warn — a stale copy there silently
        # shadows the bundled engine and can misclassify inputs.
        if [[ "$engine" != "$PROVOST_ROOT/extract/envoy-extract" ]]; then
            printf '   %swarning%s using an engine from PROVOST_HOME, not the bundled one\n' "$C_ERR" "$C_RST"
            printf '            bundled: %s\n' "$PROVOST_ROOT/extract/envoy-extract"
            printf '            if this is a stale copy, remove it or unset PROVOST_HOME.\n'
        fi
    else
        printf '%serror%s extract engine not found\n' "$C_ERR" "$C_RST"; return 1
    fi
    local analyze
    if analyze="$(_find_analyze)"; then
        printf '%s ok%s analyze engine: %s\n' "$C_OK" "$C_RST" "$analyze"
        command -v gawk >/dev/null 2>&1 || printf '   note: gawk not found; analyze uses %s (some analyses may degrade)\n' "$(awk --version 2>/dev/null | head -1 || echo awk)"
    else
        printf '   analyze engine not found (expected at %s/analyze/)\n' "$PROVOST_ROOT"
    fi
    printf '   store: %s (%s)' "$PROVOST_WORKSPACE" "${PROVOST_STORE_KIND:-global}"
    [[ -d "$PROVOST_WORKSPACE/.git" ]] && printf ' [git-backed]'; printf '\n'
    if [[ -f "$(_rowid_file)" ]]; then
        printf '   next row index: #%s\n' "$(cat "$(_rowid_file)")"
    fi
    printf '   min-confidence: %s\n' "$PROVOST_MIN_CONFIDENCE"
}

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

USAGE
    provost run [--] <cmd> [args...]                capture a command's output as a
                                                 commit, auto-detect + file it
    cmd | provost [--silent] [--min-confidence N]   inject stdin into a typed dataset —
                [--sync] [--plugin-dir DIR]        captures in the BACKGROUND by default;
                [--multi]                          --sync/--wait to file it in the foreground;
                                                 --plugin-dir adds an extra strategy dir;
                                                 --multi files EVERY dataset the input
                                                 holds (a kv block and a table and a
                                                 banner) instead of one winner
    provost [FILE]                                  inject a file instead of stdin
    provost source <#N>                             show the original source behind row #N
    provost control [<dir>]                         make <dir> its own capture instance
    provost uncontrol [<dir>]                        release a controlled directory
    provost instances                               list controlled directories
    provost export [<file>]                         bundle the store into a portable file
    provost import <file> [<dest>]                   reconstitute a bundle elsewhere to browse
    provost log                                     list captures (the git history)
    provost restore <id|latest>                     bring a capture's commit back up
    provost ls                                      list datasets
    provost show NAME                               present a dataset (leading #N = row index)
    provost rm [-f] [--no-purge] NAME...            delete dataset(s) + their captures (--no-purge keeps captures)
    provost stats NAME [analyze-args]               statistics on a dataset (--all for everything)
    provost triage                                  show low-confidence / unmatched bucket (raw input decoded)
    provost doctor                                  check engine + store
    provost help

STORE (global by default, or a controlled directory)
    Captures go to a GLOBAL git store so you can capture from any directory:
      \$PROVOST_HOME, or ~/.provost if unset. Point PROVOST_HOME at a shared,
      group-writable dir to collaborate; a lock serialises concurrent writers.
    'provost control <dir>' makes <dir> its own instance (<dir>/.provost): captures
    run inside it (or a subdir) route there, and each commit snapshots the
    directory's contents as context. Resolution: PROVOST_WORKSPACE override, else
    the nearest controlled dir walking up from CWD, else the global store.
    Every capture is one commit tagged capture-<id> ('log' / 'restore'). Every
    dataset row has a globally-unique visible index #N; 'provost source N' maps it
    back to the exact captured input that produced it.

ENV
    PROVOST_HOME            global store dir (default: ~/.provost)
    PROVOST_WORKSPACE       explicit store override (wins over PROVOST_HOME)
    PROVOST_MIN_CONFIDENCE  commit threshold 0-100 (default: 50)
    PROVOST_PLUGIN_DIR      extra strategy dir for detection (same as --plugin-dir
                          on inject/run; for wrappers like deframe — no
                          ~/.mfe/strategies symlink needed)
    PROVOST_ASYNC          background capture (default 1; set 0 or use --sync for foreground)
    PROVOST_MULTI           file every dataset an input holds (default 0; same as --multi)
    PROVOST_MAX_INPUT       max input bytes (default: 10485760)
    PROVOST_LOCK_TIMEOUT    seconds to wait for the store lock (default: 30)

MULTIPLE DATASETS PER INPUT (--multi)
    Some reports hold more than one kind of data — a key/value block, a table,
    and a banner or comment run that is best left alone. The default one-winner
    capture keeps whichever the engine ranks first and drops the rest; --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 winning strategy, so two parts that
    parse the same way but differ in shape land in NAME and NAME.v2. Row indices
    (#N) run consecutively across them and 'provost source N' still returns the
    whole original input. 'provost log' shows the comma-joined list; the capture's
    meta.tsv records 'datasets' (how many) and 'dataset_rows' (each one's range).

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

Not yet in this prototype (proposal 0001): schema reconciliation (a differing
column set currently splits a dataset into a numbered variant, e.g. free.v2).
See docs/proposals/0001-provost-rename-and-simplification.md.
EOF
}

# ---------------------------------------------------------------------------
# Dispatch — bare invocation (or a leading option / FILE) means inject.
# ---------------------------------------------------------------------------
main () {
    local verb="${1:-}"
    case "$verb" in
        run)           shift; cmd_run "$@" ;;
        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 "$@" ;;
        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 "$@" ;;
        help|-h|--help) cmd_help ;;
        -V|--version)  printf 'provost %s\n' "$PROVOST_VERSION" ;;
        inject)        shift; cmd_inject "$@" ;;   # explicit alias
        *)             cmd_inject "$@" ;;          # default verb
    esac
}
main "$@"
