#!/usr/bin/env bash
# Stack activate and rollback, health-gated (D1, #1659; docs/plans/2026-10-18_stack-release.md).
# Moves the bus, the Console and the human CLI from the pointer's copy to another copy together,
# behind a health gate, and puts the old stack back inside the same run when the gate fails.
# Design and tests: agents/darkwing/work/queue-162/BUILD.md.
#
# Who runs it. Decision 96 (Mos, mos-0716) lets this script stop and start the live units under
# four conditions:
#   1. every adopt, activate or rollback runs on Mos's go, and Jason or Mos runs it;
#   2. a run that migrates the store (activate --migrate) happens with Jason present;
#   3. a rollback --restore with events written after the backup waits for Jason, which is why it
#      needs --events N;
#   4. this script never prunes backups, rollback sets, post-attempt directories or copies.
# Decisions 94 and 102 still decide who starts the Console; webui-service.sh keeps having no start
# or stop verb. status and --plan only read, and every seat may run them.
#
# Most functions below run as `f || handler`, and bash ignores set -e inside such a call, so
# every step that can fail carries its own `|| return 1`.
set -euo pipefail
cd "$(dirname "$0")/.."
REPO="$(pwd)"
source scripts/common.sh

COPIES="${XDG_DATA_HOME:-$HOME/.local/share}/mosaic-stack/copies"
BIN="$HOME/.npm-global/bin"
UNIT_DIR="${XDG_CONFIG_HOME:-$HOME/.config}/systemd/user"
CONSOLE_UNIT=mosaic-webui.service
TIMEOUT=60 WANT_PM=0 MIGRATE=0 PLAN=0 RESTORE="" EVENTS=""

usage() {
    cat >&2 <<'USAGE'
usage: stack-release.sh adopt BUSINESS
       stack-release.sh activate ID BUSINESS [--pm] [--migrate] [--timeout S] [--plan]
       stack-release.sh rollback BUSINESS [--restore ACTIVATION [--events N]] [--timeout S] [--plan]
       stack-release.sh status BUSINESS
common options: --copies DIR, --bin DIR
USAGE
    exit "${1:-4}"
}
say() { echo "stack-release: $*"; [ -z "${STEPS:-}" ] || echo "$(date -u +%Y-%m-%dT%H:%M:%SZ) $*" >> "$STEPS" || true; }
now() { date -u +%Y-%m-%dT%H:%M:%SZ; }
now_ms() { date -u +%Y-%m-%dT%H:%M:%S.%3NZ; }

# One JSON object from KEY VALUE pairs; the value @null is JSON null.
jline() {
    node -e 'const a=process.argv.slice(1),o={};for(let i=0;i<a.length;i+=2)o[a[i]]=a[i+1]==="@null"?null:a[i+1];process.stdout.write(JSON.stringify(o)+"\n")' -- "$@"
}
log_line() { jline at "$(now)" op "$OP" from "${FROM:-@null}" to "${TO:-@null}" backup "${BACKUP:-@null}" activation "${ACT:-@null}" "$@" >> "$LOG"; }

# A refusal before anything stopped: exit 3, and one log line with the code, except for
# run-locked (the lock holder may be writing the log) and in --plan, which writes nothing.
refuse() {
    local code="$1"; shift
    echo "stack-release: refusing $code: $*" >&2
    [ "$code" = run-locked ] || [ "$PLAN" = 1 ] || log_line result refused code "$code"
    exit 3
}

parse_opts() {
    while [ $# -gt 0 ]; do
        case "$1" in
            --copies) [ $# -ge 2 ] || usage; COPIES="$2"; shift 2 ;;
            --bin) [ $# -ge 2 ] || usage; BIN="$2"; shift 2 ;;
            --timeout) [ "$OP" != status ] && [ "$OP" != adopt ] && [ $# -ge 2 ] && [[ "$2" =~ ^[1-9][0-9]{0,4}$ ]] || usage; TIMEOUT="$2"; shift 2 ;;
            --plan) [ "$OP" = activate ] || [ "$OP" = rollback ] || usage; PLAN=1; shift ;;
            --pm) [ "$OP" = activate ] || usage; WANT_PM=1; shift ;;
            --migrate) [ "$OP" = activate ] || usage; MIGRATE=1; shift ;;
            --restore) [ "$OP" = rollback ] && [ $# -ge 2 ] && [[ "$2" =~ ^[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}$ ]] || usage; RESTORE="$2"; shift 2 ;;
            --events) [ "$OP" = rollback ] && [ $# -ge 2 ] && [[ "$2" =~ ^(0|[1-9][0-9]*)$ ]] || usage; EVENTS="$2"; shift 2 ;;
            *) usage ;;
        esac
    done
    [ -z "$EVENTS" ] || [ -n "$RESTORE" ] || usage
    local d
    for d in "$COPIES" "$BIN"; do
        case "$d" in /*) ;; *) usage ;; esac
        case "$d" in *[!A-Za-z0-9/._+-]*) usage ;; esac
    done
}

# --- The pointer and the log (item 1) ---

# Sets P_COPY P_BUSINESS P_PM P_CLAUDE P_SCHEMA P_ACT and the array P_CONSOLE; returns 1 with no
# pointer. A pointer that is there but malformed stops the script: Jason or Mos settles it.
read_pointer() {
    [ -f "$POINTER" ] || return 1
    local fields
    fields="$(node -e '
const p=JSON.parse(require("fs").readFileSync(process.argv[1],"utf8"));
const ok=/^[0-9a-f]{40}$/.test(p.copy)&&typeof p.business==="string"&&typeof p.pm==="boolean"&&typeof p.claude==="string"&&typeof p.schema==="string"&&typeof p.activation==="string"&&Array.isArray(p.console)&&p.console.every((t)=>typeof t==="string"&&t.length>0&&!/[\n\r]/.test(t));
if(!ok)throw new Error("pointer fields are missing or malformed");
process.stdout.write([p.copy,p.business,p.pm?1:0,p.claude,p.schema,p.activation,...p.console].join("\n")+"\n");' -- "$POINTER")" || { echo "stack-release: $POINTER is unreadable; Jason or Mos settles it" >&2; exit 1; }
    local -a f
    mapfile -t f <<< "$fields"
    P_COPY="${f[0]}" P_BUSINESS="${f[1]}" P_PM="${f[2]}" P_CLAUDE="${f[3]}" P_SCHEMA="${f[4]}" P_ACT="${f[5]}"
    P_CONSOLE=("${f[@]:6}")
}

# Writes the pointer through a temp file in state/ and a rename. Args: copy pm claude schema console...
write_pointer() {
    local tmp
    tmp="$(mktemp "$STATE/.stack-active.XXXXXX")"
    node -e '
const [copy,business,pm,claude,schema,activation,at,...console]=process.argv.slice(1);
process.stdout.write(JSON.stringify({copy,business,pm:pm==="1",claude,console,schema,activation,at})+"\n");' -- \
        "$1" "$BUSINESS" "$2" "$3" "$4" "$ACT" "$(now)" "${@:5}" > "$tmp"
    chmod 0644 "$tmp"
    mv -f "$tmp" "$POINTER"
}

# Prints the activation id of each `started` line with no final line after it.
interrupted() {
    [ -f "$LOG" ] || return 0
    node -e '
const open=new Map();
for(const l of require("fs").readFileSync(process.argv[1],"utf8").split("\n")){
  if(!l.trim())continue;
  let o;try{o=JSON.parse(l)}catch{continue}
  if(!o.activation)continue;
  if(o.result==="started")open.set(o.activation,o);else open.delete(o.activation);
}
for(const k of open.keys())process.stdout.write(k+"\n");' -- "$LOG"
}

# Reads the log's ok lines, newest first. `previous` prints rollback's target: the first `to`
# that is not the pointer's copy (release.sh's rule). `backup` prints the backup field of the
# line that made the pointer's copy live, the backup a rollback would restore.
from_log() {
    [ -f "$LOG" ] || return 1
    node -e '
const [file,mode,current]=process.argv.slice(1);
const lines=require("fs").readFileSync(file,"utf8").split("\n").filter((l)=>l.trim());
for(const l of lines.reverse()){
  let o;try{o=JSON.parse(l)}catch{continue}
  if(o.result!=="ok"||!["adopt","activate","rollback"].includes(o.op)||!o.to)continue;
  if(mode==="previous"&&o.to!==current){process.stdout.write(o.to+"\n");process.exit(0)}
  if(mode==="backup"&&o.to===current){if(o.backup)process.stdout.write(o.backup+"\n");process.exit(o.backup?0:1)}
}
process.exit(1);' -- "$LOG" "$1" "$P_COPY"
}

# --- Reading the units ---

show() { systemctl --user show "$1" -p LoadState -p WorkingDirectory -p ExecStart -p Environment -p MainPID -p DropInPaths 2>/dev/null; }
argv_of() { sed -n 's/^ExecStart=.*argv\[\]=\([^;]*\) ;.*/\1/p' <<< "$1"; }

# Sets U_PM U_CLAUDE U_CONSOLE (array) from the effective units; returns 1 if a unit can't be read.
read_units() {
    local bus console tok
    bus="$(show "$BUS_UNIT")" && console="$(show "$CONSOLE_UNIT")" || return 1
    [ "$(sed -n 's/^LoadState=//p' <<< "$bus")" = loaded ] && [ "$(sed -n 's/^LoadState=//p' <<< "$console")" = loaded ] || return 1
    local -a argv
    read -r -a argv <<< "$(argv_of "$bus")"
    U_PM=0
    for tok in "${argv[@]:1}"; do [ "$tok" != --pm ] || U_PM=1; done
    U_CLAUDE=""
    read -r -a argv <<< "$(sed -n 's/^Environment=//p' <<< "$bus")"
    for tok in "${argv[@]}"; do case "$tok" in MOSAIC_CLAUDE_BIN=*) U_CLAUDE="${tok#MOSAIC_CLAUDE_BIN=}" ;; esac; done
    read -r -a argv <<< "$(argv_of "$console")"
    U_CONSOLE=()
    local i serve=0
    for ((i = 0; i < ${#argv[@]}; i++)); do
        if [ "$serve" = 0 ]; then [ "${argv[i]}" != serve ] || serve=1; continue; fi
        if [ "${argv[i]}" = --checkout ]; then i=$((i + 1)); continue; fi
        U_CONSOLE+=("${argv[i]}")
    done
    [ "$serve" = 1 ] || return 1
}

console_port() {
    local i
    for ((i = 0; i < ${#P_CONSOLE[@]}; i++)); do
        [ "${P_CONSOLE[i]}" != --port ] || { echo "${P_CONSOLE[i + 1]:-7332}"; return; }
    done
    echo 7332
}

# --- Copies ---

# stack-copy.sh's own helpers, in a subshell so its die, parse and globals stay out of this script.
copy_ok() ( source "$REPO/scripts/stack-copy.sh"; ID="$1"; COPIES="$COPIES_DIR"; copy_root >/dev/null && reachable "$ID" )
copy_exists() ( source "$REPO/scripts/stack-copy.sh"; ID="$1"; COPIES="$COPIES_DIR"; copy_root >/dev/null )

# Prints the copy's SCHEMA_VERSION, then each `from:` of its migration chain, one per line. Text
# only: the copy's code is never imported or run for this.
copy_schema() {
    local f="$COPIES/$1/packages/bus/src/store.mjs" v
    [ -f "$f" ] || return 1
    v="$(grep -E "^export const SCHEMA_VERSION = '[A-Za-z0-9]+';\$" "$f")" || return 1
    [ "$(wc -l <<< "$v")" = 1 ] || return 1
    v="${v#*\'}"; echo "${v%%\'*}"
    { grep -E "^[[:space:]]*from: '[A-Za-z0-9]+',\$" "$f" || true; } | sed "s/.*from: '\\([A-Za-z0-9]*\\)',\$/\\1/"
}

# Sets T_SCHEMA and T_CHAIN (array) for copy $1; returns 1 when either read fails.
target_schema() {
    local out
    out="$(copy_schema "$1")" || return 1
    local -a l
    mapfile -t l <<< "$out"
    T_SCHEMA="${l[0]}"
    T_CHAIN=("${l[@]:1}")
}

in_chain() { local v; for v in "${T_CHAIN[@]}"; do [ "$v" != "$1" ] || return 0; done; return 1; }

# The item 3 checks on the target copy, in order: copy, schema, already-active.
check_target() {
    copy_exists "$TO" 2>/dev/null || refuse copy-missing "no copy $TO in $COPIES"
    copy_ok "$TO" 2>/dev/null || refuse copy-unreachable "$TO is not reachable from refactor and origin/refactor"
    target_schema "$TO" || refuse copy-schema-unknown "cannot read SCHEMA_VERSION and its chain from $COPIES/$TO/packages/bus/src/store.mjs"
    [ "$TO" != "$P_COPY" ] || refuse already-active "$TO is the pointer's copy"
}

# --- The store ---

STORE_JS='
const {DatabaseSync}=require("node:sqlite");const {pathToFileURL}=require("node:url");
const [mode,file,stamp]=process.argv.slice(1);
// immutable=1 on a stopped store with no -wal or -shm: it creates no file and leaves the main
// file mtime and sha256 as they were (the row 162 probe). A plain readOnly open of a stopped
// WAL store creates -shm and -wal, so it is used only on a store the running bus holds.
const db=mode.startsWith("held")?new DatabaseSync(file,{readOnly:true}):new DatabaseSync(pathToFileURL(file).href+"?immutable=1",{readOnly:true});
try{
  if(mode.endsWith("version"))process.stdout.write(db.prepare("SELECT value FROM meta WHERE key=?").get("schema_version").value+"\n");
  else process.stdout.write(db.prepare("SELECT count(*) AS n FROM events WHERE at > ?").get(stamp).n+"\n");
}finally{db.close()}'

# Sets UNCLEAN and returns 0 when a -wal or -shm sits beside the store. An immutable read would
# skip that WAL and could give a stale answer, so the caller refuses store-unclean and leaves the
# store as it is. Called directly, never inside $(...), so UNCLEAN reaches the caller.
store_unclean() {
    local f
    for f in "$STORE-wal" "$STORE-shm"; do
        if [ -e "$f" ] || [ -L "$f" ]; then UNCLEAN="$f"; return 0; fi
    done
    return 1
}
# A read of the stopped store: version, or count STAMP. Call store_unclean first.
store_read() { node -e "$STORE_JS" -- "$1" "$STORE" "${2:-}"; }

# No writer.lock, and no process with a file open under <dataRoot>/bus/ (stopped-check.sh's
# test). One find over the /proc fd links, not a readlink per link: this host has about 16,000.
# find's status is ignored, since other users' fd dirs are unreadable; so were they in the loop.
stopped() {
    [ ! -e "$DATA/bus/writer.lock" ] || return 1
    local pat
    pat="$(sed 's/[][*?\\]/\\&/g' <<< "$DATA/bus")/*"
    [ -z "$(find /proc/[0-9]*/fd -mindepth 1 -maxdepth 1 -lname "$pat" -print -quit 2>/dev/null || true)" ]
}

sums() { (cd "$1" && find . -type f ! -name SHA256SUMS ! -name steps.log -print0 | sort -z | xargs -0 -r sha256sum); }
sum_of() { awk -v f="./$2" '$2 == f { print $1 }' "$1/SHA256SUMS"; }

# The backup of the stopped store (backup.sh's steps). Only bus.sqlite: a -wal or -shm was
# already refused as store-unclean. Sets TAKEN once the backup is whole.
backup_store() {
    local o="$DATA/backups/stack/$ACT"
    (
        umask 077
        mkdir -p "$DATA/backups/stack" &&
        mkdir -m 700 "$o" "$o/bus" &&
        (cd "$DATA/bus" && sha256sum bus.sqlite) > "$o/SOURCE.sha256" &&
        cp -p "$STORE" "$o/bus/" &&
        (cd "$o/bus" && sha256sum bus.sqlite) > "$o/SHA256SUMS" &&
        cmp -s "$o/SOURCE.sha256" "$o/SHA256SUMS" &&
        stat -c '%n %a %U %s %y' "$STORE" > "$o/MODES.txt" &&
        ls -1A "$DATA/bus" > "$o/LISTING.txt" &&
        echo "$DATA" > "$o/DATA_ROOT" &&
        now_ms > "$o/STAMP" &&
        echo "$STORE_SCHEMA" > "$o/SCHEMA" &&
        echo "$RS" > "$o/ROLLBACK_SET"
    ) || return 1
    stopped || return 1
    TAKEN="$ACT" BACKUP="$ACT"
    say "backup: $o"
}

# Puts backup $1 back as restore.sh does: the current bus/ moves to $POST, nothing is deleted.
# Sets RESTORED once the old bus/ has moved, so the caller knows the store was swapped.
restore_store() {
    local o="$DATA/backups/stack/$1"
    [ "$(cat "$o/DATA_ROOT" 2>/dev/null)" = "$DATA" ] || return 1
    stopped || return 1
    (cd "$o/bus" && sha256sum -c --quiet ../SHA256SUMS) || return 1
    [ ! -e "$POST" ] || return 1
    mv "$DATA/bus" "$POST" || return 1
    RESTORED=1
    (umask 077; mkdir -m 700 "$DATA/bus") || return 1
    cp -p "$o"/bus/* "$DATA/bus/" || return 1
    (cd "$DATA/bus" && sha256sum -c --quiet "$o/SHA256SUMS") || return 1
    [ "$(ls -1A "$DATA/bus")" = "$(ls -1A "$o/bus")" ] || return 1
    say "restored $o; the store from before the restore is at $POST"
}

# --- The run (item 4) ---

# Step 1: the rollback set. A failure here has changed nothing but the set itself.
capture() {
    mkdir -p -m 0700 "$STATE/stack-rollback" 2>/dev/null || return 1
    mkdir -m 0700 "$RS" 2>/dev/null || return 1
    STEPS="$RS/steps.log"
    local u d
    for u in mosaic-bus@.service "$CONSOLE_UNIT"; do
        [ -f "$UNIT_DIR/$u" ] || { say "capture: no $UNIT_DIR/$u"; return 1; }
        cp -p "$UNIT_DIR/$u" "$RS/$u" || return 1
    done
    for d in mosaic-bus@.service.d "mosaic-bus@$BUSINESS.service.d" "$CONSOLE_UNIT.d"; do
        [ ! -d "$UNIT_DIR/$d" ] || cp -pr "$UNIT_DIR/$d" "$RS/$d" || return 1
    done
    [ -f "$BIN/mosaic" ] || { say "capture: no wrapper $BIN/mosaic"; return 1; }
    cp -p "$BIN/mosaic" "$RS/wrapper" || return 1
    stat -c %a "$BIN/mosaic" > "$RS/WRAPPER_MODE" || return 1
    for u in "$BUS_UNIT" "$CONSOLE_UNIT"; do show "$u" > "$RS/show-$u.txt" || return 1; done
    sums "$RS" > "$RS/SHA256SUMS" || return 1
    say "captured the rollback set: $RS"
}

# Stops the Console, then the bus, and waits for both and the stopped check, within --timeout.
stop_both() {
    systemctl --user stop "$CONSOLE_UNIT" || return 1
    systemctl --user stop "$BUS_UNIT" || return 1
    local deadline=$((SECONDS + TIMEOUT)) u s down
    while :; do
        down=1
        for u in "$CONSOLE_UNIT" "$BUS_UNIT"; do
            s="$(systemctl --user is-active "$u" 2>/dev/null || true)"
            case "$s" in inactive|failed) ;; *) down=0 ;; esac
        done
        if [ "$down" = 1 ] && stopped; then say "stopped both units"; return 0; fi
        [ "$SECONDS" -lt "$deadline" ] || return 1
        sleep 1
    done
}

start_both() {
    START_EPOCH="$(date +%s)" && START_MS="$(date +%s%3N)" || return 1
    systemctl --user start "$BUS_UNIT" && systemctl --user start "$CONSOLE_UNIT"
}

# Step 4: installs copy $1 with pm $2, claude $3 and the Console flags after them. Sets FAIL.
install_stack() {
    local id="$1" pm="$2" claude="$3"; shift 3
    local -a pmflag=()
    [ "$pm" = 0 ] || pmflag=(--pm)
    scripts/bus-service.sh install --copy "$id" --copies "$COPIES" "${pmflag[@]}" --claude "$claude" >> "$STEPS" 2>&1 || { FAIL=install-bus; return 1; }
    scripts/webui-service.sh install --copy "$id" --copies "$COPIES" "$@" >> "$STEPS" 2>&1 || { FAIL=install-console; return 1; }
    scripts/stack-copy.sh wrapper "$id" --copies "$COPIES" --bin "$BIN" >> "$STEPS" 2>&1 || { FAIL=wrapper; return 1; }
    say "installed $id"
}

# --- The health gate (item 5) ---

# One pass of every check against copy $1 with pm $2. Sets GATE to the first failing code and
# GATE_WHY to what it saw.
gate_once() {
    local copy="$COPIES/$1" pm="$2" main out
    GATE=units-inactive GATE_WHY=""
    systemctl --user is-active --quiet "$BUS_UNIT" && systemctl --user is-active --quiet "$CONSOLE_UNIT" || return 1
    GATE=bus-status
    main="$(systemctl --user show "$BUS_UNIT" -p MainPID | sed -n 's/^MainPID=//p')" || return 1
    out="$("$copy/scripts/mosaic" bus status --json 2>/dev/null)" || { GATE_WHY="bus status failed"; return 1; }
    # Every `bus start` writes boot.at (packages/cli/src/boot-status.mjs), so a value from before
    # this start means the host on the socket is not the one this run started.
    GATE_WHY="$(node -e '
const [b,main,since]=process.argv.slice(2),s=JSON.parse(process.argv[1]);
const why=!s.host?"no host":s.host.business!==b?`host for ${s.host.business}`:s.host.live!==true?"host not live":String(s.host.pid)!==main?`host pid ${s.host.pid} is not MainPID ${main}`:!s.boot||s.boot.ok!==true?"boot status not ok":s.boot.business!==b?`boot status for ${s.boot.business}`:!(Date.parse(s.boot.at)>=Number(since))?"boot status is from before this start":"";
process.stdout.write(why);' -- "$out" "$BUSINESS" "$main" "$START_MS" 2>&1)" || { GATE_WHY="bus status unreadable"; return 1; }
    [ -z "$GATE_WHY" ] || return 1
    GATE=healthz
    node -e '
fetch(`http://127.0.0.1:${process.argv[1]}/healthz`,{signal:AbortSignal.timeout(5000)}).then((r)=>process.exit(r.status===200?0:1),()=>process.exit(1));' -- "$(console_port)" || return 1
    GATE=copy-check
    scripts/stack-copy.sh check --copies "$COPIES" --bin "$BIN" --business "$BUSINESS" > /dev/null 2>&1 || return 1
    if [ "$pm" = 1 ]; then
        GATE=pm-launch-unconfirmed
        pm_launched "$copy" || return 1
    fi
    GATE=""
}

# Ruling 2 (sage-0736, sage-0738). The PM's run is the one named on the single
# `launched <instance>: run <run>, <harness> <model>` line that `bus start --pm` prints
# (packages/cli/src/cli.mjs) to the unit's journal since this start. The gate depends on that
# wording: if it changes, this check refuses rather than passes. --since takes @<epoch seconds>,
# because this host's journalctl reads a bare time as local time.
pm_launched() {
    local journal run list
    journal="$(journalctl --user -u "$BUS_UNIT" --since "@$START_EPOCH" -o cat 2>/dev/null)" || { GATE_WHY="journal read failed"; return 1; }
    local -a lines=()
    mapfile -t lines < <(grep '^launched ' <<< "$journal" || true)
    [ "${#lines[@]}" -gt 0 ] || { GATE_WHY="no launched line since the start"; return 1; }
    [ "${#lines[@]}" = 1 ] || { GATE_WHY="${#lines[@]} launched lines since the start"; return 1; }
    [[ "${lines[0]}" =~ ^launched\ ([^[:space:]]+):\ run\ ([^[:space:]]+),\  ]] || { GATE_WHY="the launched line does not parse: ${lines[0]}"; return 1; }
    run="${BASH_REMATCH[2]}"
    list="$("$1/scripts/mosaic" launches list --json 2>/dev/null)" || { GATE_WHY="launches list failed"; return 1; }
    GATE_WHY="$(node -e '
const [list,run,b,since]=process.argv.slice(1),s=JSON.parse(list).find((x)=>x.run===run);
process.stdout.write(!s?`run ${run} is not in launches list`:s.business!==b?`run ${run} is for ${s.business}`:s.live!==true?`run ${run} is not live`:!(Date.parse(s.startedAt)>=Number(since))?`run ${run} started before this start`:"");' -- "$list" "$run" "$BUSINESS" "$START_MS" 2>&1)" || { GATE_WHY="launches list unreadable"; return 1; }
    [ -z "$GATE_WHY" ]
}

# Polls every second until every check passes or --timeout runs out.
gate() {
    local deadline=$((SECONDS + TIMEOUT))
    while :; do
        if gate_once "$1" "$2"; then say "gate passed on $1"; return 0; fi
        [ "$SECONDS" -lt "$deadline" ] || { say "gate failed on $1: $GATE${GATE_WHY:+ ($GATE_WHY)}"; return 1; }
        sleep 1
    done
}

# --- Rollback inside the run (item 6) ---

# Puts back the captured unit files, drop-ins and wrapper, checked against SHA256SUMS.
put_back() {
    local u f rel tmp
    for u in mosaic-bus@.service "$CONSOLE_UNIT"; do
        tmp="$(mktemp "$UNIT_DIR/.$u.XXXXXX")" && cp -p "$RS/$u" "$tmp" && mv -f "$tmp" "$UNIT_DIR/$u" || return 1
    done
    for u in mosaic-bus@.service.d "mosaic-bus@$BUSINESS.service.d" "$CONSOLE_UNIT.d"; do
        [ -d "$RS/$u" ] || continue
        while IFS= read -r -d '' f; do
            rel="${f#"$RS/$u/"}"
            ! cmp -s "$f" "$UNIT_DIR/$u/$rel" 2>/dev/null || continue
            mkdir -p "$(dirname "$UNIT_DIR/$u/$rel")" || return 1
            tmp="$(mktemp "$UNIT_DIR/$u/.restore.XXXXXX")" && cp -p "$f" "$tmp" && mv -f "$tmp" "$UNIT_DIR/$u/$rel" || return 1
        done < <(find "$RS/$u" -type f -print0)
    done
    tmp="$(mktemp "$BIN/.mosaic.XXXXXX")" && cp -p "$RS/wrapper" "$tmp" && chmod "$(cat "$RS/WRAPPER_MODE")" "$tmp" && mv -f "$tmp" "$BIN/mosaic" || return 1
    for u in mosaic-bus@.service "$CONSOLE_UNIT"; do
        [ "$(sha256sum < "$UNIT_DIR/$u" | cut -d' ' -f1)" = "$(sum_of "$RS" "$u")" ] || return 1
    done
    [ "$(sha256sum < "$BIN/mosaic" | cut -d' ' -f1)" = "$(sum_of "$RS" wrapper)" ] || return 1
    [ "$(stat -c %a "$BIN/mosaic")" = "$(cat "$RS/WRAPPER_MODE")" ] || return 1
    systemctl --user daemon-reload || return 1
    say "put back the unit files and the wrapper from $RS"
}

# A failure after step 2: put the pointer's stack back and gate it. $1 is the run's code. Only a
# backup this run took (TAKEN) is restored; a rollback's --restore backup never is.
rollback_in_run() {
    trap - INT TERM HUP
    local code="$1" rcode=""
    say "rolling back: $code"
    if ! stop_both; then rcode=stop-timeout
    elif ! put_back; then rcode=put-back
    elif [ -n "$TAKEN" ] && ! restore_store "$TAKEN"; then rcode=restore
    elif ! start_both; then rcode=start
    elif ! gate "$P_COPY" "$P_PM"; then rcode="$GATE"
    fi
    if [ -z "$rcode" ]; then
        log_line result gate-failed code "$code" rollback ok
        echo "stack-release: $code; rolled back to $P_COPY, which passed the gate" >&2
        exit 1
    fi
    rollback_failed "$code" "$rcode"
}

# Ruling 4 (T14b): once rollback --restore has swapped the store, any later failure skips item 6,
# since putting the newer store back would be a second swap. Before the swap, item 6 runs.
after_fail() {
    [ -z "$RESTORED" ] || rollback_failed "$1" not-attempted
    rollback_in_run "$1"
}

# Stop both units, print both results, log rollback-failed, exit 2. No retry.
rollback_failed() {
    trap - INT TERM HUP
    systemctl --user stop "$CONSOLE_UNIT" || true
    systemctl --user stop "$BUS_UNIT" || true
    say "rollback-failed: run $1, rollback $2"
    log_line result rollback-failed code "$1" rollback failed rollbackCode "$2"
    echo "stack-release: rollback-failed: the run failed with $1 and the rollback with $2; both units are stopped. Rollback set: $RS${BACKUP:+, backup $DATA/backups/stack/$BACKUP}${RESTORED:+, post-attempt store $POST}" >&2
    exit 2
}

# --- Refusals (item 3) ---

lock() {
    mkdir -p "$STATE"
    exec 9> "$STATE/stack-release.lock"
    flock -n 9 || refuse run-locked "another stack-release run holds $STATE/stack-release.lock"
    local open
    open="$(interrupted | paste -sd' ' -)"
    [ -z "$open" ] || refuse interrupted-run "activation $open started and has no final line; Jason or Mos settles it (status shows it)"
}

instance_busy() {
    local out unit load state rest
    out="$(systemctl --user list-units --all --no-legend --plain 'mosaic-bus@*.service' 2>/dev/null)" || refuse instance-busy "systemctl --user list-units failed, so the bus state is unknown"
    while read -r unit load state rest; do
        [[ "$unit" == mosaic-bus@*.service ]] || continue
        if [ "$unit" = "$BUS_UNIT" ]; then
            case "$state" in activating|deactivating|reloading) refuse instance-busy "$unit is $state" ;; esac
        else
            case "$state" in inactive|failed) ;; *) refuse instance-busy "$unit is $state; the copy and wrapper switch reach it too" ;; esac
        fi
    done <<< "$out"
}

# install_guard's drop-in scan (bus-service.sh), done before anything stops.
drop_in_override() {
    local paths directory dropdir file
    paths="$(systemd-analyze --user unit-paths 2>/dev/null)" && [ -n "$paths" ] || refuse drop-in-override "systemd unit paths are unknown"
    while IFS= read -r directory; do
        [ -n "$directory" ] && [ -d "$directory" ] || continue
        for dropdir in "$directory"/mosaic-bus@*.service.d "$directory"/mosaic-.service.d "$directory"/service.d; do
            [ -d "$dropdir" ] || continue
            for file in "$dropdir"/*.conf; do
                [ -e "$file" ] || continue
                if grep -Eq '^[[:space:]]*(ExecStart|WorkingDirectory)[[:space:]]*=' "$file" 2>/dev/null; then
                    refuse drop-in-override "$file overrides ExecStart or WorkingDirectory; the operator removes it as in TOOLS.md step 4"
                fi
            done
        done
    done <<< "$paths"
}

# unit-flags-drift: the effective units against the pointer, and --pm against the pointer's pm.
flags_drift() {
    local d=""
    [ "$U_PM" = "$P_PM" ] || d+=$'\n'"  pm: pointer $P_PM, unit $U_PM"
    [ "$U_CLAUDE" = "$P_CLAUDE" ] || d+=$'\n'"  claude: pointer $P_CLAUDE, unit $U_CLAUDE"
    [ "${P_CONSOLE[*]}" = "${U_CONSOLE[*]}" ] || d+=$'\n'"  console: pointer ${P_CONSOLE[*]}, unit ${U_CONSOLE[*]}"
    [ "$OP" != activate ] || [ "$WANT_PM" = "$P_PM" ] || d+=$'\n'"  pm: pointer $P_PM, --pm given $WANT_PM"
    [ -z "$d" ] || refuse unit-flags-drift "the units differ from the pointer; settle it first:$d"
}

# The item 3 checks after the pointer, shared by activate and rollback, for target $TO.
precheck() {
    check_target
    instance_busy
    drop_in_override
    read_units || refuse unit-flags-drift "cannot read $BUS_UNIT or $CONSOLE_UNIT"
    [ -n "$U_CLAUDE" ] || refuse no-claude-pin "$BUS_UNIT has no MOSAIC_CLAUDE_BIN"
    flags_drift
}

pointer_for_business() {
    read_pointer || refuse no-pointer "no $POINTER; run adopt first"
    FROM="$P_COPY"
    [ "$P_BUSINESS" = "$BUSINESS" ] || refuse business-mismatch "the pointer is for $P_BUSINESS"
}

new_activation() {
    ACT="$(date -u +%Y%m%dT%H%M%SZ)-${TO:0:12}"
    RS="$STATE/stack-rollback/$ACT"
    POST="$DATA/bus.post-attempt-$(date -u +%Y%m%dT%H%M%SZ)"
}

# --- Verbs ---

do_adopt() {
    lock
    [ ! -f "$POINTER" ] || { read_pointer; FROM="$P_COPY"; refuse already-adopted "$POINTER exists (copy $P_COPY)"; }
    local out
    out="$(scripts/stack-copy.sh check --copies "$COPIES" --bin "$BIN" --business "$BUSINESS" 2>&1)" || refuse check-failed "stack-copy.sh check did not pass:"$'\n'"$out"
    TO="$(sed -n 's/^bus: //p' <<< "$out")"
    TO="${TO#"$COPIES/"}"
    read_units || refuse check-failed "cannot read $BUS_UNIT or $CONSOLE_UNIT"
    [ -n "$U_CLAUDE" ] || refuse no-claude-pin "$BUS_UNIT has no MOSAIC_CLAUDE_BIN"
    target_schema "$TO" || refuse copy-schema-unknown "cannot read SCHEMA_VERSION from $COPIES/$TO/packages/bus/src/store.mjs"
    ACT="$(date -u +%Y%m%dT%H%M%SZ)-${TO:0:12}"
    write_pointer "$TO" "$U_PM" "$U_CLAUDE" "$T_SCHEMA" "${U_CONSOLE[@]}"
    log_line result ok
    echo "adopted $TO for $BUSINESS (pm $U_PM, schema $T_SCHEMA)"
}

# The store's version for --plan, from the store itself: readOnly while the bus holds it (adds no
# file), immutable when it is stopped and clean. Sets S_SCHEMA and S_FROM. Falls back to the
# pointer's schema, and says so, when there is no store file or it can't be read.
plan_store() {
    local src="" why=""
    S_SCHEMA="" S_FROM=""
    if [ ! -e "$STORE" ]; then why="there is no store file"
    elif systemctl --user is-active --quiet "$BUS_UNIT"; then
        src="the store the running bus holds"
        S_SCHEMA="$(node -e "$STORE_JS" -- heldversion "$STORE" 2>/dev/null)" || S_SCHEMA=""
    elif store_unclean; then why="the store was not read: $UNCLEAN is present"
    else
        src="the stopped store"
        S_SCHEMA="$(store_read version 2>/dev/null)" || S_SCHEMA=""
    fi
    if [ -n "$S_SCHEMA" ]; then S_FROM="read from $src"
    else S_SCHEMA="$P_SCHEMA" S_FROM="from the pointer; ${why:-$src could not be read}"
    fi
}

# sage-0774 and sage-0778: the commits the run adds (activate) or takes away (rollback), so the
# rows they carry can be named. Copies where neither is an ancestor of the other refuse the plan
# before it prints anything. Sets RANGE and VERB.
plan_range() {
    if git -C "$REPO" merge-base --is-ancestor "$P_COPY" "$TO" 2>/dev/null; then RANGE="$P_COPY..$TO" VERB=adds
    elif git -C "$REPO" merge-base --is-ancestor "$TO" "$P_COPY" 2>/dev/null; then RANGE="$TO..$P_COPY" VERB="takes away"
    else refuse copies-unrelated "neither $P_COPY nor $TO is an ancestor of the other, so the plan cannot list the commits"
    fi
}
plan_commits() {
    echo "commits: the run $VERB $(git -C "$REPO" rev-list --count --no-merges "$RANGE") (newest first, at most 200):"
    git -C "$REPO" log --oneline --no-merges --max-count=200 "$RANGE" | sed 's/^/  /'
}

# Each step from $S_SCHEMA to $T_SCHEMA, from the target store.mjs's `from:` and `to:` pairs. Text
# only, like copy_schema. Prints "cannot be read from FILE" when the lines don't pair up or the
# steps don't reach $T_SCHEMA.
plan_steps() {
    local f="$COPIES/$TO/packages/bus/src/store.mjs" v="$S_SCHEMA" out="" i
    local -a l
    local -A next=()
    mapfile -t l < <(grep -E "^[[:space:]]*(from|to): '[A-Za-z0-9]+',\$" "$f" | sed -E "s/^[[:space:]]*(from|to): '([A-Za-z0-9]*)',\$/\\1 \\2/")
    (( ${#l[@]} % 2 == 0 )) || { echo "cannot be read from $f"; return; }
    for ((i = 0; i < ${#l[@]}; i += 2)); do
        [[ "${l[i]}" == "from "* && "${l[i+1]}" == "to "* ]] || { echo "cannot be read from $f"; return; }
        next["${l[i]#from }"]="${l[i+1]#to }"
    done
    for ((i = 0; i < ${#next[@]}; i++)); do
        [ "$v" != "$T_SCHEMA" ] && [ -n "${next[$v]:-}" ] || break
        out+="${out:+, }$v to ${next[$v]}"; v="${next[$v]}"
    done
    if [ "$v" = "$T_SCHEMA" ] && [ -n "$out" ]; then echo "$out"; else echo "cannot be read from $f"; fi
}

# Item 8 and sage-0745: what the run would do, with any migration named plainly. Reads the
# pointer, the log, the target's store.mjs and the store's version. Writes nothing.
plan() {
    plan_range
    plan_store
    echo "plan: $OP for $BUSINESS"
    echo "from: $P_COPY"
    echo "to: $TO"
    echo "store schema: $S_SCHEMA ($S_FROM)"
    echo "target schema: $T_SCHEMA"
    [ "$S_SCHEMA" = "$P_SCHEMA" ] || echo "note: the pointer says $P_SCHEMA; the run reads the stopped store and goes by that"
    if [ "$S_SCHEMA" = "$T_SCHEMA" ]; then
        echo "migration: none; the store is at the target's version"
    elif [ "$OP" = activate ] && in_chain "$S_SCHEMA"; then
        echo "migration: $S_SCHEMA to $T_SCHEMA, after a backup of the stopped store"
        echo "migration steps: $(plan_steps)"
        echo "needs: --migrate$([ "$MIGRATE" = 1 ] && echo ' (given)' || echo ' (not given; the run would roll back with migration-needs-flag)')"
        echo "needs: Jason present for this run (decision 96 condition 2)"
    elif [ "$OP" = activate ]; then
        echo "migration: none possible; $S_SCHEMA is not in $T_SCHEMA's migration chain, so the run would roll back with store-newer"
    else
        echo "restore: the store must go back to $T_SCHEMA; needs --restore $(from_log backup || echo ACTIVATION)"
    fi
    plan_commits
    echo "bus: scripts/bus-service.sh install --copy $TO --copies $COPIES$([ "$P_PM" = 0 ] || echo ' --pm') --claude $P_CLAUDE"
    echo "Console: scripts/webui-service.sh install --copy $TO --copies $COPIES${P_CONSOLE[*]:+ ${P_CONSOLE[*]}}"
    echo "CLI: scripts/stack-copy.sh wrapper $TO --copies $COPIES --bin $BIN"
}

# Steps 4.1 to 4.5 and the gate for target $TO. $1 is the store step: migrate or restore.
run_steps() {
    log_line result started
    capture || refuse capture-failed "could not write the rollback set $RS; nothing was stopped"
    trap 'after_fail signal' INT TERM HUP
    stop_both || rollback_in_run stop-timeout
    if [ -e "$STORE" ]; then
        if store_unclean; then say "$UNCLEAN is present; the store is left as it is for Jason or Mos"; rollback_in_run store-unclean; fi
        STORE_SCHEMA="$(store_read version)" && [ -n "$STORE_SCHEMA" ] || rollback_in_run schema-unreadable
        say "stopped store schema: $STORE_SCHEMA"
        if [ "$1" = migrate ]; then
            if [ "$STORE_SCHEMA" != "$T_SCHEMA" ]; then
                in_chain "$STORE_SCHEMA" || rollback_in_run store-newer
                [ "$MIGRATE" = 1 ] || rollback_in_run migration-needs-flag
                backup_store || rollback_in_run backup
                node "$COPIES/$TO/packages/bus/src/migrate-schema.mjs" "$DATA" >> "$STEPS" 2>&1 || rollback_in_run migrate-failed
                say "migrated the store to $T_SCHEMA"
            fi
        else
            [ "$STORE_SCHEMA" = "$P_SCHEMA" ] || rollback_in_run store-differs
            if [ -n "$RESTORE" ]; then
                local n
                n="$(store_read count "$(cat "$DATA/backups/stack/$RESTORE/STAMP")")" || rollback_in_run schema-unreadable
                [ "$n" = "$COUNT" ] || rollback_in_run post-backup-events
                restore_store "$RESTORE" || after_fail restore
            fi
        fi
    elif [ -n "$RESTORE" ]; then
        rollback_in_run store-differs
    fi
    install_stack "$TO" "$P_PM" "$P_CLAUDE" "${P_CONSOLE[@]}" || after_fail "$FAIL"
    start_both || after_fail start
    gate "$TO" "$P_PM" || after_fail "$GATE"
    trap - INT TERM HUP
    write_pointer "$TO" "$P_PM" "$P_CLAUDE" "$T_SCHEMA" "${P_CONSOLE[@]}"
    log_line result ok
    echo "stack-release: $OP ok: $TO is live for $BUSINESS"
}

do_activate() {
    TO="$ID"
    [ "$PLAN" = 1 ] || lock
    pointer_for_business
    if [ "$PLAN" = 1 ]; then check_target; plan; exit 0; fi
    precheck
    new_activation
    run_steps migrate
}

do_rollback() {
    [ "$PLAN" = 1 ] || lock
    pointer_for_business
    TO="$(from_log previous)" || refuse no-previous "the log names no earlier copy than $P_COPY"
    if [ "$PLAN" = 1 ]; then check_target; plan; exit 0; fi
    precheck
    [ "$P_SCHEMA" = "$T_SCHEMA" ] || [ -n "$RESTORE" ] || refuse schema-differs "the store is at $P_SCHEMA and $TO expects $T_SCHEMA; pass --restore ACTIVATION"
    new_activation
    if [ -n "$RESTORE" ]; then
        local o="$DATA/backups/stack/$RESTORE" rc=0
        [ -f "$o/SHA256SUMS" ] && [ "$(cat "$o/DATA_ROOT" 2>/dev/null)" = "$DATA" ] && [ "$(cat "$o/SCHEMA" 2>/dev/null)" = "$T_SCHEMA" ] && [ -f "$o/STAMP" ] \
            || refuse restore-mismatch "$o is not a backup of this root at $T_SCHEMA"
        # Before anything stops. A store the running bus holds is read with readOnly, which adds
        # no file (the probe); a stopped one falls under the store_unclean rule.
        if systemctl --user is-active --quiet "$BUS_UNIT"; then
            COUNT="$(node -e "$STORE_JS" -- heldcount "$STORE" "$(cat "$o/STAMP")")" || rc=1
        else
            ! store_unclean || refuse store-unclean "$UNCLEAN is present; the bus did not close the store cleanly"
            COUNT="$(store_read count "$(cat "$o/STAMP")")" || rc=1
        fi
        [ "$rc" = 0 ] || refuse restore-mismatch "cannot count the events after $(cat "$o/STAMP")"
        echo "restore: $o; the store now in place moves to $POST and is kept; $COUNT event(s) were written after the backup"
        [ "$COUNT" = 0 ] || [ "$EVENTS" = "$COUNT" ] || refuse post-backup-events "$COUNT event(s) were written after the backup; Jason decides, then --events $COUNT"
        BACKUP="$RESTORE"
    fi
    run_steps restore
}

do_status() {
    echo "pointer: $POINTER"
    if read_pointer; then sed 's/^/  /' "$POINTER"; else echo "  none"; fi
    echo "log (last five lines):"
    if [ -f "$LOG" ]; then tail -n 5 "$LOG" | sed 's/^/  /'; else echo "  none"; fi
    local open
    open="$(interrupted | paste -sd' ' -)"
    [ -z "$open" ] || echo "interrupted run: $open (started, no final line; Jason or Mos settles it)"
    echo "check:"
    if scripts/stack-copy.sh check --copies "$COPIES" --bin "$BIN" --business "$BUSINESS" 2>&1 | sed 's/^/  /'; then echo "  result: pass"; else echo "  result: FAIL"; fi
    [ -n "$P_COPY" ] || return 0
    echo "bus status:"
    "$COPIES/$P_COPY/scripts/mosaic" bus status 2>&1 | sed 's/^/  /' || true
    if read_units; then
        local d=""
        [ "$U_PM" = "$P_PM" ] || d+=" pm"
        [ "$U_CLAUDE" = "$P_CLAUDE" ] || d+=" claude"
        [ "${P_CONSOLE[*]}" = "${U_CONSOLE[*]}" ] || d+=" console"
        echo "unit flags: ${d:+differ from the pointer:}${d:-match the pointer}"
    else
        echo "unit flags: cannot read the units"
    fi
    if git -C "$REPO" merge-base --is-ancestor "$P_COPY" refs/heads/refactor 2>/dev/null; then
        echo "drift: $(git -C "$REPO" rev-list --count "$P_COPY..refs/heads/refactor") commit(s) on refactor since the pointer's copy"
    else
        echo "drift: the pointer's copy is not an ancestor of refactor"
    fi
}

# --- Dispatch ---

[ $# -gt 0 ] || usage
OP="$1"; shift
ID=""
case "$OP" in
    activate) [ $# -ge 2 ] || usage; ID="$1"; BUSINESS="$2"; shift 2; [[ "$ID" =~ ^[0-9a-f]{40}$ ]] || usage ;;
    adopt|rollback|status) [ $# -ge 1 ] || usage; BUSINESS="$1"; shift ;;
    -h|--help) usage 0 ;;
    *) usage ;;
esac
[[ "$BUSINESS" =~ ^[a-z0-9][a-z0-9-]*$ ]] || usage
parse_opts "$@"
COPIES_DIR="$COPIES"
load_config
DATA="$MOSAIC_DATA_ROOT"
STATE="$DATA/state"
POINTER="$STATE/stack-active.json"
LOG="$STATE/stack-activation-log.jsonl"
STORE="$DATA/bus/bus.sqlite"
BUS_UNIT="mosaic-bus@$BUSINESS.service"
FROM="" TO="" ACT="" RS="" POST="" BACKUP="" TAKEN="" RESTORED="" STEPS="" COUNT="" UNCLEAN="" FAIL=""
STORE_SCHEMA="" START_EPOCH="" START_MS=0 GATE="" GATE_WHY=""
P_COPY="" P_PM=0 P_SCHEMA="" P_CLAUDE="" P_CONSOLE=()
case "$OP" in
    adopt) do_adopt ;;
    activate) do_activate ;;
    rollback) do_rollback ;;
    status) do_status ;;
esac