#!/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.
# Console tester listener flags follow target support (row 176, #1674); so does the row 178 proof
# pair, which activate sets or drops (row 187, #1685). activate --no-send drops --send (row 189, #1687).
# activate --sso swaps the preview set and the pair for the Console's own sign-in (row 200, #1698).
# activate --send turns send on, only beside --sso, and activate --no-sso turns --sso off with the
# Console flags from before it, dropping --send (row 273, #1778, mos-1246 F39).
# 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.
# rehearse (D2, row 163) stops and starts no unit and writes only its own rehearsal directory, but
# it reads the live config and a backup of the live store, so Jason or Mos runs it.
#
# --units external (row 232, #1734; r6 section 4 at agents/darkwing/work/queue-195/DESIGN.md) is
# for a host whose units are root's system units. This script then reads those units and never
# starts or stops one: root stops them, activate or rollback stages the copy (A3), root installs
# and starts them, and confirm gates them and writes the pointer (A5). revert and confirm
# --rollback put the old copy back when a step fails (R2, R4), and gate checks a copy root
# restarted with no activation open (A2f). Without it, everything runs as before.
#
# 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
source scripts/instance.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"
TIMEOUT=60 WANT_PM=0 AFTER_HARNESS=0 MIGRATE=0 PLAN=0 RESTORE="" EVENTS="" TESTER_PORT="" TESTER_SOCKET=0 EXPECTED_INTERPRETER=""
PROOF_HEADER="" PROOF_FILE="" NO_PROOF=0 NO_SEND=0 SEND=0 SSO=0 NO_SSO=0 R_BACKUP="" R_PORT=""
UNITS=user TO_FLAG="" EXPECT_FROM="" SINCE="" ROLLBACK_OLD="" CAUSE=""

usage() {
    cat >&2 <<'USAGE'
usage: stack-release.sh adopt BUSINESS
       stack-release.sh activate ID BUSINESS [--pm | --pm-after-harness] [--migrate] [--tester-socket | --tester-port N (legacy copies)]
           [--proof-header NAME --proof-file PATH | --no-proof | --sso | --no-sso] [--send | --no-send] [--timeout S] [--plan]
       stack-release.sh pm-on BUSINESS [--timeout S]
       stack-release.sh rollback BUSINESS [--restore ACTIVATION [--events N]] [--timeout S] [--plan]
       stack-release.sh status BUSINESS
       stack-release.sh rehearse ID BUSINESS --backup ACTIVATION|PATH --port N [--timeout S]
       stack-release.sh confirm ID BUSINESS --units external [--rollback OLD --cause A3|A4|A5] [--timeout S]
       stack-release.sh revert ID BUSINESS --units external
       stack-release.sh gate ID BUSINESS --units external --since EPOCH [--timeout S]
common options: --copies DIR, --bin DIR, --instance NAME, --units user|external
--instance NAME (row 190) runs on stack instance NAME: its config, data root, units and the wrapper
mosaic-NAME. Without it, the default instance, as before.
--events N is the total of the per-table counts the post-backup-events refusal prints: events and
every table in COUNTED.
--units external (row 232) reads root's system units and starts or stops none. adopt, activate,
rollback, status, confirm, revert and gate take it; rollback then needs --to ID, the previous copy.
activate and rollback --plan take --expect-from OLD. --pm, --pm-after-harness, --tester-port, the
proof flags, --no-proof, --send, --no-send, --sso, --no-sso and pm-on are not for it.
--send needs --sso in the run's Console flags, given or carried. --no-sso needs --sso in the
pointer's, and takes the Console flags from the rollback set of the activation that turned it on.
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() {
    [ "$UNITS" = user ] || set -- "$@" units external
    jline at "$(now)" op "$OP" from "${FROM:-@null}" to "${TO:-@null}" backup "${BACKUP:-@null}" activation "${ACT:-@null}" "$@" >> "$LOG"
}

# --- The units backend (row 232) ---

# Every systemctl, journalctl and systemd-analyze call goes through these three. user is the
# systemd user manager, as before. external reads root's system units, which needs no privilege,
# and refuses any verb that would change one.
sc() {
    if [ "$UNITS" = user ]; then systemctl --user "$@"; return; fi
    case "$1" in
        show|is-active|list-units) systemctl "$@" ;;
        *) echo "stack-release: --units external never runs systemctl $1; root does" >&2; return 1 ;;
    esac
}
# The bus unit's journal is a user journal only: under external it lands in the system journal,
# which the stack's user can't read (r6 section 4), so the external PM check reads launches list.
jctl() { [ "$UNITS" = user ] || return 1; journalctl --user "$@"; }
analyze() { [ "$UNITS" = user ] || return 1; systemd-analyze --user "$@"; }

# 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
}

# Reject tunnel/main ports and anything other than a whole high port (exit 4 via usage).
valid_tester_port() {
    [[ "$1" =~ ^[0-9]{1,5}$ ]] && [ "$((10#$1))" -ge 1024 ] && [ "$((10#$1))" -le 65535 ] &&
        [ "$((10#$1))" -ne 7332 ] && [ "$((10#$1))" -ne "$2" ] || return 1
    # Row 190: a named instance's Console never takes 7330, 7332 or 7333.
    local r
    for r in "${INSTANCE_RESERVED[@]}"; do [ "$((10#$1))" -ne "$r" ] || return 1; done
}

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 ] && [ "$OP" != revert ] && [ $# -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 ;;
            --pm-after-harness) [ "$OP" = activate ] || usage; AFTER_HARNESS=1; shift ;;
            --tester-socket) [ "$OP" = activate ] || usage; TESTER_SOCKET=1; shift ;;
            --tester-port) [ "$OP" = activate ] && [ $# -ge 2 ] && [ -n "$2" ] || usage; TESTER_PORT="$2"; shift 2 ;;
            --proof-header) [ "$OP" = activate ] && [ $# -ge 2 ] && [ -n "$2" ] || usage; PROOF_HEADER="$2"; shift 2 ;;
            --proof-file) [ "$OP" = activate ] && [ $# -ge 2 ] && [ -n "$2" ] || usage; PROOF_FILE="$2"; shift 2 ;;
            --no-proof) [ "$OP" = activate ] || usage; NO_PROOF=1; shift ;;
            --no-send) [ "$OP" = activate ] || usage; NO_SEND=1; shift ;;
            --send) [ "$OP" = activate ] || usage; SEND=1; shift ;;
            --sso) [ "$OP" = activate ] || usage; SSO=1; shift ;;
            --no-sso) [ "$OP" = activate ] || usage; NO_SSO=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}(-[1-9])?$ ]] || usage; RESTORE="$2"; shift 2 ;;
            --backup) [ "$OP" = rehearse ] && [ $# -ge 2 ] && { [ "$2" = last ] || [[ "$2" =~ ^[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}(-[1-9])?$ ]] || [[ "$2" =~ ^/[A-Za-z0-9/._+-]*$ ]]; } || usage; R_BACKUP="$2"; shift 2 ;;
            --port) [ "$OP" = rehearse ] && [ $# -ge 2 ] && [[ "$2" =~ ^[0-9]{1,5}$ ]] && [ "$((10#$2))" -ge 1024 ] && [ "$((10#$2))" -le 65535 ] || usage; R_PORT="$((10#$2))"; shift 2 ;;
            --events) [ "$OP" = rollback ] && [ $# -ge 2 ] && [[ "$2" =~ ^(0|[1-9][0-9]*)$ ]] || usage; EVENTS="$2"; shift 2 ;;
            --units) [ "$OP" != rehearse ] && [ $# -ge 2 ] && { [ "$2" = user ] || [ "$2" = external ]; } || usage; UNITS="$2"; shift 2 ;;
            --to) [ "$OP" = rollback ] && [ $# -ge 2 ] && [[ "$2" =~ ^[0-9a-f]{40}$ ]] || usage; TO_FLAG="$2"; shift 2 ;;
            --expect-from) { [ "$OP" = activate ] || [ "$OP" = rollback ]; } && [ $# -ge 2 ] && [[ "$2" =~ ^[0-9a-f]{40}$ ]] || usage; EXPECT_FROM="$2"; shift 2 ;;
            --since) [ "$OP" = gate ] && [ $# -ge 2 ] && [[ "$2" =~ ^(0|[1-9][0-9]{0,11})$ ]] || usage; SINCE="$2"; shift 2 ;;
            --rollback) [ "$OP" = confirm ] && [ $# -ge 2 ] && [[ "$2" =~ ^[0-9a-f]{40}$ ]] || usage; ROLLBACK_OLD="$2"; shift 2 ;;
            --cause) [ "$OP" = confirm ] && [ $# -ge 2 ] && [[ "$2" =~ ^A[345]$ ]] || usage; CAUSE="$2"; shift 2 ;;
            *) usage ;;
        esac
    done
    # Row 232: confirm, revert, gate, --to and --expect-from are external only, and an external
    # rollback names its target. gate needs --since, its start floor (r6 N2), and --expect-from
    # goes with --plan. The flags that change a unit's render or the PM are root's under external.
    if [ "$UNITS" = user ]; then
        [ "$OP" != confirm ] && [ "$OP" != revert ] && [ "$OP" != gate ] && [ -z "$TO_FLAG" ] && [ -z "$EXPECT_FROM" ] || usage
    else
        [ "$OP" != rollback ] || [ -n "$TO_FLAG" ] || usage
        [ "$TESTER_SOCKET" = 0 ] && [ "$WANT_PM" = 0 ] && [ -z "$TESTER_PORT" ] && [ -z "$PROOF_HEADER" ] && [ -z "$PROOF_FILE" ] && [ "$NO_PROOF" = 0 ] && [ "$NO_SEND" = 0 ] && [ "$SSO" = 0 ] &&
            [ "$SEND" = 0 ] && [ "$NO_SSO" = 0 ] || usage
    fi
    [ "$OP" != gate ] || [ -n "$SINCE" ] || usage
    [ -z "$EXPECT_FROM" ] || [ "$PLAN" = 1 ] || usage
    [ "${ROLLBACK_OLD:+1}" = "${CAUSE:+1}" ] || usage
    [ "$TESTER_SOCKET" = 0 ] || [ -z "$TESTER_PORT" ] || usage
    [ -z "$EVENTS" ] || [ -n "$RESTORE" ] || usage
    [ "$OP" != rehearse ] || { [ -n "$R_BACKUP" ] && [ -n "$R_PORT" ]; } || usage
    # A bad tester port is a usage error; --plan lists it as tester-port-invalid instead (row 180).
    [ -z "$TESTER_PORT" ] || [ "$PLAN" = 1 ] || valid_tester_port "$TESTER_PORT" 7332 || usage
    # --pm-after-harness replaces --pm (row 169 item 1), so the two never go together.
    [ "$WANT_PM" = 0 ] || [ "$AFTER_HARNESS" = 0 ] || usage
    # The proof pair goes together, and --no-proof goes alone (row 187).
    [ "${PROOF_HEADER:+1}" = "${PROOF_FILE:+1}" ] || usage
    [ "$NO_PROOF" = 0 ] || [ -z "$PROOF_HEADER" ] || usage
    # --sso drops the pair itself, so it takes neither a pair nor --no-proof (row 200).
    [ "$SSO" = 0 ] || { [ -z "$PROOF_HEADER" ] && [ "$NO_PROOF" = 0 ]; } || usage
    # Row 273: --send and --no-send contradict. --no-sso takes its Console flags from a rollback set
    # and drops --send, so it goes with no --sso, --send, pair or --no-proof.
    [ "$SEND" = 0 ] || [ "$NO_SEND" = 0 ] || usage
    [ "$NO_SSO" = 0 ] || { [ "$SSO" = 0 ] && [ "$SEND" = 0 ] && [ -z "$PROOF_HEADER" ] && [ "$NO_PROOF" = 0 ]; } || 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_PENDING 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.
# pmPending (row 169) is missing or true, and true only beside pm false.
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"&&(p.pmPending===undefined||(p.pmPending===true&&p.pm===false))&&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.pmPending?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_PENDING="${f[3]}" P_CLAUDE="${f[4]}" P_SCHEMA="${f[5]}" P_ACT="${f[6]}"
    P_CONSOLE=("${f[@]:7}")
}

# Writes the pointer through a temp file in state/ and a rename. Args: copy pm pending claude schema
# console... A pending of 1 writes pmPending true (row 169); 0 leaves the key out.
write_pointer() {
    local tmp
    tmp="$(mktemp "$STATE/.stack-active.XXXXXX")"
    node -e '
const [copy,business,pm,pending,claude,schema,activation,at,...console]=process.argv.slice(1);
process.stdout.write(JSON.stringify({copy,business,pm:pm==="1",...(pending==="1"?{pmPending:true}:{}),claude,console,schema,activation,at})+"\n");' -- \
        "$1" "$BUSINESS" "$2" "$3" "$4" "$5" "$ACT" "$(now)" "${@:6}" > "$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. `newest` prints the
# backup field of the newest line that has one, rehearse's --backup last (row 233). `arrival` prints the
# activation and `from` of the line that moved the stack onto the pointer's copy from another one,
# past any same-copy re-pin after it (row 200, sage-1389). `sets` prints the activation of each
# activate and rollback, newest first, back to the adopt line, whose run took no rollback set (row 273).
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==="sets"){if(o.op==="adopt")process.exit(0);process.stdout.write(`${o.activation??""}\n`);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)}
  if(mode==="newest"&&o.backup){process.stdout.write(o.backup+"\n");process.exit(0)}
  if(mode==="arrival"&&o.to===current&&o.from!==current){process.stdout.write(`${o.activation??""} ${o.from??""}\n`);process.exit(0)}
}
process.exit(mode==="sets"?0:1);' -- "$LOG" "$1" "$P_COPY"
}

# --- Reading the units ---

show() { sc 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"; }
# systemctl show quotes a value that holds a space and escapes with a backslash, and the read -a
# split below can't undo either, so such a line is unreadable rather than read wrong (row 166 item 4).
unquoted() { [[ "$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_argv console_argv env
    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
    bus_argv="$(argv_of "$bus")" console_argv="$(argv_of "$console")" env="$(sed -n 's/^Environment=//p' <<< "$bus")"
    unquoted "$bus_argv" && unquoted "$console_argv" || return 1
    unquoted "$env" || return 1
    local -a argv
    read -r -a argv <<< "$bus_argv"
    U_PM=0
    for tok in "${argv[@]:1}"; do [ "$tok" != --pm ] || U_PM=1; done
    U_CLAUDE=""
    read -r -a argv <<< "$env"
    for tok in "${argv[@]}"; do case "$tok" in MOSAIC_CLAUDE_BIN=*) U_CLAUDE="${tok#MOSAIC_CLAUDE_BIN=}" ;; esac; done
    serve_flags "$console_argv" || return 1
    U_CONSOLE=("${SERVE_FLAGS[@]}")
}

# Sets SERVE_FLAGS (array) to the Console argv $1 after `serve`, without --checkout and its value;
# returns 1 with no `serve`.
serve_flags() {
    local -a argv
    read -r -a argv <<< "$1"
    SERVE_FLAGS=()
    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
        SERVE_FLAGS+=("${argv[i]}")
    done
    [ "$serve" = 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. An activate that
# sets a proof pair, drops the pointer's pair with --no-proof, drops the pointer's --send with
# --no-send, turns on --sso, turns on --send, or turns off --sso may name the pointer's copy: it
# re-pins the live copy with new Console flags (row 187, Part B's B3 and B-R1; row 189, mos-0972;
# row 200, the phase 2 window; row 273). --no-sso on a pointer without --sso gets past here and
# refuses sso-not-on in no_sso_base, which names the reason.
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" ] || [ -n "$PROOF_HEADER" ] || [ "$NO_SSO" = 1 ] ||
        { [ "$NO_PROOF" = 1 ] && [[ " ${P_CONSOLE[*]} " == *" --proof-header "* ]]; } ||
        { [ "$NO_SEND" = 1 ] && [[ " ${P_CONSOLE[*]} " == *" --send "* ]]; } ||
        { [ "$SEND" = 1 ] && [[ " ${P_CONSOLE[*]} " != *" --send "* ]]; } ||
        { [ "$SSO" = 1 ] && [[ " ${P_CONSOLE[*]} " != *" --sso "* ]]; } ||
        refuse already-active "$TO is the pointer's copy"
}

# --- Plug-in kinds (row 169) ---

# Unlike copy_schema, these import the copy's own code, as a business-check does: its kinds.mjs
# (no imports) and, for `harness`, its business.mjs loader. Modes, for copy $2:
#   has KIND  exits 0 when the copy's KINDS has KIND
#   unknown   prints the business file's plugins keys the copy's KINDS lacks, from a plain JSON read
#   harness   prints plugins.harness as JSON, or `none`, from the copy's loader
#   launchby  prints launch.by, the PM's instance, from a plain JSON read (row 232); it imports
#             only business.mjs, for the business file's path
# A failure exits 1 and prints its first line.
PLUGINS_JS='
const {readFileSync}=require("node:fs");const {pathToFileURL}=require("node:url");
const [mode,copy,id,kind]=process.argv.slice(1);
const imp=(p)=>import(pathToFileURL(`${copy}/packages/${p}`).href);
(async()=>{
  if(mode==="launchby"){
    const {businessFilePath}=await imp("business/src/business.mjs");
    const by=JSON.parse(readFileSync(businessFilePath(id),"utf8")).launch?.by;
    if(typeof by!=="string"||!by)throw new Error("the business file has no launch.by");
    return process.stdout.write(by);
  }
  const {KINDS}=await imp("plugins/src/kinds.mjs");
  if(mode==="has")process.exit(Object.hasOwn(KINDS,kind)?0:1);
  const b=await imp("business/src/business.mjs");
  if(mode==="unknown"){
    const p=JSON.parse(readFileSync(b.businessFilePath(id),"utf8")).plugins;
    if(p===undefined)return;
    if(p===null||typeof p!=="object"||Array.isArray(p))throw new Error("its plugins is not an object");
    process.stdout.write(Object.keys(p).filter((k)=>!Object.hasOwn(KINDS,k)).join(" "));
  }else{
    const h=b.loadBusiness(id,{rolesDir:`${copy}/roles`}).plugins.harness;
    process.stdout.write(h===undefined?"none":JSON.stringify(h));
  }
})().catch((e)=>{process.stdout.write(String(e?.message??e).split("\n")[0]);process.exit(1)});'
plugins() { node -e "$PLUGINS_JS" -- "$1" "$COPIES/$2" "$BUSINESS" "${3:-}"; }

# A refusal --plan lists and goes on from; outside --plan it is refuse, which exits. A caller
# returns after it, so the checks that depend on this one are skipped.
check_refuse() {
    [ "$PLAN" = 1 ] || refuse "$@"
    local code="$1"; shift
    PLAN_REFUSE+=("$code: $*")
}

# Item 1 and item 4, before anything stops, for target $TO: no plug-in kind the target lacks, then
# for --pm-after-harness the target's harness kind and no harness entry yet.
plugin_checks() {
    local out fix=""
    out="$(plugins unknown "$TO" 2>&1)" || { check_refuse business-unreadable "cannot read the business file's plugins with $TO's code: $out"; return 0; }
    if [ -n "$out" ]; then
        [ "$OP" != rollback ] || fix="; set each to null with plugins set through the live bus first"
        check_refuse unknown-kind-for-target "the business file names plug-in kind(s) $out, which $TO does not know$fix"
        return 0
    fi
    [ "$AFTER_HARNESS" = 1 ] || return 0
    plugins has "$TO" harness 2>/dev/null || { check_refuse pm-after-harness-target "$TO has no harness plug-in kind, so there is nothing to wait for; use --pm"; return 0; }
    out="$(plugins harness "$TO" 2>&1)" || { check_refuse business-unreadable "$TO's loader refuses the business file: $out"; return 0; }
    [ "$out" = none ] || check_refuse pm-after-harness-set "the business file already sets harness $out; use --pm"
}

# --- The store ---

# Row 181: the tables a bus writer fills without an event in the same transaction, in schema
# order. A restore drops their rows after STAMP as it drops events, so the guard counts them too.
# BUILD.md for row 181 names each writer; the T46 guard classifies every table in the schema.
COUNTED="role_claims thread_claims decision_events messages deliveries task_snapshots projects threads sessions session_defaults thread_opens instance_names usage_records"

STORE_JS='
const {DatabaseSync}=require("node:sqlite");const {pathToFileURL}=require("node:url");
const [mode,file,stamp,tables=""]=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{
    // "TOTAL<tab>TEXT": events, then each listed table with rows after STAMP, in list order. A
    // listed table the store lacks counts 0 (no instance_names before 3g, no usage_records before
    // 3h). One that is there but cannot be read throws, as a failed events count does.
    const ev=db.prepare("SELECT count(*) AS n FROM events WHERE at > ?").get(stamp).n;let total=ev,text=ev+" event(s)";
    for(const t of tables.split(" ").filter(Boolean)){
      if(!db.prepare("SELECT 1 FROM sqlite_master WHERE type=? AND name=?").get("table",t))continue;
      const n=db.prepare("SELECT count(*) AS n FROM "+t+" WHERE at > ?").get(stamp).n;
      if(n){total+=n;text+=", "+n+" "+t+" row(s)"}
    }
    process.stdout.write(total+"\t"+text+"\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 TABLES. Call store_unclean first.
store_read() { node -e "$STORE_JS" -- "$1" "$STORE" "${2:-}" "${3:-}"; }

# 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. Every activate with a store takes one (row 185). Sets BACKUP
# once the backup is whole, for the log line; TAKEN is set by the caller only before a migration.
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
    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 "$BUS_TEMPLATE" "$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
    # This instance's own directories only; the shared prefix ones are never captured (row 190).
    for d in "$BUS_TEMPLATE.d" "$BUS_UNIT.d" "$CONSOLE_UNIT.d"; do
        [ ! -d "$UNIT_DIR/$d" ] || cp -pr "$UNIT_DIR/$d" "$RS/$d" || return 1
    done
    [ -f "$BIN/$WRAPPER_NAME" ] || { say "capture: no wrapper $BIN/$WRAPPER_NAME"; return 1; }
    cp -p "$BIN/$WRAPPER_NAME" "$RS/wrapper" || return 1
    stat -c %a "$BIN/$WRAPPER_NAME" > "$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() {
    sc stop "$CONSOLE_UNIT" || return 1
    sc 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="$(sc 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
    sc start "$BUS_UNIT" && sc 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 "${INSTANCE_FLAGS[@]}" --copy "$id" --copies "$COPIES" "${pmflag[@]}" --claude "$claude" >> "$STEPS" 2>&1 || { FAIL=install-bus; return 1; }
    scripts/webui-service.sh install "${INSTANCE_FLAGS[@]}" --copy "$id" --copies "$COPIES" "$@" >> "$STEPS" 2>&1 || { FAIL=install-console; return 1; }
    scripts/stack-copy.sh wrapper "$id" "${INSTANCE_FLAGS[@]}" --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=""
    sc is-active --quiet "$BUS_UNIT" && sc is-active --quiet "$CONSOLE_UNIT" || return 1
    if [ "$UNITS" = external ]; then
        # r6 N4: root started the units, so this start's time is the bus unit's, in whole seconds,
        # at or after FLOOR. It is a gate code, so polling goes on: Restart=on-failure can bring a
        # fresh start.
        GATE=stale-start
        out="$(sc show --timestamp=unix -p ActiveEnterTimestamp "$BUS_UNIT" 2>/dev/null | sed -n 's/^ActiveEnterTimestamp=@//p')" || return 1
        [[ "$out" =~ ^[0-9]{1,12}$ ]] || { GATE_WHY="ActiveEnterTimestamp is unreadable"; return 1; }
        [ "$out" -ge "$FLOOR" ] || { GATE_WHY="the bus started at @$out, before @$FLOOR"; return 1; }
        START_MS=$((out * 1000))
    fi
    if [ -f "$copy/packages/bus/src/interpreter.mjs" ]; then
        GATE=interpreter-receipt
        node -e 'import(require("url").pathToFileURL(process.argv[1]).href).then(m=>{const expected=m.inspectInterpreter();if(process.argv[4]&&JSON.stringify(expected)!==process.argv[4])process.exit(1);m.verifyInterpreterReceipt(process.argv[2],expected,process.argv[3]);}).catch(()=>process.exit(1));' -- "$copy/packages/bus/src/interpreter.mjs" "$DATA/bus" "$START_MS" "$EXPECTED_INTERPRETER" || return 1
    fi
    GATE=bus-status
    main="$(sc 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
    out="$(scripts/stack-copy.sh check "${INSTANCE_FLAGS[@]}" "${UNITS_FLAGS[@]}" --copies "$COPIES" --bin "$BIN" --business "$BUSINESS" 2>/dev/null)" || return 1
    # check passes when the three agree on one copy; this run needs it to be the target (row 166 item 3).
    out="$(sed -n 's/^bus: //p' <<< "$out")"
    [ "$out" = "$copy" ] || { GATE_WHY="the bus, Console and CLI agree on $out, not $copy"; return 1; }
    if [ "$pm" = 1 ]; then
        GATE=pm-launch-unconfirmed
        if [ "$UNITS" = user ]; then pm_launched "$copy" || return 1; else pm_external "$1" || return 1; fi
    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="$(jctl -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" ]
}

# r6 section 4: the PM check under --units external, where the bus unit's journal is the system
# journal and unreadable here. It passes when exactly one live launches list entry is for the
# PM's instance (launch.by, packages/cli/src/launcher.mjs:675), and that entry is for BUSINESS and
# started at or after this start. A run the PM starts for another role has another instance.
pm_external() {
    local by list
    by="$(plugins launchby "$1" 2>&1)" || { GATE_WHY="cannot read launch.by: $by"; return 1; }
    list="$("$COPIES/$1/scripts/mosaic" launches list --json 2>/dev/null)" || { GATE_WHY="launches list failed"; return 1; }
    GATE_WHY="$(node -e '
const [list,by,b,since]=process.argv.slice(1),l=JSON.parse(list).filter((x)=>x.instance===by&&x.live===true),s=l[0];
process.stdout.write(l.length===0?`no live launch of ${by}`:l.length>1?`${l.length} live launches of ${by}`:s.business!==b?`the launch of ${by} is for ${s.business}`:!(Date.parse(s.startedAt)>=Number(since))?`run ${s.run} started before this start`:"");' -- "$list" "$by" "$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 "$BUS_TEMPLATE" "$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 "$BUS_TEMPLATE.d" "$BUS_UNIT.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/.$WRAPPER_NAME.XXXXXX")" && cp -p "$RS/wrapper" "$tmp" && chmod "$(cat "$RS/WRAPPER_MODE")" "$tmp" && mv -f "$tmp" "$BIN/$WRAPPER_NAME" || return 1
    for u in "$BUS_TEMPLATE" "$CONSOLE_UNIT"; do
        [ "$(sha256sum < "$UNIT_DIR/$u" | cut -d' ' -f1)" = "$(sum_of "$RS" "$u")" ] || return 1
    done
    [ "$(sha256sum < "$BIN/$WRAPPER_NAME" | cut -d' ' -f1)" = "$(sum_of "$RS" wrapper)" ] || return 1
    [ "$(stat -c %a "$BIN/$WRAPPER_NAME")" = "$(cat "$RS/WRAPPER_MODE")" ] || return 1
    sc 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 before a migration (TAKEN) is restored; a same-schema activate's backup
# and a rollback's --restore backup never are.
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
    sc stop "$CONSOLE_UNIT" || true
    sc 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) ---

# confirm and revert act on the open activation by design, so they take the lock alone (r6 section 4).
lock_only() {
    mkdir -p "$STATE"
    exec 9> "$STATE/stack-release.lock"
    flock -n 9 || refuse run-locked "another stack-release run holds $STATE/stack-release.lock"
}

lock() {
    lock_only
    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="$(sc list-units --all --no-legend --plain "$BUS_GLOB" 2>/dev/null)" || refuse instance-busy "$SC_NAME list-units failed, so the bus state is unknown"
    while read -r unit load state rest; do
        [[ "$unit" == $BUS_GLOB ]] || 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. A path it can't read
# refuses here, as install_guard would refuse it after the stop (row 166 item 1).
drop_in_override() {
    local paths directory dropdir file content glob
    local -a dropdirs
    paths="$(analyze unit-paths 2>/dev/null)" && [ -n "$paths" ] || refuse drop-in-override "systemd unit paths are unknown"
    while IFS= read -r directory; do
        [ -n "$directory" ] || continue
        case "$directory" in /*) ;; *) refuse drop-in-override "invalid systemd unit path $directory" ;; esac
        [ -e "$directory" ] || [ -L "$directory" ] || continue
        [ -d "$directory" ] && [ -r "$directory" ] && [ -x "$directory" ] || refuse drop-in-override "cannot inspect unit path $directory"
        dropdirs=()
        for glob in "${BUS_DROPIN_GLOBS[@]}"; do dropdirs+=("$directory"/$glob); done
        for dropdir in "${dropdirs[@]}"; do
            [ -e "$dropdir" ] || [ -L "$dropdir" ] || continue
            [ -d "$dropdir" ] && [ -r "$dropdir" ] && [ -x "$dropdir" ] || refuse drop-in-override "cannot inspect drop-ins $dropdir"
            for file in "$dropdir"/*.conf; do
                [ -e "$file" ] || [ -L "$file" ] || continue
                content="$(cat "$file" 2>/dev/null)" || refuse drop-in-override "cannot read drop-in $file"
                if grep -Eq '^[[:space:]]*(ExecStart|WorkingDirectory)[[:space:]]*=' <<< "$content"; 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.
# --pm-after-harness stands in for --pm (row 169 item 1). A pending pointer (pm false, unit
# without --pm) matches its unit; only pm-on and rollback get this far with one, since activate
# refuses pm-pending first and adopt refuses already-adopted.
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[*]}"
    if [ "$AFTER_HARNESS" = 1 ]; then
        [ "$P_PM" = 1 ] || d+=$'\n'"  pm: pointer $P_PM, --pm-after-harness needs a pointer with pm 1"
    else
        # Under --units external the PM is a flag in root's unit, so there is no --pm to compare.
        [ "$UNITS" != user ] || [ "$OP" != activate ] || [ "$WANT_PM" = "$P_PM" ] || d+=$'\n'"  pm: pointer $P_PM, --pm given $WANT_PM"
    fi
    [ -z "$d" ] || refuse unit-flags-drift "the units differ from the pointer; settle it first:$d"
}

# Row 178: --proof-header and --proof-file are a pair in P_CONSOLE, which proof_for_target (row
# 187) carries unchanged, replaces or drops; never read the proof value.
# Resolve desired flags independently of the pointer and effective units, so a hand-edited
# unit still refuses drift and a failed attempt can restore exactly the captured stack.
# Read the target's own Console, because old copies cannot start with the new flag.
console_for_target() {
    local supported=0 socket_supported=0 main=7332 inherited="" inherited_socket=0 i
    local cli="$COPIES/$TO/packages/webui/src/cli.mjs"
    local -a without_tester=()
    sso_base
    if grep -q 'MOSAIC_TESTER_SOCKET_V1' "$cli"; then socket_supported=1
    elif grep -q -- '--tester-port' "$cli"; then supported=1; fi
    for ((i = 0; i < ${#BASE_CONSOLE[@]}; i++)); do
        case "${BASE_CONSOLE[i]}" in
            --tester-port) i=$((i + 1)); inherited="${BASE_CONSOLE[i]:-}" ;;
            --tester-socket) inherited_socket=1 ;;
            *) without_tester+=("${BASE_CONSOLE[i]}")
               if [ "${BASE_CONSOLE[i]}" = --port ]; then main="${BASE_CONSOLE[i+1]:-7332}"; fi ;;
        esac
    done
    RUN_CONSOLE=("${BASE_CONSOLE[@]}") DROP_TESTER="" EXPECTED_INTERPRETER=""
    if [ "$socket_supported" = 1 ]; then
        [ -z "$TESTER_PORT" ] || check_refuse tester-port-target "$TO retires tester TCP; use --tester-socket"
        if [ -n "$inherited" ] && [ "$TESTER_SOCKET" = 0 ]; then
            check_refuse tester-transport-migration "inherited tester TCP requires explicit --tester-socket conversion"
        fi
        if [ "$TESTER_SOCKET" = 1 ] || [ "$inherited_socket" = 1 ]; then
            RUN_CONSOLE=("${without_tester[@]}" --tester-socket)
            [ -z "$inherited" ] || DROP_TESTER="--tester-port $inherited (replaced by private Unix socket)"
            node -e 'const fs=require("fs"),p=require("path"),root=process.argv[1],parent=p.join(root,"console");if(!p.isAbsolute(root)||Buffer.byteLength(p.join(parent,"tester.sock"))>107)process.exit(1);try{const s=fs.lstatSync(parent);if(!s.isDirectory()||s.isSymbolicLink()||s.uid!==process.getuid()||(s.mode&0o077))process.exit(1);}catch(e){if(e.code!=="ENOENT")process.exit(1);}' -- "$DATA" || check_refuse tester-socket-path "private tester path is too long or its parent is unsafe"
        fi
        local module="$COPIES/$TO/packages/bus/src/interpreter.mjs"
        [ -f "$module" ] || check_refuse interpreter-unsafe "$TO lacks its interpreter verifier"
        if [ -f "$module" ]; then
            EXPECTED_INTERPRETER="$(node -e 'import(require("url").pathToFileURL(process.argv[1]).href).then(m=>console.log(JSON.stringify(m.inspectInterpreter())),()=>process.exit(1)).catch(()=>process.exit(1))' -- "$module")" || check_refuse interpreter-unsafe "system Python preflight refused"
        fi
    elif [ "$TESTER_SOCKET" = 1 ]; then
        check_refuse tester-socket-target "$TO does not support --tester-socket"
    elif [ -n "$TESTER_PORT" ]; then
        if ! valid_tester_port "$TESTER_PORT" "$main"; then
            [ "$PLAN" = 1 ] || usage
            local other=""
            [ "$main" = 7332 ] || other=" and the Console's --port $main"
            [ "${#INSTANCE_RESERVED[@]}" = 0 ] || other+=" and instance $MOSAIC_INSTANCE's reserved ${INSTANCE_RESERVED[*]}"
            check_refuse tester-port-invalid "--tester-port $TESTER_PORT is not a whole port from 1024 to 65535 other than 7332$other"
        fi
        [ "$supported" = 1 ] || check_refuse tester-port-target "$TO's Console does not support --tester-port"
        RUN_CONSOLE=("${without_tester[@]}" --tester-port "$TESTER_PORT")
    elif [ "$inherited_socket" = 1 ]; then
        RUN_CONSOLE=("${without_tester[@]}") DROP_TESTER="--tester-socket (tester disabled; no TCP fallback)"
    elif [ -n "$inherited" ] && [ "$supported" = 0 ]; then
        RUN_CONSOLE=("${without_tester[@]}") DROP_TESTER="--tester-port $inherited"
    fi
    if [ "$SSO" = 1 ]; then sso_for_target; else proof_for_target; fi
    send_for_target
    sso_checks
}

sso_supported() { grep -q -- '--sso' "$COPIES/$TO/packages/webui/src/cli.mjs"; }
has_flag() { local f="$1"; shift; [[ " $* " == *" $f "* ]]; }

# Row 200 (#1698, R1): a rollback from an --sso pointer to a copy whose Console has no --sso. The
# pointer no longer holds the preview set that Console ran with, so the flags come from the rollback
# set of the activation that moved the stack onto the pointer's copy from the rollback target. A
# same-copy re-pin after it has a set of its own, holding the same copy's earlier flags, so the
# pointer's own activation is the wrong set (sage-1389). The set's show-$CONSOLE_UNIT.txt is
# the Console unit as it was before that run, checked against the set's SHA256SUMS. Sets
# BASE_CONSOLE (the flags console_for_target starts from), SSO_PLAN and SSO_STEP. SSO_NO_TARGET
# skips sso_checks after a refusal here, so --plan lists that refusal alone.
sso_base() {
    BASE_CONSOLE=("${P_CONSOLE[@]}") SSO_PLAN="" SSO_STEP="" SSO_NO_TARGET=""
    if [ "$NO_SSO" = 1 ]; then no_sso_base; return 0; fi
    has_flag --sso "${P_CONSOLE[@]}" || return 0
    [ "$SSO" = 1 ] || SSO_PLAN="sso: carried"
    [ "$OP" = rollback ] && ! sso_supported || return 0
    local f="show-$CONSOLE_UNIT.txt" why="" act="" from="" set
    read -r act from <<< "$(from_log arrival || true)"
    set="$STATE/stack-rollback/$act"
    if [ "$from" != "$TO" ]; then why="the log names no activation that moved the stack from $TO onto $P_COPY"
    elif ! [[ "$act" =~ ^[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}(-[1-9])?$ ]]; then why="the activation $act that moved the stack onto $P_COPY names no rollback set"
    elif ! set_flags "$set"; then why="$SET_WHY"
    elif has_flag --sso "${SERVE_FLAGS[@]}"; then why="the Console in $set/$f already ran with --sso"
    fi
    if [ -n "$why" ]; then
        SSO_NO_TARGET=1
        check_refuse sso-rollback-flags "$TO's Console does not support --sso, and the Console flags it ran with cannot be read: $why"
        return 0
    fi
    # --send and --tester-port follow the pointer, so a --no-send or a tester port set after that
    # activation stays (row 189's rule that rollback carries --send off). A --send carried here has
    # lost its --sso, so send_for_target drops it (row 273, F39 b).
    local i
    BASE_CONSOLE=()
    for ((i = 0; i < ${#SERVE_FLAGS[@]}; i++)); do
        case "${SERVE_FLAGS[i]}" in
            --send) ;;
            --tester-port) i=$((i + 1)) ;;
            --tester-socket) ;;
            *) BASE_CONSOLE+=("${SERVE_FLAGS[i]}") ;;
        esac
    done
    for ((i = 0; i < ${#P_CONSOLE[@]}; i++)); do
        case "${P_CONSOLE[i]}" in
            --send) BASE_CONSOLE+=(--send) ;;
            --tester-port) i=$((i + 1)); BASE_CONSOLE+=(--tester-port "${P_CONSOLE[i]:-}") ;;
            --tester-socket) BASE_CONSOLE+=(--tester-socket) ;;
        esac
    done
    SSO_PLAN="sso: dropped; $TO's Console does not support --sso, so the rollback takes the Console flags from $set/$f, with --tester-port as the pointer has it"
    SSO_STEP="drop --sso; $TO's Console does not support it, so the Console flags come from $set/$f"
}

# Sets SERVE_FLAGS to the Console flags in rollback set $1's show-$CONSOLE_UNIT.txt, the Console
# unit as it was before that set's run, checked against the set's SHA256SUMS. Returns 1 with
# SET_WHY when the file is missing, fails its sum or holds no argv this script can read.
set_flags() {
    local set="$1" f="show-$CONSOLE_UNIT.txt" argv
    SET_WHY=""
    if [ ! -f "$set/$f" ] || [ ! -f "$set/SHA256SUMS" ]; then SET_WHY="$set has no $f and SHA256SUMS"
    elif [ "$(sum_of "$set" "$f")" != "$(sha256sum < "$set/$f" | cut -d' ' -f1)" ]; then SET_WHY="$set/$f does not match the set's SHA256SUMS"
    else
        argv="$(argv_of "$(cat "$set/$f")")"
        unquoted "$argv" && serve_flags "$argv" && return 0
        SET_WHY="$set/$f holds no Console argv this script can read"
    fi
    return 1
}

# Row 273 (#1778, mos-1246 F39 a): activate --no-sso turns --sso off. The Console flags come from the
# rollback set of the activation that turned --sso on. Each set holds the Console from before its
# run, so that activation is the newest one whose set holds a Console without --sso: every newer
# set ran with it. The walk reads the log's activations newest first, back to the adopt line, and
# refuses at the first set it can't read, since it can't tell whether that set is the one. --send
# is dropped (F39 a) and the tester follows the pointer. Sets BASE_CONSOLE, SSO_PLAN and SSO_STEP.
no_sso_base() {
    local f="show-$CONSOLE_UNIT.txt" why="" act set="" found=0 i restored=""
    local -a acts=()
    if ! has_flag --sso "${P_CONSOLE[@]}"; then
        check_refuse sso-not-on "the pointer's Console flags hold no --sso, so --no-sso has nothing to turn off"
        return 0
    fi
    # An empty entry is an ok line with no activation; skipping it could land on an older set.
    mapfile -t acts < <(from_log sets || true)
    for act in "${acts[@]}"; do
        set="$STATE/stack-rollback/$act"
        if [ -z "$act" ]; then why="an ok line in the log names no activation"; break; fi
        if ! [[ "$act" =~ ^[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}(-[1-9])?$ ]]; then why="the activation $act names no rollback set"; break; fi
        set_flags "$set" || { why="$SET_WHY"; break; }
        has_flag --sso "${SERVE_FLAGS[@]}" || { found=1; break; }
    done
    if [ -z "$why" ] && [ "$found" = 0 ]; then
        if [ -z "$set" ]; then why="the log names no activation with a rollback set"
        else why="the Console in $set/$f already ran with --sso, and the log names no earlier activation"
        fi
    fi
    if [ -n "$why" ]; then
        SSO_NO_TARGET=1
        check_refuse sso-off-flags "the Console flags from before --sso cannot be read: $why"
        return 0
    fi
    # A --send in the set stays here: send_for_target is the one place that drops it, so the
    # restores list leaves it out.
    BASE_CONSOLE=()
    for ((i = 0; i < ${#SERVE_FLAGS[@]}; i++)); do
        case "${SERVE_FLAGS[i]}" in
            --tester-port) i=$((i + 1)) ;;
            --tester-socket) ;;
            *) BASE_CONSOLE+=("${SERVE_FLAGS[i]}")
               [ "${SERVE_FLAGS[i]}" = --send ] || restored+="${restored:+ }${SERVE_FLAGS[i]}" ;;
        esac
    done
    for ((i = 0; i < ${#P_CONSOLE[@]}; i++)); do
        case "${P_CONSOLE[i]}" in
            --tester-port) i=$((i + 1)); BASE_CONSOLE+=(--tester-port "${P_CONSOLE[i]:-}") ;;
            --tester-socket) BASE_CONSOLE+=(--tester-socket) ;;
        esac
    done
    SSO_PLAN="sso: off; Console flags from $set/$f; restores ${restored:-nothing}"
    SSO_STEP="turn off --sso; the Console flags come from $set/$f"
}

# Row 200 (#1698): activate --sso drops the preview set and the proof pair, which serve refuses beside
# it, and puts --sso where webui-service.sh renders it: after the other flags, before --send and
# --tester-port. Sets SSO_DROP (the plan's dropped preview flags), PROOF_PLAN and PROOF_STEP.
sso_for_target() {
    local header="" file="" i
    local -a kept=()
    SSO_DROP=""
    for ((i = 0; i < ${#RUN_CONSOLE[@]}; i++)); do
        case "${RUN_CONSOLE[i]}" in
            --public-host | --auth-header | --allow-user) SSO_DROP+=" ${RUN_CONSOLE[i]} ${RUN_CONSOLE[i + 1]:-}"; i=$((i + 1)) ;;
            --proof-header) i=$((i + 1)); header="${RUN_CONSOLE[i]:-}" ;;
            --proof-file) i=$((i + 1)); file="${RUN_CONSOLE[i]:-}" ;;
            --sso) ;;
            *) kept+=("${RUN_CONSOLE[i]}") ;;
        esac
    done
    RUN_CONSOLE=()
    for ((i = 0; i < ${#kept[@]}; i++)); do
        case "${kept[i]}" in --send | --tester-port | --tester-socket) break ;; esac
        RUN_CONSOLE+=("${kept[i]}")
    done
    RUN_CONSOLE+=(--sso "${kept[@]:i}")
    PROOF_PLAN="proof: off" PROOF_STEP=""
    if [ -n "$header" ]; then
        PROOF_PLAN="proof: dropped; --sso drops header $header"
        PROOF_STEP="drop --proof-header $header --proof-file $file; --sso was given"
    fi
}

# Row 200 item 2: whenever the run's Console flags hold --sso, the checks row 192's --sso start runs,
# before anything stops: a target Console with --sso, console.sso in config.json as the target's own
# checkConsole reads it, and each provider's slot file. The slot gets lstat, stat and realpath
# only; this script never opens it. Also sets the plan's sso: line for activate --sso.
SSO_JS='
const {readFileSync}=require("node:fs");const {pathToFileURL}=require("node:url");
const [copy,file]=process.argv.slice(1);
(async()=>{
  const {checkConsole}=await import(pathToFileURL(`${copy}/packages/webui/src/sso-config.mjs`).href);
  let doc;
  // A JSON.parse message can quote the text, so it is not passed on.
  try{doc=JSON.parse(readFileSync(file,"utf8"))}catch{throw new Error(`cannot read ${file} as JSON`)}
  if(!doc?.console?.sso)throw new Error("there is no console.sso in the system config");
  checkConsole(doc.console);
  for(const p of doc.console.sso.providers)process.stdout.write(`${p.id} ${p.clientSecret.slot}\n`);
})().catch((e)=>{process.stdout.write(String(e?.message??e).split("\n")[0]);process.exit(1)});'
sso_checks() {
    has_flag --sso "${RUN_CONSOLE[@]}" || return 0
    if [ "$SSO" = 1 ]; then
        local t carried="" dropped="${SSO_DROP# }"
        for t in "${RUN_CONSOLE[@]}"; do [ "$t" = --sso ] || carried+="${carried:+ }$t"; done
        SSO_PLAN="sso: on; drops ${dropped:-nothing}; carries ${carried:-nothing}"
        SSO_STEP="turn on --sso${SSO_DROP:+; drop$SSO_DROP}"
    fi
    [ -z "$SSO_NO_TARGET" ] || return 0
    sso_supported || { check_refuse sso-target "$TO's Console does not support --sso"; return 0; }
    local cfg="${MOSAIC_CONFIG:-$HOME/.config/mosaic-dev/config.json}" out id slot path why
    # N4 (webby-0019, sage-1403): a pre-phase-1 loader refuses a console key, so with the key not yet
    # written and the stack on another copy, the first step is the re-pin without --sso. On a named
    # instance both commands carry --instance NAME (row 238).
    local first="" inst="${INSTANCE_FLAGS[*]:+ ${INSTANCE_FLAGS[*]}}"
    if ! out="$(node -e "$SSO_JS" -- "$COPIES/$TO" "$cfg" 2>&1)"; then
        out="${out%%$'\n'*}"
        [ "$out" != "there is no console.sso in the system config" ] || [ "$P_COPY" = "$TO" ] ||
            first="; activate $TO$inst without --sso first, then write console.sso, then run activate $TO$inst --sso"
        check_refuse sso-config "$TO's Console would refuse its --sso start: $out$first"
        return 0
    fi
    while read -r id slot; do
        path="$(dirname "$cfg")/secrets/mosaic-stack/$slot"
        why="$(slot_problem "$path")"
        [ -z "$why" ] || check_refuse sso-slot "console.sso provider $id client secret slot $path $why"
    done <<< "$out"
}

# Why a slot file can't serve, or nothing: readSlot's checks in packages/webui/src/sso.mjs, in its
# order, from lstat, stat and realpath only, then a size of 0 for its one-non-empty-line rule.
slot_problem() {
    local st uid mode size type real copies repo
    [ -e "$1" ] || [ -L "$1" ] || { echo "is missing"; return; }
    [ ! -L "$1" ] || { echo "is a symbolic link"; return; }
    st="$(stat -c '%u:%a:%s:%F' -- "$1" 2>/dev/null)" || { echo "cannot be read with stat"; return; }
    uid="${st%%:*}" st="${st#*:}"
    mode="${st%%:*}" st="${st#*:}"
    size="${st%%:*}" type="${st#*:}"
    case "$type" in "regular file" | "regular empty file") ;; *) echo "is not a regular file"; return ;; esac
    [ "$uid" = "$(id -u)" ] || { echo "is owned by uid $uid, not $(id -u), the uid this script runs as"; return; }
    (( (8#$mode & 8#077) == 0 )) || { echo "has mode $mode, which gives group or other permission bits"; return; }
    # REPO is $(pwd), which keeps a symlink in the path the script ran by (webby-0028 N1).
    real="$(realpath -e -- "$1")" && copies="$(realpath -m -- "$COPIES")" && repo="$(realpath -m -- "$REPO")" || { echo "cannot be resolved"; return; }
    case "$real" in "$repo" | "$repo"/* | "$copies" | "$copies"/*) echo "is inside the repository or the copies"; return ;; esac
    [ "$size" != 0 ] || echo "is empty"
}

# Row 200 item 3 and the design's phase 2 rollback order: a config.json with a `console` key must be
# one the target's loader reads, or its bus, Console and seats fail on their next read. Runs the
# target's own scripts/mosaic-config.mjs validate; its output, the resolved config, is not kept.
config_checks() {
    local cfg="${MOSAIC_CONFIG:-$HOME/.config/mosaic-dev/config.json}" out loader="$COPIES/$TO/scripts/mosaic-config.mjs" first
    # 0: a console key; 1: no key, or no file; 2: the file doesn't parse, so whether it holds the key
    # is unknown, and the run refuses rather than pass it (webby-0028 N2). load_config already exits on
    # such a file before anything is read or stopped, so this is reached only if the file changes after it.
    local key=0
    node -e 'const fs=require("fs"),f=process.argv[1];if(!fs.existsSync(f))process.exit(1);let d;try{d=JSON.parse(fs.readFileSync(f,"utf8"))}catch{process.exit(2)}
process.exit(d!==null&&typeof d==="object"&&Object.hasOwn(d,"console")?0:1)' -- "$cfg" 2>/dev/null || key=$?
    [ "$key" != 1 ] || return 0
    if [ "$key" != 0 ]; then
        check_refuse config-console-key "$cfg cannot be read as JSON, so whether $TO's loader reads it is unknown; fix the config first, then run the $OP again"
        return 0
    fi
    if [ ! -f "$loader" ]; then out="it has no scripts/mosaic-config.mjs"
    elif out="$(MOSAIC_CONFIG="$cfg" node "$loader" validate 2>&1 >/dev/null)"; then return 0
    fi
    first="${out%%$'\n'*}"
    check_refuse config-console-key "$cfg holds a console key, and $TO's loader refuses it ($first); remove the key first (restore the config.json backup from the window's step 1), then run the $OP again"
}

# Row 189 (mos-0972): --no-send drops --send from RUN_CONSOLE, so the unit and the pointer both lose
# it; without the flag the pointer's --send goes on unchanged. Rollback and pm-on never take it, so
# they carry the pointer as it is. Sets SEND_PLAN (the plan's send: line) and SEND_STEP.
# Row 273 (mos-1246 F39): under forward auth any same-uid process can send as the signed-in user
# through the Console's port, so --send turns send on only when the run's Console flags hold --sso,
# given or carried. --no-sso drops the pointer's --send, and so does a rollback to a Console without
# --sso.
send_for_target() {
    local had=0 i
    local -a without_send=() with_send=()
    for ((i = 0; i < ${#RUN_CONSOLE[@]}; i++)); do
        if [ "${RUN_CONSOLE[i]}" = --send ]; then had=1; else without_send+=("${RUN_CONSOLE[i]}"); fi
    done
    SEND_STEP=""
    if [ "$NO_SSO" = 1 ]; then
        RUN_CONSOLE=("${without_send[@]}")
        if has_flag --send "${P_CONSOLE[@]}"; then
            SEND_PLAN="send: dropped; --no-sso drops --send (F39)"
            SEND_STEP="drop --send; --no-sso was given"
        else
            SEND_PLAN="send: off; --no-sso was given and the pointer holds no --send"
        fi
    elif has_flag --sso "${P_CONSOLE[@]}" && ! has_flag --sso "${RUN_CONSOLE[@]}"; then
        # Row 273 (F39 b): a rollback to a Console without --sso drops it, and --send goes with it.
        RUN_CONSOLE=("${without_send[@]}")
        if [ "$had" = 1 ]; then
            SEND_PLAN="send: dropped; the run drops --sso, and --send needs it (F39)"
            SEND_STEP="drop --send; the run drops --sso"
        else
            SEND_PLAN="send: off"
        fi
    elif [ "$SEND" = 1 ]; then
        if ! has_flag --sso "${RUN_CONSOLE[@]}"; then
            SEND_PLAN="send: off"
            check_refuse send-needs-sso "--send needs --sso in the run's Console flags, given or carried; under forward auth any same-uid process could send (F39)"
            return 0
        fi
        if [ "$had" = 1 ]; then
            SEND_PLAN="send: carried"
            return 0
        fi
        # webui-service.sh renders --send after --sso, before the tester flags.
        for ((i = 0; i < ${#RUN_CONSOLE[@]}; i++)); do
            case "${RUN_CONSOLE[i]}" in --tester-port | --tester-socket) break ;; esac
            with_send+=("${RUN_CONSOLE[i]}")
        done
        RUN_CONSOLE=("${with_send[@]}" --send "${RUN_CONSOLE[@]:i}")
        SEND_PLAN="send: on; --send was given"
        SEND_STEP="turn on --send; --send was given"
    elif [ "$NO_SEND" = 1 ]; then
        RUN_CONSOLE=("${without_send[@]}")
        if [ "$had" = 1 ]; then
            SEND_PLAN="send: dropped; --no-send was given"
            SEND_STEP="drop --send; --no-send was given"
        else
            SEND_PLAN="send: off; --no-send was given and the pointer holds no --send"
            SEND_STEP="--no-send was given and the pointer holds no --send, so there is nothing to drop"
        fi
    elif [ "$had" = 1 ]; then
        SEND_PLAN="send: carried"
    else
        SEND_PLAN="send: off"
    fi
}

# Row 187: the row 178 proof pair in RUN_CONSOLE. A new pair replaces the pointer's and goes last;
# --no-proof drops it. With neither, a target Console that takes --proof-header carries the pair
# unchanged. One that doesn't drops it on rollback, the way back, and refuses proof-target on
# activate, so only --no-proof turns the proof off. Sets PROOF_PLAN (the plan's proof: line) and
# PROOF_STEP (a steps log line, or empty).
proof_for_target() {
    local supported=0 had=0 header="" file="" i
    local -a without_proof=()
    grep -q -- '--proof-header' "$COPIES/$TO/packages/webui/src/cli.mjs" && supported=1
    for ((i = 0; i < ${#RUN_CONSOLE[@]}; i++)); do
        case "${RUN_CONSOLE[i]}" in
            --proof-header) had=1; i=$((i + 1)); header="${RUN_CONSOLE[i]:-}" ;;
            --proof-file) had=1; i=$((i + 1)); file="${RUN_CONSOLE[i]:-}" ;;
            *) without_proof+=("${RUN_CONSOLE[i]}") ;;
        esac
    done
    PROOF_PLAN="proof: off" PROOF_STEP=""
    if [ "$NO_PROOF" = 1 ]; then
        RUN_CONSOLE=("${without_proof[@]}")
        if [ "$had" = 1 ]; then
            PROOF_PLAN="proof: dropped; --no-proof drops header $header"
            PROOF_STEP="drop --proof-header $header --proof-file $file; --no-proof was given"
        else
            PROOF_PLAN="proof: off; --no-proof was given and the pointer holds no pair"
            PROOF_STEP="--no-proof was given and the pointer holds no proof pair, so there is nothing to drop"
        fi
        return 0
    fi
    if [ -n "$PROOF_HEADER" ]; then
        # webui-service.sh renders the pair after the preview flags and before --send and
        # tester flags, and flags_drift compares the pointer with that argv as it stands.
        header="$PROOF_HEADER" file="$PROOF_FILE" RUN_CONSOLE=()
        for ((i = 0; i < ${#without_proof[@]}; i++)); do
            case "${without_proof[i]}" in --send | --tester-port | --tester-socket) break ;; esac
            RUN_CONSOLE+=("${without_proof[i]}")
        done
        RUN_CONSOLE+=(--proof-header "$header" --proof-file "$file" "${without_proof[@]:i}")
        PROOF_PLAN="proof: on, header $header, file $file"
        [ "$supported" = 1 ] || check_refuse proof-target "$TO's Console does not support --proof-header"
    elif [ "$had" = 0 ]; then
        return 0
    elif [ "$supported" = 0 ] && [ "$OP" = rollback ]; then
        RUN_CONSOLE=("${without_proof[@]}")
        PROOF_PLAN="proof: dropped; $TO's Console does not support --proof-header, so the rollback drops header $header"
        PROOF_STEP="drop --proof-header $header --proof-file $file; the target Console does not support it"
        return 0
    else
        PROOF_PLAN="proof: carried, header $header"
        [ "$supported" = 1 ] || check_refuse proof-target "$TO's Console does not support --proof-header, which the pointer's Console flags carry; give --no-proof to drop the pair"
    fi
    proof_checks "$header" "$file"
}

# Row 187: proofOptions' own rules (packages/webui/src/serve.mjs), checked before anything stops so
# the Console can't fail its start on them. The file gets stat only: this script never opens it,
# and no line it writes holds the value.
proof_checks() {
    local header="${1,,}" auth="" seen=" " missing="" f i why
    for ((i = 0; i < ${#RUN_CONSOLE[@]}; i++)); do
        seen+="${RUN_CONSOLE[i]} "
        [ "${RUN_CONSOLE[i]}" != --auth-header ] || auth="${RUN_CONSOLE[i + 1]:-}"
    done
    for f in --public-host --auth-header --allow-user; do
        [[ "$seen" == *" $f "* ]] || missing+=" $f"
    done
    [ -z "$missing" ] || check_refuse proof-needs-preview "the proof pair needs --public-host, --auth-header and --allow-user in the Console flags; the run's flags lack$missing"
    if ! [[ "$header" =~ ^[a-z0-9][a-z0-9-]{0,63}$ ]] || [ "$header" = host ] || [ "$header" = origin ] ||
        [ "$header" = x-mosaic-tester ] || [ "$header" = "${auth,,}" ]; then
        check_refuse proof-header-invalid "--proof-header $1 must be a header name of a-z, 0-9 and - (64 at most, lowercased) other than host, origin, x-mosaic-tester and the --auth-header value"
    fi
    why="$(proof_file_problem "$2")"
    [ -z "$why" ] || check_refuse proof-file-invalid "--proof-file $2 $why"
}

# Why the proof file can't serve, or nothing. lstat, stat and realpath only; the order follows
# webui-service.sh render, which refuses the same files.
proof_file_problem() {
    local st uid mode type real copies repo
    case "$1" in /*) ;; *) echo "is not an absolute path"; return ;; esac
    case "$1" in *[!A-Za-z0-9/._+-]*) echo "holds a character outside [A-Za-z0-9/._+-]"; return ;; esac
    [ ! -L "$1" ] || { echo "is a symlink"; return; }
    st="$(stat -c '%u:%a:%F' -- "$1" 2>/dev/null)" || { echo "cannot be found"; return; }
    uid="${st%%:*}" st="${st#*:}"
    mode="${st%%:*}" type="${st#*:}"
    case "$type" in "regular file" | "regular empty file") ;; *) echo "is not a regular file"; return ;; esac
    real="$(realpath -e -- "$1")" && copies="$(realpath -m -- "$COPIES")" && repo="$(realpath -m -- "$REPO")" || { echo "cannot be resolved"; return; }
    case "$real" in "$repo" | "$repo"/* | "$copies" | "$copies"/*) echo "is inside the repository or the copies"; return ;; esac
    [ "$mode" = 600 ] || [ "$mode" = 400 ] || { echo "has mode $mode, not 600 or 400"; return; }
    [ "$uid" = "$(id -u)" ] || echo "is owned by uid $uid, not $(id -u), the uid this script runs as"
}

# Row 190: the target Console's --port (7332 when absent) and --tester-port against every other
# installed Console unit in UNIT_DIR, the default's and each named instance's.
ports_unclaimed() {
    local main=7332 i holder
    local -a ports=()
    for ((i = 0; i < ${#RUN_CONSOLE[@]}; i++)); do
        case "${RUN_CONSOLE[i]}" in
            --port) main="${RUN_CONSOLE[i + 1]:-7332}" ;;
            --tester-port) ports+=("${RUN_CONSOLE[i + 1]:-}") ;;
        esac
    done
    for i in "$main" "${ports[@]}"; do
        [[ "$i" =~ ^[0-9]{1,5}$ ]] || continue
        if holder="$(instance_port_holder "$UNIT_DIR" "$i")"; then
            check_refuse port-claimed "port $i is taken by $UNIT_DIR/$holder"
        fi
    done
}

# The item 3 checks after the pointer, shared by activate and rollback, for target $TO.
precheck() {
    check_target
    console_for_target
    ports_unclaimed
    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"
}

# The activation id is the second and the target's 12-character prefix. A second run in the same
# second adds -1 to -9, so ids still sort by time and an id without one is still whole (row 180).
# Callers hold the lock. Sets ACT and POST from one free suffix: no rollback set, backup or
# post-attempt store with that name.
next_activation() {
    local second s
    second="$(date -u +%Y%m%dT%H%M%SZ)"
    for s in "" -1 -2 -3 -4 -5 -6 -7 -8 -9; do
        ACT="$second-${TO:0:12}$s" POST="$DATA/bus.post-attempt-$second$s"
        [ -e "$STATE/stack-rollback/$ACT" ] || [ -e "$DATA/backups/stack/$ACT" ] || [ -e "$POST" ] || return 0
    done
    # Every name is another run's, so the refused line names none and closes no started run.
    ACT="" POST=""
    refuse capture-failed "every activation id for $second-${TO:0:12} is taken; nothing was stopped"
}

new_activation() {
    next_activation
    RS="$STATE/stack-rollback/$ACT"
}

# --- 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 "${INSTANCE_FLAGS[@]}" "${UNITS_FLAGS[@]}" --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"
    # Under --units external the claude pin is root's render, so adopt records what the unit has.
    [ "$UNITS" = external ] || [ -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"
    next_activation
    write_pointer "$TO" "$U_PM" 0 "$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="" S_CHANGED=""
    if [ ! -e "$STORE" ]; then why="there is no store file"
    elif sc 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=""
        # A bus that stops during the readOnly open can leave -wal or -shm beside the stopped
        # store, so the version read can't be planned on (row 166 item 2).
        sc is-active --quiet "$BUS_UNIT" || S_CHANGED=1
    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_CHANGED" ]; then S_SCHEMA=unknown S_FROM="the bus stopped during the read, so the store state changed; run --plan again"
    elif [ -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"
    [ "$TO" != "$P_COPY" ] || echo "note: $TO is the pointer's copy; the run re-pins it and only the Console flags change"
    echo "store schema: $S_SCHEMA ($S_FROM)"
    echo "target schema: $T_SCHEMA"
    [ "$S_SCHEMA" = "$P_SCHEMA" ] || [ -n "$S_CHANGED" ] || echo "note: the pointer says $P_SCHEMA; the run reads the stopped store and goes by that"
    # Row 186: set when run_steps would roll back with store-newer or migration-needs-flag, which
    # come before backup_store. A read the bus stopped under judges nothing, so it keeps the backup line.
    local early=""
    if [ -n "$S_CHANGED" ]; then
        : # no migration or restore line from a read the bus stopped under
    elif [ "$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)"
        [ "$MIGRATE" = 1 ] || early=1
        # Row 233: root reads only the plan's exit, so under external a missing --migrate refuses here.
        [ "$UNITS" = user ] || [ "$MIGRATE" = 1 ] || PLAN_REFUSE+=("migrate-required: the store is $S_SCHEMA and $TO migrates it to $T_SCHEMA; the go names --migrate")
    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"
        early=1
    else
        echo "restore: the store must go back to $T_SCHEMA; needs --restore $(from_log backup || echo ACTIVATION)"
    fi
    if [ "$OP" != activate ]; then
        : # rollback takes no backup of its own
    elif [ ! -e "$STORE" ]; then
        echo "backup: none; there is no store file"
    elif [ -n "$early" ]; then
        echo "backup: none; the run rolls back before the backup"
    else
        echo "backup: the run backs up the stopped store to $DATA/backups/stack/<activation>/ before any migration"
    fi
    plan_commits
    local flags="${INSTANCE_FLAGS[*]:+${INSTANCE_FLAGS[*]} }"
    if [ "$UNITS" = external ]; then
        echo "units: external; root installs the units for $TO and starts them, then confirm gates them"
    else
        [ "$P_PENDING" = 0 ] || echo "pm: the pointer is pending (activate --pm-after-harness); the bus runs without its PM"
        [ "$RUN_PENDING" = 0 ] || echo "pm: the bus starts without --pm and the PM stays pending; next: Jason's plugins set of harness, then pm-on"
        echo "bus: scripts/bus-service.sh install $flags--copy $TO --copies $COPIES$([ "$RUN_PM" = 0 ] || echo ' --pm') --claude $P_CLAUDE"
        [ -z "$DROP_TESTER" ] || echo "drop $DROP_TESTER; the target Console does not support it"
        echo "$PROOF_PLAN"
        echo "$SEND_PLAN"
        [ -z "$SSO_PLAN" ] || echo "$SSO_PLAN"
        echo "Console: scripts/webui-service.sh install $flags--copy $TO --copies $COPIES${RUN_CONSOLE[*]:+ ${RUN_CONSOLE[*]}}"
    fi
    echo "CLI: scripts/stack-copy.sh wrapper $TO $flags--copies $COPIES --bin $BIN"
    [ -z "$EXPECTED_INTERPRETER" ] || echo "interpreter: $EXPECTED_INTERPRETER"
    local r
    for r in "${PLAN_REFUSE[@]}"; do echo "refuses $r"; done
    [ "${#PLAN_REFUSE[@]}" = 0 ] || exit 3
}

# Steps 4.1 to 4.5 and the gate for target $TO. $1 is the store step: migrate or restore.
run_steps() {
    if [ -n "$EXPECTED_INTERPRETER" ]; then
        log_line result started interpreter "$EXPECTED_INTERPRETER"
        say "interpreter: $EXPECTED_INTERPRETER"
    else log_line result started; fi
    capture || refuse capture-failed "could not write the rollback set $RS; nothing was stopped"
    [ -z "$DROP_TESTER" ] || say "drop $DROP_TESTER; the target Console does not support it"
    [ -z "$PROOF_STEP" ] || say "$PROOF_STEP"
    [ -z "$SEND_STEP" ] || say "$SEND_STEP"
    [ -z "$SSO_STEP" ] || say "$SSO_STEP"
    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
            fi
            backup_store || rollback_in_run backup
            if [ "$STORE_SCHEMA" != "$T_SCHEMA" ]; then
                TAKEN="$ACT"
                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")" "$COUNTED")" || rollback_in_run schema-unreadable
                [ "${n%%$'\t'*}" = "$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" "$RUN_PM" "$P_CLAUDE" "${RUN_CONSOLE[@]}" || after_fail "$FAIL"
    start_both || after_fail start
    gate "$TO" "$RUN_PM" || after_fail "$GATE"
    trap - INT TERM HUP
    write_pointer "$TO" "$RUN_PM" "$RUN_PENDING" "$P_CLAUDE" "$T_SCHEMA" "${RUN_CONSOLE[@]}"
    if [ "$RUN_PENDING" = 1 ]; then
        log_line result ok pm pending
        echo "stack-release: $OP ok: $TO is live for $BUSINESS without its PM; the PM is pending: Jason's plugins set of harness, then pm-on"
    else
        log_line result ok
        echo "stack-release: $OP ok: $TO is live for $BUSINESS"
    fi
}

# The pm the run installs and gates, and whether the pointer it writes is pending (row 169).
# --pm-after-harness installs without --pm and leaves the PM pending; a rollback from a pending
# pointer installs the --pm the activate replaced.
run_pm() {
    RUN_PM="$P_PM" RUN_PENDING=0
    if [ "$AFTER_HARNESS" = 1 ]; then RUN_PM=0 RUN_PENDING=1
    elif [ "$P_PENDING" = 1 ]; then RUN_PM=1
    fi
}

do_activate() {
    TO="$ID"
    [ "$PLAN" = 1 ] || lock
    pointer_for_business
    run_pm
    [ "$P_PENDING" = 0 ] || check_refuse pm-pending "the pointer is pending (activate --pm-after-harness); run pm-on or rollback first"
    if [ "$UNITS" = external ]; then
        ext_pointer_checks
        if [ "$PLAN" = 1 ]; then check_target; plugin_checks; config_checks; plan; exit 0; fi
        ext_precheck
        plugin_checks
        config_checks
        new_activation
        run_external migrate
        return
    fi
    if [ "$PLAN" = 1 ]; then check_target; console_for_target; ports_unclaimed; plugin_checks; config_checks; plan; exit 0; fi
    precheck
    plugin_checks
    config_checks
    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"
    run_pm
    if [ "$UNITS" = external ]; then
        # r6: root names the target from the go, and it must be the one rollback reads from the log.
        [ "$TO_FLAG" = "$TO" ] || refuse to-not-previous "--to $TO_FLAG is not the previous copy; the log names $TO"
        ext_pointer_checks
        if [ "$PLAN" = 1 ]; then
            check_target; plugin_checks; config_checks
            [ -z "$RESTORE" ] || { next_activation; restore_checks; }
            plan; exit 0
        fi
        ext_precheck
        plugin_checks
        config_checks
        [ "$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
        [ -z "$RESTORE" ] || restore_checks
        run_external restore
        return
    fi
    if [ "$PLAN" = 1 ]; then check_target; console_for_target; ports_unclaimed; plugin_checks; config_checks; plan; exit 0; fi
    precheck
    plugin_checks
    config_checks
    [ "$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
    [ -z "$RESTORE" ] || restore_checks
    run_steps restore
}

# rollback --restore's checks on the backup and the rows written after it, before anything stops.
# Sets COUNT and BACKUP; POST comes from new_activation.
restore_checks() {
    local o="$DATA/backups/stack/$RESTORE" rc=0 c="" text
    [ -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 sc is-active --quiet "$BUS_UNIT"; then
        c="$(node -e "$STORE_JS" -- heldcount "$STORE" "$(cat "$o/STAMP")" "$COUNTED")" || rc=1
        # plan_store's recheck (row 166 item 2), before anything stops.
        sc is-active --quiet "$BUS_UNIT" || refuse restore-mismatch "the bus stopped while the events after $(cat "$o/STAMP") were counted, so the store state changed; nothing was stopped"
    else
        ! store_unclean || refuse store-unclean "$UNCLEAN is present; the bus did not close the store cleanly"
        c="$(store_read count "$(cat "$o/STAMP")" "$COUNTED")" || rc=1
    fi
    [ "$rc" = 0 ] || refuse restore-mismatch "cannot count the rows after $(cat "$o/STAMP")"
    COUNT="${c%%$'\t'*}" text="${c#*$'\t'}"
    echo "restore: $o; the store now in place moves to $POST and is kept; $text were written after the backup"
    [ "$COUNT" = 0 ] || [ "$EVENTS" = "$COUNT" ] || refuse post-backup-events "$text were written after the backup; Jason decides, then --events $COUNT"
    BACKUP="$RESTORE"
}

# --- --units external (row 232, r6 section 4) ---

# After the pointer, for activate and rollback, in --plan and in the run. OLD is the copy root read
# from the unit files it installed (r6 A1, N5), so a pointer that names another copy stops the
# re-pin before root stops anything. The PM is a flag in root's unit, so --pm-after-harness has
# nothing to wait on. The run's own lock refuses an open activation; --plan takes no lock, so it
# lists one here.
ext_pointer_checks() {
    [ -z "$EXPECT_FROM" ] || [ "$EXPECT_FROM" = "$P_COPY" ] || check_refuse pointer-differs "the pointer names $P_COPY and root's units name $EXPECT_FROM; Jason or Mos settles it"
    [ "$AFTER_HARNESS" = 0 ] || check_refuse external-pm "under --units external the PM is a flag in root's bus unit; changing it is a re-install with its own go"
    [ "$PLAN" = 1 ] || return 0
    local open
    open="$(interrupted | paste -sd' ' -)"
    [ -z "$open" ] || check_refuse interrupted-run "activation $open started and has no final line; Jason or Mos settles it (status shows it)"
    # Row 233: root can't read the stack's state, so the plan lists any rollback set still waiting
    # for units, whatever the log says.
    local m sets=""
    local -a markers
    mapfile -t markers < <(compgen -G "$STATE/stack-rollback/*/AWAITING-UNITS" || true)
    for m in "${markers[@]}"; do sets+=" $(basename "$(dirname "$m")")"; done
    [ -z "$sets" ] || check_refuse open-activation "rollback set(s)$sets hold AWAITING-UNITS; root finishes them with confirm or revert first"
}

# A3's checks before `started`. Root has stopped the units and installs them after this run, so
# there is no Console flag, port, drop-in or claude pin of this script's to check.
ext_precheck() {
    check_target
    instance_busy
    read_units || refuse unit-flags-drift "cannot read $BUS_UNIT or $CONSOLE_UNIT"
    flags_drift
    units_down
}

# r6 A3 (N4): each unit reads inactive, or failed as stop_both accepts, and the store is closed,
# so nothing below runs against a store a live bus holds.
units_down() {
    local u s
    for u in "$CONSOLE_UNIT" "$BUS_UNIT"; do
        s="$(sc is-active "$u" 2>/dev/null || true)"
        case "$s" in inactive|failed) ;; *) refuse units-running "$u is ${s:-unknown}; root stops the units first" ;; esac
    done
    stopped || refuse units-running "the store under $DATA/bus is still open (writer.lock or an open file)"
}

# A3's capture into $RS, which run_external made: the pointer, the wrapper and its mode, and for each
# unit its show output and the sha256 of its unit file. It reads root's units and makes no directory.
capture_units() {
    local u frag
    [ ! -f "$POINTER" ] || cp -p "$POINTER" "$RS/pointer" || return 1
    [ -f "$BIN/$WRAPPER_NAME" ] || { say "capture: no wrapper $BIN/$WRAPPER_NAME"; return 1; }
    cp -p "$BIN/$WRAPPER_NAME" "$RS/wrapper" || return 1
    stat -c %a "$BIN/$WRAPPER_NAME" > "$RS/WRAPPER_MODE" || return 1
    : > "$RS/UNITS.sha256" || return 1
    for u in "$BUS_UNIT" "$CONSOLE_UNIT"; do
        show "$u" > "$RS/show-$u.txt" || return 1
        frag="$(sc show "$u" -p FragmentPath 2>/dev/null | sed -n 's/^FragmentPath=//p')" && [ -n "$frag" ] || { say "capture: no FragmentPath for $u"; return 1; }
        sha256sum "$frag" >> "$RS/UNITS.sha256" || return 1
    done
    sums "$RS" > "$RS/SHA256SUMS" || return 1
    say "captured the rollback set: $RS"
}

# A failure after the capture: FAILED holds the code, nothing is logged or cleaned up, exit 1.
# Root runs R1-R4, and revert is the one cleanup path (r6 A3).
ext_fail() {
    trap - INT TERM HUP
    { echo "$1" > "$RS/FAILED"; } 2>/dev/null || true
    say "failed: $1"
    echo "stack-release: $OP failed with $1; $ACT stays open for root's revert and confirm --rollback. Rollback set: $RS" >&2
    exit 1
}

# A3 for target $TO. $1 is the store step: migrate or restore. The marker goes in first, so every
# failure after it leaves what revert needs. Writes no pointer and starts nothing.
run_external() {
    log_line result started
    { mkdir -p -m 0700 "$STATE/stack-rollback" && mkdir -m 0700 "$RS"; } 2>/dev/null || refuse marker-failed "could not make the rollback set $RS; nothing was changed"
    { echo "$TO" > "$RS/AWAITING-UNITS"; } 2>/dev/null || { rm -f "$RS/AWAITING-UNITS"; refuse marker-failed "could not write $RS/AWAITING-UNITS; nothing was changed"; }
    STEPS="$RS/steps.log"
    capture_units || { rm -f "$RS/AWAITING-UNITS"; refuse capture-failed "could not write the rollback set $RS; nothing was changed"; }
    trap 'ext_fail signal' INT TERM HUP
    if [ -e "$STORE" ]; then
        if store_unclean; then say "$UNCLEAN is present; the store is left as it is for Jason or Mos"; ext_fail store-unclean; fi
        STORE_SCHEMA="$(store_read version)" && [ -n "$STORE_SCHEMA" ] || ext_fail schema-unreadable
        say "stopped store schema: $STORE_SCHEMA"
        if [ "$1" = migrate ]; then
            if [ "$STORE_SCHEMA" != "$T_SCHEMA" ]; then
                in_chain "$STORE_SCHEMA" || ext_fail store-newer
                [ "$MIGRATE" = 1 ] || ext_fail migration-needs-flag
            fi
            backup_store || ext_fail backup
            echo "$BACKUP" > "$RS/BACKUP" || ext_fail backup
            if [ "$STORE_SCHEMA" != "$T_SCHEMA" ]; then
                # MIGRATING and TAKEN before the migration, so revert restores a partial one (r6 B2).
                : > "$RS/MIGRATING" || ext_fail migrate-failed
                TAKEN="$ACT"
                node "$COPIES/$TO/packages/bus/src/migrate-schema.mjs" "$DATA" >> "$STEPS" 2>&1 || ext_fail migrate-failed
                say "migrated the store to $T_SCHEMA"
            fi
        else
            [ "$STORE_SCHEMA" = "$P_SCHEMA" ] || ext_fail store-differs
            if [ -n "$RESTORE" ]; then
                local n
                n="$(store_read count "$(cat "$DATA/backups/stack/$RESTORE/STAMP")" "$COUNTED")" || ext_fail schema-unreadable
                [ "${n%%$'\t'*}" = "$COUNT" ] || ext_fail post-backup-events
                echo "$RESTORE" > "$RS/BACKUP" || ext_fail restore
                # RESTORED tells confirm --rollback the store was swapped, so it skips OLD's gate
                # and logs rollbackCode not-attempted (ruling 4, T14b).
                if ! restore_store "$RESTORE"; then
                    [ -z "$RESTORED" ] || : > "$RS/RESTORED" || true
                    ext_fail restore
                fi
                : > "$RS/RESTORED" || ext_fail restore
            fi
        fi
    elif [ -n "$RESTORE" ]; then
        ext_fail store-differs
    fi
    scripts/stack-copy.sh wrapper "$TO" "${INSTANCE_FLAGS[@]}" --copies "$COPIES" --bin "$BIN" >> "$STEPS" 2>&1 || ext_fail wrapper
    trap - INT TERM HUP
    say "staged $TO"
    echo "stack-release: $OP staged $TO for $BUSINESS as $ACT; root installs and starts the units, then runs confirm $TO $BUSINESS --units external"
}

# r6 "The open activation": exactly one AWAITING-UNITS file, its activation the one the log holds
# open, holding the copy root passed. Sets O_ACT. Anything else refuses open-activation, and ACT
# stays unset, so the refused line names no activation and the open one stays open.
open_activation() {
    local -a markers open
    local m found="" seen=""
    mapfile -t markers < <(compgen -G "$STATE/stack-rollback/*/AWAITING-UNITS" || true)
    mapfile -t open < <(interrupted)
    for m in "${markers[@]}"; do
        seen+=" $(basename "$(dirname "$m")") (copy $(cat "$m" 2>/dev/null || echo unreadable))"
    done
    if [ "${#markers[@]}" = 1 ] && [ "${#open[@]}" = 1 ]; then
        m="$(basename "$(dirname "${markers[0]}")")"
        [ "$m" != "${open[0]}" ] || [ "$(cat "${markers[0]}" 2>/dev/null)" != "$ID" ] || found="$m"
    fi
    [ -n "$found" ] || refuse open-activation "need one AWAITING-UNITS for copy $ID whose activation the log holds open; markers:${seen:- none}; open: ${open[*]:-none}"
    O_ACT="$found"
}

# The op, from, to and start second of O_ACT's `started` line, one per line.
started_of() {
    node -e '
const [file,act]=process.argv.slice(1);
let s;
for(const l of require("fs").readFileSync(file,"utf8").split("\n")){
  if(!l.trim())continue;
  let o;try{o=JSON.parse(l)}catch{continue}
  if(o.activation===act&&o.result==="started")s=o;
}
const at=s&&Math.floor(Date.parse(s.at)/1000);
if(!s||!["activate","rollback"].includes(s.op)||!Number.isFinite(at))process.exit(1);
process.stdout.write([s.op,s.from??"",s.to??"",at].join("\n")+"\n");' -- "$LOG" "$O_ACT"
}

# A5, and R4 with --rollback. The lines it writes carry the open activation's op, from, to and
# activation, and the backup A3 recorded, so from_log reads them as a dragon-lin run's. Every check
# runs before ACT is set.
do_confirm() {
    lock_only
    open_activation
    local s
    local -a f
    s="$(started_of)" || refuse open-activation "the log's started line for $O_ACT is unreadable"
    mapfile -t f <<< "$s"
    [ "${f[2]}" = "$ID" ] || refuse open-activation "$O_ACT started for ${f[2]}, not $ID"
    if read_pointer; then [ "$P_BUSINESS" = "$BUSINESS" ] || refuse business-mismatch "the pointer is for $P_BUSINESS"; fi
    if [ -n "$ROLLBACK_OLD" ]; then
        [ "$ROLLBACK_OLD" = "${f[1]}" ] || refuse rollback-mismatch "$ROLLBACK_OLD is not the copy $O_ACT started from, ${f[1]:-none}"
    else
        [ ! -e "$STATE/stack-rollback/$O_ACT/FAILED" ] || refuse activation-failed "$O_ACT has a FAILED file; root runs revert and confirm --rollback"
        target_schema "$ID" || refuse copy-schema-unknown "cannot read SCHEMA_VERSION from $COPIES/$ID/packages/bus/src/store.mjs"
    fi
    OP="${f[0]}" FROM="${f[1]}" TO="${f[2]}" FLOOR="${f[3]}" ACT="$O_ACT" RS="$STATE/stack-rollback/$O_ACT"
    STEPS="$RS/steps.log"
    BACKUP="$(cat "$RS/BACKUP" 2>/dev/null || true)"
    if [ -z "$ROLLBACK_OLD" ]; then
        read_units || ext_fail units-unreadable
        P_CONSOLE=("${U_CONSOLE[@]}")
        gate "$TO" "$U_PM" || ext_fail "$GATE"
        write_pointer "$TO" "$U_PM" 0 "$U_CLAUDE" "$T_SCHEMA" "${U_CONSOLE[@]}"
        log_line result ok
        rm -f "$RS/AWAITING-UNITS"
        echo "stack-release: $OP ok: $TO is live for $BUSINESS"
        return
    fi
    local code rcode=""
    if [ "$CAUSE" = A4 ]; then code=units-start
    else code="$(head -n 1 "$RS/FAILED" 2>/dev/null || true)"; [ -n "$code" ] || code=unrecorded
    fi
    if [ -e "$RS/RESTORED" ]; then rcode=not-attempted
    elif ! read_units; then rcode=units-unreadable
    else
        P_CONSOLE=("${U_CONSOLE[@]}")
        gate "$ROLLBACK_OLD" "$U_PM" || rcode="$GATE"
    fi
    if [ -z "$rcode" ]; then
        local r=failed
        [ "$CAUSE" != A5 ] || r=gate-failed
        log_line result "$r" code "$code" rollback ok
        rm -f "$RS/AWAITING-UNITS" "$RS/FAILED"
        echo "stack-release: $code; rolled back to $ROLLBACK_OLD, which passed the gate" >&2
        exit 1
    fi
    say "rollback-failed: run $code, rollback $rcode"
    log_line result rollback-failed code "$code" rollback failed rollbackCode "$rcode"
    rm -f "$RS/AWAITING-UNITS" "$RS/FAILED"
    echo "stack-release: rollback-failed: the run failed with $code and the rollback with $rcode; root stops the units. Rollback set: $RS${BACKUP:+, backup $DATA/backups/stack/$BACKUP}" >&2
    exit 2
}

revert_fail() {
    say "revert failed: $1"
    echo "stack-release: revert failed: $1; $O_ACT stays open. Rollback set: $RS" >&2
    exit 1
}

# The store already holds backup $1 when every file checks against its SHA256SUMS and the listing
# matches, as after a revert that restored it.
store_matches() {
    local o="$DATA/backups/stack/$1"
    [ -d "$DATA/bus" ] && (cd "$DATA/bus" && sha256sum -c --quiet "$o/SHA256SUMS") >/dev/null 2>&1 &&
        [ "$(ls -1A "$DATA/bus")" = "$(ls -1A "$o/bus")" ]
}

# R2: puts the wrapper back and, after a migration, the backup taken before it. Each part can run
# again, so it serves an A3 that stopped partway. A rollback's --restore swap is confirm's to
# report, not revert's to undo. Writes no log line.
do_revert() {
    lock_only
    open_activation
    RS="$STATE/stack-rollback/$O_ACT" STEPS="$STATE/stack-rollback/$O_ACT/steps.log"
    [[ "$O_ACT" =~ ^([0-9]{8}T[0-9]{6}Z)-[0-9a-f]{12}(-[1-9])?$ ]] || revert_fail "$O_ACT is not an activation id"
    POST="$DATA/bus.post-attempt-${BASH_REMATCH[1]}${BASH_REMATCH[2]}"
    [ -f "$RS/wrapper" ] && [ -f "$RS/WRAPPER_MODE" ] || revert_fail "the rollback set has no saved wrapper"
    local tmp
    tmp="$(mktemp "$BIN/.$WRAPPER_NAME.XXXXXX")" && cp -p "$RS/wrapper" "$tmp" && chmod "$(cat "$RS/WRAPPER_MODE")" "$tmp" && mv -f "$tmp" "$BIN/$WRAPPER_NAME" ||
        revert_fail "could not put the wrapper back"
    [ "$(sha256sum < "$BIN/$WRAPPER_NAME" | cut -d' ' -f1)" = "$(sum_of "$RS" wrapper)" ] &&
        [ "$(stat -c %a "$BIN/$WRAPPER_NAME")" = "$(cat "$RS/WRAPPER_MODE")" ] || revert_fail "the wrapper does not match the rollback set"
    say "put back the wrapper from $RS"
    if [ -e "$RS/MIGRATING" ]; then
        TAKEN="$(cat "$RS/BACKUP" 2>/dev/null || true)"
        [ -n "$TAKEN" ] || revert_fail "MIGRATING is set and BACKUP names no backup"
        if store_matches "$TAKEN"; then say "the store already matches backup $TAKEN"
        else restore_store "$TAKEN" || revert_fail "could not restore backup $TAKEN"
        fi
    fi
    echo "stack-release: revert ok: $O_ACT's wrapper$([ -e "$RS/MIGRATING" ] && echo ' and store are' || echo ' is') back; root installs the old units next"
}

# A2f: gates copy ID with no open activation, on a start at or after --since. Takes no lock and
# writes nothing, so it refuses nothing: 0 on a pass, 1 otherwise.
do_gate() {
    FLOOR="$SINCE"
    read_units || { echo "stack-release: gate: cannot read $BUS_UNIT or $CONSOLE_UNIT" >&2; exit 1; }
    P_CONSOLE=("${U_CONSOLE[@]}")
    gate "$ID" "$U_PM" || exit 1
}

# --- pm-on (row 169 item 3) ---

# Stops the bus alone and waits for it and the stopped check, within --timeout. The Console runs on.
stop_bus() {
    sc stop "$BUS_UNIT" || return 1
    local deadline=$((SECONDS + TIMEOUT)) s
    while :; do
        s="$(sc is-active "$BUS_UNIT" 2>/dev/null || true)"
        case "$s" in inactive|failed) if stopped; then say "stopped $BUS_UNIT"; return 0; fi ;; esac
        [ "$SECONDS" -lt "$deadline" ] || return 1
        sleep 1
    done
}

start_bus() {
    START_EPOCH="$(date +%s)" && START_MS="$(date +%s%3N)" || return 1
    sc start "$BUS_UNIT"
}

# mos-0848 point 3: put the phase-1 bus unit back (no --pm), restart it and rerun the phase-1 gate.
# Passing leaves the bus live without its PM and the pointer pending (exit 1); failing stops both
# units (exit 2). Neither touches the copy or the store. $1 is the code pm-on failed with.
pm_on_fail() {
    trap - INT TERM HUP
    local code="$1" why="$GATE_WHY" pcode=""
    say "pm-on failed: $code${why:+ ($why)}; putting the phase-1 bus unit back"
    GATE_WHY=""
    if ! stop_bus; then pcode=stop-timeout
    elif ! put_back; then pcode=put-back
    elif ! start_bus; then pcode=start
    elif ! gate "$P_COPY" 0; then pcode="$GATE"
    fi
    if [ -z "$pcode" ]; then
        log_line result pm-on-failed code "$code" why "${why:-@null}"
        echo "stack-release: pm-on failed with $code${why:+ ($why)}; the bus runs $P_COPY without its PM again and passed the phase-1 gate, and the pointer stays pending. Fix harness with plugins set, then run pm-on again. Rollback set: $RS" >&2
        exit 1
    fi
    local pwhy="$GATE_WHY"
    sc stop "$CONSOLE_UNIT" || true
    sc stop "$BUS_UNIT" || true
    say "pm-on-down: pm-on $code, phase 1 $pcode"
    log_line result pm-on-down code "$code" why "${why:-@null}" phase1 "$pcode" phase1Why "${pwhy:-@null}"
    echo "stack-release: pm-on-down: pm-on failed with $code${why:+ ($why)} and the phase-1 bus with $pcode${pwhy:+ ($pwhy)}; both units are stopped. Rollback set: $RS" >&2
    exit 2
}

# Phase 2: the bus unit alone, on the pointer's copy with --pm, gated with pm_launched.
do_pm_on() {
    lock
    [ "$UNITS" = user ] || refuse external-pm "under --units external the PM is a flag in root's bus unit; changing it is a re-install with its own go"
    pointer_for_business
    [ "$P_PENDING" = 1 ] || refuse not-pending "the pointer is not pending; pm-on follows activate --pm-after-harness"
    local out
    out="$(plugins harness "$P_COPY" 2>&1)" || refuse harness-unset "$P_COPY's loader refuses the business file: $out"
    [ "$out" != none ] || refuse harness-unset "the business file sets no harness; Jason sets it with plugins set first"
    node -e 'const h=JSON.parse(process.argv[1]);process.exit(Array.isArray(h)&&h.length>0?0:1)' -- "$out" 2>/dev/null || refuse harness-unset "the business file's harness is $out, not a non-empty list"
    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
    TO="$P_COPY"
    new_activation
    log_line result started
    capture || refuse capture-failed "could not write the rollback set $RS; nothing was stopped"
    trap 'pm_on_fail signal' INT TERM HUP
    stop_bus || pm_on_fail stop-timeout
    scripts/bus-service.sh install "${INSTANCE_FLAGS[@]}" --copy "$P_COPY" --copies "$COPIES" --pm --claude "$P_CLAUDE" >> "$STEPS" 2>&1 || pm_on_fail install-bus
    say "installed $BUS_UNIT for $P_COPY with --pm"
    start_bus || pm_on_fail start
    gate "$P_COPY" 1 || pm_on_fail "$GATE"
    trap - INT TERM HUP
    write_pointer "$P_COPY" 1 0 "$P_CLAUDE" "$P_SCHEMA" "${P_CONSOLE[@]}"
    log_line result ok pm on
    echo "stack-release: pm-on ok: the PM is live on $P_COPY for $BUSINESS"
}

do_status() {
    echo "pointer: $POINTER"
    if read_pointer; then sed 's/^/  /' "$POINTER"; else echo "  none"; fi
    [ "$P_PENDING" = 0 ] || echo "pm: pending (activate --pm-after-harness); next: Jason's plugins set of harness, then stack-release.sh pm-on $BUSINESS"
    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)"
    if [ "$UNITS" = external ]; then
        local m
        local -a markers
        mapfile -t markers < <(compgen -G "$STATE/stack-rollback/*/AWAITING-UNITS" || true)
        for m in "${markers[@]}"; do
            echo "awaiting units: $(basename "$(dirname "$m")") (copy $(cat "$m" 2>/dev/null || echo unreadable))"
        done
    fi
    echo "check:"
    if scripts/stack-copy.sh check "${INSTANCE_FLAGS[@]}" "${UNITS_FLAGS[@]}" --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
}

# --- The rehearsal (D2, row 163) ---

# rehearse runs the new copy's bus host (--no-launch) and Console on a scratch root built from a
# backup, under <dataRoot>/rehearsals/<stamp>/, and gates them as activate does. It never writes the
# live store, the live config files, the backup, the pointer, the log or a unit. Nothing prunes
# the rehearsal directory. Design, rulings (mos-0961, mos-0962, mos-0963) and the deviations from
# the brief: agents/sage/work/queue-163/BUILD.md.

# One PASS or FAIL line per check, on stdout and in steps.log. Lines from before the directory
# exists wait in R_EARLY and go in first.
r_line() {
    echo "$*"
    if [ -z "$STEPS" ]; then R_EARLY+=("$(now) $*"); else echo "$(now) $*" >> "$STEPS" || true; fi
}
r_pass() { r_line "PASS $1${2:+: $2}"; }
r_fail() { r_line "FAIL $1${2:+: $2}"; R_RC=1; }
r_abort() { r_fail "$@"; r_end; }

r_alive() { local s; s="$(cat "/proc/$1/stat" 2>/dev/null)" || return 1; s="${s##*) }"; [ "${s:0:1}" != Z ]; }
r_gone() {
    local deadline=$((SECONDS + TIMEOUT))
    while r_alive "$1"; do [ "$SECONDS" -lt "$deadline" ] || return 1; sleep 1; done
}

# Stops the Console, then the host, with SIGTERM within --timeout; one still there gets SIGKILL
# and a FAIL line. The host has to leave the store's writer.lock behind it.
r_stop() {
    local name pid
    for name in console host; do
        if [ "$name" = console ]; then pid="$R_CONSOLE_PID"; R_CONSOLE_PID=""; else pid="$R_HOST_PID"; R_HOST_PID=""; fi
        [ -n "$pid" ] || continue
        kill -TERM "$pid" 2>/dev/null || true
        if r_gone "$pid"; then r_pass "$name-stopped" "pid $pid"
        else kill -KILL "$pid" 2>/dev/null || true; r_fail "$name-stopped" "pid $pid was still running ${TIMEOUT}s after SIGTERM; sent SIGKILL"
        fi
        wait "$pid" 2>/dev/null || true
    done
    [ -n "$R_STARTED" ] || return 0
    R_STARTED=""
    if [ -e "$R/data/bus/writer.lock" ]; then r_fail host-released-store "$R/data/bus/writer.lock is still there"; else r_pass host-released-store; fi
}

r_tree_sums() { (cd "$1" && find . -type f -print0 | sort -z | xargs -0 -r sha256sum); }

# The last checks, then the last line. The backup and the live config files are compared with the
# sums taken before the rehearsal directory was made.
r_end() {
    trap - INT TERM HUP
    r_stop
    if [ -n "$R" ]; then
        if [ "$(r_tree_sums "$R_BK")" = "$(cat "$R/backup.before")" ]; then r_pass backup-unchanged; else r_fail backup-unchanged "$R_BK differs from $R/backup.before"; fi
        if [ "$(sha256sum "$LIVE_CFG" "$LIVE_BIZ" 2>&1)" = "$(cat "$R/live.before")" ]; then r_pass live-config-unchanged; else r_fail live-config-unchanged "$LIVE_CFG or $LIVE_BIZ differs from $R/live.before"; fi
    fi
    if [ "$R_RC" = 0 ]; then r_line REHEARSAL_PASS; else r_line REHEARSAL_FAIL; fi
    [ -z "$R" ] || echo "rehearsal directory: $R"
    exit "$R_RC"
}

r_port_free() {
    node -e 'const s=require("node:net").createServer();s.once("error",()=>process.exit(1));s.listen(Number(process.argv[1]),"127.0.0.1",()=>s.close(()=>process.exit(0)));' -- "$1"
}

# The scratch config: the live one with dataRoot set to the scratch data root.
R_CONFIG_JS='
const fs=require("node:fs");const [file,data]=process.argv.slice(1);
const c=JSON.parse(fs.readFileSync(file,"utf8"));c.dataRoot=data;process.stdout.write(JSON.stringify(c,null,2)+"\n");'

# The scratch business file (mos-0962, mos-0963): the live file without its tracker.* variables,
# business and role, and with every credential reference, the roles and tracker.sync, moved to its
# own 0600 placeholder file. The service and its rotateBy or expires date stay. The tracker key
# stays because the loader requires it. Mode `normalize` prints the live file as this writes it,
# for the diff.
R_BUSINESS_JS='
const fs=require("node:fs"),path=require("node:path");const [mode,live,out,ph]=process.argv.slice(1);
const b=JSON.parse(fs.readFileSync(live,"utf8"));
if(mode==="normalize"){process.stdout.write(JSON.stringify(b,null,2)+"\n");process.exit(0)}
const DATE={gitea:"rotateBy",vikunja:"expires"};
const strip=(vars)=>{if(vars&&typeof vars==="object")for(const k of Object.keys(vars))if(k.startsWith("tracker."))delete vars[k]};
const swap=(name,refs)=>{
  if(!refs||typeof refs!=="object")return;
  for(const s of Object.keys(refs)){
    const file=path.join(ph,`${name}-${s}`),old=refs[s],ref={file};
    fs.writeFileSync(file,"rehearsal-placeholder-token",{mode:0o600,flag:"wx"});
    if(old&&typeof old==="object"&&Object.hasOwn(DATE,s)&&old[DATE[s]]!==undefined)ref[DATE[s]]=old[DATE[s]];
    refs[s]=ref;
  }
};
strip(b.vars);
for(const [name,r] of Object.entries(b.roles??{})){strip(r?.vars);swap(name,r?.credentials)}
swap("@sync",b.tracker?.sync?.credentials);
fs.writeFileSync(out,JSON.stringify(b,null,2)+"\n",{mode:0o600,flag:"wx"});'

# The guard, before the host starts. It reads the scratch config through the new copy's own
# loadSystem and bootConfig, as the host will, and prints one line per check:
#   data-root-live       the data root is not the live one (mos-0961)
#   sessions-live        no sessions.json there names a running process (mos-0961)
#   data-root-scratch    the data root is the scratch one
#   notify-off           the notifier config is {"notifyVersion":1,"binding":null}, 0600 (mos-0962)
#   tracker-configured   bootConfig sets no trackers (mos-0962)
#   credential-ref-live  every credential reference is a file under placeholders/ (mos-0963)
# Exit 0 when all pass, 1 on the first FAIL.
R_GUARD_JS='
const fs=require("node:fs"),path=require("node:path");const {pathToFileURL}=require("node:url");
const [copy,id,live,scratch,ph]=process.argv.slice(1);
const real=(p)=>{try{return fs.realpathSync(p)}catch{return null}};
const alive=(pid)=>{try{const s=fs.readFileSync(`/proc/${pid}/stat`,"utf8");return s[s.lastIndexOf(")")+2]!=="Z"}catch{return false}};
const pass=(n,w)=>console.log(`PASS ${n}${w?": "+w:""}`);
const fail=(n,w)=>{console.log(`FAIL ${n}: ${w}`);process.exit(1)};
(async()=>{
  let c,system,config;
  try{c=await import(pathToFileURL(`${copy}/packages/cli/src/config.mjs`).href);system=c.loadSystem({env:process.env})}
  catch(e){fail("scratch-config",String(e?.message??e).split("\n")[0])}
  const root=real(system.dataRoot);
  if(root===null)fail("data-root-scratch",`${system.dataRoot} does not exist`);
  if(root===real(live))fail("data-root-live",`the scratch config names dataRoot ${system.dataRoot}, which resolves to the live data root ${live}`);
  pass("data-root-live",`${system.dataRoot} is not the live data root`);
  const sf=path.join(root,"bus-host","sessions.json");
  if(fs.existsSync(sf)){
    let doc;try{doc=JSON.parse(fs.readFileSync(sf,"utf8"))}catch{fail("sessions-live",`${sf} is unreadable`)}
    if(!Array.isArray(doc?.sessions))fail("sessions-live",`${sf} is malformed`);
    const running=doc.sessions.flatMap((s)=>[s?.pid,s?.runnerPid]).filter((p)=>Number.isSafeInteger(p)&&p>0&&alive(p));
    if(running.length)fail("sessions-live",`${sf} names running pid(s) ${running.join(", ")}`);
  }
  pass("sessions-live","no running session named in the data root");
  if(root!==real(scratch))fail("data-root-scratch",`the scratch config resolves to ${system.dataRoot}, not ${scratch}`);
  pass("data-root-scratch");
  const nf=path.join(root,"notify",id,"notify.json");
  let st,doc;try{st=fs.lstatSync(nf);doc=JSON.parse(fs.readFileSync(nf,"utf8"))}catch{fail("notify-off",`${nf} is missing or not JSON`)}
  if(!st.isFile()||(st.mode&0o777)!==0o600||JSON.stringify(doc)!==JSON.stringify({notifyVersion:1,binding:null}))fail("notify-off",`${nf} is not {"notifyVersion":1,"binding":null} at mode 0600`);
  pass("notify-off","binding null, so no Discord bridge");
  try{config=c.bootConfig({system,businessId:id,env:process.env})}catch(e){fail("boot-config",String(e?.message??e).split("\n")[0])}
  if(config.trackers!==undefined)fail("tracker-configured",`bootConfig sets trackers for ${Object.keys(config.trackers).join(", ")}`);
  pass("tracker-configured","bootConfig sets no trackers");
  const b=config.businesses?.[id],phr=real(ph),refs=[];
  for(const [role,r] of Object.entries(b?.roles??{}))for(const [s,ref] of Object.entries(r?.credentials??{}))refs.push([`${role} ${s}`,ref]);
  for(const [s,ref] of Object.entries(b?.tracker?.sync?.credentials??{}))refs.push([`@sync ${s}`,ref]);
  for(const [what,ref] of refs){
    if(ref?.env!==undefined)fail("credential-ref-live",`${what} is env ${ref.env}`);
    const f=typeof ref?.file==="string"?real(ref.file):null;
    if(f===null||phr===null||path.dirname(f)!==phr)fail("credential-ref-live",`${what} is ${ref?.file}, not a file in ${ph}`);
  }
  pass("credential-ref-live",`${refs.length} reference(s), each a placeholder`);
})().catch((e)=>fail("guard",String(e?.message??e).split("\n")[0]));'

# Opens a data root's store with a copy's Store, as the bus does at start, with no migration
# (de352dfaf's verify-open.mjs). Prints OPEN_OK <version> or REFUSED <message>.
R_OPEN_JS='
const {pathToFileURL}=require("node:url");const [copy,root]=process.argv.slice(1);
(async()=>{
  const {Store}=await import(pathToFileURL(`${copy}/packages/bus/src/store.mjs`).href);
  let s;
  try{s=new Store(root);process.stdout.write(`OPEN_OK ${s.get("SELECT value FROM meta WHERE key=?","schema_version").value}\n`)}
  catch(e){process.stdout.write(`REFUSED ${e.message}\n`)}
  finally{s?.close()}
})();'
r_open() { node -e "$R_OPEN_JS" -- "$1" "$2" 2>&1 || true; }

# Every process this rehearsal starts gets only HOME, PATH and the scratch MOSAIC_CONFIG.
r_env() { env -i HOME="$HOME" PATH="$PATH" MOSAIC_CONFIG="$R/config/config.json" "$@"; }

# The de352dfaf checks, when the backup's schema is not the new copy's.
r_migration() {
    local copy="$COPIES/$TO" old="$COPIES/$P_COPY" v out rc
    v="$(node -e "$STORE_JS" -- version "$R/data/bus/bus.sqlite" 2>&1)" || r_abort store-unreadable "$v"
    # Row 233 (sage-1653): which store the rehearsal ran against, for root's step log. Output only:
    # the rehearsal's own steps.log has the backup's path on its backup line.
    echo "BACKUP_STORE $R_BACKUP schema $v"
    if [ "$v" = "$T_SCHEMA" ]; then r_pass migration "not needed: the backup is already $v"; return 0; fi
    in_chain "$v" || r_abort store-newer "the backup is $v, which $TO's migration chain does not start from"
    (umask 077; mkdir "$R/refuse" "$R/refuse/bus" && cp "$R_BK"/bus/* "$R/refuse/bus/") || r_abort scratch-build "cannot copy the backup to $R/refuse"
    out="$(r_open "$copy" "$R/refuse")"
    [ "$out" = "REFUSED schema-mismatch" ] || r_abort new-refuses-old-store "$TO opened the $v store: $out"
    r_pass new-refuses-old-store "$TO refuses the $v store at a normal open"
    rc=0; out="$(r_env node "$copy/packages/bus/src/migrate-schema.mjs" "$R/data" 2>&1)" || rc=$?
    printf '%s\n' "$out" >> "$R/migrate.log"
    [ "$rc" = 0 ] && [ "$(tail -n 1 <<< "$out")" = "bus schema $T_SCHEMA ready" ] || r_abort migrate "migrate-schema.mjs exit $rc: $(tail -n 1 <<< "$out")"
    r_pass migrate "$v to $T_SCHEMA"
    out="$(r_open "$copy" "$R/data")"
    [ "$out" = "OPEN_OK $T_SCHEMA" ] || r_abort new-opens-migrated "$out"
    r_pass new-opens-migrated
    rc=0; out="$(r_env node "$copy/packages/bus/src/migrate-schema.mjs" "$R/data" 2>&1)" || rc=$?
    printf '%s\n' "$out" >> "$R/migrate.log"
    [ "$rc" = 0 ] && [ "$(tail -n 1 <<< "$out")" = "bus schema $T_SCHEMA ready" ] || r_abort migrate-again "second migrate-schema.mjs exit $rc: $(tail -n 1 <<< "$out")"
    r_pass migrate-again "a second run is a no-op"
    out="$(r_open "$old" "$R/data")"
    [ "$out" = "REFUSED schema-mismatch" ] || r_abort old-refuses-migrated "the pointer's copy $P_COPY opened the $T_SCHEMA store: $out"
    r_pass old-refuses-migrated "$P_COPY refuses the $T_SCHEMA store"
}

# One pass of D1's gate checks but the PM's, against the processes this run started. Sets R_GATE
# to the first failing check and R_WHY to what it saw.
r_gate_once() {
    local copy="$COPIES/$TO" out
    R_GATE=host-up R_WHY=""
    r_alive "$R_HOST_PID" || { R_WHY="the host (pid $R_HOST_PID) exited; see $R/host.log"; return 1; }
    grep -q "^bus host up: business $BUSINESS, .*, launches disabled\$" "$R/host.log" || { R_WHY="no \"bus host up ... launches disabled\" line in $R/host.log"; return 1; }
    R_GATE=bus-status
    out="$(r_env "$copy/scripts/mosaic" bus status --json 2>/dev/null)" || { R_WHY="bus status failed"; return 1; }
    R_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 ${main}, the host this run started`:!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" "$R_HOST_PID" "$START_MS" 2>&1)" || { R_WHY="bus status unreadable"; return 1; }
    [ -z "$R_WHY" ] || return 1
    R_GATE=console-up
    r_alive "$R_CONSOLE_PID" || { R_WHY="the Console (pid $R_CONSOLE_PID) exited; see $R/console.log"; return 1; }
    grep -qxF "Mosaic Console: http://127.0.0.1:$R_PORT/" "$R/console.log" || { R_WHY="no \"Mosaic Console: http://127.0.0.1:$R_PORT/\" line in $R/console.log"; return 1; }
    R_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));' -- "$R_PORT" || return 1
    R_GATE=copy-match
    tr '\0' ' ' < "/proc/$R_HOST_PID/cmdline" | grep -qF " $copy/packages/cli/src/cli.mjs bus start $BUSINESS --no-launch" || { R_WHY="the host's command line is not $copy's cli.mjs bus start $BUSINESS --no-launch"; return 1; }
    tr '\0' ' ' < "/proc/$R_CONSOLE_PID/cmdline" | grep -qF " $copy/packages/webui/src/cli.mjs serve " || { R_WHY="the Console's command line is not $copy's webui cli.mjs serve"; return 1; }
    R_GATE=""
}

r_gate() {
    local deadline=$((SECONDS + TIMEOUT))
    until r_gate_once; do
        [ "$SECONDS" -lt "$deadline" ] || { r_fail "$R_GATE" "${R_WHY:-no answer within ${TIMEOUT}s}"; return 1; }
        sleep 1
    done
    r_pass gate "host-up, bus-status, console-up, healthz and copy-match on $TO"
}

# No role session: the host wrote no sessions.json and no launches directory or log.
r_no_session() {
    local found=""
    [ ! -e "$R/data/bus-host/sessions.json" ] || found+=" bus-host/sessions.json"
    [ ! -e "$R/data/launches" ] || found+=" launches"
    if [ -z "$found" ]; then r_pass no-session "no sessions.json and no launches under $R/data"; else r_fail no-session "found$found under $R/data"; fi
}

do_rehearse() {
    TO="$ID"
    trap 'r_fail signal "the rehearsal was interrupted"; r_end' INT TERM HUP
    read_pointer || r_abort no-pointer "no $POINTER; the pointer's copy is the old copy the checks use"
    [ "$P_BUSINESS" = "$BUSINESS" ] || r_abort business-mismatch "the pointer is for $P_BUSINESS"
    # Row 233 (sage-1650, sage-1653): last is the newest backup the log records. With none, exit 3
    # and no directory, the one rehearse refusal stack-upgrade.sh reads as no-backup.
    if [ "$R_BACKUP" = last ]; then
        R_BACKUP="$(from_log newest)" || { r_fail no-backup "the log records no backup"; R_RC=3; r_end; }
        [[ "$R_BACKUP" =~ ^[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}(-[1-9])?$ ]] || r_abort backup-unreadable "the log's newest backup is not an activation id"
        r_pass backup-last "the log's newest backup is $R_BACKUP"
    fi
    copy_exists "$TO" 2>/dev/null || r_abort copy-missing "no copy $TO in $COPIES"
    copy_ok "$TO" 2>/dev/null || r_abort copy-unreachable "$TO is not reachable from refactor and origin/refactor"
    target_schema "$TO" || r_abort copy-schema-unknown "cannot read SCHEMA_VERSION and its chain from $COPIES/$TO/packages/bus/src/store.mjs"
    # mos-0961: an older copy's host has no --no-launch and would start the PM on its own.
    grep -qF -- '"--no-launch"' "$COPIES/$TO/packages/cli/src/cli.mjs" 2>/dev/null || r_abort copy-without-no-launch "$TO's bus host has no --no-launch flag"
    r_pass copy "$TO, schema $T_SCHEMA"
    local i
    local -a reserved=(7332 7333 "$(console_port)")
    for ((i = 0; i < ${#P_CONSOLE[@]}; i++)); do [ "${P_CONSOLE[i]}" != --tester-port ] || reserved+=("${P_CONSOLE[i + 1]:-}"); done
    for i in "${reserved[@]}"; do [ "$((10#$R_PORT))" != "$i" ] || r_abort port-reserved "--port $R_PORT is a live Console port (7332, 7333 and the pointer's: ${reserved[*]:2})"; done
    if i="$(instance_port_holder "$UNIT_DIR" "$R_PORT")"; then r_abort port-reserved "--port $R_PORT is $UNIT_DIR/$i's Console port"; fi
    r_port_free "$R_PORT" || r_abort port-in-use "127.0.0.1:$R_PORT is in use; the rehearsal takes no other port"
    r_pass port "127.0.0.1:$R_PORT is free"
    case "$R_BACKUP" in /*) R_BK="$R_BACKUP" ;; *) R_BK="$DATA/backups/stack/$R_BACKUP" ;; esac
    [ -f "$R_BK/bus/bus.sqlite" ] && [ -f "$R_BK/SHA256SUMS" ] || r_abort backup-missing "$R_BK has no bus/bus.sqlite and SHA256SUMS"
    (cd "$R_BK/bus" && sha256sum -c --quiet ../SHA256SUMS) >/dev/null 2>&1 || r_abort backup-sums "$R_BK/bus does not match its SHA256SUMS"
    r_pass backup "$R_BK matches its SHA256SUMS"
    LIVE_CFG="$(realpath -m "${MOSAIC_CONFIG:-$HOME/.config/mosaic-dev/config.json}")"
    LIVE_BIZ="$(dirname "$LIVE_CFG")/businesses/$BUSINESS.json"
    [ -f "$LIVE_BIZ" ] || r_abort business-missing "no $LIVE_BIZ"

    local stamp before live
    before="$(r_tree_sums "$R_BK")" && live="$(sha256sum "$LIVE_CFG" "$LIVE_BIZ" 2>&1)" || r_abort sums "cannot read $R_BK or the live config files"
    stamp="$(date -u +%Y%m%dT%H%M%SZ)-${TO:0:12}"
    (umask 077; mkdir -p "$DATA/rehearsals" && mkdir "$DATA/rehearsals/$stamp") || r_abort rehearsal-dir "cannot make $DATA/rehearsals/$stamp (one rehearsal per second and copy)"
    R="$DATA/rehearsals/$stamp" STEPS="$DATA/rehearsals/$stamp/steps.log"
    printf '%s\n' "${R_EARLY[@]}" > "$STEPS" && printf '%s\n' "$before" > "$R/backup.before" && printf '%s\n' "$live" > "$R/live.before" ||
        r_abort scratch-build "cannot write the sums to $R"
    (
        umask 077
        mkdir "$R/config" "$R/config/businesses" "$R/placeholders" "$R/data" "$R/data/bus" "$R/data/notify" "$R/data/notify/$BUSINESS" &&
        cp "$R_BK"/bus/* "$R/data/bus/" &&
        node -e "$R_CONFIG_JS" -- "$LIVE_CFG" "$R/data" > "$R/config/config.json" &&
        node -e "$R_BUSINESS_JS" -- write "$LIVE_BIZ" "$R/config/businesses/$BUSINESS.json" "$R/placeholders" &&
        printf '{"notifyVersion":1,"binding":null}\n' > "$R/data/notify/$BUSINESS/notify.json"
    ) 2>> "$STEPS" || r_abort scratch-build "cannot build the scratch root under $R; see $STEPS"
    local rc=0
    diff -u --label "$LIVE_BIZ (as JSON.stringify writes it)" --label "$R/config/businesses/$BUSINESS.json" \
        <(node -e "$R_BUSINESS_JS" -- normalize "$LIVE_BIZ") "$R/config/businesses/$BUSINESS.json" > "$R/business.diff" || rc=$?
    [ "$rc" -le 1 ] || r_abort scratch-build "cannot write $R/business.diff"
    r_pass scratch-root "$R/data from the backup, config and business file under $R/config, business.diff beside them"

    local out
    rc=0; out="$(r_env node -e "$R_GUARD_JS" -- "$COPIES/$TO" "$BUSINESS" "$DATA" "$R/data" "$R/placeholders" 2>&1)" || rc=$?
    while IFS= read -r i; do r_line "$i"; done <<< "$out"
    [ "$rc" = 0 ] || { R_RC=1; r_end; }

    r_migration

    START_MS="$(date +%s%3N)"
    R_STARTED=1
    # env, not r_env: a function in the background is a subshell, and $! has to be the process
    # itself (env execs scripts/mosaic, which execs node), for the bus-status pid check and the stop.
    env -i HOME="$HOME" PATH="$PATH" MOSAIC_CONFIG="$R/config/config.json" "$COPIES/$TO/scripts/mosaic" bus start "$BUSINESS" --no-launch > "$R/host.log" 2>&1 &
    R_HOST_PID=$!
    local -a tester_rehearsal=()
    if grep -q 'MOSAIC_TESTER_SOCKET_V1' "$COPIES/$TO/packages/webui/src/cli.mjs"; then tester_rehearsal=(--tester-socket); fi
    env -i HOME="$HOME" PATH="$PATH" MOSAIC_CONFIG="$R/config/config.json" node "$COPIES/$TO/packages/webui/src/cli.mjs" serve --port "$R_PORT" --business "$BUSINESS" "${tester_rehearsal[@]}" > "$R/console.log" 2>&1 &
    R_CONSOLE_PID=$!
    if r_gate; then r_no_session; fi
    r_end
}

# --- Dispatch ---

# Row 190: --instance NAME may come anywhere; instance.sh takes it out and sets the names.
instance_args "$@"
set -- ${INSTANCE_ARGS[@]+"${INSTANCE_ARGS[@]}"}
[ $# -gt 0 ] || usage
OP="$1"; shift
ID=""
case "$OP" in
    activate|rehearse) [ $# -ge 2 ] || usage; ID="$1"; BUSINESS="$2"; shift 2; [[ "$ID" =~ ^[0-9a-f]{40}$ ]] || usage ;;
    adopt|pm-on|rollback|status) [ $# -ge 1 ] || usage; BUSINESS="$1"; shift ;;
    confirm|revert|gate) [ $# -ge 2 ] || usage; ID="$1"; BUSINESS="$2"; shift 2; [[ "$ID" =~ ^[0-9a-f]{40}$ ]] || usage ;;
    -h|--help) usage 0 ;;
    *) usage ;;
esac
[[ "$BUSINESS" =~ ^[a-z0-9][a-z0-9-]*$ ]] || usage
parse_opts "$@"
# Row 232: the stack-copy.sh check flags and the systemctl the instance-busy message names.
UNITS_FLAGS=()
SC_NAME="systemctl --user"
if [ "$UNITS" = external ]; then UNITS_FLAGS=(--units external); SC_NAME=systemctl; fi
COPIES_DIR="$COPIES"
instance_select
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="${BUS_TEMPLATE%.service}$BUSINESS.service"
FROM="" TO="" ACT="" RS="" POST="" BACKUP="" TAKEN="" RESTORED="" STEPS="" COUNT="" UNCLEAN="" FAIL=""
STORE_SCHEMA="" START_EPOCH="" START_MS=0 GATE="" GATE_WHY="" S_CHANGED=""
P_COPY="" P_PM=0 P_PENDING=0 P_SCHEMA="" P_CLAUDE="" P_CONSOLE=() RUN_PM=0 RUN_PENDING=0 RUN_CONSOLE=() DROP_TESTER="" PROOF_PLAN="" PROOF_STEP="" SEND_PLAN="" SEND_STEP="" PLAN_REFUSE=()
BASE_CONSOLE=() SERVE_FLAGS=() SSO_PLAN="" SSO_STEP="" SSO_DROP="" SSO_NO_TARGET=""
R="" R_RC=0 R_EARLY=() R_BK="" R_HOST_PID="" R_CONSOLE_PID="" R_STARTED="" LIVE_CFG="" LIVE_BIZ=""
O_ACT="" FLOOR=0
case "$OP" in
    adopt) do_adopt ;;
    activate) do_activate ;;
    pm-on) do_pm_on ;;
    rollback) do_rollback ;;
    status) do_status ;;
    rehearse) do_rehearse ;;
    confirm) do_confirm ;;
    revert) do_revert ;;
    gate) do_gate ;;
esac